auto-stock/backend/tests/test_swing_trailing_mult_20261009.py

"""[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 실접속 방지)