QuantEngineByItz/tools/build_cash_recovery_optimizer_v4.py

from __future__ import annotations

import argparse
import json
import sys
from hashlib import sha256
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_JSON = ROOT / "GatherTradingData.json"
DEFAULT_OUT = ROOT / "Temp" / "cash_recovery_optimizer_v4.json"
FORMULA_ID = "CASH_RECOVERY_OPTIMIZER_V4"

if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
    sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1)


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 _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 _canonical(obj: Any) -> str:
    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def _hash_text(text: str) -> str:
    return sha256(text.encode("utf-8")).hexdigest()[:16]


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:
            return _rows(json.loads(v))
        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(description="Build deterministic cash recovery optimizer v4.")
    ap.add_argument("--json", default=str(DEFAULT_JSON))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    json_path = Path(args.json)
    out_path = Path(args.out)
    if not json_path.is_absolute():
        json_path = ROOT / json_path
    if not out_path.is_absolute():
        out_path = ROOT / out_path

    payload = _load(json_path)
    hctx = _extract_harness_root(payload)

    scr7_path = ROOT / "Temp" / "smart_cash_recovery_v7_authoritative.json"
    scr6_path = ROOT / "Temp" / "smart_cash_recovery_v6.json"
    sell_audit_path = ROOT / "Temp" / "sell_engine_audit_v1.json"
    scr7 = _load(scr7_path)
    scr6 = _load(scr6_path)
    sell_audit = _load(sell_audit_path)

    selected_combo = _rows(scr7.get("selected_sell_combo")) or _rows(scr6.get("selected_sell_combo"))
    selected_sell_combo_source = "SMART_CASH_RECOVERY_V7_AUTH" if _rows(scr7.get("selected_sell_combo")) else "SMART_CASH_RECOVERY_V6"
    # v7 authoritative artifact already contains the K2 50/50 redesign with raw damage <= 10.
    value_damage_pct_avg = (
        scr7.get("optimized_value_damage_pct_avg")
        if scr7.get("optimized_value_damage_pct_avg") is not None
        else scr7.get("raw_value_damage_pct_avg")
        if scr7.get("raw_value_damage_pct_avg") is not None
        else scr6.get("value_damage_pct_avg")
    )
    cash_shortfall_min_krw = scr7.get("cash_shortfall_min_krw") or scr6.get("cash_shortfall_min_krw")
    cash_shortfall_covered = bool(scr7.get("cash_shortfall_covered") if scr7 else scr6.get("cash_shortfall_covered"))
    execution_allowed = bool(scr7.get("execution_allowed") if scr7 else scr6.get("execution_allowed"))
    raw_damage_pct = (
        float(selected_combo[0].get("raw_value_damage_pct") or selected_combo[0].get("adjusted_value_damage_pct") or 0.0)
        if selected_combo
        else float(scr6.get("value_damage_pct_avg") or 0.0)
    )
    adjusted_damage_pct = (
        float(selected_combo[0].get("adjusted_value_damage_pct") or selected_combo[0].get("raw_value_damage_pct") or 0.0)
        if selected_combo
        else raw_damage_pct
    )

    sell_authority_conflict_count = 0
    if isinstance(sell_audit.get("scr_plan"), dict):
        audit_combo_count = int(sell_audit["scr_plan"].get("combo_count") or 0)
        if audit_combo_count != len(selected_combo):
            sell_authority_conflict_count += 1
    if any(not isinstance(row, dict) for row in selected_combo):
        sell_authority_conflict_count += 1

    result = {
        "formula_id": FORMULA_ID,
        "status": "PASS" if sell_authority_conflict_count == 0 else "WARN",
        "execution_allowed": execution_allowed,
        "selected_sell_combo_source": selected_sell_combo_source,
        "sell_authority_conflict_count": sell_authority_conflict_count,
        "selected_sell_combo": selected_combo,
        "cash_shortfall_min_krw": cash_shortfall_min_krw,
        "cash_shortfall_covered": cash_shortfall_covered,
        "value_damage_pct_avg": value_damage_pct_avg,
        "value_damage_raw_pct": round(raw_damage_pct, 2),
        "value_damage_adjusted_pct": round(adjusted_damage_pct, 2),
        "value_damage_pct_avg_max": 10.0,
        "sell_engine_audit_gate": sell_audit.get("gate", "MISSING"),
        "optimizer_scr_divergence_count": sell_authority_conflict_count,
        "source": {
            "source_json": str(json_path),
            "smart_cash_recovery_v7_authoritative_json": str(scr7_path),
            "smart_cash_recovery_v6_json": str(scr6_path),
            "sell_engine_audit_v1_json": str(sell_audit_path),
            "generated_by_llm": False,
        },
        "input_hash": _hash_text(_canonical(payload) + _canonical(scr6) + _canonical(sell_audit)),
    }

    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())