Business Intelligence

#!/usr/bin/env python3
"""Generate dashboard layout specifications from KPI definitions.

Reads a KPI definitions file and produces a structured dashboard
specification with chart types, layout positions, filters, and
recommended visualizations.

Usage:
    python dashboard_spec_generator.py --definitions kpis.json
    python dashboard_spec_generator.py --definitions kpis.json --title "Sales Dashboard" --json
    python dashboard_spec_generator.py --definitions kpis.json --layout 2-column

KPI definition format:
[
    {
        "name": "Monthly Revenue",
        "type": "currency",
        "aggregation": "sum",
        "dimensions": ["region", "product"],
        "time_grain": "monthly",
        "target": 100000,
        "chart_hint": "trend"
    }
]
"""

import argparse
import json
import os
import sys
from datetime import datetime


# ---------------------------------------------------------------------------
# Chart type selection logic
# ---------------------------------------------------------------------------

CHART_RULES = {
    "trend": {"chart": "line", "alt": "area", "reason": "Shows metric change over time"},
    "comparison": {"chart": "bar", "alt": "column", "reason": "Compares values across categories"},
    "composition": {"chart": "donut", "alt": "stacked_bar", "reason": "Shows parts of a whole"},
    "distribution": {"chart": "histogram", "alt": "box_plot", "reason": "Shows data distribution"},
    "kpi_card": {"chart": "scorecard", "alt": "gauge", "reason": "Highlights a single key metric"},
    "table": {"chart": "data_table", "alt": "pivot_table", "reason": "Shows detailed records"},
    "geographic": {"chart": "choropleth", "alt": "bubble_map", "reason": "Maps data geographically"},
}


def _infer_chart_type(kpi: dict) -> str:
    """Infer best chart type from KPI definition."""
    if kpi.get("chart_hint"):
        return kpi["chart_hint"]
    agg = kpi.get("aggregation", "sum")
    dims = kpi.get("dimensions", [])
    time_grain = kpi.get("time_grain")

    if time_grain and not dims:
        return "trend"
    if time_grain and dims:
        return "trend"  # trend with dimension breakdown
    if len(dims) == 0:
        return "kpi_card"
    if len(dims) == 1:
        return "comparison"
    if any(d in ("region", "country", "state", "city") for d in dims):
        return "geographic"
    return "comparison"


def _generate_widget(kpi: dict, position: int, layout: str) -> dict:
    chart_key = _infer_chart_type(kpi)
    chart_info = CHART_RULES.get(chart_key, CHART_RULES["kpi_card"])

    # Calculate grid position
    if layout == "2-column":
        cols = 2
    elif layout == "3-column":
        cols = 3
    else:
        cols = 2

    row = position // cols
    col = position % cols

    widget = {
        "id": f"widget_{position + 1}",
        "title": kpi["name"],
        "chart_type": chart_info["chart"],
        "alt_chart_type": chart_info["alt"],
        "chart_rationale": chart_info["reason"],
        "metric": {
            "name": kpi["name"],
            "aggregation": kpi.get("aggregation", "sum"),
            "type": kpi.get("type", "number"),
        },
        "layout": {
            "row": row,
            "col": col,
            "width": 1,
            "height": 1,
        },
    }

    if kpi.get("dimensions"):
        widget["dimensions"] = kpi["dimensions"]

    if kpi.get("time_grain"):
        widget["time_grain"] = kpi["time_grain"]

    if kpi.get("target") is not None:
        widget["target"] = kpi["target"]
        widget["show_target_line"] = True

    if kpi.get("filters"):
        widget["filters"] = kpi["filters"]

    return widget


def generate_dashboard_spec(kpi_defs: list, title: str, layout: str) -> dict:
    widgets = []

    # KPI cards come first (top row)
    kpi_cards = []
    detail_widgets = []
    for kpi in kpi_defs:
        chart_type = _infer_chart_type(kpi)
        if chart_type == "kpi_card":
            kpi_cards.append(kpi)
        else:
            detail_widgets.append(kpi)

    position = 0
    for kpi in kpi_cards:
        w = _generate_widget(kpi, position, layout)
        w["section"] = "summary"
        widgets.append(w)
        position += 1

    for kpi in detail_widgets:
        w = _generate_widget(kpi, position, layout)
        w["section"] = "detail"
        widgets.append(w)
        position += 1

    # Global filters
    all_dims = set()
    for kpi in kpi_defs:
        all_dims.update(kpi.get("dimensions", []))

    time_grains = set()
    for kpi in kpi_defs:
        if kpi.get("time_grain"):
            time_grains.add(kpi["time_grain"])

    filters = []
    if time_grains:
        filters.append({
            "name": "date_range",
            "type": "date_range",
            "default": "last_30_days",
        })
    for dim in sorted(all_dims):
        filters.append({
            "name": dim,
            "type": "multi_select",
            "default": "all",
        })

    spec = {
        "dashboard": {
            "title": title,
            "layout": layout,
            "generated_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
            "total_widgets": len(widgets),
            "sections": {
                "summary": sum(1 for w in widgets if w.get("section") == "summary"),
                "detail": sum(1 for w in widgets if w.get("section") == "detail"),
            },
        },
        "global_filters": filters,
        "widgets": widgets,
        "design_notes": [
            "Place summary KPI cards in the top row for immediate visibility.",
            "Trend charts should show comparison to target or prior period.",
            "Limit dashboard to 5-8 widgets per page for readability.",
            "Use consistent color palette: GREEN #28A745, YELLOW #FFC107, RED #DC3545.",
            "Dashboard load time target: < 5 seconds.",
        ],
    }

    return spec


def _format_text(spec: dict) -> str:
    lines = []
    d = spec["dashboard"]
    lines.append(f"Dashboard Specification: {d['title']}")
    lines.append("=" * 60)
    lines.append(f"Layout: {d['layout']}  |  Widgets: {d['total_widgets']}  |  Generated: {d['generated_at']}")
    lines.append(f"Summary cards: {d['sections']['summary']}  |  Detail charts: {d['sections']['detail']}")
    lines.append("")

    if spec.get("global_filters"):
        lines.append("Global Filters:")
        for f in spec["global_filters"]:
            lines.append(f"  - {f['name']} ({f['type']}, default: {f['default']})")
        lines.append("")

    lines.append("Widgets:")
    lines.append("-" * 60)
    for w in spec["widgets"]:
        target_str = f"  Target: {w['target']}" if w.get("target") else ""
        dims_str = f"  Dimensions: {', '.join(w['dimensions'])}" if w.get("dimensions") else ""
        lines.append(f"  [{w['id']}] {w['title']}")
        lines.append(f"    Chart: {w['chart_type']} (alt: {w['alt_chart_type']})")
        lines.append(f"    Reason: {w['chart_rationale']}")
        lines.append(f"    Position: row={w['layout']['row']}, col={w['layout']['col']}")
        if target_str:
            lines.append(f"   {target_str}")
        if dims_str:
            lines.append(f"   {dims_str}")
        if w.get("time_grain"):
            lines.append(f"    Time grain: {w['time_grain']}")
        lines.append("")

    lines.append("Design Notes:")
    for note in spec.get("design_notes", []):
        lines.append(f"  - {note}")

    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="Generate dashboard layout specifications from KPI definitions.")
    parser.add_argument("--definitions", required=True, help="Path to KPI definitions JSON file")
    parser.add_argument("--title", default="Analytics Dashboard", help="Dashboard title")
    parser.add_argument("--layout", choices=["2-column", "3-column"], default="2-column", help="Layout style")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    if not os.path.exists(args.definitions):
        print(f"Error: File not found: {args.definitions}", file=sys.stderr)
        sys.exit(1)

    with open(args.definitions, "r") as f:
        kpi_defs = json.load(f)

    if not isinstance(kpi_defs, list) or not kpi_defs:
        print("Error: Definitions file must contain a non-empty JSON array.", file=sys.stderr)
        sys.exit(1)

    spec = generate_dashboard_spec(kpi_defs, args.title, args.layout)

    if args.json:
        print(json.dumps(spec, indent=2))
    else:
        print(_format_text(spec))

    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Calculate and track KPIs from data files against targets and thresholds.

Reads a KPI definition file (JSON) and a data file (CSV/JSON), computes
each metric, evaluates RAG status, and reports trends.

Usage:
    python kpi_tracker.py --definitions kpis.json --data sales.csv
    python kpi_tracker.py --definitions kpis.json --data sales.csv --json
    python kpi_tracker.py --definitions kpis.json --data sales.csv --period monthly

KPI definition format (kpis.json):
[
    {
        "name": "Monthly Revenue",
        "column": "revenue",
        "aggregation": "sum",
        "target": 100000,
        "warning_pct": 0.9,
        "critical_pct": 0.8,
        "higher_is_better": true
    }
]
"""

import argparse
import csv
import json
import math
import os
import sys
from collections import defaultdict


# ---------------------------------------------------------------------------
# Aggregation functions
# ---------------------------------------------------------------------------

def _agg_sum(values: list) -> float:
    return sum(values)

def _agg_mean(values: list) -> float:
    return sum(values) / len(values) if values else 0

def _agg_count(values: list) -> float:
    return float(len(values))

def _agg_count_distinct(values: list) -> float:
    return float(len(set(values)))

def _agg_min(values: list) -> float:
    return min(values) if values else 0

def _agg_max(values: list) -> float:
    return max(values) if values else 0

def _agg_median(values: list) -> float:
    s = sorted(values)
    n = len(s)
    if n == 0:
        return 0.0
    mid = n // 2
    return (s[mid - 1] + s[mid]) / 2.0 if n % 2 == 0 else s[mid]


AGG_MAP = {
    "sum": _agg_sum,
    "mean": _agg_mean,
    "average": _agg_mean,
    "count": _agg_count,
    "count_distinct": _agg_count_distinct,
    "min": _agg_min,
    "max": _agg_max,
    "median": _agg_median,
}


# ---------------------------------------------------------------------------
# Core
# ---------------------------------------------------------------------------

def load_data(file_path: str) -> list:
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".csv":
        with open(file_path, "r", newline="") as f:
            return list(csv.DictReader(f))
    elif ext == ".json":
        with open(file_path, "r") as f:
            data = json.load(f)
            return data if isinstance(data, list) else []
    else:
        print(f"Error: Unsupported file type '{ext}'.", file=sys.stderr)
        sys.exit(1)


def _extract_numeric(data: list, column: str) -> list:
    values = []
    for row in data:
        v = row.get(column)
        if v is not None:
            try:
                values.append(float(str(v)))
            except ValueError:
                pass
    return values


def _extract_raw(data: list, column: str) -> list:
    return [str(row.get(column, "")) for row in data if row.get(column) is not None]


def _rag_status(actual: float, target: float, warning_pct: float, critical_pct: float, higher_is_better: bool) -> str:
    if higher_is_better:
        if actual >= target:
            return "GREEN"
        elif actual >= target * warning_pct:
            return "YELLOW"
        elif actual >= target * critical_pct:
            return "YELLOW"
        else:
            return "RED"
    else:
        if actual <= target:
            return "GREEN"
        elif actual <= target / warning_pct:
            return "YELLOW"
        else:
            return "RED"


def compute_kpi(kpi_def: dict, data: list) -> dict:
    name = kpi_def["name"]
    column = kpi_def["column"]
    agg_name = kpi_def.get("aggregation", "sum")
    target = kpi_def.get("target")
    warning_pct = kpi_def.get("warning_pct", 0.9)
    critical_pct = kpi_def.get("critical_pct", 0.8)
    higher_is_better = kpi_def.get("higher_is_better", True)

    agg_fn = AGG_MAP.get(agg_name)
    if not agg_fn:
        return {"name": name, "error": f"Unknown aggregation: {agg_name}"}

    if agg_name == "count_distinct":
        raw = _extract_raw(data, column)
        actual = agg_fn(raw)
    else:
        values = _extract_numeric(data, column)
        if not values:
            return {"name": name, "error": f"No numeric values found in column '{column}'."}
        actual = agg_fn(values)

    result = {
        "name": name,
        "column": column,
        "aggregation": agg_name,
        "actual": round(actual, 2),
    }

    if target is not None:
        result["target"] = target
        result["variance"] = round(actual - target, 2)
        result["variance_pct"] = round((actual - target) / target * 100, 2) if target != 0 else 0
        result["status"] = _rag_status(actual, target, warning_pct, critical_pct, higher_is_better)
    else:
        result["status"] = "NO_TARGET"

    return result


def compute_all_kpis(kpi_defs: list, data: list) -> dict:
    results = []
    for kpi_def in kpi_defs:
        results.append(compute_kpi(kpi_def, data))

    green = sum(1 for r in results if r.get("status") == "GREEN")
    yellow = sum(1 for r in results if r.get("status") == "YELLOW")
    red = sum(1 for r in results if r.get("status") == "RED")

    return {
        "total_kpis": len(results),
        "green": green,
        "yellow": yellow,
        "red": red,
        "health_score": round(green / len(results) * 100, 1) if results else 0,
        "kpis": results,
    }


def main():
    parser = argparse.ArgumentParser(description="Calculate and track KPIs against targets.")
    parser.add_argument("--definitions", required=True, help="Path to KPI definitions JSON file")
    parser.add_argument("--data", required=True, help="Path to data file (CSV or JSON)")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    for path, label in [(args.definitions, "Definitions"), (args.data, "Data")]:
        if not os.path.exists(path):
            print(f"Error: {label} file not found: {path}", file=sys.stderr)
            sys.exit(1)

    with open(args.definitions, "r") as f:
        kpi_defs = json.load(f)

    if not isinstance(kpi_defs, list):
        print("Error: KPI definitions must be a JSON array.", file=sys.stderr)
        sys.exit(1)

    data = load_data(args.data)
    if not data:
        print("Error: No data rows found.", file=sys.stderr)
        sys.exit(1)

    report = compute_all_kpis(kpi_defs, data)

    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print("KPI Tracker Report")
        print("=" * 65)
        print(f"KPIs: {report['total_kpis']}  |  GREEN: {report['green']}  YELLOW: {report['yellow']}  RED: {report['red']}  |  Health: {report['health_score']}%")
        print()
        print(f"{'KPI':<30} {'Actual':>12} {'Target':>12} {'Var %':>8} {'Status':<8}")
        print("-" * 65)
        for kpi in report["kpis"]:
            if "error" in kpi:
                print(f"  {kpi['name']:<28} ERROR: {kpi['error']}")
                continue
            target_str = str(kpi.get("target", "-"))
            var_str = f"{kpi.get('variance_pct', 0):+.1f}%" if "variance_pct" in kpi else "-"
            status = kpi.get("status", "?")
            marker = {"GREEN": "[OK]", "YELLOW": "[!!]", "RED": "[XX]"}.get(status, "[ ]")
            print(f"  {kpi['name']:<28} {kpi['actual']:>12,.2f} {target_str:>12} {var_str:>8} {marker}")

    sys.exit(0)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Validate metric definitions for consistency, completeness, and correctness.

Reads a metric definitions file (JSON) and checks for common issues:
missing fields, conflicting aggregations, naming violations, undefined
dimensions, and threshold logic errors.

Usage:
    python metric_validator.py --definitions metrics.json
    python metric_validator.py --definitions metrics.json --strict --json

Metric definition format:
[
    {
        "name": "Monthly Revenue",
        "formula": "SUM(amount)",
        "data_source": "orders",
        "column": "amount",
        "aggregation": "sum",
        "owner": "Finance",
        "dimensions": ["region", "product_line"],
        "target": 100000,
        "warning_pct": 0.9,
        "critical_pct": 0.8,
        "higher_is_better": true,
        "granularity": "monthly"
    }
]
"""

import argparse
import json
import os
import re
import sys


# ---------------------------------------------------------------------------
# Validation rules
# ---------------------------------------------------------------------------

REQUIRED_FIELDS = ["name", "aggregation", "data_source"]
RECOMMENDED_FIELDS = ["owner", "formula", "granularity", "dimensions"]
VALID_AGGREGATIONS = {"sum", "mean", "average", "count", "count_distinct", "min", "max", "median", "ratio", "rate"}
VALID_GRANULARITIES = {"daily", "weekly", "monthly", "quarterly", "yearly", "hourly", "real_time"}
NAME_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9 _-]{2,60}$")


def _validate_metric(metric: dict, index: int, strict: bool = False) -> list:
    issues = []
    prefix = f"Metric #{index + 1}"
    name = metric.get("name", f"(unnamed #{index + 1})")

    # Required fields
    for field in REQUIRED_FIELDS:
        if not metric.get(field):
            issues.append({
                "metric": name,
                "severity": "error",
                "rule": "MISSING_REQUIRED_FIELD",
                "message": f"Missing required field '{field}'.",
            })

    # Recommended fields
    if strict:
        for field in RECOMMENDED_FIELDS:
            if not metric.get(field):
                issues.append({
                    "metric": name,
                    "severity": "warning",
                    "rule": "MISSING_RECOMMENDED_FIELD",
                    "message": f"Missing recommended field '{field}'.",
                })

    # Name format
    if metric.get("name") and not NAME_PATTERN.match(metric["name"]):
        issues.append({
            "metric": name,
            "severity": "warning",
            "rule": "INVALID_NAME_FORMAT",
            "message": "Metric name should be 3-60 chars, start with a letter, and use only alphanumeric, spaces, hyphens, or underscores.",
        })

    # Aggregation
    agg = metric.get("aggregation", "").lower()
    if agg and agg not in VALID_AGGREGATIONS:
        issues.append({
            "metric": name,
            "severity": "error",
            "rule": "INVALID_AGGREGATION",
            "message": f"Aggregation '{agg}' is not recognized. Valid: {', '.join(sorted(VALID_AGGREGATIONS))}.",
        })

    # Granularity
    gran = metric.get("granularity", "").lower()
    if gran and gran not in VALID_GRANULARITIES:
        issues.append({
            "metric": name,
            "severity": "warning",
            "rule": "INVALID_GRANULARITY",
            "message": f"Granularity '{gran}' is not standard. Valid: {', '.join(sorted(VALID_GRANULARITIES))}.",
        })

    # Threshold logic
    target = metric.get("target")
    warning_pct = metric.get("warning_pct")
    critical_pct = metric.get("critical_pct")
    higher = metric.get("higher_is_better", True)

    if target is not None:
        if warning_pct is not None and critical_pct is not None:
            if higher:
                if warning_pct <= critical_pct:
                    issues.append({
                        "metric": name,
                        "severity": "error",
                        "rule": "THRESHOLD_ORDER",
                        "message": f"warning_pct ({warning_pct}) should be > critical_pct ({critical_pct}) when higher_is_better=true.",
                    })
            if warning_pct is not None and (warning_pct <= 0 or warning_pct > 1):
                issues.append({
                    "metric": name,
                    "severity": "error",
                    "rule": "THRESHOLD_RANGE",
                    "message": f"warning_pct ({warning_pct}) must be between 0 and 1.",
                })
            if critical_pct is not None and (critical_pct <= 0 or critical_pct > 1):
                issues.append({
                    "metric": name,
                    "severity": "error",
                    "rule": "THRESHOLD_RANGE",
                    "message": f"critical_pct ({critical_pct}) must be between 0 and 1.",
                })

    # Dimensions validation
    dims = metric.get("dimensions", [])
    if not isinstance(dims, list):
        issues.append({
            "metric": name,
            "severity": "error",
            "rule": "INVALID_DIMENSIONS",
            "message": "Dimensions must be a list.",
        })
    elif len(dims) > 10:
        issues.append({
            "metric": name,
            "severity": "warning",
            "rule": "EXCESSIVE_DIMENSIONS",
            "message": f"Metric has {len(dims)} dimensions; consider limiting to <=10 for dashboard usability.",
        })

    # Formula vs aggregation consistency
    formula = metric.get("formula", "").upper()
    if formula and agg:
        expected_agg = None
        if formula.startswith("SUM("):
            expected_agg = "sum"
        elif formula.startswith("AVG(") or formula.startswith("AVERAGE("):
            expected_agg = "mean"
        elif formula.startswith("COUNT(DISTINCT"):
            expected_agg = "count_distinct"
        elif formula.startswith("COUNT("):
            expected_agg = "count"
        if expected_agg and expected_agg != agg:
            issues.append({
                "metric": name,
                "severity": "warning",
                "rule": "FORMULA_AGG_MISMATCH",
                "message": f"Formula suggests '{expected_agg}' but aggregation is set to '{agg}'.",
            })

    return issues


def validate_all(metrics: list, strict: bool = False) -> dict:
    all_issues = []
    for i, m in enumerate(metrics):
        all_issues.extend(_validate_metric(m, i, strict))

    # Cross-metric checks: duplicate names
    names = [m.get("name", "") for m in metrics]
    seen = set()
    for n in names:
        if n in seen:
            all_issues.append({
                "metric": n,
                "severity": "error",
                "rule": "DUPLICATE_NAME",
                "message": f"Metric name '{n}' appears more than once.",
            })
        seen.add(n)

    errors = sum(1 for i in all_issues if i["severity"] == "error")
    warnings = sum(1 for i in all_issues if i["severity"] == "warning")

    return {
        "total_metrics": len(metrics),
        "total_issues": len(all_issues),
        "errors": errors,
        "warnings": warnings,
        "valid": errors == 0,
        "issues": all_issues,
    }


def main():
    parser = argparse.ArgumentParser(description="Validate metric definitions for consistency and completeness.")
    parser.add_argument("--definitions", required=True, help="Path to metric definitions JSON file")
    parser.add_argument("--strict", action="store_true", help="Enable strict mode (flag missing recommended fields)")
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    args = parser.parse_args()

    if not os.path.exists(args.definitions):
        print(f"Error: File not found: {args.definitions}", file=sys.stderr)
        sys.exit(1)

    with open(args.definitions, "r") as f:
        metrics = json.load(f)

    if not isinstance(metrics, list):
        print("Error: Definitions must be a JSON array.", file=sys.stderr)
        sys.exit(1)

    result = validate_all(metrics, args.strict)

    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print("Metric Validation Report")
        print("=" * 55)
        status = "PASS" if result["valid"] else "FAIL"
        print(f"Status: [{status}]  |  Metrics: {result['total_metrics']}  |  Errors: {result['errors']}  Warnings: {result['warnings']}")
        print()

        if not result["issues"]:
            print("All metric definitions are valid.")
        else:
            for issue in result["issues"]:
                sev = issue["severity"].upper()
                print(f"  [{sev}] {issue['metric']}: {issue['rule']}")
                print(f"    {issue['message']}")

    sys.exit(1 if result["errors"] > 0 else 0)


if __name__ == "__main__":
    main()