refactor(phase2): Create strategy module core structure

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",
+]

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}>"

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