QuantEngineByItz/tools/build_execution_readiness_matrix_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_JSON = ROOT / "GatherTradingData.json"
DEFAULT_TRUTH = ROOT / "Temp" / "operational_truth_score_v1.json"
DEFAULT_DQ = ROOT / "Temp" / "data_quality_reconciliation_v1.json"
DEFAULT_FJ = ROOT / "Temp" / "final_judgment_gate_v1.json"
DEFAULT_SCR = ROOT / "Temp" / "smart_cash_recovery_v7.json"
DEFAULT_HARDENING = ROOT / "Temp" / "strategy_hardening_harness_v2.json"
DEFAULT_OUT = ROOT / "Temp" / "execution_readiness_matrix_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 _as_float(value: Any, default: float = 0.0) -> float:
    try:
        if value is None or value == "":
            return default
        return float(value)
    except Exception:
        return default


def _clamp_score(value: Any) -> float:
    return round(max(0.0, min(100.0, _as_float(value))), 2)


def _axis_gate(score: float, hard_block: bool = False) -> str:
    if hard_block or score < 50.0:
        return "BLOCK"
    if score < 100.0:
        return "WATCH"
    return "PASS_100"


def _axis(axis: str, score: Any, source_json: str, formula_id: str, reason: str = "", hard_block: bool = False) -> dict[str, Any]:
    s = _clamp_score(score)
    gate = _axis_gate(s, hard_block=hard_block)
    return {
        "axis": axis,
        "score_0_100": s,
        "gate": gate,
        "blocking_reason": reason if gate != "PASS_100" else "NONE",
        "source_json": source_json,
        "formula_id": formula_id,
    }


def main() -> int:
    ap = argparse.ArgumentParser(description="Build EXECUTION_READINESS_MATRIX_V1.")
    ap.add_argument("--json", default=str(DEFAULT_JSON))
    ap.add_argument("--truth", default=str(DEFAULT_TRUTH))
    ap.add_argument("--dq", default=str(DEFAULT_DQ))
    ap.add_argument("--fj", default=str(DEFAULT_FJ))
    ap.add_argument("--scr", default=str(DEFAULT_SCR))
    ap.add_argument("--hardening", default=str(DEFAULT_HARDENING))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    def _rp(path_str: str) -> Path:
        path = Path(path_str)
        return path if path.is_absolute() else ROOT / path

    truth = _load(_rp(args.truth))
    dq = _load(_rp(args.dq))
    fj = _load(_rp(args.fj))
    scr = _load(_rp(args.scr))
    hardening = _load(_rp(args.hardening))

    domain_scores = hardening.get("domain_scores") if isinstance(hardening.get("domain_scores"), dict) else {}
    meta_scores = hardening.get("meta_scores") if isinstance(hardening.get("meta_scores"), dict) else {}

    cap_basis = _clamp_score(dq.get("confidence_cap_basis_score", truth.get("data_truth_score")))
    fundamental_domain = _clamp_score(domain_scores.get("fundamental"))
    fundamental_score = min(fundamental_domain, cap_basis)

    scr_allowed = bool(scr.get("execution_allowed"))
    scr_status = str(scr.get("status") or "UNKNOWN")
    scr_damage = _as_float(scr.get("value_damage_pct_avg"))
    if scr_allowed and scr_status == "PASS" and scr_damage <= 10.0:
        cash_recovery_score = 100.0
        cash_recovery_reason = ""
        cash_recovery_hard_block = False
    else:
        cash_recovery_score = 0.0
        cash_recovery_reason = f"SCR_STATUS={scr_status};EXECUTION_ALLOWED={scr_allowed};VALUE_DAMAGE={round(scr_damage, 2)}"
        cash_recovery_hard_block = True

    final_judgment_score = 100.0
    final_judgment_reason = ""
    fj_gate = str(fj.get("gate") or "MISSING")
    fj_coverage = _as_float(fj.get("coverage_pct"))
    fj_silent = int(_as_float(fj.get("silent_pass_violations")))
    if fj_gate != "PASS" or fj_coverage < 100.0 or fj_silent > 0:
        final_judgment_score = 0.0
        final_judgment_reason = f"FJ_GATE={fj_gate};COVERAGE={round(fj_coverage, 2)};SILENT={fj_silent}"

    axes = [
        _axis("data_integrity", truth.get("data_truth_score"), "operational_truth_score_v1.json", "OPERATIONAL_TRUTH_SCORE_V1", "DATA_TRUTH_LT_100"),
        _axis("routing_serving", domain_scores.get("routing_serving"), "strategy_hardening_harness_v2.json", "STRATEGY_HARDENING_HARNESS_V2", "ROUTING_SERVING_LT_100"),
        _axis("serving_output_lock", truth.get("report_consistency_score"), "operational_truth_score_v1.json", "OPERATIONAL_TRUTH_SCORE_V1", "REPORT_CONSISTENCY_LT_100"),
        _axis("decision_governance", truth.get("decision_truth_score"), "operational_truth_score_v1.json", "OPERATIONAL_TRUTH_SCORE_V1", "DECISION_TRUTH_LT_100"),
        _axis("final_judgment_lock", final_judgment_score, "final_judgment_gate_v1.json", "FINAL_JUDGMENT_GATE_V1", final_judgment_reason),
        _axis("fundamental_basis", fundamental_score, "data_quality_reconciliation_v1.json", "DATA_QUALITY_RECONCILIATION_V1", "CONFIDENCE_CAP_BASIS_LT_100"),
        _axis("horizon_policy", domain_scores.get("horizon_short_mid_long"), "strategy_hardening_harness_v2.json", "STRATEGY_HARDENING_HARNESS_V2", "HORIZON_POLICY_LT_100"),
        _axis("smart_money_liquidity", domain_scores.get("smart_money_liquidity"), "strategy_hardening_harness_v2.json", "STRATEGY_HARDENING_HARNESS_V2", "SMART_MONEY_LIQUIDITY_LT_100"),
        _axis("cash_recovery_execution", cash_recovery_score, "smart_cash_recovery_v7.json", "SMART_CASH_RECOVERY_V7", cash_recovery_reason, cash_recovery_hard_block),
        _axis("execution_availability", truth.get("execution_truth_score"), "operational_truth_score_v1.json", "OPERATIONAL_TRUTH_SCORE_V1", "EXECUTION_TRUTH_LT_100"),
        _axis("performance_readiness", truth.get("performance_readiness_score"), "operational_truth_score_v1.json", "OPERATIONAL_TRUTH_SCORE_V1", "PERFORMANCE_READINESS_LT_100"),
        _axis("report_consistency", truth.get("report_consistency_score"), "operational_truth_score_v1.json", "OPERATIONAL_TRUTH_SCORE_V1", "REPORT_CONSISTENCY_LT_100"),
    ]

    scores = [float(row["score_0_100"]) for row in axes]
    hard_blocks = [row for row in axes if row["gate"] == "BLOCK"]
    min_axis_score = round(min(scores), 2) if scores else 0.0
    average_axis_score = round(sum(scores) / len(scores), 2) if scores else 0.0

    truth_gate = str(truth.get("gate") or "MISSING")
    readiness_gate = str(meta_scores.get("readiness_gate") or "MISSING")
    if hard_blocks:
        gate = "BLOCK_EXECUTION"
    elif min_axis_score == 100.0 and truth_gate == "PASS_100" and readiness_gate == "PERFORMANCE_READY":
        gate = "PASS_100"
    else:
        gate = "WATCH_PENDING_SAMPLE"

    result = {
        "formula_id": "EXECUTION_READINESS_MATRIX_V1",
        "gate": gate,
        "min_axis_score": min_axis_score,
        "average_axis_score": average_axis_score,
        "hard_block_count": len(hard_blocks),
        "blocking_reasons": [str(row["blocking_reason"]) for row in axes if row["gate"] != "PASS_100"],
        "axes": axes,
        "targets": {
            "pass_100_rule": "all axes score_0_100 == 100, truth gate PASS_100, readiness_gate PERFORMANCE_READY",
            "block_rule": "any axis gate BLOCK",
            "watch_rule": "no BLOCK but at least one axis below 100 or insufficient performance sample",
        },
        "metric_basis": {
            "truth_gate": truth_gate,
            "truth_score_0_100": truth.get("score_0_100"),
            "readiness_gate": readiness_gate,
            "confidence_cap_basis_score": cap_basis,
            "fundamental_domain_score": fundamental_domain,
            "smart_cash_recovery_status": scr_status,
            "smart_cash_recovery_execution_allowed": scr_allowed,
            "smart_cash_recovery_value_damage_pct": round(scr_damage, 2),
        },
    }

    out_path = _rp(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.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())