diff --git a/strategies/rotation/config.yaml b/strategies/rotation/config.yaml index 69ef507..a221899 100644 --- a/strategies/rotation/config.yaml +++ b/strategies/rotation/config.yaml @@ -125,6 +125,14 @@ diversified: true # 设置为0表示过滤负动量标的，更高阈值虽能改善回撤但可能错过正动量机会 min_score: 0.0 +# V3: 动态阈值配置（替代固定 min_score: 0.0） +# 使用短债动量作为动态 min_score：标的动量 < 短债动量 → 不持有 +bond_threshold: + enabled: true # true=V3动态阈值, false=退化为V2固定阈值 + bond_code: "931862.CSI" # 阈值参考标的（短债指数） + ratio: 1.0 # 阈值 = 短债动量 × ratio + fill_bond: true # 选出不足select_num只时，用短债填充空余仓位 + # ==================== 调仓控制 ==================== # 最低调仓周期（交易日）：持仓至少持有 N 天后才允许换仓 rebalance_days: 1 diff --git a/strategies/rotation/strategy.py b/strategies/rotation/strategy.py index 9f75129..1e9ae13 100644 --- a/strategies/rotation/strategy.py +++ b/strategies/rotation/strategy.py @@ -71,7 +71,8 @@ class RotationStrategy(StrategyBase): group_mapping=self._group_mapping, min_score=self.min_score, # 从配置读取，支持动态调整阈值 rebalance_days=self.rebalance_days, - rebalance_threshold=self.rebalance_threshold + rebalance_threshold=self.rebalance_threshold, + bond_threshold_config=self.config.get('bond_threshold', {}) # V3动态阈值配置 ) @classmethod diff --git a/strategies/shared/signals/selectors.py b/strategies/shared/signals/selectors.py index 797f845..4797ffe 100644 --- a/strategies/shared/signals/selectors.py +++ b/strategies/shared/signals/selectors.py @@ -18,6 +18,7 @@ class TopNSelector(SignalGenerator): - 按因子值排序，选出Top N标的 - 支持分组选股（先类内竞争，再跨类排序） - 支持调仓阈值检查（新组合得分需超过当前组合一定比例才调仓） + - V3: 支持动态阈值（短债动量作为过滤阈值） 参数： - select_num: 选中数量（默认3） @@ -27,6 +28,7 @@ class TopNSelector(SignalGenerator): - min_score: 最小得分阈值（可选，如0表示过滤负分） - rebalance_threshold: 调仓阈值（可选，新组合得分需超过当前组合X%才调仓） - rebalance_days: 最低调仓周期（可选，持仓至少N天才能调仓） + - bond_threshold_config: V3动态阈值配置 """ mode = "top_n" @@ -39,7 +41,8 @@ class TopNSelector(SignalGenerator): top_per_group: int = 1, min_score: Optional[float] = None, rebalance_threshold: float = 0.0, - rebalance_days: int = 1 + rebalance_days: int = 1, + bond_threshold_config: Optional[Dict] = None ): super().__init__( select_num=select_num, @@ -57,6 +60,28 @@ class TopNSelector(SignalGenerator): self.min_score = min_score self.rebalance_threshold = rebalance_threshold self.rebalance_days = rebalance_days + self.bond_threshold_config = bond_threshold_config or {} + + def _get_dynamic_threshold(self, scores: Dict[str, float]) -> float: + """获取动态阈值：短债动量 × ratio，无数据时退化为 min_score + + V3动态阈值逻辑： + - 若bond_threshold.enabled=true，阈值 = 短债动量 × ratio + - 若短债无数据或动量<0，退化为固定min_score + - 若enabled=false，退化为固定min_score + """ + cfg = self.bond_threshold_config + if not cfg.get('enabled', False): + return self.min_score if self.min_score is not None else 0.0 + + bond_code = cfg.get('bond_code', '931862.CSI') + ratio = cfg.get('ratio', 1.0) + + bond_score = scores.get(bond_code, None) + if bond_score is None or bond_score < 0: + return self.min_score if self.min_score is not None else 0.0 + + return bond_score * ratio def generate(self, factor_data: pd.DataFrame) -> pd.DataFrame: """生成Top N选股信号（支持调仓周期控制）""" @@ -80,9 +105,9 @@ class TopNSelector(SignalGenerator): if pd.notna(score): scores[col] = score - # 最小得分过滤（如过滤负分） - if self.min_score is not None: - scores = {k: v for k, v in scores.items() if v >= self.min_score} + # V3: 动态阈值过滤（替代固定 min_score） + threshold = self._get_dynamic_threshold(scores) + scores = {k: v for k, v in scores.items() if v >= threshold} # 分组选股或全局选股 if self.group_mapping: @@ -191,37 +216,48 @@ class TopNSelector(SignalGenerator): return new_total > 0 def _grouped_selection(self, scores: Dict[str, float]) -> List[str]: - """分组选股：先类内竞争（每大类选Top1），再跨类排序 + """V3分组选股：BOND不参与竞争，空余仓位填充短债 - 改进：大类冠军得分不足时跳过该大类，不强制持有弱正动量标的 + V3逻辑： + 1. BOND大类标的不参与冠军竞争（它是阈值，不是候选） + 2. 选出不足 select_num 只时，用短债填充 + 3. V2退化：若bond_threshold.enabled=false，BOND正常参与竞争 """ if not scores: return [] - # 建立 group -> (code, score) 的映射 + cfg = self.bond_threshold_config + bond_code = cfg.get('bond_code', '931862.CSI') if cfg.get('enabled') else None + + # 建立 group -> (code, score) 映射 + # V3: 排除 BOND 大类（它不参与竞争） group_champions = {} for code, score in scores.items(): - # 从group_mapping获取分组 group = self.group_mapping.get(code, 'default') + + # V3: BOND大类不参与竞争 + if cfg.get('enabled') and group == 'BOND': + continue + if group not in group_champions or score > group_champions[group][1]: group_champions[group] = (code, score) - # ⭐ 关键改进：大类冠军二次过滤 - # 只保留得分足够显著的冠军，得分不足的大类跳过 - # 这样组合中的每个标的动量都足够强 - valid_champions = [] - for group, (code, score) in group_champions.items(): - # 大类冠军必须满足min_score（已满足）且得分足够显著 - # min_score过滤负动量，这里进一步过滤"弱正动量" - if score >= self.min_score: - valid_champions.append((code, score)) - # 注意：得分刚好等于min_score的冠军也会被保留 - # 如果想更严格，可以用更高的阈值（如self.min_score + 0.02） + # 跨类排序取 Top N + sorted_champions = sorted(group_champions.values(), key=lambda x: x[1], reverse=True) + selected = [code for code, score in sorted_champions[:self.select_num]] - # 对有效冠军进行排序，选出Top N - # 持仓数量动态调整：最多select_num，最少可以是0 - sorted_champions = sorted(valid_champions, key=lambda x: x[1], reverse=True) - return [code for code, score in sorted_champions[:self.select_num]] + # V3: 空余仓位填充短债 + if cfg.get('fill_bond', False) and bond_code: + n_bond_slots = self.select_num - len(selected) + if n_bond_slots > 0: + # 检查短债是否满足阈值条件（需在原始scores中存在） + bond_score = scores.get(bond_code, None) + if bond_score is not None: + # 用短债代码填充空余仓位（可能重复填充多次） + for _ in range(n_bond_slots): + selected.append(bond_code) + + return selected class TrendFollower(SignalGenerator):