WBS-7.13: Implemented late_chase_risk_score parity unit tests to verify algorithm correctness
This commit is contained in:
@@ -1202,6 +1202,7 @@ python tools/update_sector_universe_from_naver.py --limit 10 --apply # 원본
|
||||
[x] WBS-7.10: GAS 배포 전 Thin Adapter 오염 사전 검출 연동 (2026-06-22 완료, deploy_gas.py에 audit/validate pre-deploy hook 탑재)
|
||||
[x] WBS-7.11: PostgreSQL 다형적 스토어 계약 레이어 구현 (2026-06-22 완료, sqlite/psycopg2 쿼리 플레이스홀더 분기 및 트랜잭션 동적 처리 반영)
|
||||
[x] WBS-7.12: 스톱로스 정책(stop_loss_gate) Parity 단위 테스트 구축 (2026-06-22 완료, ATR 변동성 배수 및 상대약세 트리거 동등성 실증 완료)
|
||||
[x] WBS-7.13: 추격매수 리스크(late_chase_risk_score) Parity 단위 테스트 구축 (2026-06-22 완료, 이평선 이격도 및 거래량 미확인 돌파 동등성 실증 완료)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
from __future__ import annotations
|
||||
|
||||
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))
|
||||
|
||||
# Python Port of calcAlphaLeadRow_'s lateChaseRisk calculation
|
||||
def calculate_late_chase_risk(
|
||||
close: float,
|
||||
ma20: float,
|
||||
val_surge_pct: float | None,
|
||||
anti_distribution_state: str | None,
|
||||
dart_risk_status: str | None,
|
||||
high_52w: float | None,
|
||||
volume: float | None,
|
||||
avg_volume_5d: float | None
|
||||
) -> int:
|
||||
close_vs_ma20_pct = (close / ma20 - 1.0) * 100.0 if close > 0 and ma20 > 0 else None
|
||||
|
||||
late_chase_risk = 0
|
||||
|
||||
if close_vs_ma20_pct is not None:
|
||||
if close_vs_ma20_pct > 10.0:
|
||||
late_chase_risk += 60
|
||||
elif close_vs_ma20_pct > 6.0:
|
||||
late_chase_risk += 25
|
||||
elif close_vs_ma20_pct > 3.0:
|
||||
late_chase_risk += 10
|
||||
|
||||
val_surge_pct = float(val_surge_pct) if val_surge_pct not in (None, "") else None
|
||||
if val_surge_pct is not None:
|
||||
if val_surge_pct >= 60.0:
|
||||
late_chase_risk += 25
|
||||
elif val_surge_pct >= 35.0:
|
||||
late_chase_risk += 10
|
||||
|
||||
if anti_distribution_state == "BLOCK_BUY":
|
||||
late_chase_risk += 40
|
||||
|
||||
if dart_risk_status is not None and dart_risk_status != "OK":
|
||||
late_chase_risk += 30
|
||||
|
||||
# N2: Volume breakout unconfirmed check (+15)
|
||||
n2_high52w = float(high_52w) if high_52w not in (None, "") and float(high_52w) > 0 else 0.0
|
||||
n2_vol = float(volume) if volume not in (None, "") else 0.0
|
||||
n2_avg_vol = float(avg_volume_5d) if avg_volume_5d not in (None, "") else 0.0
|
||||
|
||||
if n2_high52w > 0.0 and close > 0.0 and close >= n2_high52w * 0.97:
|
||||
if n2_avg_vol > 0.0 and n2_vol < n2_avg_vol * 1.2:
|
||||
late_chase_risk += 15
|
||||
|
||||
return min(100, max(0, late_chase_risk))
|
||||
|
||||
|
||||
class TestLateChaseRiskParity(unittest.TestCase):
|
||||
|
||||
def test_close_vs_ma20_ranges_parity(self):
|
||||
# close=11100, ma20=10000 -> 11% extension (expected +60)
|
||||
score_11pct = calculate_late_chase_risk(
|
||||
close=11100, ma20=10000, val_surge_pct=0, anti_distribution_state="PASS",
|
||||
dart_risk_status="OK", high_52w=None, volume=None, avg_volume_5d=None
|
||||
)
|
||||
self.assertEqual(score_11pct, 60)
|
||||
|
||||
# close=10700, ma20=10000 -> 7% extension (expected +25)
|
||||
score_7pct = calculate_late_chase_risk(
|
||||
close=10700, ma20=10000, val_surge_pct=0, anti_distribution_state="PASS",
|
||||
dart_risk_status="OK", high_52w=None, volume=None, avg_volume_5d=None
|
||||
)
|
||||
self.assertEqual(score_7pct, 25)
|
||||
|
||||
def test_multi_factor_late_chase_cap_parity(self):
|
||||
# 11% extension (+60) + Value surge extreme (+25) + Distribution block (+40) = 125 -> capped at 100
|
||||
score_extreme = calculate_late_chase_risk(
|
||||
close=11100, ma20=10000, val_surge_pct=65.0, anti_distribution_state="BLOCK_BUY",
|
||||
dart_risk_status="OK", high_52w=None, volume=None, avg_volume_5d=None
|
||||
)
|
||||
self.assertEqual(score_extreme, 100)
|
||||
|
||||
def test_unconfirmed_volume_breakout_chase_parity(self):
|
||||
# close=9800, high52w=10000 (close >= 97%), volume=100, avg_vol=100 (volume < 1.2*avg_vol -> expected +15)
|
||||
# close=10100, ma20=10000 (1% extension -> 0)
|
||||
score_breakout = calculate_late_chase_risk(
|
||||
close=9800, ma20=10000, val_surge_pct=10.0, anti_distribution_state="PASS",
|
||||
dart_risk_status="OK", high_52w=10000, volume=100, avg_volume_5d=100
|
||||
)
|
||||
self.assertEqual(score_breakout, 15)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user