aszerW
de31271ab3
- RotationStrategy: 继承StrategyBase，使用MomentumFactor+TopNSelector - 实现before_entry溢价过滤、dynamic_stoploss动态止损、custom_exit自定义出场 - 策略配置从类属性读取，支持config覆盖
80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""
|
||
轮动策略定制实现
|
||
|
||
使用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 |