aszerW
908b28473f
## 架构设计 - 三层架构：core(抽象接口) → shared(通用实现) → tests(验证测试) - 5个核心抽象基类：StrategyBase, FactorBase, SignalGenerator, Executor, DataFetcher - 零侵入：与现有框架并行开发，不修改生产代码 ## 已完成 ✓ 核心接口层（5个ABC类） ✓ 通用因子层（MomentumFactor完全复制现有逻辑） ✓ 对比验证测试（新旧因子输出差异=0，测试通过） ## 验证结果 - 最大差异: 0.000000e+00 - 平均差异: 0.000000e+00 - 容差: < 1e-10 ## 下一步 - 阶段3: 信号层迁移（TopNSelector, DynamicThreshold, RebalanceController） - 阶段4: 执行层迁移（BacktestRunner） - 阶段5: 数据层迁移（DataFetcher实现） - 阶段6: 完整策略对比验证 ## 设计原则 - 按需抽象，不预先设计 - 职责分离，避免框架膨胀 - 测试驱动，每个组件必须有对比测试 - 渐进式迁移，验证通过再替换
56 lines
1.3 KiB
Python
56 lines
1.3 KiB
Python
"""
|
|
ta-lib 因子基类（通用）
|
|
|
|
所有 ta-lib 因子继承此类，只需指定函数和参数
|
|
"""
|
|
|
|
import talib
|
|
import pandas as pd
|
|
import numpy as np
|
|
from framework_v2.core import FactorBase
|
|
|
|
|
|
class TALibFactorBase(FactorBase):
|
|
"""
|
|
ta-lib 因子基类
|
|
|
|
子类只需实现：
|
|
- name: 因子名称
|
|
- _talib_func: 返回 ta-lib 函数
|
|
"""
|
|
|
|
category = "technical"
|
|
|
|
def __init__(self, period: int = 14, **params):
|
|
"""
|
|
初始化
|
|
|
|
Args:
|
|
period: 周期参数
|
|
**params: 其他参数
|
|
"""
|
|
super().__init__(period=period, **params)
|
|
self.period = period
|
|
|
|
def compute(self, data: pd.DataFrame) -> pd.Series:
|
|
"""
|
|
计算因子值
|
|
|
|
Args:
|
|
data: OHLCV 数据
|
|
|
|
Returns:
|
|
因子值序列
|
|
"""
|
|
close = data['close'].values.astype(float)
|
|
|
|
# 调用子类指定的 ta-lib 函数
|
|
result = self._talib_func(close, timeperiod=self.period)
|
|
|
|
return pd.Series(result, index=data.index, name=self.name)
|
|
|
|
@property
|
|
def _talib_func(self):
|
|
"""子类必须实现，返回 ta-lib 函数"""
|
|
raise NotImplementedError("Subclasses must implement _talib_func")
|