feat(selector): 大类冠军动量排名二次过滤

新增过滤逻辑:
- 大类冠军必须在全局排名Top select_num范围内才有效
- 假设短债排名第5,select_num=3,则短债被排除
- 避免持有动量过低(排名靠后)的防御资产

示例:
- NDX (排名1) -> 选中
- 399006.SZ (排名2) -> 选中
- N225 (排名3) -> 选中
- 931862.CSI (排名5 > 3) -> 排除
This commit is contained in:
2026-05-18 22:35:41 +08:00
parent 3e6d9d1fdb
commit 740135e5fb

View File

@@ -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范围内才有效
- 假设短债排名第4select_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):