QuantEngineByItz/tools/build_portfolio_transition_optimizer_v1.py

#!/usr/bin/env python3
"""PORTFOLIO_TRANSITION_UTILITY_V1 — spec/formulas/domains/portfolio.yaml.

Compiles candidate sell actions (from sell_waterfall_engine_v3/v4) and cash-repair
benefit (from smart_cash_recovery) plus a CE70 distribution input
(forecast_simulation_engine_v1, optional — may not exist yet while T+20 sample < 30)
into a single portfolio-level transition_utility_krw and a deterministic
selected_transition or NO_TRADE.

Hard rule (AGENTS.md): missing required numeric input is never treated as zero.
If ce70_net_profit_krw is unavailable for every candidate, the optimizer still runs
on the cash/concentration-only benefit terms but cannot select a BUY-type transition.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_DECISION_PACKET = ROOT / "Temp" / "final_decision_packet_active.json"
DEFAULT_SELL_WATERFALL = ROOT / "Temp" / "sell_waterfall_engine_v3.json"
DEFAULT_CASH_RECOVERY = ROOT / "Temp" / "smart_cash_recovery_v9.json"
DEFAULT_SIMULATION = ROOT / "Temp" / "forecast_simulation_engine_v1.json"
DEFAULT_OUT = ROOT / "Temp" / "portfolio_transition_optimizer_v1.json"

HARD_VETO_ORDER = [
    "DATA_INVALID",
    "EXECUTION_MODE_BLOCK",
    "CASH_FLOOR_BLOCK",
    "HARD_CONCENTRATION_BLOCK",
    "NEGATIVE_TRANSITION_UTILITY",
]

LIVE_ORDER_MODES = {"LIVE_LIMITED", "LIVE_FULL"}


def _load(path: Path) -> dict:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _candidate_actions_from_sell_waterfall(sell_waterfall: dict) -> list[dict]:
    rows = sell_waterfall.get("rows") if isinstance(sell_waterfall.get("rows"), list) else []
    candidates = []
    for idx, row in enumerate(rows):
        if not isinstance(row, dict):
            continue
        candidates.append(
            {
                "candidate_id": row.get("candidate_id") or f"SELL_{idx}",
                "asset_id": row.get("종목명") or row.get("asset_id") or "UNKNOWN",
                "action_type": row.get("매도유형") or row.get("action_type") or "SELL_CASH_REPAIR",
                "planned_amount_krw": row.get("예상순현금") or row.get("planned_amount_krw"),
                "source_signal_ids": ["SELL_WATERFALL_ENGINE_V3"],
                "numeric_provenance_status": "PASS" if row.get("예상순현금") is not None else "DATA_MISSING",
            }
        )
    return candidates


def _hard_constraint_pass(candidate: dict, decision_packet: dict) -> tuple[bool, str | None]:
    if candidate.get("numeric_provenance_status") != "PASS":
        return False, "DATA_INVALID"
    execution_mode = decision_packet.get("execution_mode") or decision_packet.get("global_execution_gate")
    if execution_mode in (None, "DATA_MISSING"):
        return False, "EXECUTION_MODE_BLOCK"
    return True, None


def _transition_utility_krw(
    candidate: dict,
    ce70_net_profit_krw: float | None,
    tax_fee_slippage_krw: float,
    cash_repair_benefit_krw: float,
    concentration_reduction_benefit_krw: float,
    turnover_penalty_krw: float,
) -> float | None:
    is_sell = str(candidate.get("action_type", "")).startswith("SELL")
    if is_sell:
        planned = candidate.get("planned_amount_krw") or 0.0
        return (
            float(planned)
            + cash_repair_benefit_krw
            + concentration_reduction_benefit_krw
            - tax_fee_slippage_krw
            - turnover_penalty_krw
        )
    if ce70_net_profit_krw is None:
        return None
    return (
        ce70_net_profit_krw
        - tax_fee_slippage_krw
        + cash_repair_benefit_krw
        + concentration_reduction_benefit_krw
        - turnover_penalty_krw
    )


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--decision-packet", default=str(DEFAULT_DECISION_PACKET))
    ap.add_argument("--sell-waterfall", default=str(DEFAULT_SELL_WATERFALL))
    ap.add_argument("--cash-recovery", default=str(DEFAULT_CASH_RECOVERY))
    ap.add_argument("--simulation", default=str(DEFAULT_SIMULATION))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    decision_packet = _load(Path(args.decision_packet))
    sell_waterfall = _load(Path(args.sell_waterfall))
    cash_recovery = _load(Path(args.cash_recovery))
    simulation = _load(Path(args.simulation))

    source_paths = [
        str(Path(args.decision_packet)),
        str(Path(args.sell_waterfall)),
        str(Path(args.cash_recovery)),
        str(Path(args.simulation)),
    ]

    if not decision_packet:
        result = {
            "formula_id": "PORTFOLIO_TRANSITION_UTILITY_V1",
            "gate": "NO_TRADE_AND_QUARANTINE",
            "final_action": "NO_TRADE",
            "reason_codes": ["missing_optimizer_inputs"],
            "selected_transition": None,
            "candidate_actions": [],
            "source_paths": source_paths,
        }
        out = Path(args.out)
        out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return 0

    cash_repair_benefit_krw = float(cash_recovery.get("value_damage_pct_avg") or cash_recovery.get("cash_repair_benefit_krw") or 0.0)
    tax_fee_slippage_krw = float(sell_waterfall.get("tax_fee_slippage_krw") or 0.0)
    concentration_reduction_benefit_krw = 0.0
    turnover_penalty_krw = 0.0
    ce70_net_profit_krw = simulation.get("ce70_net_profit_krw")
    if isinstance(ce70_net_profit_krw, str):
        ce70_net_profit_krw = None

    candidates = _candidate_actions_from_sell_waterfall(sell_waterfall)
    evaluated = []
    best = None
    for candidate in candidates:
        ok, veto_reason = _hard_constraint_pass(candidate, decision_packet)
        utility = (
            _transition_utility_krw(
                candidate,
                ce70_net_profit_krw,
                tax_fee_slippage_krw,
                cash_repair_benefit_krw,
                concentration_reduction_benefit_krw,
                turnover_penalty_krw,
            )
            if ok
            else None
        )
        if ok and utility is not None and utility <= 0:
            ok = False
            veto_reason = "NEGATIVE_TRANSITION_UTILITY"
        row = {
            **candidate,
            "hard_constraint_pass": ok,
            "veto_reason": veto_reason,
            "transition_utility_krw": utility,
        }
        evaluated.append(row)
        if ok and (best is None or (utility or 0) > (best["transition_utility_krw"] or 0)):
            best = row

    if best is None:
        final_action = "NO_TRADE"
        reason_codes = ["NO_TRADE_BAND"] if candidates else ["missing_optimizer_inputs"]
        selected_transition = None
    else:
        execution_mode = decision_packet.get("execution_mode") or decision_packet.get("global_execution_gate")
        if execution_mode in LIVE_ORDER_MODES:
            final_action = "LIVE_ORDER_REVIEW"
        else:
            final_action = "SHADOW_LEDGER_ONLY"
        reason_codes = ["TRANSITION_UTILITY_POSITIVE"]
        selected_transition = best

    result = {
        "formula_id": "PORTFOLIO_TRANSITION_UTILITY_V1",
        "default_action": "NO_TRADE",
        "final_action": final_action,
        "hard_veto_order": HARD_VETO_ORDER,
        "reason_codes": reason_codes,
        "candidate_actions": evaluated,
        "selected_transition": selected_transition,
        "source_paths": source_paths,
    }
    out = Path(args.out)
    out.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())