etf/framework_v2/core/strategy.py

"""
策略抽象基类

所有策略必须继承此类并实现必要方法
"""

from abc import ABC, abstractmethod
from typing import Dict, Optional, Any
import pandas as pd


class StrategyBase(ABC):
    """
    策略抽象基类
    
    定义策略的标准生命周期：
    1. 初始化配置
    2. 获取数据
    3. 计算因子
    4. 生成信号
    5. 执行回测
    
    子类必须实现：
    - init_factors(): 初始化因子
    - init_signal_generator(): 初始化信号生成器
    """
    
    INTERFACE_VERSION = 2  # V2 版本
    
    name: str = "base"
    timeframe: str = "1d"
    
    def __init__(self, config: Optional[Dict] = None):
        """
        初始化策略
        
        Args:
            config: 策略配置字典
        """
        self.config = config or {}
        self._factor = None
        self._signal_generator = None
    
    @abstractmethod
    def init_factors(self) -> Any:
        """
        初始化因子组件
        
        Returns:
            因子实例（继承 FactorBase）
        """
        pass
    
    @abstractmethod
    def init_signal_generator(self) -> Any:
        """
        初始化信号生成器
        
        Returns:
            信号生成器实例（继承 SignalGenerator）
        """
        pass
    
    def get_data(self) -> Dict[str, Any]:
        """
        获取数据（可选覆盖）
        
        Returns:
            数据字典，包含：
            - index_data: 指数数据
            - etf_data: ETF数据
            - benchmark_data: 基准数据
            - valid_codes: 有效标的列表
            - trading_calendar: 交易日历
        """
        raise NotImplementedError("Subclasses must implement get_data()")
    
    def compute_factors(self, data: Dict[str, Any]) -> pd.DataFrame:
        """
        计算因子（可选覆盖）
        
        Args:
            data: 数据字典
        
        Returns:
            因子 DataFrame（日期 × 标的）
        """
        if self._factor is None:
            self._factor = self.init_factors()
        
        # 默认实现：遍历标的计算因子
        factor_values = {}
        for code in data.get('valid_codes', []):
            if code in data.get('index_data', {}):
                factor_values[code] = self._factor.compute(data['index_data'][code])
        
        return pd.DataFrame(factor_values)
    
    def generate_signals(self, factor_df: pd.DataFrame) -> pd.DataFrame:
        """
        生成信号
        
        Args:
            factor_df: 因子 DataFrame
        
        Returns:
            信号 DataFrame（包含 'signal' 列）
        """
        if self._signal_generator is None:
            self._signal_generator = self.init_signal_generator()
        
        return self._signal_generator.generate(factor_df)
    
    def run_backtest(self, data: Optional[Dict] = None) -> Dict[str, Any]:
        """
        运行完整回测流程
        
        Args:
            data: 可选，如不提供则自动获取
        
        Returns:
            回测结果字典
        """
        # 1. 获取数据
        if data is None:
            data = self.get_data()
        
        # 2. 计算因子
        factor_df = self.compute_factors(data)
        
        # 3. 生成信号
        signals = self.generate_signals(factor_df)
        
        # 4. 执行回测（子类实现）
        return self._execute_backtest(signals, data)
    
    def _execute_backtest(self, signals: pd.DataFrame, data: Dict[str, Any]) -> Dict[str, Any]:
        """
        执行回测（子类可覆盖）
        
        Args:
            signals: 信号 DataFrame
            data: 数据字典
        
        Returns:
            回测结果
        """
        raise NotImplementedError("Subclasses must implement _execute_backtest()")
    
    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name={self.name})"