diff --git a/strategies/shared/signals/selectors.py b/strategies/shared/signals/selectors.py index 797f845..7ffb3fd 100644 --- a/strategies/shared/signals/selectors.py +++ b/strategies/shared/signals/selectors.py @@ -193,7 +193,11 @@ class TopNSelector(SignalGenerator): def _grouped_selection(self, scores: Dict[str, float]) -> List[str]: """分组选股:先类内竞争(每大类选Top1),再跨类排序 - 改进:大类冠军得分不足时跳过该大类,不强制持有弱正动量标的 + 改进:大类冠军二次过滤 + 1. min_score过滤负动量 + 2. 动量排名过滤:大类冠军必须在全局Top select_num范围内才有效 + - 假设短债排名第4,select_num=3,则短债被排除 + - 这避免持有动量过低(排名靠后)的防御资产 """ if not scores: return [] @@ -206,22 +210,27 @@ class TopNSelector(SignalGenerator): if group not in group_champions or score > group_champions[group][1]: group_champions[group] = (code, score) - # ⭐ 关键改进:大类冠军二次过滤 - # 只保留得分足够显著的冠军,得分不足的大类跳过 - # 这样组合中的每个标的动量都足够强 + # ⭐ 计算全局动量排名(用于二次过滤) + # 将所有标的按得分排序,计算每个标的的排名 + all_sorted = sorted(scores.items(), key=lambda x: x[1], reverse=True) + rank_map = {code: rank + 1 for rank, (code, _) in enumerate(all_sorted)} + + # ⭐ 大类冠军二次过滤 + # 只保留全局排名 <= select_num 的大类冠军 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) + rank = rank_map.get(code, len(all_sorted) + 1) # 未找到则排名为最后 + + # 过滤条件: + # 1. 得分 >= min_score(过滤负动量) + # 2. 全局排名 <= select_num(过滤排名靠后的冠军) + if score >= self.min_score and rank <= self.select_num: + valid_champions.append((code, score, rank)) - # 对有效冠军进行排序,选出Top N + # 对有效冠军按得分排序,选出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]] + return [code for code, score, rank in sorted_champions[:self.select_num]] class TrendFollower(SignalGenerator):