"""하드닝 R2 (2026-10-09) — 백테스트 진입 판정 결정론화 회귀 테스트. 결함: kr_backtest_engine 의 진입 판정이 라이브 check_buy_signal 을 호출해, 실행 시각의 라이브 적응형 모드(get_current_adaptive_params)와 운영 DB(strategy_cfg)로 RSI 임계·%B 스케일이 정해졌다. 수정: check_buy_signal_backtest_v2(params, ...) 는 둘 다 읽지 않고 params > settings. """ import ast import os from types import SimpleNamespace import pytest from app.core.config import settings from app.services import mean_reversion_strategy as mrs from app.services.mean_reversion_strategy import MarketRegime BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) class LiveStateRead(BaseException): """백테스트 경로가 라이브 상태를 읽으면 발생 — except Exception 폴백에 삼켜지지 않게 BaseException.""" def _boom(*a, **k): raise LiveStateRead("백테스트 진입 판정이 라이브 상태(adaptive/strategy_cfg)를 읽었다") @pytest.fixture def forbid_live_state(monkeypatch): import app.services.adaptive_strategy as adp monkeypatch.setattr(adp, "get_current_adaptive_params", _boom) monkeypatch.setattr(mrs.strategy_cfg, "get", _boom) def _bull_kwargs(rsi, close, rsi_prev=None): return dict( regime=MarketRegime.BULL, rsi=rsi, close=close, bb_upper=11000.0, bb_middle=10500.0, bb_lower=10000.0, avg_volume_20d=1_000_000, current_volume=1_200_000, prev_volume=900_000, is_reversal=False, rsi_prev=rsi_prev, ) def _bear_kwargs(rsi, volume_mult, reversal=True, rsi_prev=None): return dict( regime=MarketRegime.BEAR, rsi=rsi, close=10100.0, # %B = 0.10 bb_upper=11000.0, bb_middle=10500.0, bb_lower=10000.0, avg_volume_20d=1_000_000, current_volume=int(1_000_000 * volume_mult), prev_volume=900_000, is_reversal=reversal, rsi_prev=rsi_prev, ) # ── 1. 라이브 상태를 전혀 읽지 않는다 ──────────────────────────────────────── def test_backtest_entry_never_reads_adaptive_or_strategy_cfg(forbid_live_state): thr = float(settings.MEAN_REVERSION_RSI_THRESHOLD) bear_thr = float(settings.MEAN_REVERSION_BEAR_RSI_THRESHOLD) # BULL 발화 / BULL 미발화 / BEAR 발화 / BEAR v2 경로 — 어느 것도 라이브 상태를 읽지 않아야 한다 assert mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(thr - 20, 10100.0)) is not None assert mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(thr + 1, 10100.0)) is None vm = float(settings.MEAN_REVERSION_BEAR_VOLUME_MULT) + 0.5 assert mrs.check_buy_signal_backtest_v2(None, **_bear_kwargs(bear_thr - 1, vm)) is not None mrs.check_buy_signal_backtest_v2({"bear_v2_enabled": True}, **_bear_kwargs(bear_thr - 1, vm)) def test_live_check_buy_signal_still_reads_live_state(forbid_live_state): """대조군: 라이브 함수는 여전히 adaptive/strategy_cfg 를 읽는다(§0 우선순위 유지).""" with pytest.raises(LiveStateRead): mrs.check_buy_signal(**_bull_kwargs(float(settings.MEAN_REVERSION_RSI_THRESHOLD) - 20, 10100.0)) # ── 2. 같은 입력 → 라이브 모드와 무관하게 같은 결과 ────────────────────────── @pytest.mark.parametrize("live_mode", [ SimpleNamespace(fallback_used=False, rsi_entry=50.0, bb_pct_b=0.39, bear_rsi_entry=40.0), # Aggressive 류 SimpleNamespace(fallback_used=False, rsi_entry=38.0, bb_pct_b=0.25, bear_rsi_entry=25.0), # Defensive 류 SimpleNamespace(fallback_used=True, rsi_entry=99.0, bb_pct_b=0.99, bear_rsi_entry=99.0), ]) def test_backtest_entry_independent_of_live_mode(monkeypatch, live_mode): import app.services.adaptive_strategy as adp monkeypatch.setattr(adp, "get_current_adaptive_params", lambda: live_mode) monkeypatch.setattr(mrs.strategy_cfg, "get", lambda k, *a, **kw: 1.0) # 운영 DB 의 이상값 흉내 thr = float(settings.MEAN_REVERSION_RSI_THRESHOLD) # thr+2 는 settings 기준 미발화 — 라이브 Aggressive(50) 가 새면 발화한다 assert mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(thr + 2, 10100.0)) is None # 기본 스케일 1.0: RSI 30~임계 구간 %B 상한 0.20 → %B 0.25 는 미발화 (스케일 1.3 이 새면 발화) assert mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(thr - 5, 10250.0)) is None r = mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(thr - 5, 10150.0)) assert r is not None and f"<{thr:.0f}" in r["description"] # ── 3. params 주입이 반영된다 (모드별 백테스트 가능) ──────────────────────── def test_params_rsi_entry_and_bb_pct_b_injected(forbid_live_state): thr = float(settings.MEAN_REVERSION_RSI_THRESHOLD) rsi = thr + 2 assert mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(rsi, 10100.0)) is None r = mrs.check_buy_signal_backtest_v2({"rsi_entry": thr + 4}, **_bull_kwargs(rsi, 10100.0)) assert r is not None and f"<{thr + 4:.0f}" in r["description"] # bb_pct_b=0.39 → 스케일 1.3(상한) → %B 상한 0.26 → %B 0.25 발화 assert mrs.check_buy_signal_backtest_v2({}, **_bull_kwargs(thr - 5, 10250.0)) is None assert mrs.check_buy_signal_backtest_v2({"bb_pct_b": 0.39}, **_bull_kwargs(thr - 5, 10250.0)) is not None # 상한 1.3 유지: bb_pct_b=0.9 여도 0.26 을 넘는 %B 0.30 은 미발화 assert mrs.check_buy_signal_backtest_v2({"bb_pct_b": 0.9}, **_bull_kwargs(thr - 5, 10300.0)) is None def test_params_bear_threshold_and_volume_injected(forbid_live_state): bear_thr = float(settings.MEAN_REVERSION_BEAR_RSI_THRESHOLD) base_vm = float(settings.MEAN_REVERSION_BEAR_VOLUME_MULT) ok_vm = base_vm + 0.5 assert mrs.check_buy_signal_backtest_v2({}, **_bear_kwargs(bear_thr + 1, ok_vm)) is None assert mrs.check_buy_signal_backtest_v2( {"bear_rsi_entry": bear_thr + 3}, **_bear_kwargs(bear_thr + 1, ok_vm)) is not None # 거래량 배수 상향 주입 시 미발화 assert mrs.check_buy_signal_backtest_v2( {"bear_volume_mult": ok_vm + 1.0}, **_bear_kwargs(bear_thr - 1, ok_vm)) is None @pytest.mark.parametrize("bad", ["abc", float("nan"), float("inf"), -5.0, 0.0, 150.0]) def test_invalid_rsi_entry_falls_back_to_settings(forbid_live_state, bad): thr = float(settings.MEAN_REVERSION_RSI_THRESHOLD) # 비정상 값은 settings 로 폴백 — 150 이 그대로 쓰이면 모든 봉이 진입한다 assert mrs.check_buy_signal_backtest_v2({"rsi_entry": bad}, **_bull_kwargs(thr + 2, 10100.0)) is None # ── 4. 라이브 함수와 동일 로직 (출처만 다르다) ─────────────────────────────── def test_live_and_backtest_parity_for_same_effective_params(monkeypatch): import app.services.adaptive_strategy as adp mode = SimpleNamespace(fallback_used=False, rsi_entry=50.0, bb_pct_b=0.36, bear_rsi_entry=30.0) monkeypatch.setattr(adp, "get_current_adaptive_params", lambda: mode) monkeypatch.setattr(mrs.strategy_cfg, "get", lambda k, *a, **kw: None) params = {"rsi_entry": 50.0, "bb_pct_b": 0.36, "bear_rsi_entry": 30.0} cases = [_bull_kwargs(r, c, rp) for r in (22.0, 28.0, 45.0, 49.0, 51.0) for c in (10050.0, 10250.0, 10400.0) for rp in (None, 10.0, 60.0)] cases += [_bear_kwargs(r, vm, rev) for r in (20.0, 29.0, 31.0) for vm in (1.0, 2.5) for rev in (True, False)] for kw in cases: assert mrs.check_buy_signal(**kw) == mrs.check_buy_signal_backtest_v2(params, **kw), kw # ── 5. 엔진 배선: kr_backtest_engine 은 라이브 check_buy_signal 을 호출하지 않는다 ── def test_kr_backtest_engine_uses_backtest_entry(): p = os.path.join(BACKEND_DIR, "app", "services", "kr_backtest_engine.py") with open(p, encoding="utf-8") as f: tree = ast.parse(f.read()) called = set() for node in ast.walk(tree): if isinstance(node, ast.Call): fn = node.func called.add(fn.id if isinstance(fn, ast.Name) else getattr(fn, "attr", None)) assert "check_buy_signal" not in called assert "check_buy_signal_backtest_v2" in called assert "get_current_adaptive_params" not in called def test_kr_backtest_on_bar_passes_params(monkeypatch): import app.services.kr_backtest_engine as eng captured = [] def spy(params, **kw): captured.append(params) return None monkeypatch.setattr(eng, "check_buy_signal_backtest_v2", spy) monkeypatch.setattr(eng, "calculate_rsi", lambda c, p=14: 30.0) monkeypatch.setattr(eng, "calculate_bollinger", lambda c, p=20: (110.0, 100.0, 90.0)) from datetime import date, timedelta s = eng.MeanReversionBacktest() p = s.default_params() p["rsi_entry"] = 41.0 n = 80 d0 = date(2025, 1, 2) bars = [eng.BarData(trade_date=d0 + timedelta(days=i), open_price=10000 + i, high_price=10100 + i, low_price=9900 + i, close_price=10000 + i, volume=1000, stock_code="A") for i in range(n)] s.on_bar(n - 1, {"A": bars}, {}, 1e6, 1e6, p) assert captured and captured[0] is p and captured[0]["rsi_entry"] == 41.0