"""박스권(SIDEWAYS) 레인지 매매 전략 — CLAUDE.md §8.5 항목 9 전반부 ⚠️ **백테스트 미검증 — `RANGE_TRADE_ENABLED=True` 전환 금지 (CLAUDE.md §8.1-4)** 현재 상태: 코드만 존재. 실계좌·모의계좌 어느 쪽에서도 검증되지 않았다. `settings.RANGE_TRADE_ENABLED` 는 기본 False 이며, False 인 동안 `swing/entry_service.scan_range_trading_signals()` 는 함수 진입 즉시 `[]` 를 반환한다 (기존 동작 완전 불변). **켜기 전에 반드시 통과해야 하는 검증 항목** 1. SIDEWAYS 구간 라벨링 정확도 — `detect_market_regime()` 이 SIDEWAYS 로 판정한 구간(2023-03~2026-03)을 추출하고, 그 구간에서만 이 전략을 돌린 결과를 측정한다. SIDEWAYS 표본이 60거래일 미만이면 통계적 의미 없음 → 판정 불가로 간주. 2. In-sample 백테스트(2023-03~2025-03): 승률 ≥ 50%, Profit Factor ≥ 1.3, 평균 보유 5~15거래일, MDD ≥ -8%(더 얕아야 함), 거래비용 0.5%(왕복) 반영 후 기준. 3. Walk-forward (6개월 학습 / 3개월 검증, 최소 4구간): 모든 검증 구간에서 PF ≥ 1.0, 어느 구간도 MDD -10% 초과 금지. 한 구간이라도 깨지면 불합격. 4. 박스 이탈 손절의 실효성: 전체 거래 중 `box_break` 손절 비중과 그때의 평균 손실률을 따로 집계. 평균 손실이 -RANGE_TRADE_STOP_LOSS_PERCENT 보다 크면(더 많이 잃으면) 박스 이탈 판정 임계값(RANGE_TRADE_BOX_BREAK_PCT)을 재조정해야 한다. 5. 파라미터 민감도: RANGE_TRADE_LOOKBACK_DAYS(15/20/25), RANGE_TRADE_MIN_WIDTH_PCT(5/6/8), RANGE_TRADE_RSI_ENTRY(35/40/45) 그리드에서 PF 가 1.0 아래로 떨어지는 조합이 절반을 넘으면 과최적화로 판단 → 폐기. 6. 기존 전략과의 중복/상관: 같은 날 평균회귀·눌림목과 동일 종목을 잡는 비율이 30%를 넘으면 신규 신호원으로서의 가치가 없음 → 배선 재검토. 7. 미검증 상태로 스케줄러에 배선하지 말 것(swing/scheduler.py 는 본 작업 범위 밖이며 현재 호출부가 없다 — 보고서의 후속 작업 항목 참조). ───────────────────────────────────────────────────────────────────────────── 전략 개요 (순수 로직 — DB·네트워크 접근 없음) 진입: 성립된 박스권의 **하단 지지 반등** - 박스 성립 검증 (detect_range_box): 최근 N일 고/저 레인지 폭이 일정 범위, 상단·하단을 각각 min_touches 회 이상 접촉, 접촉 순서가 교대(왕복 이력)해야 한다. - ADX 상한으로 추세장 배제 (박스권 전제 = 추세 없음). - 가격이 박스 하단 zone 안 + BB 하단 부근(%B) + RSI 과매도. - **반등 확인 필수**: RSI 상승 전환(rsi > rsi_prev) 또는 양봉. 둘 다 불가하면 진입 거부. - 거래량 ≥ 20일 평균 × 배수 (지지 반등의 수급 확인). 청산: 박스 상단 접근 시 익절 / 박스 이탈 시 즉시 손절 - 1) 박스 하단 하향 이탈 → 즉시 전량 (레인지 전략의 최대 리스크) - 2) 일반 손절 (기본%·ATR 중 타이트한 쪽) → 전량 - 3) 박스 상단 zone 도달 / BB 상단(99%) / RSI 전량임계 / 트레일링 → 전량 - 4) 박스 중단(mid) 통과 또는 RSI 부분임계 → 50% 부분 익절 (수수료 가드 후) 손절 (CLAUDE.md §5.2): stop_price = max(atr_stop, pct_stop) ← 가격 기준에서 **높은 값이 타이트한 쪽** atr_stop = entry_price − RANGE_TRADE_ATR_STOP_MULTIPLIER × ATR pct_stop = entry_price × (1 + stop_loss_pct/100), stop_loss_pct = -abs(...) 보장 박스 하단 이탈 손절은 위 stop_price 와 **별개의 OR 트리거**로 동작한다 (박스 하단이 stop_price 보다 낮을 수도, 높을 수도 있어 max() 에 섞지 않는다). 모든 파라미터는 `app.core.config.settings` 의 `RANGE_TRADE_*` 설정에서 읽는다 (하드코딩 금지 — 함수 인자로 오버라이드 가능하게 두어 백테스트 그리드 탐색에 사용). """ import logging from dataclasses import dataclass from typing import Dict, List, Optional from app.core.log_throttle import warn_throttled # noqa: E402 logger = logging.getLogger(__name__) # ActiveTrade.strategy_name 에 기록되는 값. # ⚠️ mdd_manager.SWING_STRATEGY_NAMES 에 이 값이 추가되어야 MDD 축소가 반영된다 # (본 작업 범위 밖 — 보고서 후속 작업 항목 참조). RANGE_TRADE_STRATEGY_NAME = "RANGE_TRADE" @dataclass class RangeBox: """성립이 검증된 박스권.""" high: float # 박스 상단 (lookback 구간 최고가) low: float # 박스 하단 (lookback 구간 최저가) mid: float # 박스 중앙 width_pct: float # 박스 폭 / 중앙 × 100 top_touches: int # 상단 접촉 봉 수 bottom_touches: int # 하단 접촉 봉 수 alternations: int # 상단↔하단 접촉 교대 횟수 (왕복 이력) lookback: int # 판정에 사용한 봉 수 position_pct: float # 현재가의 박스 내 위치 (0=하단, 100=상단) @dataclass class RangeExitSignal: """청산 신호. mean_reversion_strategy.ExitSignal 과 동일한 필드 구성 (action/reason/sell_pct) — swing/exit_service 의 소비 코드와 duck-typing 호환.""" action: str # "SELL_PARTIAL" | "SELL_ALL" reason: str sell_pct: float # 0.5 = 50%, 1.0 = 전량 def _cfg(name: str, override: Optional[float], default: float) -> float: """설정값 조회: 명시 오버라이드 > settings.RANGE_TRADE_* > 하드 폴백. 하드 폴백은 config.py 가 아직 해당 키를 갖지 않는 경우(구버전 이미지)에만 쓰인다. """ if override is not None: return float(override) try: from app.core.config import settings val = getattr(settings, name, None) if val is not None: return float(val) except Exception as _swallowed: warn_throttled(logger, "range_trading_strategy:103", "[RangeTrade] settings.%s 조회 실패(기본값 사용): %s", name, _swallowed) return float(default) # ═══════════════════════════════════════════════════════════════════════════ # 1. 박스권 성립 검증 # ═══════════════════════════════════════════════════════════════════════════ def detect_range_box( highs: List[float], lows: List[float], closes: List[float], lookback: Optional[int] = None, min_width_pct: Optional[float] = None, max_width_pct: Optional[float] = None, touch_tolerance_pct: Optional[float] = None, min_touches: Optional[int] = None, min_alternations: Optional[int] = None, box_break_pct: Optional[float] = None, ) -> Optional[RangeBox]: """최근 lookback 봉이 '박스권'인지 검증하고 박스를 반환. 실패 시 None. 검증 조건 (모두 AND): ① 데이터 충분: len ≥ lookback ② 박스 폭: min_width_pct ≤ (high-low)/mid×100 ≤ max_width_pct - 너무 좁으면 왕복 이익이 거래비용(왕복 ~0.5%)을 못 넘고, 너무 넓으면 '박스'가 아니라 추세/고변동 구간이다. ③ 상단·하단 접촉 각각 min_touches 회 이상 (tolerance 이내 접근을 접촉으로 간주) ④ 접촉 순서 교대 min_alternations 회 이상 → 레인지 안에서 실제로 왕복한 이력 (모든 하단 접촉이 앞쪽에만 몰려 있으면 하락 후 횡보일 뿐 박스가 아니다) ⑤ 현재 종가가 박스를 이탈하지 않았음 (상·하단 ±box_break_pct 이내) """ _lookback = int(_cfg("RANGE_TRADE_LOOKBACK_DAYS", lookback, 20)) _min_w = _cfg("RANGE_TRADE_MIN_WIDTH_PCT", min_width_pct, 6.0) _max_w = _cfg("RANGE_TRADE_MAX_WIDTH_PCT", max_width_pct, 20.0) _tol = _cfg("RANGE_TRADE_TOUCH_TOLERANCE_PCT", touch_tolerance_pct, 2.0) _min_touch = int(_cfg("RANGE_TRADE_MIN_TOUCHES", min_touches, 2)) _min_alt = int(_cfg("RANGE_TRADE_MIN_ALTERNATIONS", min_alternations, 2)) _break = _cfg("RANGE_TRADE_BOX_BREAK_PCT", box_break_pct, 1.0) if _lookback <= 0: return None if len(highs) < _lookback or len(lows) < _lookback or len(closes) < 1: return None win_h = highs[-_lookback:] win_l = lows[-_lookback:] box_high = max(win_h) box_low = min(win_l) if box_high <= 0 or box_low <= 0 or box_high <= box_low: return None mid = (box_high + box_low) / 2.0 if mid <= 0: return None width_pct = (box_high - box_low) / mid * 100.0 if width_pct < _min_w or width_pct > _max_w: logger.debug( "[RangeBox] 폭 조건 미충족: width=%.2f%% (허용 %.1f~%.1f%%)", width_pct, _min_w, _max_w, ) return None # ③④ 접촉 및 교대 판정 (시간 순서 유지) top_line = box_high * (1.0 - _tol / 100.0) bottom_line = box_low * (1.0 + _tol / 100.0) events: List[str] = [] top_touches = 0 bottom_touches = 0 for h, l in zip(win_h, win_l): is_top = h >= top_line is_bottom = l <= bottom_line if is_top and not is_bottom: top_touches += 1 events.append("T") elif is_bottom and not is_top: bottom_touches += 1 events.append("B") elif is_top and is_bottom: # 한 봉이 박스 전체를 커버 = 변동성 급증, 박스 판정 무효 logger.debug("[RangeBox] 단일 봉이 박스 전폭 커버 — 박스 무효") return None if top_touches < _min_touch or bottom_touches < _min_touch: logger.debug( "[RangeBox] 접촉 부족: top=%d, bottom=%d (각 %d회 이상 필요)", top_touches, bottom_touches, _min_touch, ) return None alternations = 0 prev = None for e in events: if prev is not None and e != prev: alternations += 1 prev = e if alternations < _min_alt: logger.debug( "[RangeBox] 왕복 이력 부족: alternations=%d (%d회 이상 필요) events=%s", alternations, _min_alt, "".join(events), ) return None # ⑤ 현재 종가의 박스 이탈 여부 close = closes[-1] if close < box_low * (1.0 - _break / 100.0): logger.debug("[RangeBox] 하단 이탈 상태 (close=%.1f < low=%.1f) — 박스 무효", close, box_low) return None if close > box_high * (1.0 + _break / 100.0): logger.debug("[RangeBox] 상단 이탈 상태 (close=%.1f > high=%.1f) — 박스 무효", close, box_high) return None position_pct = (close - box_low) / (box_high - box_low) * 100.0 return RangeBox( high=box_high, low=box_low, mid=mid, width_pct=width_pct, top_touches=top_touches, bottom_touches=bottom_touches, alternations=alternations, lookback=_lookback, position_pct=position_pct, ) # ═══════════════════════════════════════════════════════════════════════════ # 2. 진입 판정 # ═══════════════════════════════════════════════════════════════════════════ def check_range_buy_signal( highs: List[float], lows: List[float], closes: List[float], volumes: List[int], current_close: float, current_volume: int, rsi: float, bb_upper: float, bb_middle: float, bb_lower: float, adx: Optional[float], rsi_prev: Optional[float] = None, opens: Optional[List[float]] = None, box: Optional[RangeBox] = None, rsi_entry: Optional[float] = None, bb_pct_b_max: Optional[float] = None, entry_zone_pct: Optional[float] = None, max_adx: Optional[float] = None, volume_ratio: Optional[float] = None, ) -> Optional[Dict]: """박스권 하단 지지 반등 매수 신호. 조건 미충족 시 None. 반환 dict 키는 mean_reversion_strategy.check_buy_signal() 과 동일한 관례를 따른다 (signal/strategy/description) + 레인지 전용 필드(box_high/box_low/target_price). Args: adx: 일봉 ADX(14). **None 이면 진입 거부**(추세장 배제 판정 불가 = fail-closed). rsi_prev: N봉 전 RSI. RSI 상승 전환 확인용. opens: 시가 리스트. 양봉 확인용. rsi_prev·opens 둘 다 없으면 반등 확인 불가 → 거부. """ _rsi_entry = _cfg("RANGE_TRADE_RSI_ENTRY", rsi_entry, 40.0) _pct_b_max = _cfg("RANGE_TRADE_BB_PCT_B_MAX", bb_pct_b_max, 0.25) _zone = _cfg("RANGE_TRADE_ENTRY_ZONE_PCT", entry_zone_pct, 25.0) _max_adx = _cfg("RANGE_TRADE_MAX_ADX", max_adx, 20.0) _vol_ratio = _cfg("RANGE_TRADE_VOLUME_RATIO", volume_ratio, 1.0) if current_close <= 0 or rsi is None: return None # ── ① 추세장 배제 (ADX 상한) ────────────────────────────────────────── if adx is None: logger.debug("[RangeBuy] ADX 없음 — 추세 판정 불가로 진입 거부(fail-closed)") return None if adx >= _max_adx: logger.debug("[RangeBuy] ADX(%.1f) >= %.1f — 추세장, 레인지 진입 거부", adx, _max_adx) return None # ── ② 박스 성립 검증 ──────────────────────────────────────────────── _box = box if box is not None else detect_range_box(highs, lows, closes) if _box is None: return None # ── ③ 박스 하단 zone 안인지 ────────────────────────────────────────── if _box.position_pct > _zone: logger.debug( "[RangeBuy] 박스 내 위치 %.1f%% > %.1f%% — 하단 지지대 아님", _box.position_pct, _zone, ) return None # ── ④ BB 하단 부근 (%B) ───────────────────────────────────────────── bb_width = bb_upper - bb_lower pct_b = (current_close - bb_lower) / bb_width if bb_width > 0 else 0.5 if pct_b > _pct_b_max: logger.debug("[RangeBuy] %%B(%.2f) > %.2f — BB 하단 부근 아님", pct_b, _pct_b_max) return None # ── ⑤ RSI 과매도 ──────────────────────────────────────────────────── if rsi >= _rsi_entry: logger.debug("[RangeBuy] RSI(%.1f) >= %.1f — 과매도 아님", rsi, _rsi_entry) return None # ── ⑥ 반등 확인 필수 (RSI 상승 전환 OR 양봉) ───────────────────────── rsi_turning_up = rsi_prev is not None and rsi > rsi_prev bullish_candle = False if opens and len(opens) >= 1 and len(closes) >= 1: bullish_candle = closes[-1] > opens[-1] if not (rsi_turning_up or bullish_candle): logger.debug( "[RangeBuy] 반등 미확인 — 진입 거부 (rsi=%.1f, rsi_prev=%s, 양봉=%s)", rsi, f"{rsi_prev:.1f}" if rsi_prev is not None else "N/A", bullish_candle, ) return None # ── ⑦ 거래량 확인 ─────────────────────────────────────────────────── if volumes and len(volumes) >= 20: avg_vol_20 = sum(volumes[-20:]) / 20.0 if avg_vol_20 > 0 and current_volume < avg_vol_20 * _vol_ratio: logger.debug( "[RangeBuy] 거래량 부족: %d < %.0f×%.2f", current_volume, avg_vol_20, _vol_ratio, ) return None elif current_volume <= 0: return None # 목표가: 박스 상단과 BB 상단 중 보수적(낮은) 쪽 target_price = min(_box.high, bb_upper) if bb_upper > 0 else _box.high reversal_desc = [] if rsi_turning_up: reversal_desc.append(f"RSI반등({rsi_prev:.1f}→{rsi:.1f})") if bullish_candle: reversal_desc.append("양봉") return { "signal": "BUY", "strategy": RANGE_TRADE_STRATEGY_NAME, "confidence": 0.55, # 미검증 전략 — 보수적 고정값 (신뢰도 상향은 백테스트 후) "box_high": _box.high, "box_low": _box.low, "box_mid": _box.mid, "box_width_pct": _box.width_pct, "position_pct": _box.position_pct, "target_price": target_price, "description": ( f"박스권 하단 지지 반등: 박스[{_box.low:.0f}~{_box.high:.0f}] " f"폭{_box.width_pct:.1f}% 접촉(T{_box.top_touches}/B{_box.bottom_touches}) " f"왕복{_box.alternations} 위치{_box.position_pct:.0f}% " f"RSI({rsi:.1f})<{_rsi_entry:.0f} %B({pct_b:.2f}) ADX({adx:.1f})<{_max_adx:.0f} " f"{'+'.join(reversal_desc)}" ), } # ═══════════════════════════════════════════════════════════════════════════ # 3. 손절가 계산 (CLAUDE.md §5.2) # ═══════════════════════════════════════════════════════════════════════════ def calculate_range_stop_loss( entry_price: float, atr: float, box_low: float, atr_multiplier: Optional[float] = None, stop_loss_percent: Optional[float] = None, box_break_pct: Optional[float] = None, ) -> Dict[str, float]: """레인지 전략 손절가. CLAUDE.md §5.2: 기본 SL 과 ATR SL 중 **타이트한 쪽**. 가격 기준 비교에서는 더 **높은** 손절가가 타이트하므로 `max(atr_stop, pct_stop)`. (수익률 기준으로 비교할 때의 `min()` 과 방향이 반대인 점 주의) 박스 하단 이탈 손절(box_break_stop)은 별개의 OR 트리거로 반환한다 — max() 에 섞으면 박스가 넓을 때 손절이 과도하게 느슨해지거나(하단이 낮음), 박스가 좁을 때 정상 노이즈에서 조기 손절(하단이 높음)될 수 있다. """ _atr_mult = _cfg("RANGE_TRADE_ATR_STOP_MULTIPLIER", atr_multiplier, 2.0) _sl_pct_raw = _cfg("RANGE_TRADE_STOP_LOSS_PERCENT", stop_loss_percent, 5.0) _break = _cfg("RANGE_TRADE_BOX_BREAK_PCT", box_break_pct, 1.0) stop_loss_pct = -abs(_sl_pct_raw) # 부호 가드 — 양수 입력 시 전종목 매도 방지 atr_stop = entry_price - _atr_mult * max(0.0, float(atr or 0.0)) pct_stop = entry_price * (1.0 + stop_loss_pct / 100.0) stop_price = max(atr_stop, pct_stop) # 높은 값 = 타이트한 손절 box_break_stop = box_low * (1.0 - _break / 100.0) if box_low > 0 else 0.0 return { "stop_price": stop_price, "atr_stop": atr_stop, "pct_stop": pct_stop, "stop_loss_pct": stop_loss_pct, "box_break_stop": box_break_stop, } # ═══════════════════════════════════════════════════════════════════════════ # 4. 청산 판정 # ═══════════════════════════════════════════════════════════════════════════ def check_range_exit_signal( rsi: float, close: float, bb_upper: float, bb_middle: float, bb_lower: float, entry_price: float, atr: float, box_high: float, box_low: float, remaining_pct: float = 1.0, trailing_high: Optional[float] = None, rsi_partial_exit: Optional[float] = None, rsi_full_exit: Optional[float] = None, exit_zone_pct: Optional[float] = None, trailing_atr_mult: Optional[float] = None, min_profit_pct: Optional[float] = None, ) -> Optional[RangeExitSignal]: """레인지 청산 판정. 우선순위대로 평가한다. 1. 박스 하단 하향 이탈 → 즉시 전량 (레인지 전략의 핵심 리스크) 2. 일반 손절 max(atr_stop, pct_stop) → 전량 3. 박스 상단 zone / BB 상단(99%) / RSI 전량임계 / 트레일링 → 전량 4. 박스 중앙 상회 또는 RSI 부분임계 → 50% 부분 익절 (수수료 가드 통과 시) """ _rsi_partial = _cfg("RANGE_TRADE_RSI_PARTIAL_EXIT", rsi_partial_exit, 55.0) _rsi_full = _cfg("RANGE_TRADE_RSI_FULL_EXIT", rsi_full_exit, 65.0) _exit_zone = _cfg("RANGE_TRADE_EXIT_ZONE_PCT", exit_zone_pct, 80.0) _trail_mult = _cfg("RANGE_TRADE_TRAILING_ATR_MULT", trailing_atr_mult, 1.5) _min_profit = _cfg("RANGE_TRADE_MIN_PROFIT_PCT", min_profit_pct, 0.5) if close <= 0 or entry_price <= 0: return None stops = calculate_range_stop_loss(entry_price, atr, box_low) # ── 1. 박스 하단 이탈 → 즉시 전량 손절 ────────────────────────────── if stops["box_break_stop"] > 0 and close <= stops["box_break_stop"]: return RangeExitSignal( action="SELL_ALL", reason=( f"박스 이탈 손절 (close={close:.0f} <= 박스하단이탈선" f"={stops['box_break_stop']:.0f}, box_low={box_low:.0f})" ), sell_pct=1.0, ) # ── 2. 일반 손절 (기본%·ATR 중 타이트한 쪽) ───────────────────────── if close <= stops["stop_price"]: return RangeExitSignal( action="SELL_ALL", reason=( f"손절 (close={close:.0f} <= stop={stops['stop_price']:.0f}, " f"ATR기준={stops['atr_stop']:.0f}, {stops['stop_loss_pct']:.1f}%기준" f"={stops['pct_stop']:.0f})" ), sell_pct=1.0, ) profit_pct = (close - entry_price) / entry_price * 100.0 # ── 3. 전량 익절 조건 ────────────────────────────────────────────── full_reasons: List[str] = [] if box_high > box_low: pos_pct = (close - box_low) / (box_high - box_low) * 100.0 if pos_pct >= _exit_zone: full_reasons.append(f"박스상단zone(위치{pos_pct:.0f}%>={_exit_zone:.0f}%)") if bb_upper > 0 and close >= bb_upper * 0.99: full_reasons.append(f"BB상단({bb_upper:.0f})도달") if rsi is not None and rsi >= _rsi_full: full_reasons.append(f"RSI({rsi:.1f})>={_rsi_full:.0f}") if trailing_high is not None and trailing_high > entry_price and atr > 0: trail_stop = trailing_high - _trail_mult * atr if close < trail_stop: full_reasons.append(f"트레일링(close={close:.0f}<{trail_stop:.0f})") # 익절은 수수료(왕복 ~0.5%)를 커버할 때만. 트레일링은 손실 보호이므로 예외. _has_trailing = any(r.startswith("트레일링") for r in full_reasons) if full_reasons and (profit_pct >= _min_profit or _has_trailing): return RangeExitSignal( action="SELL_ALL", reason=f"전량청산: {' OR '.join(full_reasons)}", sell_pct=1.0, ) # ── 4. 부분 익절 (50%) ───────────────────────────────────────────── if remaining_pct > 0.50 and profit_pct >= _min_profit: partial_reasons: List[str] = [] if rsi is not None and rsi >= _rsi_partial: partial_reasons.append(f"RSI({rsi:.1f})>={_rsi_partial:.0f}") if bb_middle > 0 and close > bb_middle: partial_reasons.append(f"BB중앙({bb_middle:.0f})상회") if box_high > box_low and close > (box_high + box_low) / 2.0: partial_reasons.append("박스중앙상회") if partial_reasons: return RangeExitSignal( action="SELL_PARTIAL", reason=f"1차 익절(50%): {' OR '.join(partial_reasons)}", sell_pct=0.50, ) return None