132 lines
6.1 KiB
Python
132 lines
6.1 KiB
Python
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()
|