程序员量化交易实战 08：把原始 K 线清洗成可信行情

第 7 篇得到了一批干净的股票。接下来要处理行情。

行情数据最容易让人掉以轻心。看起来就是日期、开高低收、成交量、成交额几列，但供应商格式、单位、缺失值和异常行只要有一点没处理好，回测就会悄悄偏掉。

原始 K 线不能直接入库

同样是日线，不同来源可能长这样：

日期, 开盘, 最高, 最低, 收盘, 成交量
2026-06-15, 10.0, 10.8, 9.9, 10.5, 120

也可能长这样：

{"date": "20260615", "open": "10.0", "high": "10.8", "low": "9.9", "close": "10.5", "volume": "12000"}

成交量的单位还可能是“手”，也可能是“股”。如果不统一，成交量参与率、流动性过滤和回测成交约束都会错。

先定义清洗后的对象

第 8 章新增 app/market_data.py。清洗后的行情对象叫 CleanMarketBar：

@dataclass(frozen=True)
class CleanMarketBar:
    symbol: str
    trade_date: date
    open: float
    high: float
    low: float
    close: float
    volume: float
    amount: float
    source: str
    payload: dict[str, Any] = field(default_factory=dict)

它还不是 ORM 对象。它位于“供应商原始数据”和“数据库写入”之间，职责很清楚：把一批原始行变成统一口径。

日期和数字解析要宽容

供应商字段很少完全一致。日期可能是 2026-06-16、20260616 或 2026/06/16。数字里可能有逗号、百分号、空值和 --。

所以先写两个很小的解析函数：

def parse_trade_date(value: Any) -> date | None:
    if isinstance(value, date):
        return value
    text = str(value or "").strip().replace("/", "-")
    for fmt in ("%Y-%m-%d", "%Y%m%d"):
        try:
            return datetime.strptime(text[:10] if fmt == "%Y-%m-%d" else text, fmt).date()
        except ValueError:
            continue
    return None

def parse_number(value: Any) -> float | None:
    text = str(value).replace(",", "").replace("%", "").strip()
    if not text or text in {"-", "--", "nan", "None"}:
        return None
    return float(text)

工程上要注意：解析可以宽容，但后续校验要严格。

OHLC 必须自洽

第 8 章的 clean_market_bars() 会检查几类问题：

• 没有交易日：missing_trade_date
• 同一天重复：duplicate_trade_date
• 收盘价缺失或小于等于 0：invalid_close
• 最高价、最低价、开盘价、收盘价关系不成立：invalid_ohlc_range

这里最关键的是不要静默丢弃。函数返回两个结果：清洗后的 bars 和被拒绝的 rejected。

bars, rejected = clean_market_bars("600519.SH", rows, source="eastmoney", volume_unit="lot")

被拒绝行会带上原因和原始行：

{"reason": "invalid_ohlc_range", "row": row}

这对排查供应商字段变化很有用。真实系统里，rejected 后续可以进入审计日志或数据任务结果。

成交量单位统一成股

A 股里常见的“手”是 100 股。如果供应商返回的是手，系统内部要统一转成股：

volume_multiplier = 100 if volume_unit == "lot" else 1
normalized_volume = round(volume * volume_multiplier, 4)

这一步会写入 payload：

payload={"raw": row, "volume_unit": "shares"}

也就是说，内部对象明确表达“成交量已经是股”。原始行仍然保留在 payload，后面发现问题可以追溯。

覆盖率报告

清洗完以后，还要有一个最小覆盖率摘要：

def coverage_report(bars: Iterable[CleanMarketBar]) -> dict[str, object]:
    rows = list(bars)
    dates = [row.trade_date for row in rows]
    symbols = sorted({row.symbol for row in rows})
    return {
        "rows": len(rows),
        "symbols": len(symbols),
        "first_date": min(dates).isoformat() if dates else None,
        "latest_date": max(dates).isoformat() if dates else None,
        "sources": sorted({row.source for row in rows}),
    }

这是后续 /api/data/quality 的小版本。写策略之前，先知道自己有多少行、覆盖多少股票、最早和最新日期是什么。

本篇实战任务

拉取第 8 章代码：

git clone https://github.com/ax2/zi-quant-platform.git
cd zi-quant-platform
git checkout chapter-08
uv sync --extra dev
uv run pytest

只跑行情清洗测试：

uv run pytest tests/test_market_data.py

第 8 章全量测试通过：159 passed，仍只有既有 FastAPI deprecation warning。

本章更新与代码仓库

本章更新内容：

• 新增 app/market_data.py。
• 实现交易日解析、数字解析、OHLC 校验、重复行拒绝、成交量单位统一和覆盖率摘要。
• 新增 tests/test_market_data.py，验证清洗结果和拒绝原因。

代码仓库：

https://github.com/ax2/zi-quant-platform

本章代码：

git clone https://github.com/ax2/zi-quant-platform.git
cd zi-quant-platform
git checkout chapter-08
uv sync --extra dev
uv run pytest tests/test_market_data.py

本篇小结

行情数据的重点不是“下载到了”，而是“清洗后能不能解释”。

这一篇把日期解析、数字解析、OHLC 校验、重复行处理、成交量单位和覆盖率摘要做成纯函数。下一篇开始在这些干净 K 线上计算因子。