diff --git a/strategies/__init__.py b/strategies/__init__.py index e69de29..7bac080 100644 --- a/strategies/__init__.py +++ b/strategies/__init__.py @@ -0,0 +1,9 @@ +""" +strategies模块入口 + +包含所有策略实现 +""" + +from strategies.rotation import RotationStrategy + +__all__ = ['RotationStrategy'] \ No newline at end of file diff --git a/strategies/rotation/__init__.py b/strategies/rotation/__init__.py index e69de29..fb303df 100644 --- a/strategies/rotation/__init__.py +++ b/strategies/rotation/__init__.py @@ -0,0 +1,7 @@ +""" +轮动策略模块入口 +""" + +from strategies.rotation.strategy import RotationStrategy + +__all__ = ['RotationStrategy'] \ No newline at end of file diff --git a/strategies/rotation/strategy.py b/strategies/rotation/strategy.py new file mode 100644 index 0000000..13f7463 --- /dev/null +++ b/strategies/rotation/strategy.py @@ -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 \ No newline at end of file