etf/docs/V3动态阈值实施方案.md

V3 动态阈值实施方案

目标

将 generate_bond_threshold_report.py 中验证的动态阈值逻辑落地到正式策略代码，使 RotationStrategy.run_backtest() 直接产出 V3 效果（CAGR 28.17%, 回撤 -24.35%）。

---

当前代码流程

config.yaml (min_score: 0.0)
    ↓
strategy.py:96  → self.min_score = config.get('min_score', 0.0)
    ↓
strategy.py:72  → TopNSelector(..., min_score=self.min_score)
    ↓
selectors.py:84 → scores = {k: v for k, v in scores.items() if v >= self.min_score}
    ↓
selectors.py:216 → if score >= self.min_score: valid_champions.append(...)
    ↓
selectors.py:224 → return sorted_champions[:self.select_num]  (不足N只时返回少于N只)

问题：当前 min_score=0.0 是固定值，且选出不足 select_num 只时直接返回少数标的（没有用短债填充空余仓位）。

---

需要修改的文件（3个）

文件1：strategies/rotation/config.yaml

改动：新增 bond_threshold 配置块，替代固定 min_score

# ==================== 轮动参数 ====================
select_num: 3
diversified: true

# 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只时，用短债填充空余仓位

# 保留 min_score 作为 fallback（bond_threshold.enabled=false 或短债无数据时使用）
min_score: 0.0

位置：第 120~132 行区域

---

文件2：strategies/shared/signals/selectors.py

改动点1：TopNSelector.__init__ 新增参数（第 34~59 行）

def __init__(
    self,
    select_num: int = 3,
    group_by: Optional[str] = None,
    group_mapping: Optional[Dict[str, str]] = None,
    top_per_group: int = 1,
    min_score: Optional[float] = None,
    rebalance_threshold: float = 0.0,
    rebalance_days: int = 1,
    # V3 新增
    bond_threshold_config: Optional[Dict] = None,
):
    ...
    self.bond_threshold_config = bond_threshold_config or {}

改动点2：generate() 方法中替换固定 min_score 过滤（第 83~85 行）

当前：

# 最小得分过滤（如过滤负分）
if self.min_score is not None:
    scores = {k: v for k, v in scores.items() if v >= self.min_score}

改为：

# V3: 动态阈值 = 短债动量; V2 fallback: 固定 min_score
threshold = self._get_dynamic_threshold(scores)
scores = {k: v for k, v in scores.items() if v >= threshold}

改动点3：新增 _get_dynamic_threshold 方法

def _get_dynamic_threshold(self, scores: Dict[str, float]) -> float:
    """获取动态阈值：短债动量 × ratio，无数据时退化为 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

改动点4：_grouped_selection() 方法（第 193~224 行）

当前逻辑已排除 BOND 大类的标的与其他类竞争吗？没有——当前代码所有通过 min_score 的标的都参与分组选股，BOND 冠军和其他大类冠军一起排名。

需要改为：

1. BOND 大类标的不参与冠军竞争（它是阈值，不是候选）
2. 选出不足 select_num 只时，用短债填充

def _grouped_selection(self, scores: Dict[str, float]) -> List[str]:
    """V3分组选股：BOND不参与竞争，空余仓位填充短债"""
    if not scores:
        return []

    cfg = self.bond_threshold_config
    bond_code = cfg.get('bond_code', '931862.CSI') if cfg.get('enabled') else None

    # 建立 group -> (code, score) 映射，排除 BOND 大类
    group_champions = {}
    for code, score in scores.items():
        group = self.group_mapping.get(code, 'default')
        if group == 'BOND':
            continue  # BOND 不参与竞争
        if group not in group_champions or score > group_champions[group][1]:
            group_champions[group] = (code, score)

    # 跨类排序取 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]]

    # 空余仓位填充短债
    if cfg.get('fill_bond', False) and bond_code:
        n_bond_slots = self.select_num - len(selected)
        for _ in range(n_bond_slots):
            selected.append(bond_code)

    return selected

---

文件3：strategies/rotation/strategy.py

改动点：初始化 TopNSelector 时传入 bond_threshold_config（第 69~75 行）

当前：

self._selector = TopNSelector(
    select_num=self.select_num,
    group_mapping=self._group_mapping,
    min_score=self.min_score,
    rebalance_days=self.rebalance_days,
    rebalance_threshold=self.rebalance_threshold
)

改为：

self._selector = TopNSelector(
    select_num=self.select_num,
    group_mapping=self._group_mapping,
    min_score=self.min_score,
    rebalance_days=self.rebalance_days,
    rebalance_threshold=self.rebalance_threshold,
    bond_threshold_config=self.config.get('bond_threshold', {}),
)

---

逻辑对照表

步骤	V2（当前）	V3（目标）
阈值来源	config.yaml 固定 min_score=0.0	每日从因子中读取短债动量
阈值计算	常量 0	scores['931862.CSI'] × ratio
BOND 竞争	BOND冠军与其他大类一起排名	BOND 不参与竞争（仅作阈值+填充）
不足N只时	返回少于N只（BacktestExecutor等权分配）	用短债代码填充至N只
信号格式	"NDX,GC=F" (2只)	"NDX,GC=F,931862.CSI" (3只)
仓位效果	2只各50%	NDX 33%, GC=F 33%, 短债 33%

---

关键注意事项

1. BOND 大类从"候选"变为"工具"

V2 中 931862.CSI 是普通候选标的，与其他大类一样参与 Top3 排名。V3 中它变为双重角色：

• 角色A：阈值（它的动量决定其他标的是否值得持有）
• 角色B：填充物（空余仓位用它填充）

它不再参与 _grouped_selection 的冠军竞争。

2. 信号格式兼容

V3 信号中短债出现方式："NDX,931862.CSI,931862.CSI" 表示 NDX 1/3 + 短债 2/3。BacktestExecutor 已支持重复代码等权分配（generate_bond_threshold_report.py 已验证）。

3. 短债无数据时的退化

2002-2007年 931862.CSI 无数据，scores.get('931862.CSI') 返回 None → 阈值退化为 min_score=0.0 → 策略行为与 V2 完全一致。无需特殊处理。

4. min_score 过滤的时机

动态阈值替代了第84行的全局过滤。但第216行的大类冠军二次过滤也需要用动态阈值，否则存在不一致：

# 第216行当前逻辑
if score >= self.min_score:
    valid_champions.append((code, score))

V3 中这行逻辑被合并进新的 _grouped_selection：在 scores 已经过动态阈值过滤后，所有剩余标的天然满足 >= threshold，不需要二次检查。

---

验证方式

修改完成后运行：

# 用正式策略流程跑回测
python -c "
from strategies.rotation.strategy import RotationStrategy
strategy = RotationStrategy.from_yaml('strategies/rotation/config.yaml')
strategy.run_backtest()
"

预期结果应与 generate_bond_threshold_report.py --ratio 1.0 一致：

• CAGR ≈ 28%
• 最大回撤 ≈ -24%
• 夏普 ≈ 1.40

如果数值有差异，检查：

1. 调仓控制逻辑是否一致（_apply_rebalance_control vs _apply_rebalance）
2. 溢价率过滤是否影响了部分标的的入选

---

回滚方案

# config.yaml 一行改动即可回退到V2
bond_threshold:
  enabled: false    # 关闭动态阈值，退化为 min_score=0.0

代码中 _get_dynamic_threshold 会返回 self.min_score，_grouped_selection 中 BOND 排除逻辑不生效（因为 cfg.get('enabled') 为 false），行为与 V2 完全一致。