QuantEngineByItz/tools/build_sell_execution_timing_lock_v2.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" / "sell_execution_timing_lock_v2.json"
FORMULA_ID = "SELL_EXECUTION_TIMING_LOCK_V2"

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 _first_non_null(*values: Any) -> Any:
    for value in values:
        if value is not None:
            return value
    return None


def main() -> int:
    ap = argparse.ArgumentParser(description="Build deterministic sell timing lock output.")
    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)

    sell_timing_verdict = _first_non_null(
        hctx.get("sell_timing_verdict"),
        (hctx.get("sell_execution_timing_json") or {}).get("sell_timing_verdict")
        if isinstance(hctx.get("sell_execution_timing_json"), dict)
        else None,
    )
    if sell_timing_verdict is None:
        sell_timing_verdict = "DATA_MISSING"

    cash_plan = hctx.get("cash_recovery_plan_json")
    if isinstance(cash_plan, str):
        try:
            cash_plan = json.loads(cash_plan)
        except Exception:
            cash_plan = {}
    if not isinstance(cash_plan, dict):
        cash_plan = {}

    sell_sequence = cash_plan.get("sell_sequence") if isinstance(cash_plan.get("sell_sequence"), list) else []
    first_row = next((r for r in sell_sequence if isinstance(r, dict)), {})
    execution_style = str(first_row.get("execution_style") or "")
    immediate_qty = first_row.get("immediate_qty")
    rebound_wait_qty = first_row.get("rebound_wait_qty")
    rebound_trigger_price = first_row.get("rebound_trigger_price")

    if sell_timing_verdict == "SELL_READY":
        recommended_order_type = "LIMIT_SELL"
        timing_reason_code = "SELL_READY_FROM_HARNESS"
    elif sell_timing_verdict == "TIMING_BLOCKED_INTRADAY":
        recommended_order_type = "NEXT_DAY_OPEN"
        timing_reason_code = "INTRADAY_BLOCK"
    elif sell_timing_verdict == "SELL_BLOCKED_DATA":
        recommended_order_type = "DATA_MISSING"
        timing_reason_code = "DATA_MISSING"
    else:
        recommended_order_type = "DATA_MISSING"
        timing_reason_code = "DATA_MISSING"

    result = {
        "formula_id": FORMULA_ID,
        "gate": "PASS" if sell_timing_verdict != "DATA_MISSING" else "WARN",
        "sell_timing_verdict": sell_timing_verdict,
        "recommended_order_type": recommended_order_type,
        "timing_reason_code": timing_reason_code,
        "execution_style": execution_style or "DATA_MISSING",
        "immediate_sell_qty": immediate_qty if immediate_qty is not None else "DATA_MISSING",
        "rebound_wait_qty": rebound_wait_qty if rebound_wait_qty is not None else "DATA_MISSING",
        "rebound_trigger_price": rebound_trigger_price if rebound_trigger_price is not None else "DATA_MISSING",
        "source": {
            "source_json": str(json_path),
            "json_path": "$.data._harness_context.sell_timing_verdict",
            "formula_id": FORMULA_ID,
            "generated_by_llm": False,
        },
        "input_hash": _hash_text(_canonical(payload)),
    }

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