""" 15분봉 단기 전략 (CLAUDE.md 섹션 8.2 — 승률 최적화) 전체 자본의 최대 25% 이내에서 운용. 일봉 RSI ≤ 40 관심 종목에 한해 15분봉 신호를 감지한다. [승률 개선 항목] 1. 가격 하한 필터 → ATR 비율 필터 (ATR/close ≥ 0.8%) — 거래비용을 커버할 수 있는 변동성 종목만 선별 2. 거래량 배수 1.3 → 1.5 (허위 신호 감소) 3. RSI 방향 확인 — RSI가 바닥에서 반등 중인 경우만 진입 4. 반전 캔들 확인 — 양봉 or 긴 하단 꼬리 (optional) 5. 손절 조건 추가 — entry - 1.0×ATR (CRITICAL: 기존 코드에 누락되어 있던 것) 6. 장 초반 30분 / 마감 15분 진입 금지 (노이즈 구간) ⚠️ 슬리피지 주의: 수수료 0.3% + 슬리피지 0.2% 포함 백테스트 필수 """ import logging from dataclasses import dataclass, field from datetime import time as dtime from enum import Enum from typing import List, Optional from .mean_reversion_strategy import calculate_rsi, calculate_bollinger logger = logging.getLogger(__name__) # 자본 비율 상한 MAX_CAPITAL_RATIO = 0.25 # 비용 가정 — 백테스트 시 왕복으로 적용 COMMISSION_RATE = 0.003 SLIPPAGE_RATE = 0.002 TOTAL_COST_RATE = COMMISSION_RATE + SLIPPAGE_RATE # 승률 개선 파라미터 MIN_ATR_RATIO = 0.008 # ATR/close ≥ 0.8% : 거래비용 커버 보장 RSI_DIRECTION_LOOKBACK = 3 # RSI 반등 확인용 과거 봉 수 # 장 운영 시간 필터 _SKIP_OPEN_UNTIL = dtime(9, 30) # 장 초반 30분 제외 (개장 직후 변동성 과다) _SKIP_CLOSE_FROM = dtime(15, 15) # 마감 15분 전 제외 (종가 청산 압력) class TimeFrame(str, Enum): M15 = "15m" H1 = "60m" @dataclass class EntrySignal: ticker: str timeframe: TimeFrame reason: str rsi: float close: float atr_ratio: float = 0.0 # ATR/close — 수익 가능성 지표 @dataclass class ExitSignal: action: str # "SELL_ALL" reason: str sell_pct: float # 1.0 = 전량 class ShortTermStrategy: """ 15분봉 단기 평균회귀 전략 (승률 우선 최적화) 진입 조건 (모두 충족 필요): 1. 일봉 RSI(14) ≤ 40 — 관심 종목 사전 필터 2. 15분봉 RSI(14) ≤ 30 — 과매도 3. 15분봉 RSI 반등 중 — RSI_now > RSI_3봉전 (바닥 확인) 4. 15분봉 종가 ≤ BB 하단 × 1.01 5. 15분봉 종가 > MA5 — Fake-out 방지 6. 15분봉 거래량 > MA20 × 1.5 — 1.3에서 강화 7. ATR/close ≥ 0.8% — 거래비용 커버 가능한 변동성 8. 반전 캔들 (optional) — 양봉 or 핵머 캔들 9. 시간 필터 (optional) — 09:30~15:15만 허용 청산 조건 (우선순위 순): 1. 손절: close ≤ entry - 1.0×ATR [승률 핵심 — 기존 코드에 누락] 2. RSI(14) ≥ 55 회복 3. 수익률 ≥ 1.5 × ATR_15m_pct """ TIMEFRAME = TimeFrame.M15 DAILY_RSI_THRESHOLD = 40.0 M15_RSI_ENTRY = 30.0 M15_RSI_EXIT = 55.0 VOLUME_MULTIPLIER = 1.5 # 1.3 → 1.5 강화 MIN_ATR_RATIO = MIN_ATR_RATIO def __init__(self, max_capital_ratio: float = MAX_CAPITAL_RATIO): self.max_capital_ratio = max_capital_ratio # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def entry_signal( self, ticker: str, daily_rsi: float, m15_closes: List[float], m15_volumes: List[int], current_volume: int, atr_15m: Optional[float] = None, m15_opens: Optional[List[float]] = None, m15_lows: Optional[List[float]] = None, trade_time: Optional[dtime] = None, ) -> Optional[EntrySignal]: """ 15분봉 진입 신호 감지 Args: ticker: 종목코드 daily_rsi: 일봉 RSI(14) m15_closes: 15분봉 종가 리스트 (최소 28개) m15_volumes: 15분봉 거래량 리스트 (최소 20개) current_volume: 현재 봉 거래량 atr_15m: 15분봉 ATR — None이면 ATR 비율 필터 스킵 m15_opens: 15분봉 시가 — None이면 반전 캔들 스킵 m15_lows: 15분봉 저가 — None이면 반전 캔들 스킵 trade_time: 현재 시각 — None이면 시간 필터 스킵 Returns: EntrySignal if 조건 충족, else None """ # 조건 1: 일봉 사전 필터 if daily_rsi > self.DAILY_RSI_THRESHOLD: return None # 조건 9: 시간 필터 (제공된 경우) if trade_time is not None: if not (_SKIP_OPEN_UNTIL <= trade_time < _SKIP_CLOSE_FROM): return None min_len = 25 + RSI_DIRECTION_LOOKBACK if len(m15_closes) < min_len or len(m15_volumes) < 20: return None close = m15_closes[-1] # 조건 2: 15분봉 RSI 과매도 rsi_15m = calculate_rsi(m15_closes, 14) if rsi_15m is None or rsi_15m > self.M15_RSI_ENTRY: return None # 조건 3: RSI 반등 방향 확인 — 바닥에서 회복 중인지 확인 (승률 햵심) rsi_prev = calculate_rsi(m15_closes[:-RSI_DIRECTION_LOOKBACK], 14) if rsi_prev is not None and rsi_15m <= rsi_prev: # RSI가 여전히 하락 중 → 변곡점 전이므로 진입 금지 return None # 조건 4: BB 하단 근처 bb = calculate_bollinger(m15_closes, 20) if bb is None: return None _, _, bb_lower = bb if close > bb_lower * 1.01: return None # 조건 5: MA5 상향 (Fake-out 방지) # BB 하단 ±1% 구간에서는 최근 급락으로 MA5가 눌려있을 수 있으므로 완화 허용 # (close < bb_lower * 0.99 → BB 하단을 크게 이탈 중 = 완전 차단, 그 외는 허용) ma5 = sum(m15_closes[-5:]) / 5 if close < ma5 and close < bb_lower * 0.99: return None # 조건 6: 거래량 급증 (1.5배 강화) vol_ma20 = sum(m15_volumes[-20:]) / 20 if vol_ma20 == 0 or current_volume <= vol_ma20 * self.VOLUME_MULTIPLIER: return None # 조건 7: ATR 비율 필터 — 5,000원 필터 대체 atr_ratio = 0.0 if atr_15m is not None: atr_ratio = atr_15m / close if close > 0 else 0.0 if atr_ratio < self.MIN_ATR_RATIO: # 변동성 부족 → 1.5×ATR 익절 목표가 왕복 비용에 도달안함 return None # 조건 8: 반전 캔들 (데이터 제공 시) if m15_opens is not None and m15_lows is not None: if not _is_reversal_candle(m15_opens, m15_lows, m15_closes): return None return EntrySignal( ticker=ticker, timeframe=self.TIMEFRAME, reason=( f"RSI({rsi_15m:.1f})≤30↑ + BB하단 + MA5돌파" f" + 거래량({current_volume / vol_ma20:.1f}x)" + (f" + ATR비율({atr_ratio * 100:.1f}%)" if atr_15m else "") ), rsi=round(rsi_15m, 2), close=close, atr_ratio=round(atr_ratio, 4), ) def exit_signal( self, ticker: str, m15_closes: List[float], entry_price: float, atr_15m: Optional[float] = None, ) -> Optional[ExitSignal]: """ 15분봉 청산 신호 청산 우선순위: 1. 손절 : close ≤ max(entry - 1.0×ATR, entry × 0.98) [핵심 추가] 2. RSI ≥ 55 회복 3. 수익률 ≥ 1.5 × ATR_15m_pct """ if len(m15_closes) < 15: return None close = m15_closes[-1] rsi_15m = calculate_rsi(m15_closes, 14) # 청산 우선순위 1: 손절 [기존 코드에서 누락된 핸심 로직] if entry_price > 0: if atr_15m: atr_stop = entry_price - 1.0 * atr_15m hard_stop = entry_price * 0.98 # -2% 하드 스탑 stop_price = max(atr_stop, hard_stop) # 더 타이트한 쪽 else: stop_price = entry_price * 0.98 # ATR 없으면 -2% 고정 if close <= stop_price: loss_pct = (close - entry_price) / entry_price * 100 return ExitSignal( action="SELL_ALL", reason=f"손절 close({close:.0f}) ≤ stop({stop_price:.0f}) | {loss_pct:.1f}%", sell_pct=1.0, ) # 청산 우선순위 2: RSI ≥ 55 회복 if rsi_15m is not None and rsi_15m >= self.M15_RSI_EXIT: return ExitSignal( action="SELL_ALL", reason=f"15분봉 RSI({rsi_15m:.1f})≥55 회복", sell_pct=1.0, ) # 청산 우선순위 3: 수익률 ≥ 1.5 × ATR_15m_pct if atr_15m and entry_price > 0: profit_pct = (close - entry_price) / entry_price atr_pct = atr_15m / entry_price if profit_pct >= 1.5 * atr_pct: return ExitSignal( action="SELL_ALL", reason=( f"익절 수익률({profit_pct * 100:.1f}%)" f" ≥ 1.5×ATR({atr_pct * 100:.1f}%)" ), sell_pct=1.0, ) return None def is_capital_available( self, total_assets: float, current_short_term_allocation: float ) -> bool: """CLAUDE.md 8.3: 단기 전략 자본 한도 충족 여부""" return current_short_term_allocation < total_assets * self.max_capital_ratio @staticmethod def net_profit_pct(gross_profit_pct: float) -> float: """수수료 + 슬리피지 차감 후 순수익률 (왕복 기준)""" return gross_profit_pct - TOTAL_COST_RATE * 2 # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _is_reversal_candle( opens: List[float], lows: List[float], closes: List[float], ) -> bool: """반전 캔들 확인 — 양봉 또는 핸머(카드) 여부 조건 (둘 중 하나): - 양봉: close > open - 해머: 하단 꼬리 > 몸통의 2배 (핀바 / 행장형) """ if not opens or not lows or not closes: return False o, l, c = opens[-1], lows[-1], closes[-1] body = abs(c - o) lower_shadow = min(o, c) - l return (c > o) or (lower_shadow > max(body * 2.0, 1e-6))