From 025ecb5f4f3306e6884e874d5790f5994d268fa5 Mon Sep 17 00:00:00 2001 From: YoungWooSeo Date: Fri, 6 Mar 2026 10:52:40 +0900 Subject: [PATCH] refactor(phase2): Add StrategyManager for strategy orchestration --- backend/app/strategy/strategy_manager.py | 253 +++++++++++++++++++++++ 1 file changed, 253 insertions(+) create mode 100644 backend/app/strategy/strategy_manager.py 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()