from __future__ import annotations import sys import unittest import math from pathlib import Path ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) # Python Port simulation of GAS runRouteFlow_ logic def run_route_flow_simulation( h: dict, df: dict, h1: dict ) -> tuple[str, list[dict]]: base_fa = str(df.get("finalAction") or "INSUFFICIENT_DATA").upper() final_fa = base_fa trace = [] # ── Gate 1a: Stop_Price Breach 감지 if h.get("stopBreach"): if h1.get("intradayLock"): final_fa = "TRIM_50" trace.append({"gate": "STOP_BREACH", "result": "DOWNGRADE_P4", "reason": "장중(P4): stop_breach→TRIM_50 완화"}) else: final_fa = "EXIT_100" trace.append({"gate": "STOP_BREACH", "result": "FORCE_EXIT", "reason": f"close({h.get('close')})<=stop({h.get('stop_price')})"}) else: trace.append({"gate": "STOP_BREACH", "result": "PASS", "reason": "no_breach"}) # ── Gate 1a-bis: Relative Stop if final_fa != "EXIT_100": rs_ret20d = df.get("ret20d") rs_atr20 = df.get("atr20") rs_close = h.get("close") or df.get("close") or 0.0 rs_pft = h.get("profitPct") rs_hdays = h.get("holdingDays") or 0 rs_kospi = h1.get("kospiRet20d") or 0.0 if rs_ret20d is not None and rs_atr20 is not None and rs_close > 0: rs_beta = min(3.0, max(0.3, rs_ret20d / rs_kospi)) if abs(rs_kospi) >= 0.5 else 1.0 rs_excess = rs_ret20d - rs_beta * rs_kospi rs_sigma = (rs_atr20 / rs_close * 100.0) * math.sqrt(20) rs_thresh = -2.0 * rs_sigma rs_abs_fl = rs_pft is not None and rs_pft < -20.0 rs_time_st = rs_hdays >= 60 and rs_excess < 0 if rs_abs_fl or (rs_excess < rs_thresh) or rs_time_st: trace.append({"gate": "RELATIVE_STOP", "result": "TRIM_50", "reason": "relative_stop_breached"}) if final_fa == "HOLD" or "BUY" in final_fa: final_fa = "TRIM_50" else: trace.append({"gate": "RELATIVE_STOP", "result": "PASS", "reason": "relative_stop_passed"}) else: trace.append({"gate": "RELATIVE_STOP", "result": "SKIP", "reason": "insufficient_data"}) # ── Gate 1b: Intraday_Lock if h1.get("intradayLock"): intraday_blocked_keywords = ["BUY", "BUY_LADDER", "EXIT_100"] intraday_allowed_actions = ["WATCH", "TRIM_50", "HOLD", "TRIM_33"] if any(keyword in final_fa for keyword in intraday_blocked_keywords): downgraded = "WATCH" if "BUY" in final_fa else "TRIM_50" trace.append({"gate": "INTRADAY_LOCK", "result": "DOWNGRADE", "reason": f"P4: {final_fa}→{downgraded}"}) final_fa = downgraded if final_fa not in intraday_allowed_actions: trace.append({"gate": "INTRADAY_LOCK", "result": "FORCE_WATCH", "reason": f"P4_ALLOWLIST: {final_fa} not allowed→WATCH"}) final_fa = "WATCH" else: trace.append({"gate": "INTRADAY_LOCK", "result": "PASS", "reason": "action_in_allowlist"}) else: trace.append({"gate": "INTRADAY_LOCK", "result": "INACTIVE", "reason": "post_market"}) # ── Gate 1c: Heat Gate if h1.get("heatGate") == "BLOCK_NEW_BUY" and "BUY" in final_fa: trace.append({"gate": "HEAT_GATE", "result": "BLOCK_BUY", "reason": "total_heat>=10%: BUY→WATCH"}) final_fa = "WATCH" elif h1.get("heatGate") == "HALVE_NEW_BUY_QUANTITY" and "BUY" in final_fa: trace.append({"gate": "HEAT_GATE", "result": "HALVE_QTY", "reason": "total_heat>=7%: Qty halved"}) else: trace.append({"gate": "HEAT_GATE", "result": "PASS", "reason": "heat_gate_pass"}) # ── Gate 1d: Mean Reversion Gate if "BUY" in final_fa: mrg_close = df.get("close") or 0.0 mrg_ma20 = df.get("ma20") or 0.0 if mrg_close > 0.0 and mrg_ma20 > 0.0: dev_ratio = mrg_close / mrg_ma20 if dev_ratio >= 1.15: trace.append({"gate": "MEAN_REVERSION_GATE", "result": "BUY_HARD_BLOCK", "reason": f"deviation_ratio={dev_ratio:.2f}>=1.15"}) final_fa = "WATCH" else: trace.append({"gate": "MEAN_REVERSION_GATE", "result": "PASS", "reason": "mean_reversion_pass"}) # ── Gate 2: Cash Floor if h1.get("cashFloorStatus") == "HARD_BLOCK" and "BUY" in final_fa: trace.append({"gate": "CASH_FLOOR", "result": "HARD_BLOCK", "reason": "immediate_cash EXIT_100 h_1 = {"stopBreach": True, "close": 9000, "stop_price": 10000} df_1 = {"finalAction": "HOLD"} h1_1 = {"intradayLock": False} final_fa, trace = run_route_flow_simulation(h_1, df_1, h1_1) self.assertEqual(final_fa, "EXIT_100") self.assertEqual(trace[0]["result"], "FORCE_EXIT") # Scenario 2: Stop breach during intraday market (intraday lock active) # Expected: DOWNGRADE_P4 -> TRIM_50 h1_2 = {"intradayLock": True} final_fa_2, trace_2 = run_route_flow_simulation(h_1, df_1, h1_2) self.assertEqual(final_fa_2, "TRIM_50") self.assertEqual(trace_2[0]["result"], "DOWNGRADE_P4") def test_heat_gate_and_mr_gating(self): # Scenario 1: Heat gate BLOCK_NEW_BUY overrides BUY_LADDER h_3 = {"stopBreach": False} df_3 = {"finalAction": "BUY_LADDER", "close": 10000, "ma20": 10000} h1_3 = {"intradayLock": False, "heatGate": "BLOCK_NEW_BUY"} final_fa, trace = run_route_flow_simulation(h_3, df_3, h1_3) self.assertEqual(final_fa, "WATCH") # Scenario 2: Mean Reversion Gate (MRG001) close/ma20 = 12000/10000 = 1.20 >= 1.15 df_4 = {"finalAction": "BUY_LADDER", "close": 12000, "ma20": 10000} h1_4 = {"intradayLock": False, "heatGate": "PASS"} final_fa_4, trace_4 = run_route_flow_simulation(h_3, df_4, h1_4) self.assertEqual(final_fa_4, "WATCH") self.assertTrue(any(t["gate"] == "MEAN_REVERSION_GATE" and t["result"] == "BUY_HARD_BLOCK" for t in trace_4)) def test_cash_floor_routes_hold_to_trim_and_preserves_exit(self): h_5 = {"stopBreach": False} df_5 = {"finalAction": "HOLD"} h1_5 = {"intradayLock": False, "cashFloorStatus": "TRIM_REQUIRED"} final_fa_5, trace_5 = run_route_flow_simulation(h_5, df_5, h1_5) self.assertEqual(final_fa_5, "TRIM_33") self.assertTrue(any(t["gate"] == "CASH_FLOOR" and t["result"] == "NUDGE_TRIM" for t in trace_5)) df_6 = {"finalAction": "EXIT_REVIEW"} final_fa_6, trace_6 = run_route_flow_simulation(h_5, df_6, {"intradayLock": False, "cashFloorStatus": "HARD_BLOCK"}) self.assertEqual(final_fa_6, "EXIT_REVIEW") self.assertTrue(any(t["gate"] == "CASH_FLOOR" and t["result"] == "PASS" for t in trace_6)) if __name__ == "__main__": unittest.main()