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/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})" + )