refactor(phase2): Add StrategyManager for strategy orchestration

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()