#!/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())