diff --git a/backend/app/strategy/README.md b/backend/app/strategy/README.md new file mode 100644 index 0000000..30cffb6 --- /dev/null +++ b/backend/app/strategy/README.md @@ -0,0 +1,288 @@ +# Strategy Module + +매매 전략을 관리하고 실행하는 핵심 모듈 + +## 모듈 구조 + +``` +strategy/ +├── __init__.py # 모듈 내보내기 +├── base_strategy.py # BaseStrategy 추상 클래스 +├── signal.py # Signal 데이터 클래스 +├── strategy_manager.py # 전략 관리자 (등록/실행) +├── indicators.py # 보조지표 (RSI, BB, MACD 등) +├── scanner.py # 종목 스캔 +├── strategies/ # 구체적 전략 구현 +│ ├── __init__.py +│ └── day_trading.py # 데이트레이딩 전략 +└── README.md +``` + +## 사용 방법 + +### 1. 전략 구현 + +`BaseStrategy`를 상속받아 `analyze()` 메서드 구현: + +```python +from app.strategy import BaseStrategy, Signal, SignalType +from datetime import datetime +import pytz + +class MyStrategy(BaseStrategy): + def __init__(self): + super().__init__(name="MyStrategy", config={"param1": 10}) + + async def analyze(self, stock_code, market_data, portfolio_data=None): + current_price = market_data.get("current_price") + + # 매수 조건 체크 + if self._should_buy(market_data): + return Signal( + stock_code=stock_code, + signal_type=SignalType.BUY, + strategy_name=self.name, + timestamp=datetime.now(pytz.timezone("Asia/Seoul")), + price=current_price, + confidence=0.8, + reason="매수 조건 충족" + ) + + return None + + def _should_buy(self, market_data): + # 실제 로직 구현 + return True +``` + +### 2. 전략 등록 및 실행 + +```python +from app.strategy import strategy_manager +from app.strategy.strategies import DayTradingStrategy + +# 전략 인스턴스 생성 +day_trading = DayTradingStrategy( + name="DayTrading_ORB", + config={ + "max_positions": 3, + "stop_loss_pct": 0.025 + } +) + +# 전략 등록 (priority 낮을수록 먼저 실행) +strategy_manager.register_strategy(day_trading, priority=10) + +# 단일 전략 실행 +signal = await strategy_manager.run_strategy( + "DayTrading_ORB", + stock_code="005930", + market_data={ + "current_price": 75000, + "minute_chart": [...], + } +) + +if signal: + print(f"신호 발생: {signal}") + +# 모든 전략 동시 실행 +signals = await strategy_manager.run_all_strategies( + stock_code="005930", + market_data={...}, + parallel=True # 병렬 실행 +) + +# 신호 충돌 해결 +final_signal = strategy_manager.resolve_conflicts( + signals, + method="highest_confidence" # or "priority", "majority_vote" +) +``` + +### 3. Signal 데이터 구조 + +```python +from app.strategy import Signal, SignalType +from datetime import datetime +import pytz + +signal = Signal( + stock_code="005930", + signal_type=SignalType.BUY, + strategy_name="MyStrategy", + timestamp=datetime.now(pytz.timezone("Asia/Seoul")), + + # 주문 정보 + quantity=10, + price=75000, # None = 시장가 + confidence=0.85, + + # 손익 관리 + stop_loss_price=73000, + take_profit_price=78000, + trailing_stop_pct=0.02, # 2% 트레일링 + + # 메타데이터 + reason="RSI 과매도 반등", + metadata={"rsi": 25, "bb_lower": 74000} +) + +# JSON 변환 +data = signal.to_dict() +``` + +### 4. 보조지표 사용 + +```python +from app.strategy.indicators import rsi, bollinger_bands, sma, macd, atr + +# 종가 리스트 +closes = [75000, 75500, 76000, 74500, 75000, ...] + +# RSI +rsi_value = rsi(closes, period=14) +if rsi_value and rsi_value < 30: + print("과매도 구간") + +# 볼린저 밴드 +bb = bollinger_bands(closes, period=20, std_mult=2.0) +if bb: + upper, middle, lower = bb + if closes[-1] < lower: + print("하단 돌파") + +# SMA +sma_20 = sma(closes, 20) +sma_60 = sma(closes, 60) +if sma_20 and sma_60 and sma_20 > sma_60: + print("골든 크로스 발생") + +# ATR +highs = [76000, 76500, ...] +lows = [74500, 75000, ...] +atr_value = atr(highs, lows, closes, period=14) +``` + +### 5. 종목 스캔 + +```python +from app.strategy.scanner import Scanner, filter_price_range, filter_volume_min + +# 스캀너 생성 +scanner = Scanner() + +# 필터 추가 +scanner.add_filter(filter_price_range(10000, 100000)) # 1만~10만원 +scanner.add_filter(filter_volume_min(100000)) # 최소 거래량 10만주 + +# 커스텀 필터 +def custom_filter(data): + # 등락률 5% 이상 + return data.get("change_rate", 0) >= 5.0 + +scanner.add_filter(custom_filter) + +# 스캔 실행 +kospi_codes = ["005930", "000660", "035420", ...] # KOSPI 종목 리스트 +matched = await scanner.scan(kospi_codes, max_concurrent=10) + +for stock in matched: + print(f"{stock['stock_name']} ({stock['stock_code']}): " + f"{stock['current_price']}원, +{stock['change_rate']}%") +``` + +## 주요 기능 + +### BaseStrategy +- ✅ 전략 추상화 (템플릿 메서드 패턴) +- ✅ `analyze()` 구현 강제 +- ✅ `on_signal_executed()` 콜백 +- ✅ `on_error()` 에러 핸들링 +- ✅ 설정 관리 (config) +- ✅ 활성화/비활성화 + +### StrategyManager +- ✅ 전략 등록/삭제 +- ✅ 우선순위 관리 +- ✅ 단일/복수 전략 실행 +- ✅ 병렬 실행 지원 +- ✅ 신호 충돌 해결 + - `highest_confidence`: 가장 높은 신뢰도 + - `priority`: 전략 우선순위 + - `majority_vote`: 다수결 +- ✅ 신호 이력 관리 +- ✅ 통계 정보 + +### Signal +- ✅ 표준화된 신호 데이터 구조 +- ✅ 6가지 신호 타입 (BUY/SELL/HOLD/CLOSE/STOP_LOSS/TAKE_PROFIT) +- ✅ 손익 관리 필드 +- ✅ 신뢰도 평가 +- ✅ JSON 직렬화 + +### Indicators +- ✅ SMA (Simple Moving Average) +- ✅ EMA (Exponential Moving Average) +- ✅ RSI (Relative Strength Index) +- ✅ Bollinger Bands +- ✅ MACD +- ✅ ATR (Average True Range) +- ✅ VWAP (Volume Weighted Average Price) + +### Scanner +- ✅ 조건 기반 종목 필터링 +- ✅ 동시 다중 종목 스캔 +- ✅ 사전 정의 필터 + - 가격 범위 + - 거래량 최소값 + - 등락률 범위 + - 시가총액 최소값 +- ✅ 커스텀 필터 추가 가능 + +## 기존 코드 마이그레이션 + +기존 `services/` 폴더의 전략 파일들을 새 구조로 변환: + +| 기존 파일 | 새 위치 | +|----------|----------| +| `day_trading_strategy.py` | `strategy/strategies/day_trading.py` | +| `swing_trading_strategy.py` | `strategy/strategies/swing_trading.py` | +| `momentum_strategy.py` | `strategy/strategies/momentum.py` | + +변환 패턴: +```python +# 기존 +def check_signal(stock_code, market_data): + if condition: + return DayTradeSignal(...) + return None + +# 새 구조 +class DayTradingStrategy(BaseStrategy): + async def analyze(self, stock_code, market_data, portfolio_data=None): + if condition: + return Signal(...) + return None +``` + +## Phase 2 완료 체크리스트 + +- [x] `strategy/base_strategy.py` 생성 +- [x] `strategy/signal.py` 생성 +- [x] `strategy/strategy_manager.py` 생성 +- [x] `strategy/indicators.py` 생성 +- [x] `strategy/scanner.py` 생성 +- [x] `strategy/strategies/day_trading.py` 예시 생성 +- [x] README 문서 작성 + +## 다음 단계 (Phase 3) + +- [ ] WebSocket 모듈 분리 (`realtime/`) +- [ ] 실시간 시세 수신 구조 개선 +- [ ] 장 스케줄 관리 + +--- + +**작성일**: 2026-03-06 +**작성자**: 서영우 diff --git a/backend/app/strategy/__init__.py b/backend/app/strategy/__init__.py new file mode 100644 index 0000000..5714e8f --- /dev/null +++ b/backend/app/strategy/__init__.py @@ -0,0 +1,25 @@ +""" +Strategy Engine Module + +매매 전략을 관리하고 실행하는 핵심 모듈 +- 전략 추상화 (BaseStrategy) +- 신호 생성 (Signal) +- 전략 등록·실행·관리 (StrategyManager) +- 보조지표 (Indicators) +- 종목 스캔 (Scanner) +""" + +from app.strategy.base_strategy import BaseStrategy +from app.strategy.signal import Signal, SignalType +from app.strategy.strategy_manager import StrategyManager, strategy_manager + +__all__ = [ + # Core classes + "BaseStrategy", + "Signal", + "SignalType", + "StrategyManager", + + # Singleton + "strategy_manager", +] diff --git a/backend/app/strategy/base_strategy.py b/backend/app/strategy/base_strategy.py new file mode 100644 index 0000000..e567e64 --- /dev/null +++ b/backend/app/strategy/base_strategy.py @@ -0,0 +1,101 @@ +""" +Base Strategy Abstract Class + +모든 전략이 상속받아야 하는 추상 클래스 +""" +from abc import ABC, abstractmethod +from typing import List, Dict, Any, Optional +from datetime import datetime +import logging + +from app.strategy.signal import Signal + +logger = logging.getLogger(__name__) + + +class BaseStrategy(ABC): + """ + 전략 기본 추상 클래스 + + 모든 전략은 이 클래스를 상속받아 analyze() 메서드를 구현해야 함 + """ + + def __init__(self, name: str, config: Optional[Dict[str, Any]] = None): + """ + Args: + name: 전략 이름 (고유 식별자) + config: 전략 설정 딕셔너리 + """ + self.name = name + self.config = config or {} + self.enabled = True + self.last_signals: Dict[str, Signal] = {} # 종목별 마지막 신호 + + logger.info(f"📊 전략 초기화: {self.name}") + + @abstractmethod + async def analyze( + self, + stock_code: str, + market_data: Dict[str, Any], + portfolio_data: Optional[Dict[str, Any]] = None + ) -> Optional[Signal]: + """ + 시장 데이터를 분석하여 매매 신호 생성 + + Args: + stock_code: 종목코드 + market_data: 시장 데이터 (현재가, 차트, 호가 등) + portfolio_data: 포트폴리오 데이터 (보유 수량, 평단가 등) + + Returns: + Signal 객체 또는 None (신호 없음) + """ + pass + + async def on_signal_executed(self, signal: Signal, result: Dict[str, Any]): + """ + 신호 실행 후 콜백 + + Args: + signal: 실행된 신호 + result: 실행 결과 (주문 응답) + """ + logger.info(f"✅ [{self.name}] 신호 실행 완료: {signal.signal_type.value} {signal.stock_code}") + self.last_signals[signal.stock_code] = signal + + async def on_error(self, error: Exception, context: Dict[str, Any]): + """ + 에러 발생 시 콜백 + + Args: + error: 발생한 예외 + context: 에러 컨텍스트 + """ + logger.error(f"❌ [{self.name}] 에러 발생: {error}", exc_info=True) + + def get_config(self, key: str, default: Any = None) -> Any: + """설정값 조회""" + return self.config.get(key, default) + + def update_config(self, config: Dict[str, Any]): + """설정 업데이트""" + self.config.update(config) + logger.info(f"🔧 [{self.name}] 설정 업데이트: {config}") + + def enable(self): + """전략 활성화""" + self.enabled = True + logger.info(f"✅ [{self.name}] 활성화") + + def disable(self): + """전략 비활성화""" + self.enabled = False + logger.info(f"⏸️ [{self.name}] 비활성화") + + def get_last_signal(self, stock_code: str) -> Optional[Signal]: + """종목의 마지막 신호 조회""" + return self.last_signals.get(stock_code) + + def __repr__(self) -> str: + return f"<{self.__class__.__name__} name={self.name} enabled={self.enabled}>" 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/scanner.py b/backend/app/strategy/scanner.py new file mode 100644 index 0000000..25e1d7f --- /dev/null +++ b/backend/app/strategy/scanner.py @@ -0,0 +1,119 @@ +""" +Stock Scanner + +시장에서 특정 조건에 맞는 종목을 스캔하는 모듈 +""" +from typing import List, Dict, Any, Optional, Callable +import logging +import asyncio + +from app.trading import market_data_provider + +logger = logging.getLogger(__name__) + + +class Scanner: + """ + 종목 스캔 엔진 + + 특정 조건을 만족하는 종목을 찾는다. + """ + + def __init__(self): + self.filters: List[Callable] = [] + + def add_filter(self, filter_func: Callable[[Dict[str, Any]], bool]): + """ + 필터 추가 + + Args: + filter_func: market_data를 받아 True/False 반환하는 함수 + """ + self.filters.append(filter_func) + + async def scan( + self, + stock_codes: List[str], + max_concurrent: int = 5 + ) -> List[Dict[str, Any]]: + """ + 종목 목록을 스캔하여 조건에 맞는 종목 반환 + + Args: + stock_codes: 스캔할 종목코드 목록 + max_concurrent: 동시 처리 수 + + Returns: + 조건에 맞는 종목 데이터 목록 + """ + semaphore = asyncio.Semaphore(max_concurrent) + + async def scan_one(stock_code: str) -> Optional[Dict[str, Any]]: + async with semaphore: + try: + # 시세 데이터 수집 + price_data = await market_data_provider.get_current_price(stock_code) + + if not price_data or price_data.get("rt_cd") != "0": + return None + + output = price_data.get("output", {}) + market_data = { + "stock_code": stock_code, + "stock_name": output.get("hts_kor_isnm", ""), + "current_price": float(output.get("stck_prpr", 0)), + "change_rate": float(output.get("prdy_ctrt", 0)), + "volume": int(output.get("acml_vol", 0)), + "market_cap": int(output.get("hts_avls", 0)) * 100000000, # 억 → 원 + } + + # 필터 적용 + for filter_func in self.filters: + if not filter_func(market_data): + return None + + return market_data + + except Exception as e: + logger.warning(f"⚠️ 스캔 실패 {stock_code}: {e}") + return None + + tasks = [scan_one(code) for code in stock_codes] + results = await asyncio.gather(*tasks) + + matched = [r for r in results if r is not None] + logger.info(f"🔍 스캔 완료: {len(matched)}/{len(stock_codes)} 종목 매칭") + + return matched + + +# 사전 정의 필터 + +def filter_price_range(min_price: float, max_price: float) -> Callable: + """가격 범위 필터""" + def _filter(data: Dict[str, Any]) -> bool: + price = data.get("current_price", 0) + return min_price <= price <= max_price + return _filter + + +def filter_volume_min(min_volume: int) -> Callable: + """최소 거래량 필터""" + def _filter(data: Dict[str, Any]) -> bool: + return data.get("volume", 0) >= min_volume + return _filter + + +def filter_change_rate(min_rate: float, max_rate: float) -> Callable: + """등락률 범위 필터""" + def _filter(data: Dict[str, Any]) -> bool: + rate = data.get("change_rate", 0) + return min_rate <= rate <= max_rate + return _filter + + +def filter_market_cap_min(min_cap: int) -> Callable: + """최소 시가총액 필터 (원)""" + def _filter(data: Dict[str, Any]) -> bool: + return data.get("market_cap", 0) >= min_cap + return _filter diff --git a/backend/app/strategy/signal.py b/backend/app/strategy/signal.py new file mode 100644 index 0000000..f799556 --- /dev/null +++ b/backend/app/strategy/signal.py @@ -0,0 +1,79 @@ +""" +Signal Data Class + +전략이 생성하는 매매 신호 데이터 구조 +""" +from enum import Enum +from dataclasses import dataclass +from typing import Optional, Dict, Any +from datetime import datetime + + +class SignalType(Enum): + """신호 타입""" + BUY = "BUY" # 매수 + SELL = "SELL" # 매도 + HOLD = "HOLD" # 보유 + CLOSE = "CLOSE" # 청산 + STOP_LOSS = "STOP_LOSS" # 손절 + TAKE_PROFIT = "TAKE_PROFIT" # 익절 + + +@dataclass +class Signal: + """ + 매매 신호 데이터 클래스 + + 전략이 생성한 매매 신호를 표준화된 형식으로 전달 + """ + # 필수 필드 + stock_code: str # 종목코드 + signal_type: SignalType # 신호 타입 + strategy_name: str # 전략 이름 + timestamp: datetime # 신호 생성 시각 + + # 주문 정보 + quantity: Optional[int] = None # 수량 + price: Optional[float] = None # 가격 (None = 시장가) + confidence: float = 1.0 # 신뢰도 (0.0 ~ 1.0) + + # 손익 관리 + stop_loss_price: Optional[float] = None # 손절가 + take_profit_price: Optional[float] = None # 익절가 + trailing_stop_pct: Optional[float] = None # 트레일링 스탑 % + + # 메타데이터 + reason: Optional[str] = None # 신호 발생 이유 + metadata: Optional[Dict[str, Any]] = None # 추가 데이터 + + def __post_init__(self): + """유효성 검증""" + if self.confidence < 0.0 or self.confidence > 1.0: + raise ValueError(f"Confidence must be 0.0~1.0, got {self.confidence}") + + if isinstance(self.signal_type, str): + self.signal_type = SignalType(self.signal_type) + + def to_dict(self) -> Dict[str, Any]: + """딕셔너리 변환 (DB 저장용)""" + return { + "stock_code": self.stock_code, + "signal_type": self.signal_type.value, + "strategy_name": self.strategy_name, + "timestamp": self.timestamp.isoformat(), + "quantity": self.quantity, + "price": self.price, + "confidence": self.confidence, + "stop_loss_price": self.stop_loss_price, + "take_profit_price": self.take_profit_price, + "trailing_stop_pct": self.trailing_stop_pct, + "reason": self.reason, + "metadata": self.metadata, + } + + def __repr__(self) -> str: + return ( + f"Signal({self.signal_type.value} {self.stock_code} " + f"@{self.price or 'MARKET'} x{self.quantity} " + f"[{self.strategy_name}] conf={self.confidence:.2f})" + ) 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 diff --git a/backend/app/strategy/strategy_manager.py b/backend/app/strategy/strategy_manager.py new file mode 100644 index 0000000..21ed21c --- /dev/null +++ b/backend/app/strategy/strategy_manager.py @@ -0,0 +1,253 @@ +""" +Strategy Manager + +전략 등록·실행·관리를 담당하는 오케스트레이터 +""" +from typing import Dict, List, Optional, Any +import logging +import asyncio +from datetime import datetime + +from app.strategy.base_strategy import BaseStrategy +from app.strategy.signal import Signal, SignalType + +logger = logging.getLogger(__name__) + + +class StrategyManager: + """ + 전략 관리자 + + - 전략 등록/삭제 + - 복수 전략 동시 실행 + - 신호 충돌 해결 + - 전략 우선순위 관리 + """ + + def __init__(self): + self.strategies: Dict[str, BaseStrategy] = {} # 전략 등록부 + self.priority_order: List[str] = [] # 우선순위 + self.signal_history: List[Signal] = [] # 신호 이력 + + logger.info("🎯 StrategyManager 초기화") + + def register_strategy(self, strategy: BaseStrategy, priority: int = 100): + """ + 전략 등록 + + Args: + strategy: 등록할 전략 인스턴스 + priority: 우선순위 (낮을수록 먼저 실행) + """ + if strategy.name in self.strategies: + logger.warning(f"⚠️ 전략 이미 등록됨: {strategy.name}") + return + + self.strategies[strategy.name] = strategy + + # 우선순위 삽입 + inserted = False + for i, name in enumerate(self.priority_order): + existing_strategy = self.strategies[name] + existing_priority = getattr(existing_strategy, '_priority', 100) + if priority < existing_priority: + self.priority_order.insert(i, strategy.name) + inserted = True + break + + if not inserted: + self.priority_order.append(strategy.name) + + strategy._priority = priority + logger.info(f"✅ 전략 등록: {strategy.name} (priority={priority})") + + def unregister_strategy(self, strategy_name: str): + """전략 등록 해제""" + if strategy_name in self.strategies: + del self.strategies[strategy_name] + self.priority_order.remove(strategy_name) + logger.info(f"❌ 전략 삭제: {strategy_name}") + + def get_strategy(self, strategy_name: str) -> Optional[BaseStrategy]: + """전략 조회""" + return self.strategies.get(strategy_name) + + def list_strategies(self) -> List[str]: + """등록된 전략 목록""" + return list(self.strategies.keys()) + + async def run_strategy( + self, + strategy_name: str, + stock_code: str, + market_data: Dict[str, Any], + portfolio_data: Optional[Dict[str, Any]] = None + ) -> Optional[Signal]: + """ + 특정 전략 단독 실행 + + Args: + strategy_name: 실행할 전략 이름 + stock_code: 종목코드 + market_data: 시장 데이터 + portfolio_data: 포트폴리오 데이터 + + Returns: + Signal 또는 None + """ + strategy = self.strategies.get(strategy_name) + if not strategy: + logger.error(f"❌ 전략 찾을 수 없음: {strategy_name}") + return None + + if not strategy.enabled: + logger.debug(f"⏸️ 전략 비활성화됨: {strategy_name}") + return None + + try: + signal = await strategy.analyze(stock_code, market_data, portfolio_data) + + if signal: + self.signal_history.append(signal) + logger.info(f"📡 신호 생성: {signal}") + + return signal + + except Exception as e: + logger.error(f"❌ [{strategy_name}] 실행 중 오류: {e}", exc_info=True) + await strategy.on_error(e, {"stock_code": stock_code}) + return None + + async def run_all_strategies( + self, + stock_code: str, + market_data: Dict[str, Any], + portfolio_data: Optional[Dict[str, Any]] = None, + parallel: bool = False + ) -> List[Signal]: + """ + 모든 전략 동시 실행 + + Args: + stock_code: 종목코드 + market_data: 시장 데이터 + portfolio_data: 포트폴리오 데이터 + parallel: True = 병렬 실행, False = 순차 실행 + + Returns: + 생성된 Signal 목록 + """ + signals = [] + + if parallel: + # 병렬 실행 + tasks = [ + self.run_strategy(name, stock_code, market_data, portfolio_data) + for name in self.priority_order + ] + results = await asyncio.gather(*tasks, return_exceptions=True) + signals = [s for s in results if isinstance(s, Signal)] + else: + # 순차 실행 (우선순위) + for strategy_name in self.priority_order: + signal = await self.run_strategy( + strategy_name, stock_code, market_data, portfolio_data + ) + if signal: + signals.append(signal) + + return signals + + def resolve_conflicts( + self, + signals: List[Signal], + method: str = "highest_confidence" + ) -> Optional[Signal]: + """ + 신호 충돌 해결 + + Args: + signals: 신호 목록 + method: 해결 방법 + - "highest_confidence": 가장 높은 신뢰도 + - "priority": 전략 우선순위 + - "majority_vote": 다수결 + + Returns: + 최종 선택된 Signal + """ + if not signals: + return None + + if len(signals) == 1: + return signals[0] + + if method == "highest_confidence": + return max(signals, key=lambda s: s.confidence) + + elif method == "priority": + # 우선순위 높은 전략의 신호 선택 + for strategy_name in self.priority_order: + for signal in signals: + if signal.strategy_name == strategy_name: + return signal + return signals[0] + + elif method == "majority_vote": + # 가장 많이 나온 신호 타입 선택 + from collections import Counter + signal_types = [s.signal_type for s in signals] + most_common = Counter(signal_types).most_common(1)[0][0] + + # 해당 타입 중 가장 높은 신뢰도 + matching_signals = [s for s in signals if s.signal_type == most_common] + return max(matching_signals, key=lambda s: s.confidence) + + return signals[0] + + def get_signal_history( + self, + stock_code: Optional[str] = None, + strategy_name: Optional[str] = None, + limit: int = 100 + ) -> List[Signal]: + """ + 신호 이력 조회 + + Args: + stock_code: 종목코드 필터 + strategy_name: 전략 이름 필터 + limit: 최대 개수 + + Returns: + Signal 목록 + """ + history = self.signal_history[:] + + if stock_code: + history = [s for s in history if s.stock_code == stock_code] + + if strategy_name: + history = [s for s in history if s.strategy_name == strategy_name] + + return history[-limit:] + + def get_stats(self) -> Dict[str, Any]: + """전략 통계 정보""" + return { + "total_strategies": len(self.strategies), + "enabled_strategies": sum(1 for s in self.strategies.values() if s.enabled), + "total_signals": len(self.signal_history), + "strategies": { + name: { + "enabled": strategy.enabled, + "priority": getattr(strategy, '_priority', 100), + "signals_count": len([s for s in self.signal_history if s.strategy_name == name]) + } + for name, strategy in self.strategies.items() + } + } + + +# 싱글톤 +strategy_manager = StrategyManager()