102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""
|
|
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}>"
|