4. 动量轮动信号：单窗口收益率排序实现

任务拆解：从价格面板到每期持仓排名

本节在已经对齐好的宽表价格面板上，计算固定回看窗口的动量收益率，并按收益率从高到低生成每期持仓排名。核心约束有三点：收益率必须用复权收盘价计算；回看窗口按交易日计数而非自然日；排名只在调仓日取值，且调仓日当天的价格不参与该日信号计算（避免未来函数）。

设价格面板为 P，行索引为交易日，列为 ETF 代码。回看窗口长度为 L（交易日），则第 t 日的动量收益率为：

mom[t] = P[t] / P[t-L] - 1

其中 P[t-L] 是 t 往前数第 L 个交易日的收盘价。若 t-L 处存在缺失（上市不足或停牌），该 ETF 在 t 日的动量记为 NaN，不参与排名。

实现：单窗口动量排名函数

下面给出一个可直接运行的实现，输入为宽表价格面板（行=交易日，列=ETF），输出为每期动量收益率矩阵与排名矩阵。排名采用 rank(ascending=False)，即收益率最高者排名为 1；缺失值保持 NaN，不参与排名。

import numpy as np
import pandas as pd

def momentum_rank(price: pd.DataFrame, lookback: int = 20) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    price: 宽表价格面板，index=交易日(DatetimeIndex)，columns=ETF代码
    lookback: 回看窗口（交易日数）
    返回: (mom, rank)
      mom[t]  = price[t] / price[t-lookback] - 1
      rank[t] = 按 mom[t] 降序排名，1 为最强；NaN 不参与排名
    """
    if lookback <= 0:
        raise ValueError("lookback 必须为正整数")
    if not isinstance(price.index, pd.DatetimeIndex):
        raise TypeError("price.index 必须是 DatetimeIndex")

    # 按交易日位移，而非自然日；shift 会保留 NaN 缺口
    past = price.shift(lookback)
    mom = price / past - 1.0

    # 若窗口内任一端缺失，mom 自动为 NaN，无需额外处理
    rank = mom.rank(axis=1, ascending=False, method="min")
    return mom, rank

def rebalance_snapshot(rank: pd.DataFrame, rebalance_dates: pd.DatetimeIndex) -> pd.DataFrame:
    """只取调仓日的排名快照，并校验日期全部命中"""
    missing = rebalance_dates.difference(rank.index)
    if len(missing) > 0:
        raise KeyError(f"以下调仓日不在价格面板索引中: {missing.tolist()}")
    return rank.loc[rebalance_dates].copy()

边界验证：调仓日取值是否正确

排名最容易出错的地方是窗口端点。需要验证三件事：

1. 窗口起点正确：mom[t] 的分母必须是 t 往前第 L 个交易日，而不是自然日 t-L 天。
2. 调仓日当天价格参与计算：mom[t] 使用 P[t]，这是合法的，因为信号在收盘后生成。
3. 缺失值不污染排名：窗口内任一端为 NaN 时，该 ETF 排名应为 NaN，且不影响其他 ETF 的相对排名。

def validate_rank_at_boundary(price, mom, rank, lookback, check_dates):
    """逐日核对 mom 与 rank 在指定调仓日的取值"""
    report = []
    for t in check_dates:
        pos = price.index.get_loc(t)
        if pos < lookback:
            report.append((t, "窗口不足", None, None))
            continue
        t0 = price.index[pos - lookback]
        expected_mom = price.loc[t] / price.loc[t0] - 1.0
        actual_mom = mom.loc[t]
        # 允许浮点误差
        ok_mom = np.allclose(expected_mom.fillna(-999), actual_mom.fillna(-999), atol=1e-10)
        # 排名校验：按实际 mom 重排一次
        expected_rank = actual_mom.rank(ascending=False, method="min")
        ok_rank = expected_rank.fillna(-1).equals(rank.loc[t].fillna(-1))
        report.append((t, "通过" if (ok_mom and ok_rank) else "失败", ok_mom, ok_rank))
    return pd.DataFrame(report, columns=["调仓日", "结论", "mom一致", "rank一致"])

示例数据与预期结果

用 5 个交易日、3 只 ETF 的极小面板做手工核对，lookback=2：

以 D3 为调仓日、lookback=2：分母取 D1 价格。A: 1.20/1.00-1=+20.0%；B: 2.05/2.00-1=+2.5%；C: 1.60/1.50-1=+6.67%。排名为 A=1、C=2、B=3。若误用自然日位移或误把 D3 当天排除，结果会明显偏离，这正是边界验证要抓的错误。

执行检查清单

• 价格面板索引为 DatetimeIndex 且已排序、无重复交易日。
• lookback 为正整数，且小于等于面板有效样本长度。
• 调仓日集合是面板索引的子集，缺失时抛错而非静默跳过。
• 排名矩阵中 NaN 数量与窗口缺失的 ETF 数量一致。
• 抽样 3 个调仓日，手工复算 P[t]/P[t-L]-1 与排名，与程序输出一致。