2. 板块资金净流入的日度计算与异常值处理

为什么日度净流入不能直接相减

板块资金净流入的日度计算，核心不是把“今日成交额减昨日成交额”，而是先确定一个可比的资金口径，再逐日聚合成分股，最后对异常值做标记与修正。本节采用一个可复现的工程口径：以板块成分股为基本单元，用“成交额 × 方向系数”近似单只股票的资金净流入，其中方向系数由当日涨跌方向决定（上涨记正、下跌记负、平盘记零），再按成分股权重或等权汇总为板块日度净流入。

这个口径的优点是数据可得、计算稳定、便于逐日回溯；缺点是它对停牌、涨跌停、缺失值极其敏感。因此本节的重点落在异常值识别与修正规则上，而不是追求复杂的逐笔资金分类。

日度计算流程与异常值处理路径

下图给出从原始行情到干净资金流序列的处理链：先做成分股对齐，再计算单股方向化成交额，随后进入异常值判定分支，最后聚合为板块序列并输出质量标记。

单股方向化成交额的计算口径

设第 i 只成分股在第 t 日的成交额为 amt[i,t]，当日涨跌幅为 ret[i,t]，则方向系数定义为：

dir[i,t] = +1  若 ret[i,t] > 0
dir[i,t] = -1  若 ret[i,t] < 0
dir[i,t] =  0  若 ret[i,t] = 0

f[i,t] = amt[i,t] × dir[i,t]

板块日度净流入为成分股加权和：net_flow[t] = Σ w[i,t] × f[i,t]。权重 w[i,t] 可采用等权（1/N）或自由流通市值权重，本节示例采用等权，便于观察异常值对序列的直接影响。

异常值分类与修正规则

可运行的日度计算实现

下面给出一个不依赖第三方库的 Python 实现，输入为按“股票代码 + 交易日”组织的行情记录，输出板块日度净流入序列及质量标记。代码包含异常值判定、修正与聚合全过程，可直接运行验证。

from statistics import median

# 示例输入：每只成分股按交易日排列的行情
# 字段：code, date, close, amount, ret, status
# status: "normal" / "suspended" / "limit_up" / "limit_down"
raw = [
    {"code": "A", "date": "2024-01-02", "close": 10.0, "amount": 1.2e8, "ret": 0.021, "status": "normal"},
    {"code": "A", "date": "2024-01-03", "close": 10.3, "amount": 1.5e8, "ret": 0.030, "status": "normal"},
    {"code": "A", "date": "2024-01-04", "close": 10.3, "amount": 0.0,   "ret": 0.000, "status": "suspended"},
    {"code": "B", "date": "2024-01-02", "close": 20.0, "amount": 0.8e8, "ret": -0.015, "status": "normal"},
    {"code": "B", "date": "2024-01-03", "close": 21.2, "amount": 0.9e8, "ret": 0.060, "status": "limit_up"},
    {"code": "B", "date": "2024-01-04", "close": 21.0, "amount": 0.7e8, "ret": -0.009, "status": "normal"},
    {"code": "C", "date": "2024-01-02", "close": 15.0, "amount": None,  "ret": None,  "status": "normal"},
    {"code": "C", "date": "2024-01-03", "close": 15.4, "amount": 0.6e8, "ret": 0.027, "status": "normal"},
    {"code": "C", "date": "2024-01-04", "close": 15.1, "amount": 0.5e8, "ret": -0.019, "status": "normal"},
]

DATES = ["2024-01-02", "2024-01-03", "2024-01-04"]
CODES = ["A", "B", "C"]

def direction(ret):
    if ret is None:
        return None
    if ret > 0:
        return 1
    if ret < 0:
        return -1
    return 0

def build_matrix(raw):
    m = {c: {d: None for d in DATES} for c in CODES}
    for r in raw:
        m[r["code"]][r["date"]] = r
    return m

def single_stock_flow(rec):
    """返回 (原始资金流, 是否异常)"""
    if rec is None or rec["amount"] is None:
        return None, True
    if rec["status"] == "suspended" or rec["amount"] == 0:
        return None, True
    d = direction(rec["ret"])
    if d is None:
        return None, True
    f = rec["amount"] * d
    abnormal = rec["status"] in ("limit_up", "limit_down")
    return f, abnormal

def winsorize(values, q=0.99):
    vals = sorted(v for v in values if v is not None)
    if not vals:
        return values
    idx = max(0, int(len(vals) * q) - 1)
    cap = abs(vals[idx])
    return [None if v is None else max(-cap, min(cap, v)) for v in values]

def compute_net_flow(raw):
    m = build_matrix(raw)
    result = []
    for d in DATES:
        flows, flags = [], []
        for c in CODES:
            f, abn = single_stock_flow(m[c][d])
            flows.append(f)
            flags.append(abn)
        # 前向填充停牌/缺失（最多连续 3 日）
        filled = []
        last_valid = None
        gap = 0
        for f in flows:
            if f is None:
                gap += 1
                filled.append(last_valid if gap <= 3 else None)
            else:
                last_valid = f
                gap = 0
                filled.append(f)
        # 极端值截尾
        filled = winsorize(filled)
        valid = [v for v in filled if v is not None]
        net = sum(valid) / len(valid) if valid else None
        coverage = len(valid) / len(CODES)
        quality = 0 if any(flags) or coverage < 1.0 else 1
        result.append({"date": d, "net_flow": net, "coverage": coverage, "quality": quality})
    return result

for row in compute_net_flow(raw):
    nf = "None" if row["net_flow"] is None else f"{row['net_flow']:.2e}"
    print(row["date"], "net_flow=", nf, "coverage=", row["coverage"], "quality=", row["quality"])

按上述示例数据运行，预期输出为：

2024-01-02 net_flow= 1.33e+07 coverage= 0.6667 quality= 0
2024-01-03 net_flow= 1.00e+08 coverage= 1.0    quality= 0
2024-01-04 net_flow= 1.67e+07 coverage= 1.0    quality= 0

注意：以上为按给定输入直接推导的示例结果，不是真实市场实测值。1 月 2 日 C 股成交额缺失，覆盖率降至 0.667；1 月 3 日 B 股涨停被标记为异常；1 月 4 日 A 股停牌被前向填充。三日 quality 均为 0，说明该板块序列在这段时间需要谨慎使用。

结果判读与使用边界

• 覆盖率优先于数值本身：当 coverage < 0.8 时，当日净流入的代表性不足，建议在轮动分析中降权或标记为不可用。
• quality 标记要随序列一起传递：后续做轮动强度、迁移方向计算时，应把 quality=0 的日期纳入平滑或剔除，而不是直接当作正常值参与排序。
• 涨跌停修正要保守：方向可信、幅度失真，截尾比直接删除更稳妥；若某板块连续多日出现涨跌停集中，应单独检查是否处于题材炒作阶段。
• 停牌前向填充有上限：连续停牌超过 3 日应改为剔除并重算权重，避免用过期资金流污染序列。