diff --git a/docs/ROADMAP_WBS.md b/docs/ROADMAP_WBS.md index 1b79184..d00ea9c 100644 --- a/docs/ROADMAP_WBS.md +++ b/docs/ROADMAP_WBS.md @@ -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 완료, 이평선 이격도 및 거래량 미확인 돌파 동등성 실증 완료) ``` --- diff --git a/tests/parity/test_late_chase_risk_parity.py b/tests/parity/test_late_chase_risk_parity.py new file mode 100644 index 0000000..2f4afcf --- /dev/null +++ b/tests/parity/test_late_chase_risk_parity.py @@ -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()