254 lines
8.1 KiB
Python
254 lines
8.1 KiB
Python
"""
|
|
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()
|