QuantEngineByItz/tools/build_prediction_lift_dashboard_v1.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import math
import statistics
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_HISTORY = ROOT / "Temp" / "proposal_evaluation_history.json"
DEFAULT_CONTINUOUS = ROOT / "Temp" / "continuous_evaluation_dashboard_v1.json"
DEFAULT_OUT = ROOT / "Temp" / "prediction_lift_dashboard_v1.json"


def _load_json(path: Path) -> object:
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}


def _records(payload: object) -> list[dict]:
    if isinstance(payload, dict) and isinstance(payload.get("records"), list):
        return [r for r in payload["records"] if isinstance(r, dict)]
    if isinstance(payload, list):
        return [r for r in payload if isinstance(r, dict)]
    return []


def _confidence_interval_pct(rate_pct: float, n: int) -> tuple[float, float]:
    if n <= 0:
        return 0.0, 0.0
    p = rate_pct / 100.0
    z = 1.96
    denom = 1.0 + z * z / n
    center = (p + z * z / (2 * n)) / denom
    margin = z * math.sqrt((p * (1 - p) + z * z / (4 * n)) / n) / denom
    return round(max(0.0, (center - margin) * 100.0), 2), round(min(100.0, (center + margin) * 100.0), 2)


def _horizon_stats(records: list[dict], status_key: str, outcome_key: str, return_key: str) -> dict:
    total = len(records)
    matched = [r for r in records if str(r.get(outcome_key) or "").upper() == "MATCHED"]
    pass_rate = round(len(matched) / total * 100.0, 2) if total else 0.0
    returns = [float(r.get(return_key)) for r in records if r.get(return_key) is not None]
    avg_ret = round(statistics.mean(returns), 2) if returns else None
    ci_low, ci_high = _confidence_interval_pct(pass_rate, total)
    return {
        "sample_count": total,
        "pass_rate_pct": pass_rate,
        "avg_return_pct": avg_ret,
        "confidence_interval_pct": [ci_low, ci_high],
        "status_key": status_key,
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--history", default=str(DEFAULT_HISTORY))
    ap.add_argument("--continuous", default=str(DEFAULT_CONTINUOUS))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    history = _load_json(Path(args.history))
    continuous = _load_json(Path(args.continuous))
    records = _records(history)

    t5 = [r for r in records if str(r.get("t5_evaluation_status") or "") == "EVALUATED_T5"]
    t20 = [r for r in records if str(r.get("t20_evaluation_status") or "") == "EVALUATED_T20"]
    t60 = [r for r in records if str(r.get("evaluation_status") or "").startswith("EVALUATED_")]

    t5_stats = _horizon_stats(t5, "t5_evaluation_status", "t5_outcome", "t5_return_pct")
    t20_stats = _horizon_stats(t20, "t20_evaluation_status", "t20_outcome", "t20_return_pct")
    t60_stats = {
        "sample_count": len(t60),
        "pass_rate_pct": round((len([r for r in t60 if str(r.get("evaluation_status") or "").startswith("EVALUATED_")]) / max(1, len(t60))) * 100.0, 2),
        "avg_return_pct": round(statistics.mean([float(r.get("t20_return_pct")) for r in t60 if r.get("t20_return_pct") is not None]), 2) if any(r.get("t20_return_pct") is not None for r in t60) else None,
        "confidence_interval_pct": [0.0, 0.0],
        "status_key": "evaluation_status",
    }

    baseline_random_pct = 50.0
    benchmark_sector_neutral_pct = round(float((continuous.get("performance_readiness_score") or 0.0)), 2)
    lift_vs_random = round(t5_stats["pass_rate_pct"] - baseline_random_pct, 2)
    lift_vs_benchmark = round(t5_stats["pass_rate_pct"] - benchmark_sector_neutral_pct, 2)
    after_slippage_pct = round((t5_stats["avg_return_pct"] or 0.0) - 0.35, 2) if t5_stats["avg_return_pct"] is not None else None

    result = {
        "formula_id": "PREDICTION_LIFT_DASHBOARD_V1",
        "gate": "PASS" if t5_stats["sample_count"] > 0 else "INSUFFICIENT_DATA",
        "baseline_random_pct": baseline_random_pct,
        "benchmark_sector_neutral_pct": benchmark_sector_neutral_pct,
        "transaction_cost_bps": 35,
        "prediction_lift_vs_baseline_ppt": lift_vs_random,
        "prediction_lift_vs_benchmark_ppt": lift_vs_benchmark,
        "after_slippage_pct": after_slippage_pct,
        "horizons": {
            "t5": t5_stats,
            "t20": t20_stats,
            "t60": t60_stats,
        },
        "sample_count": len(records),
        "confidence_policy": "Wilson 95% CI",
        "source_paths": [str(Path(args.history)), str(Path(args.continuous))],
    }
    out = Path(args.out)
    out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())