165 lines
7.7 KiB
Python
165 lines
7.7 KiB
Python
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<floor: BUY→WATCH"})
|
|
final_fa = "WATCH"
|
|
elif h1.get("cashFloorStatus") == "TRIM_REQUIRED" and "BUY" in final_fa:
|
|
trace.append({"gate": "CASH_FLOOR", "result": "BUY_BLOCKED", "reason": "TRIM_REQUIRED: BUY→WATCH"})
|
|
final_fa = "WATCH"
|
|
elif h1.get("cashFloorStatus") == "TRIM_REQUIRED" and final_fa == "HOLD":
|
|
trace.append({"gate": "CASH_FLOOR", "result": "NUDGE_TRIM", "reason": "TRIM_REQUIRED: HOLD→TRIM_33"})
|
|
final_fa = "TRIM_33"
|
|
else:
|
|
trace.append({"gate": "CASH_FLOOR", "result": "PASS", "reason": "cash_floor_pass"})
|
|
|
|
return final_fa, trace
|
|
|
|
|
|
class TestRoutingDecisionParity(unittest.TestCase):
|
|
|
|
def test_stop_breach_routing_gating(self):
|
|
# Scenario 1: Stop breach during post-market (no intraday lock)
|
|
# Expected: FORCE_EXIT -> 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()
|