auto-stock/backend/tests/test_mr_backtest_entry_deterministic_20261009.py

"""하드닝 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