QuantEngineByItz/tools/validate_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_PATH = ROOT / "Temp" / "execution_readiness_matrix_v1.json"
REQUIRED_AXES = {
    "data_integrity",
    "routing_serving",
    "serving_output_lock",
    "decision_governance",
    "final_judgment_lock",
    "fundamental_basis",
    "horizon_policy",
    "smart_money_liquidity",
    "cash_recovery_execution",
    "execution_availability",
    "performance_readiness",
    "report_consistency",
}


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 main() -> int:
    ap = argparse.ArgumentParser(description="Validate EXECUTION_READINESS_MATRIX_V1.")
    ap.add_argument("--json", default=str(DEFAULT_PATH))
    args = ap.parse_args()

    path = Path(args.json)
    if not path.is_absolute():
        path = ROOT / path

    payload = _load(path)
    errors: list[str] = []

    if str(payload.get("formula_id") or "") != "EXECUTION_READINESS_MATRIX_V1":
        errors.append("formula_id mismatch")

    gate = str(payload.get("gate") or "")
    if gate not in {"PASS_100", "WATCH_PENDING_SAMPLE", "BLOCK_EXECUTION"}:
        errors.append(f"gate={gate}")

    axes = payload.get("axes")
    if not isinstance(axes, list) or not axes:
        errors.append("axes must be non-empty list")
        axes = []

    seen: set[str] = set()
    scores: list[float] = []
    block_count = 0
    for idx, row in enumerate(axes):
        if not isinstance(row, dict):
            errors.append(f"axes[{idx}] must be object")
            continue
        axis = str(row.get("axis") or "")
        seen.add(axis)
        score = row.get("score_0_100")
        if not isinstance(score, (int, float)):
            errors.append(f"axes[{idx}].score_0_100 must be numeric")
            continue
        score_f = float(score)
        if score_f < 0 or score_f > 100:
            errors.append(f"axes[{idx}].score_0_100 out of range")
        scores.append(score_f)
        row_gate = str(row.get("gate") or "")
        if row_gate not in {"PASS_100", "WATCH", "BLOCK"}:
            errors.append(f"axes[{idx}].gate={row_gate}")
        if row_gate == "BLOCK":
            block_count += 1
        if row_gate != "PASS_100" and not str(row.get("blocking_reason") or "").strip():
            errors.append(f"axes[{idx}].blocking_reason missing")
        if not str(row.get("source_json") or "").strip():
            errors.append(f"axes[{idx}].source_json missing")
        if not str(row.get("formula_id") or "").strip():
            errors.append(f"axes[{idx}].formula_id missing")

    missing_axes = sorted(REQUIRED_AXES - seen)
    if missing_axes:
        errors.append(f"missing_axes={missing_axes}")

    if scores:
        expected_min = round(min(scores), 2)
        actual_min = payload.get("min_axis_score")
        if not isinstance(actual_min, (int, float)) or round(float(actual_min), 2) != expected_min:
            errors.append(f"min_axis_score mismatch expected={expected_min} actual={actual_min}")
        expected_avg = round(sum(scores) / len(scores), 2)
        actual_avg = payload.get("average_axis_score")
        if not isinstance(actual_avg, (int, float)) or round(float(actual_avg), 2) != expected_avg:
            errors.append(f"average_axis_score mismatch expected={expected_avg} actual={actual_avg}")

    actual_blocks = payload.get("hard_block_count")
    if not isinstance(actual_blocks, int) or actual_blocks != block_count:
        errors.append(f"hard_block_count mismatch expected={block_count} actual={actual_blocks}")

    if gate == "PASS_100":
        if block_count != 0:
            errors.append("PASS_100 cannot have BLOCK axes")
        if scores and min(scores) < 100.0:
            errors.append("PASS_100 requires all axes score_0_100=100")
    if gate == "BLOCK_EXECUTION" and block_count == 0:
        errors.append("BLOCK_EXECUTION requires at least one BLOCK axis")

    if errors:
        print("EXECUTION_READINESS_MATRIX_V1_FAIL")
        for err in errors:
            print(f"  {err}")
        return 1

    print("EXECUTION_READINESS_MATRIX_V1_OK")
    print(f"  gate: {gate}")
    print(f"  min_axis_score: {float(payload.get('min_axis_score')):.2f}")
    print(f"  hard_block_count: {block_count}")
    return 0


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