494 lines
17 KiB
Python
494 lines
17 KiB
Python
"""
|
||
데이트레이딩 전략 (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)
|
||
|
||
|
||
# ── 데이터 클래스 ─────────────────────────────────────────────────────────────
|
||
|
||
@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
|