""" 데이트레이딩 전략 (CLAUDE.md 섹션 8.2 확장) 전략 3종: 1. ORB — Opening Range Breakout (09:00~09:30 고점 돌파) 2. VWAP — VWAP 지지 + 과매도 반등 3. RSI_BB — 5분봉 RSI+볼린저 과매도 반등 공통 규칙: - EOD 강제 청산: 15:20 이후 전량 시장가 매도 - 일일 손실 한도: 전체 자본 -3% 초과 시 신규 진입 중단 - 동시 최대 3종목, 종목당 최대 10% 자본 - 전체 데이트레이딩 자본: 총자산의 최대 25% (CLAUDE.md 8.2) [.env 설정] DAY_TRADE_MAX_CAPITAL_RATIO DAY_TRADE_MAX_RATIO_DT DAY_TRADE_MAX_POSITIONS (== MAX_DAY_TRADE_POSITIONS) DAY_TRADE_DAILY_LOSS_LIMIT DAY_TRADE_ORB_STOP_LOSS_PCT DAY_TRADE_REVERSION_STOP_LOSS_PCT DAY_TRADE_TRAILING_TRIGGER DAY_TRADE_TRAILING_ATR_MULT DAY_TRADE_ORB_BREAKOUT_MARGIN DAY_TRADE_COST_PCT DAY_TRADE_ENTRY_CUTOFF_HOUR, DAY_TRADE_ENTRY_CUTOFF_MIN DAY_TRADE_EOD_HOUR, DAY_TRADE_EOD_MIN """ import logging from typing import Optional, Dict, List, Tuple from dataclasses import dataclass from datetime import time from app.core.config import settings logger = logging.getLogger(__name__) # ── 장 시간 상수 (KST) ────────────────────────────────────────────────────── MARKET_OPEN = time(9, 0) ORB_END = time(9, 30) # Opening Range 집계 종료 # 동적 시간 설정 (settings에서 읽음) def _get_entry_cutoff() -> time: """신규 진입 마감 시간 (.env: DAY_TRADE_ENTRY_CUTOFF_HOUR, DAY_TRADE_ENTRY_CUTOFF_MIN)""" return time(settings.DAY_TRADE_ENTRY_CUTOFF_HOUR, settings.DAY_TRADE_ENTRY_CUTOFF_MIN) def _get_eod_flat() -> time: """EOD 강제 청산 시작 시간 (.env: DAY_TRADE_EOD_HOUR, DAY_TRADE_EOD_MIN)""" return time(settings.DAY_TRADE_EOD_HOUR, settings.DAY_TRADE_EOD_MIN) MARKET_CLOSE = time(15, 30) # ── 데이터 클래스 ───────────────────────────────────────────────────────────── ENTRY_CUTOFF = _get_entry_cutoff() EOD_FLAT = _get_eod_flat() MAX_POSITIONS = settings.DAY_TRADE_MAX_POSITIONS MAX_DAY_TRADE_RATIO = settings.DAY_TRADE_MAX_CAPITAL_RATIO MAX_CAPITAL_RATIO = settings.DAY_TRADE_MAX_RATIO_DT DAILY_LOSS_LIMIT = -settings.DAY_TRADE_DAILY_LOSS_LIMIT / 100 MARKET_CLOSE = time(15, 30) @dataclass class MinuteCandle: """KIS API output2 데이터 정규화 결과""" date: str # YYYYMMDD time: str # HHMMSS open: float high: float low: float close: float volume: int cum_volume: int # 누적 거래량 (acml_vol) cum_amount: float # 누적 거래대금 원 (acml_tr_pbmn) @dataclass class OpeningRange: high: float low: float total_volume: int avg_vol_per_min: float # 분당 평균 거래량 (ORB 기간) @dataclass class DayTradeSignal: stock_code: str stock_name: str strategy: str # "ORB" | "VWAP_BOUNCE" | "RSI_BB" entry_price: float stop_loss: float target_price: float reason: str confidence: float # 0.0 ~ 1.0 @dataclass class DayTradeExit: action: str # "SELL_ALL" reason: str sell_pct: float = 1.0 # ── 분봉 파싱 헬퍼 ──────────────────────────────────────────────────────────── def parse_minute_candles(raw_output2: list) -> List[MinuteCandle]: """KIS API output2 리스트를 MinuteCandle 리스트로 변환 KIS 응답 필드: stck_bsop_date / stck_cntg_hour / stck_oprc / stck_hgpr / stck_lwpr / stck_prpr (종가) / cntg_vol / acml_vol / acml_tr_pbmn """ candles: List[MinuteCandle] = [] for row in raw_output2: try: candles.append(MinuteCandle( date=row.get("stck_bsop_date", ""), time=row.get("stck_cntg_hour", ""), open=float(row.get("stck_oprc", 0) or 0), high=float(row.get("stck_hgpr", 0) or 0), low=float(row.get("stck_lwpr", 0) or 0), close=float(row.get("stck_prpr", 0) or 0), volume=int(row.get("cntg_vol", 0) or 0), cum_volume=int(row.get("acml_vol", 0) or 0), cum_amount=float(row.get("acml_tr_pbmn", 0) or 0), )) except (ValueError, TypeError): continue return candles def aggregate_to_5min(candles: List[MinuteCandle]) -> List[MinuteCandle]: """1분봉 → 5분봉 집계 (09:00 기준 5분 버킷)""" if not candles: return [] buckets: Dict[tuple, List[MinuteCandle]] = {} for c in candles: if not c.time: continue h = int(c.time[:2]) m = int(c.time[2:4]) key = (c.date, (h * 60 + m) // 5) # 날짜 포함 → 다른 날 캔들 혼합 방지 buckets.setdefault(key, []).append(c) result: List[MinuteCandle] = [] for key in sorted(buckets): bkt = buckets[key] result.append(MinuteCandle( date=bkt[0].date, time=bkt[0].time, open=bkt[0].open, high=max(b.high for b in bkt), low=min(b.low for b in bkt), close=bkt[-1].close, volume=sum(b.volume for b in bkt), cum_volume=bkt[-1].cum_volume, cum_amount=bkt[-1].cum_amount, )) return result # ── 지표 계산 ───────────────────────────────────────────────────────────────── def calculate_vwap(candles: List[MinuteCandle]) -> Optional[float]: """VWAP = 누적거래대금 / 누적거래량 (당일 09:00 기준 누적)""" if not candles: return None last = candles[-1] if last.cum_volume <= 0: return None return last.cum_amount / last.cum_volume def calculate_opening_range(candles_1min: List[MinuteCandle]) -> Optional[OpeningRange]: """09:00~09:29 분봉으로 Opening Range (H/L) 계산""" orb = [c for c in candles_1min if "090000" <= c.time <= "092959"] if len(orb) < 5: return None total_vol = sum(c.volume for c in orb) return OpeningRange( high=max(c.high for c in orb), low=min(c.low for c in orb), total_volume=total_vol, avg_vol_per_min=total_vol / len(orb), ) def _rsi(candles: List[MinuteCandle], period: int = 14) -> Optional[float]: closes = [c.close for c in candles] if len(closes) < period + 1: return None gains, losses = [], [] for i in range(-period, 0): d = closes[i] - closes[i - 1] gains.append(max(d, 0)) losses.append(max(-d, 0)) avg_gain = sum(gains) / period avg_loss = sum(losses) / period if avg_loss == 0: return 100.0 return 100 - (100 / (1 + avg_gain / avg_loss)) def _bollinger(candles: List[MinuteCandle], period: int = 20) -> Optional[Tuple[float, float, float]]: closes = [c.close for c in candles] if len(closes) < period: return None window = closes[-period:] mid = sum(window) / period std = (sum((p - mid) ** 2 for p in window) / period) ** 0.5 return (mid + 2 * std, mid, mid - 2 * std) def _atr(candles: List[MinuteCandle], period: int = 14) -> Optional[float]: if len(candles) < period + 1: return None trs = [] for i in range(1, len(candles)): h, l, cp = candles[i].high, candles[i].low, candles[i - 1].close trs.append(max(h - l, abs(h - cp), abs(l - cp))) if len(trs) < period: return None return sum(trs[-period:]) / period # ── 진입 신호 ───────────────────────────────────────────────────────────────── def check_orb_signal( stock_code: str, stock_name: str, candles_1min: List[MinuteCandle], orb: OpeningRange, current_price: float, ) -> Optional[DayTradeSignal]: """ORB 상향 돌파 매수 신호 조건: 1. 현재가 > ORB 고점 × 1.002 (0.2% 이상 돌파) 2. 최근 1분봉 거래량 ≥ ORB 분당평균 × 1.5 (거래량 폭발) 3. 직전 5분봉이 양봉 (음봉이면 Fake-out 위험) """ if not orb or current_price <= 0: return None if current_price < orb.high * (1 + settings.DAY_TRADE_ORB_BREAKOUT_MARGIN): return None # 최근 1분봉 거래량 체크 if not candles_1min: return None recent_vol = candles_1min[-1].volume if recent_vol < orb.avg_vol_per_min * 1.5: return None # 직전 5분봉 양봉 확인 (5분봉 2개 이상 없으면 보수적으로 차단) candles_5m = aggregate_to_5min(candles_1min) if len(candles_5m) < 2: return None # 데이터 부족 시 진입 차단 if candles_5m[-2].close < candles_5m[-2].open: return None # 직전 5분봉 음봉 → 스킵 stop_loss = max(orb.low, current_price * (1 - settings.DAY_TRADE_ORB_STOP_LOSS_PCT)) risk = current_price - stop_loss # 수수료 고려: 목표 = 진입가 + 2R + 수수료 보전 target = current_price + max(risk * 2.0, current_price * 0.03) vol_ratio = recent_vol / orb.avg_vol_per_min confidence = min(0.85, 0.65 + vol_ratio * 0.05) return DayTradeSignal( stock_code=stock_code, stock_name=stock_name, strategy="ORB", entry_price=current_price, stop_loss=round(stop_loss, 0), target_price=round(target, 0), reason=( f"ORB 상향 돌파: ORB고={orb.high:.0f}, " f"현재={current_price:.0f}, 거래량={vol_ratio:.1f}배" ), confidence=round(confidence, 2), ) def check_vwap_bounce_signal( stock_code: str, stock_name: str, candles_5min: List[MinuteCandle], vwap: float, current_price: float, ) -> Optional[DayTradeSignal]: """VWAP 지지 반등 매수 신호 조건: 1. 현재가가 VWAP의 -1% ~ 0.5% 구간 (VWAP 근처) 2. 5분봉 RSI(14) ≤ 40 3. 최근 봉 거래량 ≥ 직전 봉 × 1.3 4. 최근 5분봉 양봉 전환 (close > open) """ if not candles_5min or vwap <= 0 or current_price <= 0: return None if not (vwap * 0.99 <= current_price <= vwap * 1.005): return None rsi = _rsi(candles_5min, 14) if rsi is None or rsi > 40: return None if len(candles_5min) < 2: return None if candles_5min[-1].volume < candles_5min[-2].volume * 1.3: return None # 최근 봉 양봉 확인 last = candles_5min[-1] if last.close <= last.open: return None atr = _atr(candles_5min, 14) raw_stop = (current_price - 1.5 * atr) if atr else current_price * (1 - settings.DAY_TRADE_ORB_STOP_LOSS_PCT) stop_loss = max(raw_stop, current_price * (1 - settings.DAY_TRADE_REVERSION_STOP_LOSS_PCT), 1.0) # 하드 스탑 -2.5% (일일 손실 한도 -3% 이내) target = (current_price + 2.0 * atr) if atr else vwap * 1.03 return DayTradeSignal( stock_code=stock_code, stock_name=stock_name, strategy="VWAP_BOUNCE", entry_price=current_price, stop_loss=round(stop_loss, 0), target_price=round(target, 0), reason=( f"VWAP 반등: VWAP={vwap:.0f}, RSI={rsi:.1f}, " f"현재={current_price:.0f}" ), confidence=0.65, ) def check_rsi_bb_signal( stock_code: str, stock_name: str, candles_5min: List[MinuteCandle], current_price: float, ) -> Optional[DayTradeSignal]: """5분봉 RSI + 볼린저 과매도 반등 신호 조건: 1. RSI(14) ≤ 30 (강한 과매도) 2. 종가 ≤ BB 하단 3. 직전 봉 대비 거래량 ≥ 1.3배 4. 최근 봉 양봉 (음봉이면 스킵) """ if not candles_5min or current_price <= 0: return None rsi = _rsi(candles_5min, 14) if rsi is None or rsi > 30: return None bb = _bollinger(candles_5min, 20) if bb is None: return None bb_upper, bb_middle, bb_lower = bb if current_price > bb_lower * 1.01: return None if len(candles_5min) < 2: return None if candles_5min[-1].volume < candles_5min[-2].volume * 1.3: return None last = candles_5min[-1] if last.close <= last.open: return None atr = _atr(candles_5min, 14) raw_stop = (current_price - 1.5 * atr) if atr else current_price * (1 - settings.DAY_TRADE_ORB_STOP_LOSS_PCT) stop_loss = max(raw_stop, current_price * (1 - settings.DAY_TRADE_REVERSION_STOP_LOSS_PCT), 1.0) # 하드 스탑 -2.5% (일일 손실 한도 -3% 이내) # 최소 1.5R 보장: bb_middle이 너무 가까우면 R:R 확보를 위해 목표 상향 risk = current_price - stop_loss target = max(bb_middle, current_price + risk * 1.5) if target <= current_price: # 방어 로직 (stop_loss ≥ current_price 극단 케이스) return None return DayTradeSignal( stock_code=stock_code, stock_name=stock_name, strategy="RSI_BB", entry_price=current_price, stop_loss=round(stop_loss, 0), target_price=round(target, 0), reason=( f"5분봉 과매도 반등: RSI={rsi:.1f}, " f"BB하단={bb_lower:.0f}, 현재={current_price:.0f}" ), confidence=0.60, ) # ── 청산 신호 ───────────────────────────────────────────────────────────────── def check_day_trade_exit( entry_price: float, current_price: float, stop_loss: float, target_price: float, highest_since_entry: float, candles_5min: List[MinuteCandle], current_time: time, ) -> Optional[DayTradeExit]: """데이트레이딩 청산 조건 체크 (우선순위 순) 1. EOD 강제 청산 (15:20 이후) 2. 손절 (현재가 ≤ stop_loss) 3. 목표가 달성 (현재가 ≥ target_price) 4. 트레일링 스탑 (1.5% 수익 후 ATR 트레일) 5. 5분봉 RSI ≥ 70 (과매수 청산) """ # 1. EOD 강제 청산 if current_time >= _get_eod_flat(): return DayTradeExit( action="SELL_ALL", reason=f"EOD 강제 청산 ({current_time.strftime('%H:%M')} ≥ {_get_eod_flat().strftime('%H:%M')})", ) # 2. 손절 if current_price <= stop_loss: loss_pct = (current_price - entry_price) / entry_price * 100 return DayTradeExit( action="SELL_ALL", reason=f"손절 {loss_pct:.1f}% (현재={current_price:.0f} ≤ 손절가={stop_loss:.0f})", ) # 3. 목표가 달성 if current_price >= target_price: gain_pct = (current_price - entry_price) / entry_price * 100 return DayTradeExit( action="SELL_ALL", reason=f"목표가 달성 +{gain_pct:.1f}% (현재={current_price:.0f} ≥ 목표={target_price:.0f})", ) # 4. 트레일링 스탑 (1.5% 이상 수익 시 발동) profit_pct = (highest_since_entry - entry_price) / entry_price if profit_pct >= settings.DAY_TRADE_TRAILING_TRIGGER: atr = _atr(candles_5min, 14) if atr: trailing_stop = highest_since_entry - settings.DAY_TRADE_TRAILING_ATR_MULT * atr else: trailing_stop = highest_since_entry * (1 - 0.02) # fallback: 최고가 -2% (5분봉 데이터 부족 시) if current_price <= trailing_stop: return DayTradeExit( action="SELL_ALL", reason=( f"트레일링 스탑: 최고={highest_since_entry:.0f}, " f"트레일={trailing_stop:.0f}, 현재={current_price:.0f}" ), ) # 5. 과매수 청산 (RSI ≥ 70) rsi = _rsi(candles_5min, 14) if rsi is not None and rsi >= 70: gain_pct = (current_price - entry_price) / entry_price * 100 if gain_pct > settings.DAY_TRADE_COST_PCT * 100: # 수수료 이상 수익인 경우만 return DayTradeExit( action="SELL_ALL", reason=f"과매수 청산: RSI={rsi:.1f}≥70, 수익={gain_pct:.1f}%", ) return None # ── 포지션 사이징 ───────────────────────────────────────────────────────────── def calculate_day_trade_position_size( total_assets: float, available_cash: float, current_day_trade_amount: float, ) -> float: """데이트레이딩 포지션 크기 계산 - 종목당 최대: 총자산 × 10% - 전체 데이트레이딩 합산: 총자산 × 25% (CLAUDE.md 8.2) - 동시 최대 3종목 (MAX_POSITIONS) """ max_total = total_assets * MAX_DAY_TRADE_RATIO remaining_budget = max_total - current_day_trade_amount per_position_max = total_assets * MAX_CAPITAL_RATIO return max(0.0, min(per_position_max, remaining_budget, available_cash)) def is_daily_loss_exceeded( total_assets: float, realized_pnl_today: float, ) -> bool: """일일 손실 한도 초과 여부 (전체 자본 -3%)""" if total_assets <= 0: return False return (realized_pnl_today / total_assets) <= DAILY_LOSS_LIMIT # 동적 시간 설정 (settings에서 읽음) def _get_entry_cutoff() -> time: """신규 진입 마감 시간 (.env: DAY_TRADE_ENTRY_CUTOFF_HOUR, DAY_TRADE_ENTRY_CUTOFF_MIN)""" return time(settings.DAY_TRADE_ENTRY_CUTOFF_HOUR, settings.DAY_TRADE_ENTRY_CUTOFF_MIN) def _get_eod_flat() -> time: """EOD 강제 청산 시작 시간 (.env: DAY_TRADE_EOD_HOUR, DAY_TRADE_EOD_MIN)""" return time(settings.DAY_TRADE_EOD_HOUR, settings.DAY_TRADE_EOD_MIN)