shuim
a264561576
- 스윙 체결가 확인 전 executed_price NULL + fill_sweep, 진입가 resolve(ActiveTrade↔Portfolio), 0체결/부분체결 매수 장부 원복 - DT 고아 감지(근거 한정)·0체결 매도 복원·stale 점검 확대, 시간손절 -0.3% 원복 - 폴백 손절 ATR 캐시, 거래량부족-차단 신호 매수 차단, 현금 5% 유보 재적용, MDD peak/reset 수정 - KIS 서킷브레이커 청산 우회·경보, db_worker 재큐잉 순서, heartbeat·health-watch 사이드카 - 실시간 WS 재구독·41건 우선순위·다건 프레임 분할, 차트 샘플 제거·당일봉 합성, WS 서브프로토콜 인증 - 프론트: 미관리 그룹·확인 모달·차트 실시간 반영·예수금 표시 정정 - 전문가 에이전트 6종 추가(.claude/agents) - 유닛 스위트 2667 passed (소켓 차단 하), tsc OK Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y83Za98XsNw9zuA4FTUytN
158 lines
6.7 KiB
Python
158 lines
6.7 KiB
Python
"""[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
|