"""[R3 2026-10-09] 스윙 ATR 트레일링 배수 단일 출처 회귀 테스트. 결함: 평균회귀 check_exit_signal·모멘텀 _check_momentum_exit 가 리터럴 1.5 를 써 config / strategy_cfg(DB) / 적응형 어느 값도 소비되지 않았다. 수정: resolve_swing_trailing_atr_mult_v2 (strategy_cfg > config, 적응형은 SWING_ADAPTIVE_TRAILING_ENABLED 일 때만 — 백테스트 전 기본 OFF). """ import os import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from types import SimpleNamespace from unittest.mock import MagicMock, patch import pytest import app.services.mean_reversion_strategy as mr from app.core.config import settings from app.services.strategy_config_service import DEFAULT_PARAMS class _AdaptiveCalled(Exception): """플래그 OFF 인데 적응형을 조회하면 실패시키는 예외 발생형 목.""" def _cfg(values): m = MagicMock() m.get.side_effect = lambda key, default=None: values.get(key, default) return m _FALLBACK_ADP = SimpleNamespace(fallback_used=True) @pytest.fixture def no_adaptive(): """적응형 오버레이 없음(fallback_used=True) — 손절/RSI 경로 결정론.""" with patch("app.services.adaptive_strategy.get_current_adaptive_params", return_value=_FALLBACK_ADP): yield def _exit(close, cfg_values, trailing_high=11000.0, atr=200.0): with patch.object(mr, "strategy_cfg", _cfg(cfg_values)): return mr.check_exit_signal( regime=mr.MarketRegime.BULL, rsi=50.0, close=close, bb_upper=20000.0, bb_middle=9000.0, bb_lower=8000.0, entry_price=10000.0, atr=atr, lowest_since_entry=9800.0, remaining_pct=0.50, trailing_high=trailing_high, ) def test_default_uses_config_value(no_adaptive): with patch.object(mr, "strategy_cfg", _cfg({})): mult, source = mr._resolve_swing_trailing_atr_mult_impl() assert mult == float(settings.ATR_TRAILING_MULTIPLIER) assert source == "config" def test_strategy_cfg_overrides_config_and_reaches_exit(no_adaptive): cfg_mult = 2.0 assert cfg_mult != float(settings.ATR_TRAILING_MULTIPLIER) atr, high = 200.0, 11000.0 # config 배수 트레일과 DB 배수 트레일 사이의 종가 → DB 값이면 미청산이어야 한다 close = high - ((float(settings.ATR_TRAILING_MULTIPLIER) + cfg_mult) / 2) * atr if cfg_mult > float(settings.ATR_TRAILING_MULTIPLIER): assert _exit(close, {"ATR_TRAILING_MULTIPLIER": cfg_mult}) is None sig = _exit(close, {}) assert sig is not None and sig.action == "SELL_ALL" and "트레일링" in sig.reason assert f"{float(settings.ATR_TRAILING_MULTIPLIER):g}×ATR" in sig.reason def test_out_of_range_db_value_falls_back_to_config(no_adaptive): _hi = float(DEFAULT_PARAMS["ATR_TRAILING_MULTIPLIER"][6]) for bad in (_hi + 1, 0, -1.5, "abc", float("nan")): with patch.object(mr, "strategy_cfg", _cfg({"ATR_TRAILING_MULTIPLIER": bad})): mult, source = mr._resolve_swing_trailing_atr_mult_impl() assert source == "config", bad assert mult == float(settings.ATR_TRAILING_MULTIPLIER) def test_adaptive_not_consulted_when_flag_off(): with patch("app.services.adaptive_strategy.get_current_adaptive_params", side_effect=_AdaptiveCalled("적응형 조회 금지")), \ patch.object(mr, "_adaptive_trailing_enabled_v2", return_value=False), \ patch.object(mr, "strategy_cfg", _cfg({})): mult, source = mr._resolve_swing_trailing_atr_mult_impl() status = mr.get_swing_trailing_mult_status_v2() assert source == "config" assert mult == float(settings.ATR_TRAILING_MULTIPLIER) assert status["adaptive_trailing_applied"] is False assert status["adaptive_trailing_enabled"] is False def test_flag_default_is_off(): # 백테스트 전에는 적응형 배수(2.0/1.6/1.3/1.0)가 실전에 적용되면 안 된다 assert getattr(settings, "SWING_ADAPTIVE_TRAILING_ENABLED", False) is not True assert mr._adaptive_trailing_enabled_v2() is False def test_adaptive_applied_when_flag_on(): adp = SimpleNamespace(fallback_used=False, atr_trailing_multiplier=1.0) with patch("app.services.adaptive_strategy.get_current_adaptive_params", return_value=adp), \ patch.object(mr, "_adaptive_trailing_enabled_v2", return_value=True), \ patch.object(mr, "strategy_cfg", _cfg({"ATR_TRAILING_MULTIPLIER": 2.0})): mult, source = mr._resolve_swing_trailing_atr_mult_impl() assert (mult, source) == (1.0, "adaptive") def test_adaptive_failure_falls_back_when_flag_on(): with patch("app.services.adaptive_strategy.get_current_adaptive_params", side_effect=RuntimeError("adaptive down")), \ patch.object(mr, "_adaptive_trailing_enabled_v2", return_value=True), \ patch.object(mr, "strategy_cfg", _cfg({"ATR_TRAILING_MULTIPLIER": 2.0})): mult, source = mr._resolve_swing_trailing_atr_mult_impl() assert (mult, source) == (2.0, "strategy_cfg") def test_invalid_config_uses_meta_default(monkeypatch): monkeypatch.setattr(settings, "ATR_TRAILING_MULTIPLIER", 0.0) with patch.object(mr, "strategy_cfg", _cfg({})): mult, source = mr._resolve_swing_trailing_atr_mult_impl() assert source == "meta_default" assert mult == float(DEFAULT_PARAMS["ATR_TRAILING_MULTIPLIER"][4]) > 0 def test_momentum_trailing_uses_resolver(no_adaptive): """공통 종단(resolve_swing_trailing_atr_mult_v2)을 예외 발생형으로 치환 → 모멘텀 경로가 반드시 호출.""" from app.services.swing import exit_service class _ResolverHit(Exception): pass with patch.object(mr, "strategy_cfg", _cfg({})), \ patch.object(mr, "resolve_swing_trailing_atr_mult_v2", side_effect=_ResolverHit()): with pytest.raises(_ResolverHit): exit_service._check_momentum_exit( current_price=10900.0, entry_price=10000.0, atr=200.0, remaining_pct=0.40, trailing_high=12000.0, ) def test_momentum_trailing_honors_db_value(no_adaptive): from app.services.swing import exit_service atr, high = 200.0, 12000.0 base = float(settings.ATR_TRAILING_MULTIPLIER) cfg_mult = 2.0 close = high - ((base + cfg_mult) / 2) * atr # 두 트레일 사이 kwargs = dict(current_price=close, entry_price=10000.0, atr=atr, remaining_pct=0.40, trailing_high=high) with patch.object(mr, "strategy_cfg", _cfg({})): sig = exit_service._check_momentum_exit(**kwargs) assert sig is not None and "트레일링" in sig.reason with patch.object(mr, "strategy_cfg", _cfg({"ATR_TRAILING_MULTIPLIER": cfg_mult})): sig2 = exit_service._check_momentum_exit(**kwargs) assert sig2 is None or "트레일링" not in sig2.reason from _swing_redis_guard import _swing_redis_guard # noqa: E402,F401 (R9: 단독 실행 시 Redis 실접속 방지)