QuantEngineByItz/tools/build_shadow_promotion_scorecard_v1.py

"""build_shadow_promotion_scorecard_v1.py — spec/57: H007_SHADOW_PROMOTION_SCORECARD

Evaluates shadow-stage factors against live sample count, edge improvement,
drawdown constraint, false positive reduction, conflict rate, and provenance
coverage. Blocks promotion if any criterion is unmet.

formula_id: BUILD_SHADOW_PROMOTION_SCORECARD_V1
contract:   spec/57_shadow_promotion_scorecard.yaml
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]

DEFAULT_SHADOW         = ROOT / "Temp" / "shadow_ledger_v2.json"
DEFAULT_LIVE_REPLAY    = ROOT / "Temp" / "live_replay_separation_v3.json"
OUTPUT_PATH            = ROOT / "Temp" / "shadow_promotion_scorecard_v1.json"

# Gate criteria from spec/57
LIVE_SAMPLE_MIN          = 30
EDGE_IMPROVEMENT_MIN_PCT = 2.0
DRAWDOWN_TOLERANCE_PCT   = 0.5
FALSE_POSITIVE_REDUCTION_MIN_PCT = 5.0
CONFLICT_RATE_CAP_PCT    = 10.0
PROVENANCE_REQUIRED_PCT  = 100.0


def _load_json(path: Path) -> dict:
    if not path.exists():
        return {"_missing": True, "_path": str(path)}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        return {"_error": str(e), "_path": str(path)}


def _evaluate_shadow_factor(factor: dict, live_data: dict) -> dict:
    """Evaluate a single shadow factor against promotion criteria."""
    fid = factor.get("formula_id") or factor.get("id") or "UNKNOWN"
    blocked_reasons = []

    # Live sample count
    live_n = factor.get("live_sample_count") or 0
    if live_n < LIVE_SAMPLE_MIN:
        blocked_reasons.append(
            f"live_sample_count={live_n} < {LIVE_SAMPLE_MIN}"
        )

    # Edge improvement
    edge_delta = factor.get("prediction_match_rate_improvement") or 0
    if edge_delta < EDGE_IMPROVEMENT_MIN_PCT:
        blocked_reasons.append(
            f"edge_improvement={edge_delta}% < {EDGE_IMPROVEMENT_MIN_PCT}%"
        )

    # Drawdown constraint
    mdd_delta = factor.get("mdd_delta_pct") or 0
    if mdd_delta > DRAWDOWN_TOLERANCE_PCT:
        blocked_reasons.append(
            f"mdd_delta={mdd_delta}% > tolerance={DRAWDOWN_TOLERANCE_PCT}%"
        )

    # False positive reduction
    fp_reduction = factor.get("false_positive_reduction_pct") or 0
    if fp_reduction < FALSE_POSITIVE_REDUCTION_MIN_PCT:
        blocked_reasons.append(
            f"fp_reduction={fp_reduction}% < {FALSE_POSITIVE_REDUCTION_MIN_PCT}%"
        )

    # Conflict rate
    conflict_rate = factor.get("conflict_rate_pct") or 0
    if conflict_rate > CONFLICT_RATE_CAP_PCT:
        blocked_reasons.append(
            f"conflict_rate={conflict_rate}% > cap={CONFLICT_RATE_CAP_PCT}%"
        )

    # Provenance coverage
    prov_pct = factor.get("provenance_coverage_pct") or 0
    if prov_pct < PROVENANCE_REQUIRED_PCT:
        blocked_reasons.append(
            f"provenance={prov_pct}% < {PROVENANCE_REQUIRED_PCT}%"
        )

    promotion_allowed = len(blocked_reasons) == 0
    return {
        "formula_id": fid,
        "live_sample_count": live_n,
        "promotion_allowed": promotion_allowed,
        "blocked_reasons": blocked_reasons,
        "criteria_checked": 6,
        "criteria_passed": 6 - len(blocked_reasons),
    }


def run(shadow_path: Path, live_replay_path: Path) -> dict:
    shadow      = _load_json(shadow_path)
    live_replay = _load_json(live_replay_path)

    if shadow.get("_missing"):
        result = {
            "gate": "SKIP",
            "reason": f"shadow_ledger missing: {shadow_path}",
            "promotion_candidates": [],
            "blocked_factors": [],
            "contract": "spec/57_shadow_promotion_scorecard.yaml",
        }
        OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
        OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
        return result

    shadow_factors = shadow.get("shadow_factors") or shadow.get("factors") or []
    if not isinstance(shadow_factors, list):
        shadow_factors = []

    evaluations = [_evaluate_shadow_factor(f, live_replay) for f in shadow_factors]
    candidates  = [e for e in evaluations if e["promotion_allowed"]]
    blocked     = [e for e in evaluations if not e["promotion_allowed"]]

    gate = "PASS" if len(blocked) == 0 else "WARN"
    # FAIL only if a factor that was previously promoted (lifecycle=active) now fails
    # WARN if shadow factors are blocked (expected for new factors)

    result = {
        "gate": gate,
        "promotion_candidates": candidates,
        "blocked_factors": blocked,
        "shadow_factor_count": len(shadow_factors),
        "live_sample_minimum": LIVE_SAMPLE_MIN,
        "contract": "spec/57_shadow_promotion_scorecard.yaml",
    }

    OUTPUT_PATH.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2))
    return result


def main() -> None:
    import argparse
    parser = argparse.ArgumentParser(description="H007 Shadow Promotion Scorecard")
    parser.add_argument("--shadow",      default=str(DEFAULT_SHADOW))
    parser.add_argument("--live-replay", default=str(DEFAULT_LIVE_REPLAY))
    args = parser.parse_args()

    result = run(Path(args.shadow), Path(args.live_replay))
    gate = result.get("gate", "FAIL")
    print(f"[H007_SHADOW_PROMOTION_SCORECARD] gate={gate} "
          f"candidates={len(result.get('promotion_candidates', []))} "
          f"blocked={len(result.get('blocked_factors', []))}")
    if gate == "FAIL":
        sys.exit(1)


if __name__ == "__main__":
    main()