1. 行业轮动数据基座构建与质量校验

任务边界与数据对象

本例要交付的不是某个因子，而是一张可直接进入轮动分析的宽表：行索引为交易日，列索引为申万一级行业，单元格为后复权收盘价。同时附一份数据质量报告，记录每个行业的有效样本数、缺失区间、异常跳变与复权一致性检查结果。后续所有动量、相对强弱、分层回测都从这张宽表出发，因此基座一旦错位，后面全部结论都会失真。

关键约束有三条：其一，各行业指数上市日期不同，早期存在结构性缺失，不能简单用 0 填充；其二，交易日历必须统一，停牌或未上市造成的缺口要区分对待；其三，复权口径必须一致，若混用不复权与后复权价格，跨期收益率会出现虚假跳变。

加载与字段规范化

申万行业指数的原始导出通常包含 date、code、name、open、high、low、close、volume 等字段，且不同来源的日期格式可能是 2020-01-02 或 20200102。第一步是把所有行业文件读入后统一成 trade_date（datetime64）与 close（float）两列，并剔除重复行。下面用 pandas 完成这一层，示例数据以 3 个行业演示，实际替换为 31 个行业即可。

import pandas as pd
import numpy as np

# 示例：三个行业的原始日行情（实际从文件/接口读取）
raw = {
    "801010": pd.DataFrame({
        "date": ["2020-01-02", "2020-01-03", "2020-01-06", "2020-01-07"],
        "close": [3200.5, 3225.1, 3210.8, 3250.6],
    }),
    "801020": pd.DataFrame({
        "date": ["2020-01-02", "2020-01-03", "2020-01-06", "2020-01-07"],
        "close": [2800.0, 2795.4, 2810.2, 2822.9],
    }),
    "801030": pd.DataFrame({
        "date": ["2020-01-03", "2020-01-06", "2020-01-07"],  # 首日缺失
        "close": [1500.2, 1512.7, 1508.3],
    }),
}

def normalize(df):
    df = df.copy()
    df["trade_date"] = pd.to_datetime(df["date"])
    df = df[["trade_date", "close"]].drop_duplicates("trade_date")
    return df.sort_values("trade_date").set_index("trade_date")

frames = {code: normalize(df) for code, df in raw.items()}
print({k: len(v) for k, v in frames.items()})
# {'801010': 4, '801020': 4, '801030': 3}

日历对齐与缺失值处理

对齐的核心是先建立统一交易日历，再 reindex。这样每个行业在日历上的位置一致，缺失位置为 NaN，可以明确区分“未上市/停牌”与“数据丢失”。随后做前向填充，但必须限制填充长度：连续缺失超过 5 个交易日说明该行业可能长期停牌或数据源中断，继续填充会制造虚假的零波动区间，应保留 NaN 并在质量报告中标记。

# 统一交易日历：取所有行业日期的并集（实际应使用交易所日历）
calendar = pd.DatetimeIndex(
    sorted(set().union(*[f.index for f in frames.values()]))
)

wide = pd.DataFrame(index=calendar)
for code, df in frames.items():
    wide[code] = df["close"].reindex(calendar)

# 前向填充，最多 5 个交易日
wide_filled = wide.ffill(limit=5)

# 记录填充情况
fill_report = pd.DataFrame({
    "原始缺失数": wide.isna().sum(),
    "填充后缺失数": wide_filled.isna().sum(),
    "填充数": wide.isna().sum() - wide_filled.isna().sum(),
})
print(fill_report)

示例中 801030 在 2020-01-02 缺失，属于上市前结构性缺失，前向填充无法处理（前面没有值），因此填充后仍为 NaN，这是正确行为。若某行业在中间连续缺失 6 天，则第 6 天起保持 NaN，质量报告会将其列为“长缺口待核查”。

复权校验与异常跳变检查

复权校验分两步。第一步用复权因子表反推：后复权价 ≈ 不复权价 × 累计因子，若相对偏差超过 1e-6 则判定口径不一致。第二步做单日涨跌幅扫描，行业指数单日涨跌幅超过 11%（排除涨跌停日）通常意味着复权因子切换或数据拼接错误。下面给出可运行的校验函数。

def check_adjust(wide_df, factor_df, tol=1e-6):
    """factor_df: index=trade_date, columns=行业代码, 值为累计后复权因子"""
    issues = []
    for code in wide_df.columns:
        if code not in factor_df.columns:
            issues.append((code, "缺少复权因子"))
            continue
        ratio = wide_df[code] / factor_df[code]
        # 同一行业 ratio 应近似常数（不复权价/因子 = 常数）
        rel_dev = (ratio - ratio.median()).abs() / ratio.median()
        bad = rel_dev[rel_dev > tol]
        if len(bad) > 0:
            issues.append((code, f"复权偏差 {len(bad)} 天，首日 {bad.index[0].date()}"))
    return issues

def check_jumps(wide_df, threshold=0.11):
    """返回单日涨跌幅超过阈值的 (行业, 日期, 涨跌幅)"""
    ret = wide_df.pct_change()
    mask = ret.abs() > threshold
    out = []
    for code in ret.columns:
        for dt in ret.index[mask[code].fillna(False)]:
            out.append((code, dt.date(), round(float(ret.loc[dt, code]), 4)))
    return out

# 示例：构造一个与价格匹配的因子表（仅演示校验逻辑）
factor_demo = pd.DataFrame(1.0, index=calendar, columns=wide.columns)
print("复权问题:", check_adjust(wide_filled, factor_demo))
print("异常跳变:", check_jumps(wide_filled))

输出宽表与质量报告

最终交付两张表：wide_filled（宽表，index 为交易日，columns 为行业代码，值为后复权收盘价）和一份质量报告。质量报告建议至少包含以下字段，便于后续排查：

def build_quality_report(wide_raw, wide_filled, adjust_issues, jump_list):
    rows = []
    for code in wide_raw.columns:
        s_raw = wide_raw[code]
        s_fill = wide_filled[code]
        # 最长连续缺失
        is_na = s_raw.isna().astype(int)
        groups = (is_na != is_na.shift()).cumsum()
        max_gap = is_na.groupby(groups).sum().max() if len(is_na) else 0
        rows.append({
            "行业代码": code,
            "有效样本数": int(s_fill.notna().sum()),
            "缺失率": round(float(s_raw.isna().mean()), 4),
            "最长连续缺失": int(max_gap),
            "复权校验": "通过" if not any(i[0] == code for i in adjust_issues) else "偏差",
            "异常跳变": sum(1 for j in jump_list if j[0] == code),
        })
    return pd.DataFrame(rows)

report = build_quality_report(wide, wide_filled, [], [])
print(report.to_string(index=False))

运行后应得到每个行业一行记录。判读规则：缺失率超过 5% 或最长连续缺失超过 5 的行业，需回到原始数据确认是否上市较晚或长期停牌；复权校验为“偏差”的行业，必须修正因子表后重跑；异常跳变数大于 0 的，逐条核对日期与公告。三项全部通过后，wide_filled 才可作为后续轮动分析的输入基座。