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_FJ = ROOT / "Temp" / "final_judgment_gate_v1.json" DEFAULT_SCR = ROOT / "Temp" / "smart_cash_recovery_v5.json" DEFAULT_HARDENING = ROOT / "Temp" / "strategy_hardening_harness_v2.json" DEFAULT_MATRIX = ROOT / "Temp" / "execution_readiness_matrix_v1.json" DEFAULT_OUT = ROOT / "Temp" / "final_execution_decision_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: return float(value) except Exception: return default def _as_dict(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value if isinstance(value, str) and value.strip(): try: parsed = json.loads(value) return parsed if isinstance(parsed, dict) else {} except Exception: return {} return {} def _extract_harness_root(payload: dict[str, Any]) -> dict[str, Any]: h_apex = payload.get("hApex") data_apex = ((payload.get("data") or {}).get("_harness_context")) if isinstance(payload.get("data"), dict) else None if isinstance(h_apex, dict) and isinstance(data_apex, dict): merged = dict(data_apex) merged.update(h_apex) return merged if isinstance(h_apex, dict): return h_apex if isinstance(data_apex, dict): return data_apex return payload def _rows(v: Any) -> list[dict[str, Any]]: if isinstance(v, list): return [x for x in v if isinstance(x, dict)] if isinstance(v, str) and v.strip(): try: parsed = json.loads(v) return _rows(parsed) except Exception: return [] if isinstance(v, dict): for key in ("rows", "data", "tickers"): candidate = v.get(key) if isinstance(candidate, list): return [x for x in candidate if isinstance(x, dict)] return [] def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--json", default=str(DEFAULT_JSON)) ap.add_argument("--truth", default=str(DEFAULT_TRUTH)) 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("--matrix", default=str(DEFAULT_MATRIX)) 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 payload = _load(_rp(args.json)) hctx = _extract_harness_root(payload) truth = _load(_rp(args.truth)) fj = _load(_rp(args.fj)) scr = _load(_rp(args.scr)) hardening = _load(_rp(args.hardening)) matrix = _load(_rp(args.matrix)) export_gate = _as_dict(hctx.get("export_gate_json")) export_status = str(_first_non_null(export_gate.get("json_validation_status"), hctx.get("json_validation_status")) or "UNKNOWN") export_allowed = export_gate.get("hts_entry_allowed") truth_gate = str(truth.get("gate") or "MISSING") truth_score = _as_float(truth.get("score_0_100")) truth_blockers = truth.get("blocking_reasons") if isinstance(truth.get("blocking_reasons"), list) else [] 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"))) fj_late = int(len(fj.get("late_chase_buy_violations") or [])) 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")) hard_meta = hardening.get("meta_scores") if isinstance(hardening.get("meta_scores"), dict) else {} readiness_gate = str(hard_meta.get("readiness_gate") or "MISSING") matrix_gate = str(matrix.get("gate") or "MISSING") matrix_min_axis = _as_float(matrix.get("min_axis_score")) order_rows = _rows(hctx.get("order_blueprint_json")) hts_candidate_rows = [ r for r in order_rows if str(r.get("validation_status") or "").upper() == "PASS" ] hts_order_count = len(hts_candidate_rows) blocking_reasons: list[str] = [] if truth_gate != "PASS_100": blocking_reasons.append(f"TRUTH_GATE={truth_gate}") if truth_score < 100.0: blocking_reasons.append(f"TRUTH_SCORE={truth_score:.2f}") if export_status != "EXPORT_READY" or export_allowed is not True: blocking_reasons.append(f"EXPORT_GATE={export_status}") if fj_gate != "PASS" or fj_silent > 0 or fj_coverage < 100.0 or fj_late > 0: blocking_reasons.append("FINAL_JUDGMENT_NOT_STABLE") if not scr_allowed or scr_status != "PASS": blocking_reasons.append("SMART_CASH_RECOVERY_BLOCKED") if scr_damage > 10.0: blocking_reasons.append("VALUE_DAMAGE_GT_10") if readiness_gate != "PERFORMANCE_READY": blocking_reasons.append(f"READINESS_GATE={readiness_gate}") if matrix_gate != "PASS_100" or matrix_min_axis < 100.0: blocking_reasons.append(f"EXECUTION_READINESS_MATRIX={matrix_gate}:{matrix_min_axis:.2f}") if not blocking_reasons and hts_order_count > 0: global_gate = "HTS_READY" buy_allowed = True sell_allowed = True llm_allowed_actions = ["HTS_READY"] else: global_gate = "AUDIT_ONLY" buy_allowed = False sell_allowed = False llm_allowed_actions = ["AUDIT_ONLY", "RENDER_LEDGER_ONLY", "SHADOW_LEDGER_ONLY"] result = { "formula_id": "FINAL_EXECUTION_DECISION_V1", "global_execution_gate": global_gate, "buy_allowed": buy_allowed, "sell_allowed": sell_allowed, "hts_order_count": hts_order_count if global_gate == "HTS_READY" else 0, "reason_codes": blocking_reasons, "llm_allowed_actions": llm_allowed_actions, "decision_basis": { "truth_gate": truth_gate, "truth_score_0_100": truth_score, "final_judgment_gate": fj_gate, "final_judgment_coverage_pct": fj_coverage, "smart_cash_recovery_status": scr_status, "smart_cash_recovery_execution_allowed": scr_allowed, "smart_cash_recovery_value_damage_pct": scr_damage, "export_status": export_status, "export_allowed": export_allowed, "readiness_gate": readiness_gate, "execution_readiness_matrix_gate": matrix_gate, "execution_readiness_min_axis_score": matrix_min_axis, "hts_candidate_rows": hts_order_count, }, } 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 def _first_non_null(*values: Any) -> Any: for value in values: if value is not None: return value return None if __name__ == "__main__": raise SystemExit(main())