Archive legacy framework and utility modules that are no longer referenced by the active core (datasource/ and rotation/): - framework/ -> archive/framework/ - framework_v2/ -> archive/framework_v2/ - strategies/ -> archive/strategies/ - config/ -> archive/config/ - visualization/ -> archive/visualization/ - scripts/ -> archive/scripts/ - tests/ -> archive/tests/ - run_rotation.py, run_us_rotation.py -> archive/single_files/ - compare_*.py, test_api_dates.py -> archive/single_files/
58 lines
1.3 KiB
Python
58 lines
1.3 KiB
Python
"""
|
||
信号生成器抽象基类
|
||
"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
import pandas as pd
|
||
|
||
|
||
class SignalGenerator(ABC):
|
||
"""
|
||
信号生成器抽象基类
|
||
|
||
所有信号生成器必须实现 generate 方法
|
||
"""
|
||
|
||
mode: str = "base"
|
||
|
||
def __init__(self, **params):
|
||
"""
|
||
初始化信号生成器参数
|
||
|
||
Args:
|
||
**params: 信号参数(如 select_num, rebalance_days 等)
|
||
"""
|
||
self._params = params
|
||
|
||
@abstractmethod
|
||
def generate(self, factor_data: pd.DataFrame) -> pd.DataFrame:
|
||
"""
|
||
生成交易信号
|
||
|
||
Args:
|
||
factor_data: 因子数据 DataFrame
|
||
|
||
Returns:
|
||
信号 DataFrame,必须包含 'signal' 列
|
||
"""
|
||
pass
|
||
|
||
def validate_factor_data(self, factor_data: pd.DataFrame) -> bool:
|
||
"""
|
||
验证因子数据是否有效
|
||
|
||
Args:
|
||
factor_data: 因子数据
|
||
|
||
Returns:
|
||
True 如果数据有效
|
||
"""
|
||
if factor_data.empty:
|
||
return False
|
||
|
||
return True
|
||
|
||
def __repr__(self) -> str:
|
||
params_str = ', '.join([f"{k}={v}" for k, v in self._params.items()])
|
||
return f"{self.__class__.__name__}({params_str})"
|