Files
auto-stock/backend/tests/test_mean_reversion.py
shuim a264561576 fix: 하드닝 루프 8회차 — 체결 보정·고아 감지·청산 경보·실시간/차트·리스크 가드 일괄 수정
- 스윙 체결가 확인 전 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
2026-10-10 12:17:49 +09:00

1181 lines
47 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
평균회귀 전략 (mean_reversion_strategy.py) 단위 테스트
테스트 시나리오:
1. ATR (Average True Range) 계산
2. ADX (Average Directional Index) 계산
3. 시장 국면 감지 (detect_market_regime)
4. RSI 계산 (calculate_rsi)
5. 볼린저 밴드 계산 (calculate_bollinger)
6. 되돌림 캔들 감지 (is_reversal_candle)
7. 매수 신호 (check_buy_signal) — Bull/Bear 시장별
8. 청산 신호 (check_exit_signal) — 손절/1차 익절/2차 익절/트레일링
9. 포지션 사이징 (calculate_position_size)
10. 포지션 사이징 v2 (calculate_position_size_v2)
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import pytest
from unittest.mock import patch, MagicMock
# ---------------------------------------------------------------------------
# strategy_cfg 모킹 — 모듈 임포트 전에 패치
# ---------------------------------------------------------------------------
_mock_strategy_cfg = MagicMock()
_mock_strategy_cfg.get.side_effect = lambda key, default=None: {
"MEAN_REVERSION_RSI_THRESHOLD": 30.0,
"ATR_STOP_MULTIPLIER": 2.5,
"STOP_LOSS_PERCENT": -6.0,
"MEAN_REVERSION_MULTIPLIER_STRONG": 1.50,
"MEAN_REVERSION_MULTIPLIER_NORMAL": 1.25,
"MEAN_REVERSION_MULTIPLIER_WEAK": 1.00,
}.get(key, default)
# ---------------------------------------------------------------------------
# [2026-09-27] import 순서 의존 제거
#
# 아래 patch.dict("sys.modules", ...) 는 **이 파일이 먼저 import될 때만** 먹는다.
# 전체 스위트에서는 알파벳 앞선 테스트들이 이미 app.services.mean_reversion_strategy 를
# 로드해 두므로, 그 모듈은 **실제** strategy_cfg 를 전역에 바인딩한 상태가 된다.
# 그러면 목이 새어 strategy_cfg.get() → None → settings.MEAN_REVERSION_RSI_THRESHOLD(46)
# 이 쓰이고, RSI 임계 30 을 전제한 케이스(test_bull_rsi_too_high: rsi=45)가 깨졌다.
# (단독 실행 74 passed / 전체 스위트 1 failed 로 나타난 순서 의존 버그)
#
# → 모듈 전역 참조를 **테스트 시점에** 덮는 autouse 픽스처로 바꿔, import 순서와 무관하게
# 항상 같은 목이 쓰이도록 한다. 기대값은 하나도 바꾸지 않았다.
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _bind_mock_strategy_cfg():
import app.services.mean_reversion_strategy as _m
with patch.object(_m, "strategy_cfg", _mock_strategy_cfg):
yield
@pytest.fixture(autouse=True)
def _reset_regime_hysteresis_state():
"""[2026-10-02 3회차] detect_market_regime 의 ADX 히스테리시스 전역(_previous_regime)을
테스트마다 초기화한다. 앞선 테스트(또는 다른 파일)가 SIDEWAYS 를 남기면 국면 테스트
결과가 실행 순서에 따라 달라진다."""
import app.services.mean_reversion_strategy as _m
with patch.object(_m, "_previous_regime", None):
yield
with patch.dict("sys.modules", {
"app.core.config": MagicMock(),
"app.services.strategy_config_service": MagicMock(strategy_cfg=_mock_strategy_cfg),
}):
from app.services.mean_reversion_strategy import (
calculate_atr,
calculate_adx,
detect_market_regime,
calculate_rsi,
calculate_bollinger,
is_reversal_candle,
check_buy_signal,
check_exit_signal,
calculate_position_size,
calculate_position_size_v2,
MarketRegime,
ExitSignal,
)
# ---------------------------------------------------------------------------
# 헬퍼 함수
# ---------------------------------------------------------------------------
def _rising(start, count, step=100):
"""상승 가격 리스트"""
return [start + i * step for i in range(count)]
def _falling(start, count, step=100):
"""하락 가격 리스트"""
return [start - i * step for i in range(count)]
def _flat(price, count):
"""일정 가격 리스트"""
return [price] * count
# ═══════════════════════════════════════════════════════════════════════════
# 1. calculate_atr
# ═══════════════════════════════════════════════════════════════════════════
class TestCalculateATR:
"""ATR(Average True Range) 계산 테스트"""
def test_basic_calculation(self):
"""기본 ATR 계산 검증"""
n = 20
highs = [100 + i for i in range(n)]
lows = [90 + i for i in range(n)]
closes = [95 + i for i in range(n)]
result = calculate_atr(highs, lows, closes, period=14)
assert result is not None
assert result > 0
def test_insufficient_data(self):
"""데이터 부족 시 None 반환"""
result = calculate_atr([100, 110], [90, 95], [95, 105], period=14)
assert result is None
def test_constant_prices_atr_near_zero(self):
"""일정한 가격 → ATR ≈ H-L (매 봉 동일)"""
n = 20
highs = [101] * n
lows = [99] * n
closes = [100] * n
result = calculate_atr(highs, lows, closes, period=14)
assert result is not None
assert result == pytest.approx(2.0, abs=0.01) # H-L = 2
def test_period_parameter(self):
"""period 변경 시 정상 동작"""
n = 30
highs = [100 + i for i in range(n)]
lows = [90 + i for i in range(n)]
closes = [95 + i for i in range(n)]
result_5 = calculate_atr(highs, lows, closes, period=5)
result_14 = calculate_atr(highs, lows, closes, period=14)
assert result_5 is not None
assert result_14 is not None
def test_gap_up_increases_atr(self):
"""갭 상승 시 TR이 커져 ATR 상승"""
n = 20
highs = _flat(110, n)
lows = _flat(100, n)
closes = _flat(105, n)
# 갭 상승 삽입
highs[-1] = 130
lows[-1] = 120
closes[-1] = 125
result = calculate_atr(highs, lows, closes, period=14)
assert result is not None
assert result > 10 # 기본 H-L=10 보다 큰 ATR
def test_minimum_data_for_period(self):
"""period+1개 데이터 → 계산 가능"""
period = 5
n = period + 1
highs = [100 + i * 2 for i in range(n)]
lows = [90 + i * 2 for i in range(n)]
closes = [95 + i * 2 for i in range(n)]
result = calculate_atr(highs, lows, closes, period=period)
assert result is not None
# ═══════════════════════════════════════════════════════════════════════════
# 2. calculate_adx
# ═══════════════════════════════════════════════════════════════════════════
class TestCalculateADX:
"""ADX(Average Directional Index) 계산 테스트"""
def test_basic_calculation(self):
"""기본 ADX 계산 — 충분한 데이터, 결과는 숫자"""
# 실제 주식 시장처럼 변동하는 가격 데이터 사용
import math
n = 40
highs = [1000 + 50 * math.sin(i * 0.5) + i * 2 + 30 for i in range(n)]
lows = [1000 + 50 * math.sin(i * 0.5) + i * 2 - 30 for i in range(n)]
closes = [1000 + 50 * math.sin(i * 0.5) + i * 2 for i in range(n)]
result = calculate_adx(highs, lows, closes, period=14)
assert result is not None
assert isinstance(result, float)
def test_insufficient_data(self):
"""데이터 부족 시 None 반환 — 최소 2*period+1 필요"""
result = calculate_adx([100] * 10, [90] * 10, [95] * 10, period=14)
assert result is None
def test_strong_trend_high_adx(self):
"""강한 상승 추세 → ADX 높음"""
n = 50
highs = [100 + i * 10 for i in range(n)]
lows = [90 + i * 10 for i in range(n)]
closes = [95 + i * 10 for i in range(n)]
result = calculate_adx(highs, lows, closes, period=14)
assert result is not None
assert result > 20 # 강한 추세
def test_sideways_market_low_adx(self):
"""횡보 시장 → ADX 낮음"""
n = 40
# 좁은 범위에서 오르내림
highs = [101 + (i % 3) for i in range(n)]
lows = [99 - (i % 3) for i in range(n)]
closes = [100 + (i % 2) - 0.5 for i in range(n)]
result = calculate_adx(highs, lows, closes, period=14)
assert result is not None
# 횡보 시 ADX는 보통 낮지만 정확한 수치는 데이터에 의존
def test_minimum_required_data(self):
"""최소 데이터 (2*period+1)에서 정상 작동"""
period = 5
n = period * 2 + 1
highs = [100 + i for i in range(n)]
lows = [90 + i for i in range(n)]
closes = [95 + i for i in range(n)]
result = calculate_adx(highs, lows, closes, period=period)
assert result is not None
def test_result_is_rounded(self):
"""결과값이 소수점 2자리로 반올림"""
n = 40
highs = [100 + i * 3 for i in range(n)]
lows = [90 + i * 3 for i in range(n)]
closes = [95 + i * 3 for i in range(n)]
result = calculate_adx(highs, lows, closes, period=14)
assert result is not None
# 소수점 2자리 이내 확인
assert result == round(result, 2)
# ═══════════════════════════════════════════════════════════════════════════
# 3. detect_market_regime
# ═══════════════════════════════════════════════════════════════════════════
class TestDetectMarketRegime:
"""시장 국면 감지 테스트"""
def test_insufficient_data_returns_bear(self):
"""데이터 부족 → 보수적으로 BEAR 반환"""
result = detect_market_regime([2500] * 100)
assert result == MarketRegime.BEAR
def test_bull_market(self):
"""상승장: 지수 > MA200 AND 기울기 양(+)"""
prices = _rising(2000, 210, 5)
result = detect_market_regime(prices)
assert result == MarketRegime.BULL
def test_bear_market(self):
"""하락장: 지수 < MA200"""
prices = _falling(3000, 210, 5)
result = detect_market_regime(prices)
assert result == MarketRegime.BEAR
def test_flat_market_bear(self):
"""횡보 → 기울기 ≈ 0 → SIDEWAYS (CLAUDE.md 6.1: 기울기 ≈ 0 → 횡보장)"""
prices = _flat(2500, 210)
result = detect_market_regime(prices)
# [2026-03-15 수정] CLAUDE.md 6.1: MA200 기울기 0 → SIDEWAYS (slope_ratio < 0.001)
assert result == MarketRegime.SIDEWAYS
def test_price_above_ma_but_declining(self):
"""가격 > MA200이지만 기울기 하락 → BEAR"""
# 초반 상승 → 후반 하락 (MA200 기울기가 음수가 되도록)
prices = _rising(2000, 150, 10) + _falling(3500, 60, 15)
result = detect_market_regime(prices)
# 최근 가격은 하락 중이므로 MA200 기울기 음수일 가능성
assert result in (MarketRegime.BULL, MarketRegime.BEAR)
def test_exactly_205_data_points(self):
"""최소 필요 데이터 (205개)에서 정상 작동"""
prices = _rising(2000, 205, 3)
result = detect_market_regime(prices)
assert result == MarketRegime.BULL
# ═══════════════════════════════════════════════════════════════════════════
# 4. calculate_rsi
# ═══════════════════════════════════════════════════════════════════════════
class TestCalculateRSI:
"""RSI(Relative Strength Index) 계산 테스트"""
def test_insufficient_data(self):
"""데이터 부족 → None"""
result = calculate_rsi([100, 101, 102], period=14)
assert result is None
def test_all_gains_rsi_100(self):
"""모든 봉 상승 → RSI = 100"""
prices = _rising(10000, 20, 100)
result = calculate_rsi(prices, period=14)
assert result == 100.0
def test_all_losses_rsi_0(self):
"""모든 봉 하락 → RSI = 0"""
prices = _falling(20000, 20, 100)
result = calculate_rsi(prices, period=14)
assert result == pytest.approx(0.0, abs=0.01)
def test_balanced_rsi_near_50(self):
"""상승/하락 균등 → RSI ≈ 50"""
prices = []
base = 10000
for i in range(20):
if i % 2 == 0:
base += 100
else:
base -= 100
prices.append(base)
result = calculate_rsi(prices, period=14)
assert result is not None
assert 30 < result < 70 # 대략 50 근처
def test_period_parameter(self):
"""period 변경 시 다른 결과"""
prices = _rising(10000, 30, 50) + _falling(11500, 10, 100)
rsi_7 = calculate_rsi(prices, period=7)
rsi_14 = calculate_rsi(prices, period=14)
assert rsi_7 is not None
assert rsi_14 is not None
# 다른 period → 다른 값 (동일할 수도 있지만 일반적으로 다름)
def test_minimum_data(self):
"""period+1개 데이터에서 정상 작동"""
prices = _rising(10000, 15, 50) # 15 = 14 + 1
result = calculate_rsi(prices, period=14)
assert result is not None
# ═══════════════════════════════════════════════════════════════════════════
# 5. calculate_bollinger
# ═══════════════════════════════════════════════════════════════════════════
class TestCalculateBollinger:
"""볼린저 밴드 계산 테스트"""
def test_insufficient_data(self):
"""데이터 부족 → None"""
result = calculate_bollinger([100, 101], period=20)
assert result is None
def test_basic_calculation(self):
"""기본 볼린저 밴드 계산 — (upper, middle, lower) 반환"""
prices = _flat(10000, 20)
result = calculate_bollinger(prices, period=20)
assert result is not None
upper, middle, lower = result
assert middle == pytest.approx(10000.0)
# 표준편차 0 → upper == middle == lower
assert upper == pytest.approx(10000.0)
assert lower == pytest.approx(10000.0)
def test_band_width_with_volatility(self):
"""변동성 있으면 밴드 폭이 넓어짐"""
prices = [10000 + (i % 2) * 500 - 250 for i in range(20)]
result = calculate_bollinger(prices, period=20)
assert result is not None
upper, middle, lower = result
assert upper > middle > lower
band_width = upper - lower
assert band_width > 0
def test_upper_above_lower(self):
"""항상 upper >= lower"""
prices = _rising(9000, 25, 100)
result = calculate_bollinger(prices, period=20)
assert result is not None
upper, middle, lower = result
assert upper >= lower
def test_symmetric_bands(self):
"""밴드는 중앙선 기준 대칭"""
prices = [10000, 10100, 9900, 10050, 9950] * 4
result = calculate_bollinger(prices, period=20)
assert result is not None
upper, middle, lower = result
assert upper - middle == pytest.approx(middle - lower, abs=0.01)
def test_period_parameter(self):
"""다른 period에서 정상 작동"""
prices = _rising(10000, 30, 50)
result = calculate_bollinger(prices, period=10)
assert result is not None
# ═══════════════════════════════════════════════════════════════════════════
# 6. is_reversal_candle
# ═══════════════════════════════════════════════════════════════════════════
class TestIsReversalCandle:
"""되돌림 캔들 감지 테스트"""
def test_bullish_candle(self):
"""양봉 (종가 > 시가) → True"""
result = is_reversal_candle(
opens=[10000], highs=[10500], lows=[9800], closes=[10300]
)
assert result is True
def test_long_lower_shadow(self):
"""긴 하단 꼬리 (하단 그림자 > 2× 몸통) → True"""
# 음봉이지만 긴 꼬리: 시가=10300, 종가=10200 (body=100)
# 하단 그림자 = min(10300,10200) - 9800 = 400 > 2*100
result = is_reversal_candle(
opens=[10300], highs=[10500], lows=[9800], closes=[10200]
)
assert result is True
def test_bearish_no_shadow(self):
"""음봉 + 짧은 꼬리 → False"""
# 시가=10500, 종가=10000 (body=500), 하단 그림자 = 10000-9900=100 < 2*500
result = is_reversal_candle(
opens=[10500], highs=[10600], lows=[9900], closes=[10000]
)
assert result is False
def test_empty_closes_returns_false(self):
"""빈 closes → False"""
result = is_reversal_candle(opens=[], highs=[], lows=[], closes=[])
assert result is False
def test_doji_with_long_lower_shadow(self):
"""도지 (몸통 ≈ 0) + 긴 하단 꼬리 → True"""
result = is_reversal_candle(
opens=[10000], highs=[10010], lows=[9500], closes=[10000]
)
assert result is True
def test_uses_last_candle_only(self):
"""여러 캔들 데이터 중 마지막 캔들만 사용"""
result = is_reversal_candle(
opens=[9000, 10000], # 마지막: 10000
highs=[9500, 10500],
lows=[8500, 9800],
closes=[8800, 10300], # 마지막: 양봉
)
assert result is True
# ═══════════════════════════════════════════════════════════════════════════
# 7. check_buy_signal
# ═══════════════════════════════════════════════════════════════════════════
class TestCheckBuySignal:
"""매수 신호 판별 테스트 — Bull/Bear 시장별"""
def test_bull_buy_signal(self):
"""상승장 매수 조건 충족: RSI<=30 + BB 하단 구간 + 거래량 충분 + RSI 반등"""
result = check_buy_signal(
regime=MarketRegime.BULL,
rsi=25.0,
close=9750, # bb_lower*0.97=9700 ~ bb_lower*1.015=10150
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=900000,
is_reversal=False,
rsi_prev=22.0, # RSI 반등 중 (22 → 25)
)
assert result is not None
assert result["signal"] == "BUY"
assert result["strategy"] == "MEAN_REVERSION_BULL"
def test_bull_rsi_too_high(self):
"""상승장 RSI > 30 → 신호 없음"""
result = check_buy_signal(
regime=MarketRegime.BULL,
rsi=45.0,
close=9800,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=900000,
)
assert result is None
def test_bull_insufficient_volume(self):
"""상승장 거래량 < 20일 평균 → 신호 없음"""
result = check_buy_signal(
regime=MarketRegime.BULL,
rsi=25.0,
close=9800,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=500000,
prev_volume=900000,
rsi_prev=22.0,
)
assert result is None
def test_bull_price_outside_bb_lower_range(self):
"""상승장 가격이 BB 하단 -3%~+1.5% 범위 밖 → 신호 없음"""
result = check_buy_signal(
regime=MarketRegime.BULL,
rsi=25.0,
close=10500, # BB 하단(10000) 대비 너무 높음
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=900000,
rsi_prev=22.0,
)
assert result is None
def test_bull_rsi_still_falling_blocked(self):
"""상승장 RSI가 하락 중 (낙하하는 칼) → 진입 차단"""
result = check_buy_signal(
regime=MarketRegime.BULL,
rsi=25.0,
close=9800,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=900000,
rsi_prev=28.0, # RSI 하락 중 (28 → 25)
)
assert result is None
def test_bear_buy_signal(self):
"""하락장 매수 조건: RSI<30 + BB 하단 이하 + 되돌림 + 거래량 > 1.5×20MA
[2026-03-15 수정] CLAUDE.md 6.2: BEAR RSI < 30 (이하 아닌 미만), 거래량 > 1.5× 20일 평균
- rsi=29.0 (< 30 충족)
- current_volume=1600000 (> 1.5×1000000=1500000 충족)
"""
result = check_buy_signal(
regime=MarketRegime.BEAR,
rsi=29.0,
close=9900,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1600000,
prev_volume=1100000,
is_reversal=True,
rsi_prev=27.0, # RSI 반등 중 (27→29)
)
assert result is not None
assert result["signal"] == "BUY"
assert result["strategy"] == "MEAN_REVERSION_BEAR"
assert len(result["split_plan"]) == 3 # 3회 분할 매수
def test_bear_no_reversal_candle(self):
"""하락장 되돌림 캔들 없음 → 신호 없음"""
result = check_buy_signal(
regime=MarketRegime.BEAR,
rsi=30.0,
close=9900,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=1100000,
is_reversal=False,
rsi_prev=27.0,
)
assert result is None
def test_bear_volume_not_increasing(self):
"""하락장 전일 대비 거래량 감소 → 신호 없음"""
result = check_buy_signal(
regime=MarketRegime.BEAR,
rsi=30.0,
close=9900,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=1300000, # 전일보다 감소
is_reversal=True,
rsi_prev=27.0,
)
assert result is None
def test_bear_rsi_still_falling_blocked(self):
"""하락장 RSI 하락 중 → 진입 차단"""
result = check_buy_signal(
regime=MarketRegime.BEAR,
rsi=30.0,
close=9900,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=1100000,
is_reversal=True,
rsi_prev=33.0, # RSI 하락 중 (33 → 30)
)
assert result is None
def test_bull_split_plan(self):
"""상승장 매수 → 2회 분할 매수 [0.50, 0.50]"""
result = check_buy_signal(
regime=MarketRegime.BULL,
rsi=25.0,
close=9800,
bb_upper=11000,
bb_middle=10500,
bb_lower=10000,
avg_volume_20d=1000000,
current_volume=1200000,
prev_volume=900000,
rsi_prev=22.0,
)
assert result is not None
assert result["split_plan"] == [0.50, 0.50]
# ═══════════════════════════════════════════════════════════════════════════
# 8. check_exit_signal
# ═══════════════════════════════════════════════════════════════════════════
class TestCheckExitSignal:
"""청산 신호 판별 테스트"""
def test_stop_loss_pct_triggered(self):
"""손절 도달 → SELL_ALL
[2026-03-15 수정] 변동성 판별 로직: ATR/종가 > 3% → -10% 손절 적용
- atr=100, close=9200 → _atr_pct=100/9200*100=1.09% <= 3% → 안정적 종목 (-6%)
- pct_stop = 10000×(1-6/100) = 9400
- atr_stop = 10000 - 2.5×100 = 9750
- stop_price = max(9750, 9400) = 9750
- close=9200 <= 9750 → 손절 발동
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=35.0,
close=9200, # stop_price=9750보다 낮아 손절 발동
bb_upper=11000,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=100, # ATR 작게 → 변동성 낮음 → -6% 손절 적용
lowest_since_entry=9000,
remaining_pct=1.0,
)
assert result is not None
assert result.action == "SELL_ALL"
assert result.sell_pct == 1.0
def test_atr_stop_triggered(self):
"""ATR 손절 도달 → SELL_ALL"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=30.0,
close=9200,
bb_upper=11000,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=300, # ATR 손절: 10000 - 2.5*300 = 9250
lowest_since_entry=9100,
remaining_pct=1.0,
)
assert result is not None
assert result.action == "SELL_ALL"
def test_first_profit_taking_bull(self):
"""Bull 1차 익절: RSI>=50 OR BB중앙 상회 OR 2×ATR 수익 → 50% 매도
[2026-03-15 수정] CLAUDE.md 6.3: 1차 익절은 OR 로직, 분할매도 50% (Bull/Bear 동일)
- RSI(58) >= 50: True
- close(10800) > bb_middle(10500): True
- close(10800) >= entry(10000)+2×ATR(300)=10600: True
→ sell_pct=0.50
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=58.0,
close=10800, # > bb_middle(10500), > entry+2*ATR(10600)
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=300,
lowest_since_entry=9800,
remaining_pct=1.0, # 아직 1차 익절 안 함
)
assert result is not None
assert result.action == "SELL_PARTIAL"
assert result.sell_pct == 0.50 # CLAUDE.md 5.2: 50% 통일
def test_first_profit_taking_bear(self):
"""Bear 1차 익절: RSI>=55 + BB중앙 상회 → 50% 매도"""
result = check_exit_signal(
regime=MarketRegime.BEAR,
rsi=60.0,
close=10800,
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=300,
lowest_since_entry=9800,
remaining_pct=1.0,
)
assert result is not None
assert result.action == "SELL_PARTIAL"
assert result.sell_pct == 0.50 # Bear 50%
def test_no_profit_if_rsi_below_55(self):
"""RSI < 50 AND BB중앙 미하회 AND 2×ATR 수익 미달 → 1차 익절 미발동
[2026-03-15 수정] CLAUDE.md 6.3: 1차 익절 OR 조건:
- RSI(49) < 50: False
- close(10200) <= bb_middle(10500): False (10200 < 10500 → BB중앙 미초과)
- close(10200) < entry(10000)+2×ATR(200)=10400: False
→ 세 조건 모두 불충족 → 익절 미발동
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=49.0,
close=10200, # bb_middle(10500) 미달, 2×ATR(10400) 미달
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=200,
lowest_since_entry=9800,
remaining_pct=1.0,
)
assert result is None
def test_no_profit_below_min_profit(self):
"""OR 조건 모두 미충족 → 1차 익절 미발동
[2026-03-15 수정] CLAUDE.md 6.3: OR 조건이므로 세 조건 모두 False여야 미발동.
- RSI(47) < 50: False
- close(9900) <= bb_middle(10200): False (9900 < 10200 → 미초과)
- close(9900) < entry(10000)+2×ATR(50)=10100: False
→ 전부 False → 익절 미발동
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=47.0,
close=9900, # bb_middle(10200) 미달, RSI(47) < 50, 2×ATR(10100) 미달
bb_upper=11500,
bb_middle=10200,
bb_lower=9500,
entry_price=10000,
atr=50,
lowest_since_entry=9800,
remaining_pct=1.0,
)
assert result is None
def test_trailing_stop(self):
"""트레일링 스탑: 고점 - 1.5×ATR 하회 → SELL_ALL
[2026-03-15 수정] CLAUDE.md 5.2: 트레일링 = 고점 - 1.5×ATR (비율이 아닌 ATR 기반)
- remaining_pct=0.4 (1차 익절 완료, threshold=0.50 이하)
- 1차 익절 섹션: remaining_pct(0.4) > 0.50 → False → 통과
- 전량매도 섹션:
- rsi=45 < 65 → False
- close=10100 < bb_upper×0.99=11385 → False
- trailing_high(11000) > entry(10000), trail_stop=11000-1.5×300=10550
- close=10100 < 10550 → True → 트레일링 발동
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=45.0,
close=10100, # trail_stop=11000-1.5×300=10550 하회
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=300,
lowest_since_entry=9800,
remaining_pct=0.4, # 1차 익절 완료 (threshold=0.50 이하)
trailing_high=11000,
)
assert result is not None
assert result.action == "SELL_ALL"
assert "트레일링" in result.reason
def test_second_profit_bb_upper(self):
"""BB 상단 도달 + RSI>=65 → 전량청산 (SELL_ALL)
[2026-03-15 수정] CLAUDE.md 6.3: BB 상단 도달 시 전량매도(SELL_ALL) 조건.
- remaining_pct=0.4 (1차 익절 완료, threshold=0.50 이하)
- 1차 익절 섹션: remaining_pct(0.4) > 0.50 → False → 스킵
- 전량매도 섹션:
- rsi=68 >= 65 → True → SELL_ALL 반환
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=68.0,
close=11400, # bb_upper*0.99=11385 상회
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=300,
lowest_since_entry=9800,
remaining_pct=0.4, # 1차 완료 (threshold=0.50 이하)
)
assert result is not None
assert result.action == "SELL_ALL"
assert result.sell_pct == 1.0
def test_no_exit_normal_hold(self):
"""청산 조건 모두 미충족 → 신호 없음
[2026-03-15 수정] 모든 OR 조건 False여야 None 반환:
- 손절: stop_price < close(10150) → 미발동
- _atr_pct=50/10150*100=0.49% → 안정적 → -6%
- pct_stop=10000×0.94=9400, atr_stop=10000-2.5×50=9875
- stop_price=9875 < 10150 → 미발동
- 1차 익절 (remaining=1.0 > 0.50):
- RSI(47) < 50: False
- close(10150) <= bb_middle(10500): False
- close(10150) < entry+2×ATR(10100): False (10150 >= 10100 → True!)
→ atr_exit=True → 발동됨... atr=50: min_profit=10000+2×50=10100 → close=10150 >= 10100
- 수정: close를 ATR 이익 미달로 조정: close=10050, atr=50 → min_profit=10100 → 10050<10100 OK
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=47.0,
close=10050, # RSI<50, BB중앙(10500) 미달, 2×ATR(10100) 미달
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=50,
lowest_since_entry=9800,
remaining_pct=1.0,
)
assert result is None
def test_stop_loss_has_priority(self):
"""손절은 익절보다 우선
[2026-03-15 수정] 변동성 판별: ATR/종가 <= 3% → -6% 손절 적용
- atr=80, close=9200 → _atr_pct=80/9200*100=0.87% → 안정적 (-6%)
- pct_stop=10000×0.94=9400, atr_stop=10000-2.5×80=9800
- stop_price=max(9800,9400)=9800
- close=9200 <= 9800 → 손절 발동 (RSI>=50 익절 조건보다 우선)
"""
result = check_exit_signal(
regime=MarketRegime.BULL,
rsi=60.0, # 1차 익절 조건(RSI>=50) 충족이지만 손절 우선
close=9200, # stop_price=9800 하회 → 손절
bb_upper=11500,
bb_middle=10500,
bb_lower=9500,
entry_price=10000,
atr=80, # 변동성 낮음 → -6% 손절 적용
lowest_since_entry=9000,
remaining_pct=1.0,
)
assert result is not None
assert result.action == "SELL_ALL"
assert "손절" in result.reason
# ── R4 2026-10-09: 1단계(분할 전)에서 전량 조건은 50% 분할로 처리된다 ──────
# 문서(docstring)가 "단계 무관 전량"이라 서술했으나 실제로는 분할 분기가 먼저 return 한다.
# 동작은 유지(백테스트 없이 청산 타이밍 변경 금지 — §8.1-5)하고 계약을 테스트로 고정한다.
# 적응형 오버레이는 예외 발생형 목으로 끊어 config(settings) 임계를 쓰게 한다.
@staticmethod
def _block_adaptive(monkeypatch):
import app.services.adaptive_strategy as _adp_mod
def _raise(*_a, **_k):
raise RuntimeError("adaptive blocked in test")
monkeypatch.setattr(_adp_mod, "get_current_adaptive_params", _raise)
@staticmethod
def _full_exit_kwargs(remaining_pct):
from app.core.config import settings
entry = 10000.0
bb_upper = 10500.0
# BB상단×임계 이상 + RSI ≥ 전량 임계 → 전량 조건 2개 모두 성립. 수익 ≥ 0.5%.
close = bb_upper * settings.MEAN_REVERSION_BB_UPPER_THRESHOLD + 10
return dict(
regime=MarketRegime.BULL,
rsi=settings.MEAN_REVERSION_RSI_FULL_EXIT + 1.0,
close=close,
bb_upper=bb_upper,
bb_middle=10100.0,
bb_lower=9500.0,
entry_price=entry,
atr=50.0, # 손절선(≥ entry−2.5×ATR) 훨씬 아래 → 손절 미발동
lowest_since_entry=9900.0,
remaining_pct=remaining_pct,
)
def test_full_exit_conditions_before_partial_are_split_r4(self, monkeypatch):
"""remaining_pct=1.0 + 전량 조건(RSI≥full·BB상단) → SELL_PARTIAL 50% (현재 계약)"""
self._block_adaptive(monkeypatch)
kw = self._full_exit_kwargs(remaining_pct=1.0)
assert (kw["close"] - kw["entry_price"]) / kw["entry_price"] * 100 >= 0.5
result = check_exit_signal(**kw)
assert result is not None
assert result.action == "SELL_PARTIAL"
assert result.sell_pct == 0.50
def test_full_exit_conditions_after_partial_sell_all_r4(self, monkeypatch):
"""remaining_pct=0.50(1차 익절 후) + 같은 조건 → SELL_ALL, 사유에 RSI·BB상단 모두"""
self._block_adaptive(monkeypatch)
result = check_exit_signal(**self._full_exit_kwargs(remaining_pct=0.50))
assert result is not None
assert result.action == "SELL_ALL"
assert result.sell_pct == 1.0
assert "RSI(" in result.reason and "BB상단" in result.reason
def test_full_exit_before_partial_below_min_profit_holds_r6(self, monkeypatch):
"""[R6 2026-10-10] 수익 < 0.5% 이면 RSI/BB상단 전량 '익절'도 나가지 않는다.
(R4 판은 이 구간 SELL_ALL 을 고정했으나, 손실 구간 '익절'이 연패로 쌓이는 결함이었다.)
손실 구간은 손절·시간손절이 맡는다."""
self._block_adaptive(monkeypatch)
kw = self._full_exit_kwargs(remaining_pct=1.0)
kw["entry_price"] = kw["close"] # 수익 0% → 분할·전량 익절 모두 skip
kw["lowest_since_entry"] = kw["close"] - 10
assert check_exit_signal(**kw) is None
def test_docstring_no_longer_claims_stage_independent_r4(self):
"""docstring 이 '단계 무관 전량'을 다시 주장하지 않도록 고정"""
doc = check_exit_signal.__doc__ or ""
assert "도달 (단계 무관)" not in doc
assert "단계 우선순위" in doc
# ═══════════════════════════════════════════════════════════════════════════
# 9. calculate_position_size
# ═══════════════════════════════════════════════════════════════════════════
class TestCalculatePositionSize:
"""포지션 사이징 테스트"""
def test_basic_bull_market(self):
"""상승장 기본 포지션 사이즈"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
)
# 기본: 100M * 4% * 1.0 = 4M
# 예산: 100M * 70% - 0 = 70M
# min(4M, 70M, 50M) = 4M
assert result == pytest.approx(4_000_000)
def test_bear_market_lower_budget(self):
"""하락장 전략 자본 한도 30%"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BEAR,
)
# 기본: 100M * 4% * 1.0 = 4M
# 예산: 100M * 30% - 0 = 30M
# min(4M, 30M, 50M) = 4M
assert result == pytest.approx(4_000_000)
def test_rsi_extreme_oversold_multiplier(self):
"""RSI <= 20 → multiplier 1.5×이지만 CLAUDE.md 5.1: 절대 상한 4%"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=18.0,
)
# [2026-03-15] CLAUDE.md 5.1 준수: 100M * 4% = 4M (multiplier로 상한 초과 불가)
assert result == pytest.approx(4_000_000)
def test_rsi_strong_oversold_multiplier(self):
"""RSI <= 25 → multiplier 1.25×이지만 절대 상한 4%"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=23.0,
)
assert result == pytest.approx(4_000_000) # CLAUDE.md 5.1: 4% 상한
def test_adx_low_bonus(self):
"""ADX < 15 → +0.25×이지만 절대 상한 4%"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=28.0, # 1.0×
adx=10.0, # +0.25 → 1.25× (but capped at 4%)
)
assert result == pytest.approx(4_000_000)
def test_adx_low_with_strong_rsi(self):
"""RSI<=20 + ADX<15 → multiplier 1.75×이지만 절대 상한 4%"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=18.0,
adx=10.0,
)
# CLAUDE.md 5.1: 절대 상한 4% = 4M
assert result == pytest.approx(4_000_000)
def test_remaining_budget_constraint(self):
"""전략 배분 한도 소진 시 0 반환"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=70_000_000, # 70% 이미 배분
regime=MarketRegime.BULL,
)
assert result == 0.0
def test_cash_constraint(self):
"""가용 현금 < 포지션 사이즈 → 현금만큼만"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=2_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
)
assert result == pytest.approx(2_000_000)
def test_zero_assets(self):
"""자산 0 → 0"""
result = calculate_position_size(
total_assets=0,
available_cash=0,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
)
assert result == 0.0
def test_multiplier_cap_at_4pct(self):
"""[CLAUDE.md 5.1] multiplier와 무관하게 절대 상한 4%"""
result = calculate_position_size(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=18.0, # 1.5×
adx=10.0, # +0.25 → 1.75
)
# CLAUDE.md 5.1: 4% 절대 상한 = 4M
assert result == pytest.approx(4_000_000)
# ═══════════════════════════════════════════════════════════════════════════
# 10. calculate_position_size_v2
# ═══════════════════════════════════════════════════════════════════════════
class TestCalculatePositionSizeV2:
"""포지션 사이징 v2 테스트 — RSI/ADX 보정 추가"""
def test_basic_no_adjustment(self):
"""RSI/ADX 없으면 base와 동일"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
)
# base = 100M * 4% * 1.0 = 4M, mult=1.0 → 4M
assert result == pytest.approx(4_000_000)
def test_rsi_deep_oversold_boost(self):
"""RSI <= 20 → ×1.2 보정"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=18.0,
)
# base = 4M (v1에서 rsi=None), mult = 1.2
# 4M * 1.2 = 4.8M, max_allowed = 100M * 4% = 4M → min(4.8M, 4M) = 4M
assert result == pytest.approx(4_000_000)
def test_rsi_moderate_oversold_boost(self):
"""RSI <= 25 → ×1.1 보정"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=23.0,
)
# base = 4M, mult = 1.1 → 4.4M, max_allowed = 4M → 4M
assert result == pytest.approx(4_000_000)
def test_adx_strong_trend_reduction(self):
"""ADX > 30 → ×0.8 감소"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
adx=35.0,
)
# base = 4M, mult = 0.8 → 3.2M, max_allowed = 4M → 3.2M
assert result == pytest.approx(3_200_000)
def test_combined_rsi_adx(self):
"""RSI <= 20 (×1.2) + ADX > 30 (×0.8) → ×0.96"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=18.0,
adx=35.0,
)
# base = 4M, mult = 1.2 * 0.8 = 0.96 → 3.84M
assert result == pytest.approx(3_840_000)
def test_max_allowed_cap(self):
"""보정 후에도 종목당 4% 상한 재적용"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
rsi=15.0, # ×1.2
)
max_allowed = 100_000_000 * 0.04 # 4M
assert result <= max_allowed
def test_zero_base_returns_zero(self):
"""base = 0 (예산 소진) → 0"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=50_000_000,
current_strategy_allocation=70_000_000,
regime=MarketRegime.BULL,
rsi=18.0,
)
assert result == 0.0
def test_cash_limited(self):
"""가용 현금 < 보정 사이즈 → 현금 제한"""
result = calculate_position_size_v2(
total_assets=100_000_000,
available_cash=1_000_000,
current_strategy_allocation=0,
regime=MarketRegime.BULL,
)
assert result == pytest.approx(1_000_000)