QuantEngineByItz/tools/build_execution_quality_harness_v1.py

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


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


def _load(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        obj = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return obj if isinstance(obj, dict) else {}


def _to_float(v: Any) -> float:
    try:
        return float(v)
    except Exception:
        return 0.0


def _max_drawdown_pct(returns_pct: list[float]) -> float:
    equity = 1.0
    peak = 1.0
    max_dd = 0.0
    for r in returns_pct:
        equity *= (1.0 + r / 100.0)
        if equity > peak:
            peak = equity
        dd = 0.0 if peak <= 0 else (peak - equity) / peak * 100.0
        if dd > max_dd:
            max_dd = dd
    return round(max_dd, 2)


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

    hp = Path(args.history)
    op = Path(args.out)
    if not hp.is_absolute():
        hp = ROOT / hp
    if not op.is_absolute():
        op = ROOT / op

    history = _load(hp)
    records = history.get("records") if isinstance(history.get("records"), list) else []
    t20 = [r for r in records if isinstance(r, dict) and r.get("t20_evaluation_status") == "EVALUATED_T20"]
    operational = [r for r in t20 if str(r.get("validation_status") or "").upper() != "REPLAY_BACKFILL"]
    replay = [r for r in t20 if str(r.get("validation_status") or "").upper() == "REPLAY_BACKFILL"]

    def eval_set(rows: list[dict[str, Any]]) -> dict[str, float]:
        rets = [_to_float(r.get("t20_return_pct")) for r in rows]
        wins = [x for x in rets if x > 0]
        losses = [x for x in rets if x < 0]
        n = len(rets)
        win_rate = (len(wins) / n * 100.0) if n else 0.0
        avg_win = (sum(wins) / len(wins)) if wins else 0.0
        avg_loss = (sum(losses) / len(losses)) if losses else 0.0
        expectancy = (win_rate / 100.0) * avg_win + (1.0 - win_rate / 100.0) * avg_loss
        return {
            "samples": n,
            "win_rate_pct": round(win_rate, 2),
            "avg_win_pct": round(avg_win, 2),
            "avg_loss_pct": round(avg_loss, 2),
            "expectancy_pct": round(expectancy, 3),
            "max_drawdown_pct": _max_drawdown_pct(rets),
        }

    opm = eval_set(operational)
    rpm = eval_set(replay)

    gate = "PASS"
    reasons: list[str] = []
    if opm["samples"] < args.min_samples:
        gate = "WATCH_PENDING_SAMPLE"
        reasons.append("OPERATIONAL_SAMPLE_INSUFFICIENT")
    else:
        if opm["expectancy_pct"] <= 0:
            gate = "FAIL"
            reasons.append("NEGATIVE_EXPECTANCY")
        if opm["max_drawdown_pct"] > 12.0:
            gate = "FAIL"
            reasons.append("MDD_ABOVE_12")
        if opm["win_rate_pct"] < 45.0:
            gate = "FAIL"
            reasons.append("WIN_RATE_BELOW_45")

    res = {
        "formula_id": "EXECUTION_QUALITY_HARNESS_V1",
        "gate": gate,
        "reasons": reasons,
        "targets": {
            "min_samples": int(args.min_samples),
            "expectancy_pct_min": 0.0,
            "max_drawdown_pct_max": 12.0,
            "win_rate_pct_min": 45.0,
        },
        "metrics": {
            "operational_t20": opm,
            "replay_t20": rpm,
        },
    }

    op.parent.mkdir(parents=True, exist_ok=True)
    op.write_text(json.dumps(res, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(res, ensure_ascii=False, indent=2))
    return 0


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