from __future__ import annotations import math import sys import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) 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 = [] if h.get("stopBreach"): if h1.get("intradayLock"): final_fa = "TRIM_50" trace.append({"gate": "STOP_BREACH", "result": "DOWNGRADE_P4"}) else: final_fa = "EXIT_100" trace.append({"gate": "STOP_BREACH", "result": "FORCE_EXIT"}) else: trace.append({"gate": "STOP_BREACH", "result": "PASS"}) 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"}) if final_fa == "HOLD" or "BUY" in final_fa: final_fa = "TRIM_50" else: trace.append({"gate": "RELATIVE_STOP", "result": "PASS"}) else: trace.append({"gate": "RELATIVE_STOP", "result": "SKIP"}) 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): final_fa = "WATCH" if "BUY" in final_fa else "TRIM_50" if final_fa not in intraday_allowed_actions: final_fa = "WATCH" if h1.get("heatGate") == "BLOCK_NEW_BUY" and "BUY" in final_fa: final_fa = "WATCH" elif h1.get("heatGate") == "HALVE_NEW_BUY_QUANTITY" and "BUY" in final_fa: pass 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 and (mrg_close / mrg_ma20) >= 1.15: final_fa = "WATCH" if h1.get("cashFloorStatus") == "HARD_BLOCK" and "BUY" in final_fa: final_fa = "WATCH" elif h1.get("cashFloorStatus") == "TRIM_REQUIRED" and "BUY" in final_fa: final_fa = "WATCH" elif h1.get("cashFloorStatus") == "TRIM_REQUIRED" and final_fa == "HOLD": final_fa = "TRIM_33" return final_fa, trace class TestRoutingGateParityV1(unittest.TestCase): def test_stop_breach_routes_exit_or_trim(self): final_fa, trace = run_route_flow_simulation({"stopBreach": True}, {"finalAction": "HOLD"}, {"intradayLock": False}) self.assertEqual(final_fa, "EXIT_100") self.assertEqual(trace[0]["result"], "FORCE_EXIT") final_fa_lock, trace_lock = run_route_flow_simulation({"stopBreach": True}, {"finalAction": "HOLD"}, {"intradayLock": True}) self.assertEqual(final_fa_lock, "TRIM_50") self.assertEqual(trace_lock[0]["result"], "DOWNGRADE_P4") def test_heat_gate_blocks_buy_and_allows_half_quantity_trace(self): final_fa_buy, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "BUY_LADDER", "close": 12000, "ma20": 10000}, {"heatGate": "PASS"}) self.assertEqual(final_fa_buy, "WATCH") final_fa_half, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "BUY_LADDER", "close": 11000, "ma20": 10000}, {"heatGate": "HALVE_NEW_BUY_QUANTITY"}) self.assertEqual(final_fa_half, "BUY_LADDER") def test_cash_floor_routes_hold_and_buy_separately(self): final_fa, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "HOLD"}, {"cashFloorStatus": "TRIM_REQUIRED"}) self.assertEqual(final_fa, "TRIM_33") final_fa_cash_block, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "BUY_LADDER"}, {"cashFloorStatus": "HARD_BLOCK"}) self.assertEqual(final_fa_cash_block, "WATCH") final_fa_trim_buy, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "BUY_LADDER"}, {"cashFloorStatus": "TRIM_REQUIRED"}) self.assertEqual(final_fa_trim_buy, "WATCH") def test_relative_stop_and_mean_reversion(self): final_fa_rw, trace_rw = run_route_flow_simulation( {"stopBreach": False, "profitPct": -25.0}, {"finalAction": "HOLD", "ret20d": -5.0, "atr20": 100.0, "close": 10000.0}, {"intradayLock": False, "kospiRet20d": 1.0}, ) self.assertEqual(final_fa_rw, "TRIM_50") self.assertTrue(any(item["gate"] == "RELATIVE_STOP" for item in trace_rw)) final_fa_mr, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "BUY_LADDER", "close": 12000, "ma20": 10000}, {"heatGate": "PASS", "cashFloorStatus": "PASS"}) self.assertEqual(final_fa_mr, "WATCH") def test_cash_floor_hard_block_preserves_non_buy_actions(self): final_fa, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "EXIT_REVIEW"}, {"cashFloorStatus": "HARD_BLOCK"}) self.assertEqual(final_fa, "EXIT_REVIEW") final_fa_hard_hold, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "HOLD"}, {"cashFloorStatus": "HARD_BLOCK"}) self.assertEqual(final_fa_hard_hold, "HOLD") final_fa_trim_hold, _ = run_route_flow_simulation({"stopBreach": False}, {"finalAction": "HOLD"}, {"cashFloorStatus": "TRIM_REQUIRED", "heatGate": "PASS"}) self.assertEqual(final_fa_trim_hold, "TRIM_33") if __name__ == "__main__": unittest.main()