kjh2064
468ad73c52
F05 (execution_decision_v1.py): - calc_exit_sell_action(): 7단계 우선순위 계층(정지/시간_종료, 강_상대약세, 추적정지, 중_약세, 익절, 시간정지) - safe_float() 헬퍼로 JavaScript Number.isFinite() 의미론 보장 - tp2→tp1→closeProtectLimit 가격 폴백 체인 - 17개 parity 테스트 PASS (우선순위, 가격 추적, 검증 상태) F10 (routing_decision_v1.py): - run_route_flow(): 5개 게이트 순차 필터링 1. Stop_Breach: EXIT_100 또는 P4 인트라데이 락시 TRIM_50 2. Relative_Stop: 베타조정 시장정지(절대_바닥, 상대_초과, 시간조건) 3. Intraday_Lock: P4 제약(BUY→WATCH, ADD→TRIM_50, 허용목록 강제) 4. Heat_Gate: 포트폴리오 열기제어(BLOCK_NEW_BUY/HALVE_NEW_BUY_QUANTITY) 5. Mean_Reversion: MRG001(close/ma20 > 1.10이면 BUY 차단) - 17개 parity 테스트 PASS (5개 게이트 모두 테스트됨) 마이그레이션 레저 업데이트: - F05: TODO → DONE - F10: TODO → DONE - 누적 parity 테스트: 64/64 PASS Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
237 lines
11 KiB
Python
237 lines
11 KiB
Python
"""
|
|
Parity test for routing_decision_v1.py against GAS source.
|
|
|
|
F10: Portfolio routing through multi-gate decision framework.
|
|
Tests run_route_flow() with all 5 gates: stop_breach, relative_stop,
|
|
intraday_lock, heat_gate, mean_reversion.
|
|
|
|
Source: src/gas_adapter_parts/gdf_03_portfolio_gates.gs:runRouteFlow_
|
|
"""
|
|
|
|
import pytest
|
|
from formulas.routing_decision_v1 import run_route_flow
|
|
|
|
|
|
class TestRoutingDecisionGates:
|
|
"""Test routing decision multi-gate filtering."""
|
|
|
|
def test_gate1_stop_breach_normal(self):
|
|
"""Gate 1: stop breach without intraday lock → EXIT_100."""
|
|
holdings = [{"ticker": "000660", "stopBreach": True, "close": 95, "stopPrice": 98}]
|
|
df_map = {"000660": {"finalAction": "HOLD", "ret20d": 0.10}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
assert result["decisions"][0]["final_action"] == "EXIT_100"
|
|
gates = result["traces"][0]["gates"]
|
|
assert gates[0]["gate"] == "STOP_BREACH"
|
|
assert gates[0]["result"] == "FORCE_EXIT"
|
|
|
|
def test_gate1_stop_breach_with_intraday_lock(self):
|
|
"""Gate 1: stop breach with intraday lock → TRIM_50."""
|
|
holdings = [{"ticker": "005380", "stopBreach": True, "close": 50, "stopPrice": 52}]
|
|
df_map = {"005380": {"finalAction": "HOLD"}}
|
|
h1_ctx = {"intradayLock": True, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
assert result["decisions"][0]["final_action"] == "TRIM_50"
|
|
gates = result["traces"][0]["gates"]
|
|
assert gates[0]["result"] == "DOWNGRADE_P4"
|
|
|
|
def test_gate1_no_breach(self):
|
|
"""Gate 1: no stop breach → PASS."""
|
|
holdings = [{"ticker": "051910", "stopBreach": False, "close": 100, "stopPrice": 90}]
|
|
df_map = {"051910": {"finalAction": "BUY_TIER1"}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
# Gate 1 passes, checks other gates
|
|
gates = result["traces"][0]["gates"]
|
|
assert gates[0]["result"] == "PASS"
|
|
|
|
def test_gate2_relative_stop_abs_floor(self):
|
|
"""Gate 2: profit < -20% → TRIM_50."""
|
|
holdings = [{"ticker": "006800", "stopBreach": False, "close": 80, "profitPct": -25, "holdingDays": 30}]
|
|
df_map = {"006800": {"finalAction": "HOLD", "ret20d": -0.10, "atr20": 5.0}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "TRIM_50"
|
|
gates = result["traces"][0]["gates"]
|
|
rel_gate = [g for g in gates if g["gate"] == "RELATIVE_STOP"][0]
|
|
assert rel_gate["result"] == "TRIM_50"
|
|
assert "ABS_FLOOR" in rel_gate["reason"]
|
|
|
|
def test_gate2_relative_stop_time_stop(self):
|
|
"""Gate 2: holding >= 60 days + excess < 0 → TRIM_50."""
|
|
holdings = [{"ticker": "035720", "stopBreach": False, "close": 100, "profitPct": 5, "holdingDays": 65}]
|
|
df_map = {"035720": {"finalAction": "HOLD", "ret20d": 0.05, "atr20": 4.0}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.10}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "TRIM_50"
|
|
|
|
def test_gate2_relative_stop_skip(self):
|
|
"""Gate 2: insufficient data (no atr20) → SKIP."""
|
|
holdings = [{"ticker": "000020", "stopBreach": False, "close": 100, "holdingDays": 30}]
|
|
df_map = {"000020": {"finalAction": "HOLD", "ret20d": 0.10}} # no atr20
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
gates = result["traces"][0]["gates"]
|
|
rel_gate = [g for g in gates if g["gate"] == "RELATIVE_STOP"][0]
|
|
assert rel_gate["result"] == "SKIP"
|
|
|
|
def test_gate3_intraday_lock_downgrade_buy(self):
|
|
"""Gate 3: intraday lock with BUY → downgrade to WATCH."""
|
|
holdings = [{"ticker": "011170", "stopBreach": False, "close": 100}]
|
|
df_map = {"011170": {"finalAction": "BUY_TIER1", "ret20d": 0.10, "atr20": 3.0}}
|
|
h1_ctx = {"intradayLock": True, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "WATCH"
|
|
gates = result["traces"][0]["gates"]
|
|
intraday_gate = [g for g in gates if g["gate"] == "INTRADAY_LOCK"][0]
|
|
assert "DOWNGRADE" in intraday_gate["result"]
|
|
|
|
def test_gate3_intraday_lock_downgrade_add(self):
|
|
"""Gate 3: intraday lock with ADD → downgrade to TRIM_50."""
|
|
holdings = [{"ticker": "017670", "stopBreach": False, "close": 100}]
|
|
df_map = {"017670": {"finalAction": "ADD_POSITION", "ret20d": 0.10, "atr20": 3.0}}
|
|
h1_ctx = {"intradayLock": True, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "TRIM_50"
|
|
|
|
def test_gate3_intraday_lock_allowlist_pass(self):
|
|
"""Gate 3: intraday lock with allowed action (HOLD) → PASS."""
|
|
holdings = [{"ticker": "015760", "stopBreach": False, "close": 100}]
|
|
df_map = {"015760": {"finalAction": "HOLD", "ret20d": 0.10, "atr20": 3.0}}
|
|
h1_ctx = {"intradayLock": True, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "HOLD"
|
|
gates = result["traces"][0]["gates"]
|
|
intraday_gate = [g for g in gates if g["gate"] == "INTRADAY_LOCK"][0]
|
|
assert intraday_gate["result"] == "PASS"
|
|
|
|
def test_gate4_heat_gate_block_new_buy(self):
|
|
"""Gate 4: heat_gate=BLOCK_NEW_BUY with BUY → WATCH."""
|
|
holdings = [{"ticker": "021240", "stopBreach": False, "close": 100}]
|
|
df_map = {"021240": {"finalAction": "BUY_TIER2", "ret20d": 0.10, "atr20": 3.0, "close": 100, "ma20": 95}}
|
|
h1_ctx = {"intradayLock": False, "heatGate": "BLOCK_NEW_BUY", "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "WATCH"
|
|
gates = result["traces"][0]["gates"]
|
|
heat_gate = [g for g in gates if g["gate"] == "HEAT_GATE"][0]
|
|
assert heat_gate["result"] == "BLOCK_BUY"
|
|
|
|
def test_gate4_heat_gate_halve_qty(self):
|
|
"""Gate 4: heat_gate=HALVE_NEW_BUY_QUANTITY with BUY → HALVE_QTY."""
|
|
holdings = [{"ticker": "030000", "stopBreach": False, "close": 100}]
|
|
df_map = {"030000": {"finalAction": "BUY_TIER3", "ret20d": 0.10, "atr20": 3.0, "close": 100, "ma20": 95}}
|
|
h1_ctx = {"intradayLock": False, "heatGate": "HALVE_NEW_BUY_QUANTITY", "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
gates = result["traces"][0]["gates"]
|
|
heat_gate = [g for g in gates if g["gate"] == "HEAT_GATE"][0]
|
|
assert heat_gate["result"] == "HALVE_QTY"
|
|
|
|
def test_gate4_heat_gate_hold_pass(self):
|
|
"""Gate 4: heat_gate with HOLD → PASS (not BUY)."""
|
|
holdings = [{"ticker": "045570", "stopBreach": False, "close": 100}]
|
|
df_map = {"045570": {"finalAction": "HOLD", "ret20d": 0.10, "atr20": 3.0}}
|
|
h1_ctx = {"intradayLock": False, "heatGate": "BLOCK_NEW_BUY", "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
gates = result["traces"][0]["gates"]
|
|
heat_gate = [g for g in gates if g["gate"] == "HEAT_GATE"][0]
|
|
assert heat_gate["result"] == "PASS"
|
|
|
|
def test_gate5_mean_reversion_block(self):
|
|
"""Gate 5: close/ma20 > 1.10 with BUY → WATCH."""
|
|
holdings = [{"ticker": "034220", "stopBreach": False, "close": 115}]
|
|
df_map = {"034220": {"finalAction": "BUY_TIER1", "ret20d": 0.10, "atr20": 3.0, "close": 115, "ma20": 100}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
decisions = result["decisions"][0]
|
|
assert decisions["final_action"] == "WATCH"
|
|
gates = result["traces"][0]["gates"]
|
|
mrg_gate = [g for g in gates if g["gate"] == "MEAN_REVERSION"][0]
|
|
assert mrg_gate["result"] == "BLOCK"
|
|
|
|
def test_gate5_mean_reversion_pass(self):
|
|
"""Gate 5: close/ma20 <= 1.10 with BUY → PASS."""
|
|
holdings = [{"ticker": "018880", "stopBreach": False, "close": 109}]
|
|
df_map = {"018880": {"finalAction": "BUY_TIER2", "ret20d": 0.10, "atr20": 3.0, "close": 109, "ma20": 100}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
gates = result["traces"][0]["gates"]
|
|
mrg_gate = [g for g in gates if g["gate"] == "MEAN_REVERSION"][0]
|
|
assert mrg_gate["result"] == "PASS"
|
|
|
|
def test_gate5_mean_reversion_skip(self):
|
|
"""Gate 5: insufficient data (no ma20) with BUY → SKIP."""
|
|
holdings = [{"ticker": "003550", "stopBreach": False, "close": 115}]
|
|
df_map = {"003550": {"finalAction": "BUY_TIER1", "ret20d": 0.10, "atr20": 3.0, "close": 115}} # no ma20
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
gates = result["traces"][0]["gates"]
|
|
mrg_gate = [g for g in gates if g["gate"] == "MEAN_REVERSION"][0]
|
|
assert mrg_gate["result"] == "SKIP"
|
|
|
|
def test_gate5_mean_reversion_hold_pass(self):
|
|
"""Gate 5: HOLD action (not BUY) → PASS."""
|
|
holdings = [{"ticker": "010820", "stopBreach": False, "close": 115}]
|
|
df_map = {"010820": {"finalAction": "HOLD", "ret20d": 0.10, "atr20": 3.0, "close": 115, "ma20": 100}}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
gates = result["traces"][0]["gates"]
|
|
mrg_gate = [g for g in gates if g["gate"] == "MEAN_REVERSION"][0]
|
|
assert mrg_gate["result"] == "PASS"
|
|
|
|
def test_multiple_holdings(self):
|
|
"""Test multi-holding routing with different outcomes."""
|
|
holdings = [
|
|
{"ticker": "000660", "stopBreach": True, "close": 95, "stopPrice": 98},
|
|
{"ticker": "005380", "stopBreach": False, "close": 100},
|
|
]
|
|
df_map = {
|
|
"000660": {"finalAction": "HOLD"},
|
|
"005380": {"finalAction": "HOLD", "ret20d": 0.10, "atr20": 3.0},
|
|
}
|
|
h1_ctx = {"intradayLock": False, "kospiRet20d": 0.05}
|
|
|
|
result = run_route_flow(holdings, df_map, h1_ctx)
|
|
|
|
assert len(result["decisions"]) == 2
|
|
assert result["decisions"][0]["final_action"] == "EXIT_100"
|
|
assert result["decisions"][1]["final_action"] == "HOLD"
|