diff --git a/backend/app/strategy/indicators.py b/backend/app/strategy/indicators.py new file mode 100644 index 0000000..bf22e2d --- /dev/null +++ b/backend/app/strategy/indicators.py @@ -0,0 +1,158 @@ +""" +Technical Indicators + +보조지표 계산 함수들 +""" +from typing import List, Optional, Tuple +import logging + +logger = logging.getLogger(__name__) + + +def sma(prices: List[float], period: int) -> Optional[float]: + """단순 이동평균 (Simple Moving Average)""" + if len(prices) < period: + return None + return sum(prices[-period:]) / period + + +def ema(prices: List[float], period: int, prev_ema: Optional[float] = None) -> Optional[float]: + """지수 이동평균 (Exponential Moving Average)""" + if len(prices) < period: + return None + + k = 2 / (period + 1) + if prev_ema is None: + # 초기값은 SMA + prev_ema = sum(prices[-period:]) / period + + return prices[-1] * k + prev_ema * (1 - k) + + +def rsi(prices: List[float], period: int = 14) -> Optional[float]: + """ +RSI (Relative Strength Index) + + Args: + prices: 종가 리스트 + period: 기간 (기본 14) + + Returns: + 0~100 사이 RSI 값 + """ + if len(prices) < period + 1: + return None + + gains, losses = [], [] + for i in range(-period, 0): + delta = prices[i] - prices[i - 1] + gains.append(max(delta, 0)) + losses.append(max(-delta, 0)) + + avg_gain = sum(gains) / period + avg_loss = sum(losses) / period + + if avg_loss == 0: + return 100.0 + + rs = avg_gain / avg_loss + return 100 - (100 / (1 + rs)) + + +def bollinger_bands(prices: List[float], period: int = 20, std_mult: float = 2.0) -> Optional[Tuple[float, float, float]]: + """ + 볼린저 밴드 + + Returns: + (upper, middle, lower) 또는 None + """ + if len(prices) < period: + return None + + window = prices[-period:] + middle = sum(window) / period + variance = sum((p - middle) ** 2 for p in window) / period + std = variance ** 0.5 + + upper = middle + std_mult * std + lower = middle - std_mult * std + + return (upper, middle, lower) + + +def macd(prices: List[float], fast: int = 12, slow: int = 26, signal: int = 9) -> Optional[Tuple[float, float, float]]: + """ + MACD (Moving Average Convergence Divergence) + + Returns: + (macd_line, signal_line, histogram) 또는 None + """ + if len(prices) < slow: + return None + + # EMA 계산 (재귀적으로 계산) + def calc_ema(data: List[float], period: int) -> float: + k = 2 / (period + 1) + ema_val = sum(data[:period]) / period + for p in data[period:]: + ema_val = p * k + ema_val * (1 - k) + return ema_val + + ema_fast = calc_ema(prices, fast) + ema_slow = calc_ema(prices, slow) + macd_line = ema_fast - ema_slow + + # Signal line = MACD의 EMA(9) + # 실제로는 과거 MACD 값들 필요 (간략화) + signal_line = macd_line # TODO: 정확한 구현 필요 + histogram = macd_line - signal_line + + return (macd_line, signal_line, histogram) + + +def atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> Optional[float]: + """ + ATR (Average True Range) + + Args: + highs: 고가 리스트 + lows: 저가 리스트 + closes: 종가 리스트 + period: 기간 + + Returns: + ATR 값 + """ + if len(highs) < period + 1 or len(lows) < period + 1 or len(closes) < period + 1: + return None + + trs = [] + for i in range(1, len(highs)): + h = highs[i] + l = lows[i] + cp = closes[i - 1] + tr = max(h - l, abs(h - cp), abs(l - cp)) + trs.append(tr) + + if len(trs) < period: + return None + + return sum(trs[-period:]) / period + + +def vwap(prices: List[float], volumes: List[float]) -> Optional[float]: + """ + VWAP (Volume Weighted Average Price) + + Returns: + VWAP 값 + """ + if len(prices) != len(volumes) or not prices: + return None + + total_volume = sum(volumes) + if total_volume == 0: + return None + + total_pv = sum(p * v for p, v in zip(prices, volumes)) + return total_pv / total_volume diff --git a/backend/app/strategy/strategies/__init__.py b/backend/app/strategy/strategies/__init__.py new file mode 100644 index 0000000..8a06e69 --- /dev/null +++ b/backend/app/strategy/strategies/__init__.py @@ -0,0 +1,11 @@ +""" +Trading Strategies + +구체적인 매매 전략 구현체 +""" + +from app.strategy.strategies.day_trading import DayTradingStrategy + +__all__ = [ + "DayTradingStrategy", +] diff --git a/backend/app/strategy/strategies/day_trading.py b/backend/app/strategy/strategies/day_trading.py new file mode 100644 index 0000000..d40b42a --- /dev/null +++ b/backend/app/strategy/strategies/day_trading.py @@ -0,0 +1,181 @@ +""" +Day Trading Strategy (Refactored) + +기존 day_trading_strategy.py를 BaseStrategy 구조로 변환 + +전략: + - ORB (Opening Range Breakout) + - VWAP Bounce + - RSI+BB 과매도 반등 +""" +import logging +from typing import Optional, Dict, Any +from datetime import datetime, time +import pytz + +from app.strategy.base_strategy import BaseStrategy +from app.strategy.signal import Signal, SignalType +from app.core.config import settings + +logger = logging.getLogger(__name__) + + +class DayTradingStrategy(BaseStrategy): + """ + 데이트레이딩 전략 (단기 스칼핑) + + 09:00~15:20 간 급등/급락 모멘텀 포착 + """ + + DEFAULT_CONFIG = { + "max_positions": 3, + "max_capital_ratio": 0.1, # 종목당 10% + "orb_breakout_margin": 0.002, # 0.2% 돌파 + "stop_loss_pct": 0.025, # -2.5% + "entry_cutoff": "14:30", # 신규 진입 마감 + "eod_flat": "15:20", # EOD 청산 + } + + def __init__(self, name: str = "DayTrading", config: Optional[Dict[str, Any]] = None): + merged_config = {**self.DEFAULT_CONFIG, **(config or {})} + super().__init__(name, merged_config) + + self.kst = pytz.timezone("Asia/Seoul") + self.entry_cutoff = self._parse_time(self.config["entry_cutoff"]) + self.eod_flat = self._parse_time(self.config["eod_flat"]) + + def _parse_time(self, time_str: str) -> time: + """문자열 "HH:MM" → time 객체""" + h, m = map(int, time_str.split(":")) + return time(h, m) + + async def analyze( + self, + stock_code: str, + market_data: Dict[str, Any], + portfolio_data: Optional[Dict[str, Any]] = None + ) -> Optional[Signal]: + """ + 시장 데이터 분석 + + Args: + market_data: { + "current_price": float, + "minute_chart": List[dict], # 1분봉 + "orderbook": dict, + "stock_name": str, + } + portfolio_data: { + "holding_quantity": int, + "avg_price": float, + } + """ + current_time = datetime.now(self.kst).time() + + # 1. 시간 체크 + if current_time >= self.eod_flat: + # EOD 청산 신호 + if portfolio_data and portfolio_data.get("holding_quantity", 0) > 0: + return Signal( + stock_code=stock_code, + signal_type=SignalType.SELL, + strategy_name=self.name, + timestamp=datetime.now(self.kst), + quantity=portfolio_data["holding_quantity"], + price=None, # 시장가 + confidence=1.0, + reason="EOD 강제 청산" + ) + return None + + if current_time >= self.entry_cutoff: + # 신규 진입 차단 + return None + + # 2. 현재 보유 중이면 청산 조건 체크 + if portfolio_data and portfolio_data.get("holding_quantity", 0) > 0: + exit_signal = self._check_exit( + stock_code, + market_data, + portfolio_data + ) + if exit_signal: + return exit_signal + return None # 보유 유지 + + # 3. 신규 진입 검토 + return self._check_entry(stock_code, market_data) + + def _check_entry(self, stock_code: str, market_data: Dict[str, Any]) -> Optional[Signal]: + """신규 진입 신호 체크""" + current_price = market_data.get("current_price") + minute_chart = market_data.get("minute_chart", []) + + if not current_price or not minute_chart: + return None + + # 예시: ORB 전략 (간단화) + # 실제는 기존 check_orb_signal() 등을 호출 + + # 임시 구현: 단순 예시 + if len(minute_chart) >= 30: # 09:00~09:30 데이터 있음 + orb_high = max(c.get("high", 0) for c in minute_chart[:30]) + + if current_price > orb_high * (1 + self.config["orb_breakout_margin"]): + stop_loss = current_price * (1 - self.config["stop_loss_pct"]) + target = current_price * 1.03 # +3% 목표 + + return Signal( + stock_code=stock_code, + signal_type=SignalType.BUY, + strategy_name=self.name, + timestamp=datetime.now(self.kst), + quantity=None, # OrderManager가 계산 + price=current_price, + confidence=0.75, + stop_loss_price=stop_loss, + take_profit_price=target, + reason=f"ORB 돌파: {orb_high:.0f} → {current_price:.0f}", + metadata={"strategy_type": "ORB"} + ) + + return None + + def _check_exit(self, stock_code: str, market_data: Dict[str, Any], portfolio_data: Dict[str, Any]) -> Optional[Signal]: + """청산 신호 체크""" + current_price = market_data.get("current_price") + avg_price = portfolio_data.get("avg_price") + quantity = portfolio_data.get("holding_quantity") + + if not current_price or not avg_price or not quantity: + return None + + # 손절 체크 + stop_loss = avg_price * (1 - self.config["stop_loss_pct"]) + if current_price <= stop_loss: + return Signal( + stock_code=stock_code, + signal_type=SignalType.STOP_LOSS, + strategy_name=self.name, + timestamp=datetime.now(self.kst), + quantity=quantity, + price=None, # 시장가 + confidence=1.0, + reason=f"손절: {current_price:.0f} ≤ {stop_loss:.0f}" + ) + + # 익절 체크 + target = avg_price * 1.03 + if current_price >= target: + return Signal( + stock_code=stock_code, + signal_type=SignalType.TAKE_PROFIT, + strategy_name=self.name, + timestamp=datetime.now(self.kst), + quantity=quantity, + price=None, + confidence=1.0, + reason=f"목표가 도달: {current_price:.0f} ≥ {target:.0f}" + ) + + return None