120 lines
5.3 KiB
Python
120 lines
5.3 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))
|
|
|
|
# Test target functions directly or simulate the exact formula logic matching tools/build_relative_underperformance_alert_v1.py
|
|
def calculate_absolute_risk_stop(close: float, avg_cost: float, atr20: float) -> tuple[float, str]:
|
|
if not atr20 or close <= 0:
|
|
return 0.0, "INSUFFICIENT_DATA"
|
|
|
|
# ATR20_Pct >= 8% -> 2.0x ATR, else 1.5x ATR
|
|
atr_pct = atr20 / close * 100.0
|
|
atr_mul = 2.0 if atr_pct >= 8.0 else 1.5
|
|
recommended_stop = max(avg_cost * 0.92, avg_cost - atr20 * atr_mul)
|
|
recommended_stop = round(recommended_stop)
|
|
|
|
# Assuming adequacy status check logic from tool
|
|
return recommended_stop, "PASS"
|
|
|
|
def calculate_relative_underperf_signal(
|
|
close: float,
|
|
ret20d: float,
|
|
atr20: float,
|
|
kospi_ret20d: float,
|
|
profit_pct: float,
|
|
hold_days: int
|
|
) -> tuple[str, bool]:
|
|
if not atr20 or close <= 0 or ret20d is None or kospi_ret20d is None:
|
|
return "INSUFFICIENT_DATA", False
|
|
|
|
# Beta estimation
|
|
beta = 1.0
|
|
if abs(kospi_ret20d) >= 0.5:
|
|
beta = min(3.0, max(0.3, ret20d / kospi_ret20d))
|
|
|
|
excess_ret = ret20d - beta * kospi_ret20d
|
|
sigma_proxy = (atr20 / close * 100.0) * math.sqrt(20)
|
|
threshold = -2.0 * sigma_proxy
|
|
|
|
rel_trigger = excess_ret < threshold
|
|
abs_floor = profit_pct is not None and profit_pct < -20.0
|
|
time_stop = hold_days >= 60 and excess_ret < 0
|
|
|
|
signal_type = "ABS_FLOOR" if abs_floor else ("REL_EXCESS" if rel_trigger else ("TIME_STOP" if time_stop else "PASS"))
|
|
signal = bool(signal_type != "PASS" and signal_type != "INSUFFICIENT_DATA")
|
|
|
|
return signal_type, signal
|
|
|
|
|
|
class TestStopLossPolicyParity(unittest.TestCase):
|
|
|
|
def test_absolute_risk_stop_logic_parity(self):
|
|
# Scenario 1: Low volatility stock (ATR Pct < 8%), average cost = 10000, atr = 500 (5%)
|
|
# Expected multiplier = 1.5. recommended_stop = max(9200, 10000 - 750) = 9250
|
|
stop_price, status = calculate_absolute_risk_stop(close=10000, avg_cost=10000, atr20=500)
|
|
self.assertEqual(stop_price, 9250)
|
|
self.assertEqual(status, "PASS")
|
|
|
|
# Scenario 2: High volatility stock (ATR Pct >= 8%), close = 10000, average cost = 10000, atr = 900 (9%)
|
|
# Expected multiplier = 2.0. recommended_stop = max(9200, 10000 - 1800) = 9200 (max bound matches 0.92)
|
|
stop_price_high, status_high = calculate_absolute_risk_stop(close=10000, avg_cost=10000, atr20=900)
|
|
self.assertEqual(stop_price_high, 9200)
|
|
|
|
def test_relative_underperformance_trigger_parity(self):
|
|
# Scenario 1: No trigger
|
|
signal_type, signal = calculate_relative_underperf_signal(
|
|
close=10000, ret20d=2.0, atr20=200, kospi_ret20d=1.0, profit_pct=-2.0, hold_days=10
|
|
)
|
|
self.assertEqual(signal_type, "PASS")
|
|
self.assertFalse(signal)
|
|
|
|
# Scenario 2: Absolute floor trigger (profit_pct < -20%)
|
|
signal_type_floor, signal_floor = calculate_relative_underperf_signal(
|
|
close=10000, ret20d=-5.0, atr20=200, kospi_ret20d=1.0, profit_pct=-22.0, hold_days=10
|
|
)
|
|
self.assertEqual(signal_type_floor, "ABS_FLOOR")
|
|
self.assertTrue(signal_floor)
|
|
|
|
# Scenario 3: Relative excess trigger (excess_ret < threshold)
|
|
# close=10000, atr20=500 -> sigma_proxy = 5.0 * sqrt(20) = 22.36. threshold = -44.72
|
|
# kospi_ret20d = 10.0 -> beta=0.3. excess_ret = -70.0 - 3.0 = -73.0 < -44.72 (triggered)
|
|
signal_type_rel, signal_rel = calculate_relative_underperf_signal(
|
|
close=10000, ret20d=-70.0, atr20=500, kospi_ret20d=10.0, profit_pct=-10.0, hold_days=10
|
|
)
|
|
def test_stop_loss_gate_decision_routing_f11_parity(self):
|
|
from src.quant_engine.exit_decisions import compute_stop_action_ladder
|
|
|
|
# Test case: holding.stopBreach is True -> EXIT_100 (due to timingAction or rw_partial >= 4, here we simulate the action routing)
|
|
# In exit_decisions.py, if timing_action == "STOP_OR_TIME_EXIT_READY" or rw_partial >= 4, it routes to EXIT_100
|
|
res1 = compute_stop_action_ladder({"timingAction": "STOP_OR_TIME_EXIT_READY"})
|
|
self.assertEqual(res1["action"], "EXIT_100")
|
|
self.assertEqual(res1["reason"], "STOP_OR_TIME_EXIT_READY")
|
|
|
|
def test_late_chase_gate_f15_parity(self):
|
|
from src.quant_engine.exit_decisions import compute_final_decision
|
|
|
|
# F15 check: breakout_quality_gate === 'BLOCKED_LATE_CHASE' or late_chase_risk_score >= 70
|
|
# In compute_final_decision: allowed_action is checked. Let's make sure it handles decisions properly.
|
|
# If allowed_action = "BUY_STAGE1_READY" but ac_gate is BLOCK, it downgrades.
|
|
# Let's verify compute_final_decision handles timing_action = "NO_BUY_OVERHEATED" (which maps to ac_gate=BLOCK or entry_gate=BLOCK in compute_timing_decision)
|
|
res = compute_final_decision({
|
|
"sellAction": "HOLD",
|
|
"allowedAction": "",
|
|
"timingAction": "NO_BUY_OVERHEATED",
|
|
"dartRisk": False
|
|
})
|
|
self.assertEqual(res["final_action"], "NO_BUY_OVERHEATED")
|
|
self.assertEqual(res["action_priority"], 50)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|