QuantEngineByItz/tools/validate_strategy_execution_locks_regression.py

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
TMP_JSON = ROOT / "Temp" / "_strategy_execution_locks_regression_input.json"
RESULT_JSON = ROOT / "Temp" / "strategy_execution_locks_regression_result.json"
LADDER_JSON = ROOT / "Temp" / "execution_method_ladder_v1.json"
TEMP_FILES = {
    "late": ROOT / "Temp" / "late_chase_attribution_v1.json",
    "reb": ROOT / "Temp" / "rebound_sell_efficiency_v1.json",
    "di": ROOT / "Temp" / "data_integrity_score_v1.json",
    "dv": ROOT / "Temp" / "derivation_validity_score_v1.json",
    "de": ROOT / "Temp" / "decision_evidence_score_v1.json",
    "oq": ROOT / "Temp" / "outcome_quality_score_v1.json",
}


def _write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")


def _run_apply() -> dict[str, Any]:
    cmd = [sys.executable, str(ROOT / "tools" / "apply_strategy_execution_locks.py"), "--json", str(TMP_JSON)]
    proc = subprocess.run(cmd, cwd=str(ROOT), text=True, capture_output=True, encoding="utf-8", errors="replace")
    if proc.returncode != 0:
        raise RuntimeError(f"apply_strategy_execution_locks failed: {proc.stdout}\n{proc.stderr}")
    payload = json.loads(TMP_JSON.read_text(encoding="utf-8"))
    h = payload.get("hApex") if isinstance(payload.get("hApex"), dict) else {}
    sel = h.get("strategy_execution_locks_v1_json") if isinstance(h.get("strategy_execution_locks_v1_json"), dict) else {}
    return sel


def _build_execution_method_ladder() -> dict[str, Any]:
    cmd = [
        sys.executable,
        str(ROOT / "tools" / "build_execution_method_ladder_v1.py"),
        "--json",
        str(ROOT / "GatherTradingData.json"),
        "--out",
        str(LADDER_JSON),
    ]
    proc = subprocess.run(cmd, cwd=str(ROOT), text=True, capture_output=True, encoding="utf-8", errors="replace")
    if proc.returncode != 0:
        raise RuntimeError(f"build_execution_method_ladder_v1 failed: {proc.stdout}\n{proc.stderr}")
    payload = json.loads(LADDER_JSON.read_text(encoding="utf-8"))
    return payload


def _build_input() -> dict[str, Any]:
    rows = [
        {"ticker": "AAA001", "order_type": "BUY", "validation_status": "PASS", "quantity": 10, "order_qty": 10, "buy_qty": 10},
        {"ticker": "AAA002", "order_type": "SELL", "validation_status": "PASS", "quantity": 10, "order_qty": 10, "sell_qty": 10},
        {"ticker": "AAA003", "order_type": "STOP_LOSS", "validation_status": "PASS", "quantity": 8, "order_qty": 8, "sell_qty": 8},
    ]
    return {
        "data": {"_harness_context": {"order_blueprint_json": rows}},
        "hApex": {"order_blueprint_json": rows},
    }


def main() -> int:
    backups: dict[str, str | None] = {}
    for key, path in TEMP_FILES.items():
        backups[key] = path.read_text(encoding="utf-8") if path.exists() else None

    try:
        _write_json(TMP_JSON, _build_input())

        # Case 1: outcome low with sufficient eval -> buy block + sell scale
        _write_json(TEMP_FILES["late"], {"formula_id": "LATE_CHASE_ATTRIBUTION_V1", "status": "PASS"})
        _write_json(TEMP_FILES["reb"], {"formula_id": "REBOUND_SELL_EFFICIENCY_V1", "metrics": {"rebound_efficiency_score": 70.0}})
        _write_json(TEMP_FILES["di"], {"formula_id": "DATA_INTEGRITY_SCORE_V1", "score": 100.0, "gate": "PASS"})
        _write_json(TEMP_FILES["dv"], {"formula_id": "DERIVATION_VALIDITY_SCORE_V1", "score": 100.0, "gate": "PASS"})
        _write_json(TEMP_FILES["de"], {"formula_id": "DECISION_EVIDENCE_SCORE_V1", "score": 100.0, "gate": "PASS"})
        _write_json(
            TEMP_FILES["oq"],
            {
                "formula_id": "OUTCOME_QUALITY_SCORE_V1",
                "score": 40.0,
                "gate": "CRITICAL_MODE",
                "metrics": {"has_sufficient_eval": True},
            },
        )
        sel1 = _run_apply()
        if int(sel1.get("buy_block_count") or 0) <= 0:
            raise RuntimeError("REGRESSION_FAIL: expected buy_block_count > 0 for low outcome score")
        if int(sel1.get("sell_scale_count") or 0) <= 0:
            raise RuntimeError("REGRESSION_FAIL: expected sell_scale_count > 0 for low outcome score")

        # Case 2: decision evidence gate block -> hard block
        _write_json(TEMP_FILES["de"], {"formula_id": "DECISION_EVIDENCE_SCORE_V1", "score": 70.0, "gate": "BLOCK"})
        _write_json(TEMP_FILES["oq"], {"formula_id": "OUTCOME_QUALITY_SCORE_V1", "score": 95.0, "gate": "PASS"})
        sel2 = _run_apply()
        if int(sel2.get("hard_block_count") or 0) <= 0:
            raise RuntimeError("REGRESSION_FAIL: expected hard_block_count > 0 for decision evidence block")

        # Case 3: insufficient eval should suspend outcome locks
        _write_json(
            TEMP_FILES["oq"],
            {
                "formula_id": "OUTCOME_QUALITY_SCORE_V1",
                "score": 30.0,
                "gate": "INSUFFICIENT_EVAL",
                "metrics": {"has_sufficient_eval": False},
            },
        )
        _write_json(TEMP_FILES["de"], {"formula_id": "DECISION_EVIDENCE_SCORE_V1", "score": 100.0, "gate": "PASS"})
        sel3 = _run_apply()
        if str(sel3.get("outcome_lock_mode") or "") != "SUSPENDED_DUE_TO_INSUFFICIENT_EVAL":
            raise RuntimeError("REGRESSION_FAIL: expected suspended outcome lock mode under insufficient eval")

        ladder = _build_execution_method_ladder()
        if int(ladder.get("market_order_default_count") or 0) != 0:
            raise RuntimeError("REGRESSION_FAIL: expected market_order_default_count == 0")
        if int(ladder.get("emergency_full_sell_without_flag_count") or 0) != 0:
            raise RuntimeError("REGRESSION_FAIL: expected emergency_full_sell_without_flag_count == 0")

        result = {
            "status": "OK",
            "formula_id": "STRATEGY_EXECUTION_LOCKS_REGRESSION_V1",
            "case1": sel1,
            "case2": sel2,
            "case3": sel3,
            "execution_method_ladder": ladder,
            "assertions": {
                "case1_buy_block_count_gt_0": int(sel1.get("buy_block_count") or 0) > 0,
                "case1_sell_scale_count_gt_0": int(sel1.get("sell_scale_count") or 0) > 0,
                "case2_hard_block_count_gt_0": int(sel2.get("hard_block_count") or 0) > 0,
                "case3_outcome_lock_suspended": str(sel3.get("outcome_lock_mode") or "") == "SUSPENDED_DUE_TO_INSUFFICIENT_EVAL",
                "ladder_market_order_default_count_eq_0": int(ladder.get("market_order_default_count") or 0) == 0,
                "ladder_emergency_flag_missing_count_eq_0": int(ladder.get("emergency_full_sell_without_flag_count") or 0) == 0,
            },
        }
        _write_json(RESULT_JSON, result)
        print("STRATEGY_EXEC_LOCKS_REGRESSION_OK")
        print(json.dumps(result, ensure_ascii=False))
        return 0
    finally:
        for key, path in TEMP_FILES.items():
            if backups[key] is None:
                if path.exists():
                    path.unlink()
            else:
                path.write_text(backups[key] or "", encoding="utf-8")
        if TMP_JSON.exists():
            try:
                TMP_JSON.unlink()
            except FileNotFoundError:
                pass


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