aszerW
30ea2970bd
- FactorBase/FactorRegistry/FactorCombiner: 因子抽象接口 - SignalGenerator: 信号生成抽象接口 - RiskControl/Position/CallbackHook: 风控抽象接口 - StrategyBase: 策略抽象基类 - Executor/Portfolio: 执行器抽象接口 - ConfigLoader: 配置加载器 - 删除framework/factors/momentum.py（具体实现）
144 lines
4.0 KiB
Python
144 lines
4.0 KiB
Python
"""
|
||
策略层抽象基类（通用）
|
||
|
||
只提供抽象接口，具体策略实现在strategies/
|
||
"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
from typing import Dict, Optional, Any
|
||
import pandas as pd
|
||
|
||
from framework.factors import FactorCombiner
|
||
from framework.signals import SignalGenerator
|
||
from framework.risk import CallbackHook, Position
|
||
|
||
|
||
class StrategyBase(ABC):
|
||
"""
|
||
策略抽象基类
|
||
|
||
所有策略必须实现init_factors和init_signal_generator方法
|
||
"""
|
||
|
||
INTERFACE_VERSION = 1
|
||
name: str = "base"
|
||
timeframe: str = "1d"
|
||
|
||
# 类属性（可被配置覆盖）
|
||
select_num: int = 3
|
||
stoploss: float = -0.05
|
||
|
||
def __init__(self, config: Optional[Dict] = None):
|
||
"""
|
||
初始化策略
|
||
|
||
Args:
|
||
config: 配置字典（可选，用于覆盖类属性）
|
||
"""
|
||
if config:
|
||
self._apply_config(config)
|
||
|
||
self._callbacks = CallbackHook()
|
||
self._register_default_callbacks()
|
||
|
||
self._factors = self.init_factors()
|
||
self._signal_gen = self.init_signal_generator()
|
||
|
||
def _apply_config(self, config: Dict) -> None:
|
||
"""应用配置覆盖类属性"""
|
||
for key, value in config.items():
|
||
if hasattr(self, key):
|
||
setattr(self, key, value)
|
||
|
||
def _register_default_callbacks(self) -> None:
|
||
"""注册默认回调方法"""
|
||
if hasattr(self, 'before_entry'):
|
||
self._callbacks.register('before_entry', self.before_entry)
|
||
|
||
if hasattr(self, 'after_entry'):
|
||
self._callbacks.register('after_entry', self.after_entry)
|
||
|
||
if hasattr(self, 'before_exit'):
|
||
self._callbacks.register('before_exit', self.before_exit)
|
||
|
||
if hasattr(self, 'after_exit'):
|
||
self._callbacks.register('after_exit', self.after_exit)
|
||
|
||
if hasattr(self, 'dynamic_stoploss'):
|
||
self._callbacks.register('dynamic_stoploss', self.dynamic_stoploss)
|
||
|
||
if hasattr(self, 'custom_exit'):
|
||
self._callbacks.register('custom_exit', self.custom_exit)
|
||
|
||
@abstractmethod
|
||
def init_factors(self) -> FactorCombiner:
|
||
"""
|
||
初始化因子组合器
|
||
|
||
Returns:
|
||
FactorCombiner实例
|
||
"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def init_signal_generator(self) -> SignalGenerator:
|
||
"""
|
||
初始化信号生成器
|
||
|
||
Returns:
|
||
SignalGenerator实例
|
||
"""
|
||
pass
|
||
|
||
def run(self, data: pd.DataFrame) -> pd.DataFrame:
|
||
"""
|
||
运行策略
|
||
|
||
Args:
|
||
data: OHLCV数据
|
||
|
||
Returns:
|
||
包含信号的DataFrame
|
||
"""
|
||
factor_data = self._factors.compute(data)
|
||
signals = self._signal_gen.generate(factor_data)
|
||
|
||
signals = self._apply_callbacks(signals, data)
|
||
|
||
return signals
|
||
|
||
def _apply_callbacks(self, signals: pd.DataFrame, data: pd.DataFrame) -> pd.DataFrame:
|
||
"""应用回调处理"""
|
||
return signals
|
||
|
||
# 可选回调方法（子类可覆盖）
|
||
def before_entry(self, code: str, price: float, **kwargs) -> bool:
|
||
"""入场前检查"""
|
||
return True
|
||
|
||
def after_entry(self, code: str, price: float, **kwargs) -> None:
|
||
"""入场后处理"""
|
||
pass
|
||
|
||
def before_exit(self, position: Position, **kwargs) -> bool:
|
||
"""出场前检查"""
|
||
return True
|
||
|
||
def after_exit(self, position: Position, **kwargs) -> None:
|
||
"""出场后处理"""
|
||
pass
|
||
|
||
def dynamic_stoploss(self, position: Position) -> float:
|
||
"""动态止损"""
|
||
return self.stoploss
|
||
|
||
def custom_exit(self, position: Position) -> bool:
|
||
"""自定义出场条件"""
|
||
return False
|
||
|
||
def __repr__(self) -> str:
|
||
return f"{self.__class__.__name__}(name={self.name})"
|
||
|
||
|
||
# 导出抽象接口
|
||
__all__ = ['StrategyBase'] |