feat(rotation): 实现轮动策略(使用框架抽象+定制组件)

- RotationStrategy: 继承StrategyBase,使用MomentumFactor+TopNSelector
- 实现before_entry溢价过滤、dynamic_stoploss动态止损、custom_exit自定义出场
- 策略配置从类属性读取,支持config覆盖
This commit is contained in:
2026-05-11 23:09:49 +08:00
parent 69081297c5
commit de31271ab3
3 changed files with 96 additions and 0 deletions

View File

@@ -0,0 +1,9 @@
"""
strategies模块入口
包含所有策略实现
"""
from strategies.rotation import RotationStrategy
__all__ = ['RotationStrategy']

View File

@@ -0,0 +1,7 @@
"""
轮动策略模块入口
"""
from strategies.rotation.strategy import RotationStrategy
__all__ = ['RotationStrategy']

View File

@@ -0,0 +1,80 @@
"""
轮动策略定制实现
使用framework通用能力 + 定制组件
"""
import pandas as pd
import yaml
from datetime import datetime
from framework.factors import FactorBase, FactorRegistry, FactorCombiner
from framework.signals import SignalGenerator
from framework.risk import CallbackHook, Position
from framework.strategy import StrategyBase
from framework.config import ConfigLoader
# 导入定制组件
from strategies.shared.factors.momentum import MomentumFactor
from strategies.shared.signals.selectors import TopNSelector
from strategies.shared.risk.controls import premium_filter_callback, holding_time_stoploss_callback
class RotationStrategy(StrategyBase):
"""
ETF轮动策略定制实现
基于动量因子 + Top N选股 + 溢价过滤
"""
name = "rotation"
select_num = 3
stoploss = -0.05
def init_factors(self) -> FactorCombiner:
"""初始化动量因子"""
# 清空注册表(避免重复注册)
FactorRegistry.clear()
# 注册定制因子
FactorRegistry.register(MomentumFactor)
return FactorCombiner([
FactorRegistry.get('momentum', n_days=25, crash_filter=True)
])
def init_signal_generator(self) -> SignalGenerator:
"""初始化Top N选股器定制"""
return TopNSelector(
select_num=self.select_num,
min_score=0.0,
group_by='market' # 定制:按大类分组
)
def before_entry(self, code: str, price: float, **kwargs) -> bool:
"""入场前:溢价过滤(定制)"""
premium = kwargs.get('premium', 0)
# 定制阈值10%
if premium > 0.10:
print(f"溢价过高,拒绝入场: {code} (溢价={premium:.2%})")
return False
return True
def dynamic_stoploss(self, position: Position) -> float:
"""动态止损:根据持仓时间调整(定制)"""
# 定制规则5天/10天阈值
if position.holding_days >= 10:
return -0.03
elif position.holding_days >= 5:
return -0.05
return -0.10
def custom_exit(self, position: Position) -> bool:
"""自定义出场条件(定制)"""
# 定制规则:亏损超过阈值强制出场
if position.profit_ratio < -0.10:
print(f"亏损超阈值,强制出场: {position.code}")
return True
return False