refactor(phase2): Add DayTradingStrategy example and indicators module
This commit is contained in:
158
backend/app/strategy/indicators.py
Normal file
158
backend/app/strategy/indicators.py
Normal file
@@ -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
|
||||
11
backend/app/strategy/strategies/__init__.py
Normal file
11
backend/app/strategy/strategies/__init__.py
Normal file
@@ -0,0 +1,11 @@
|
||||
"""
|
||||
Trading Strategies
|
||||
|
||||
구체적인 매매 전략 구현체
|
||||
"""
|
||||
|
||||
from app.strategy.strategies.day_trading import DayTradingStrategy
|
||||
|
||||
__all__ = [
|
||||
"DayTradingStrategy",
|
||||
]
|
||||
181
backend/app/strategy/strategies/day_trading.py
Normal file
181
backend/app/strategy/strategies/day_trading.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user