程序员量化交易实战 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的小版本。写策略之前，先知道自己有多少行、覆盖多少股票、最早和最新日期是什么。

可运行基础校验

行情清洗的校验要同时看到覆盖率和拒绝原因。当前统一用这条命令复现第 01-08 篇的基础能力：

uv run python -m scripts.chapter_examples foundation-check

本章对应输出如下：

截图里的覆盖率说明清洗后剩下 2 行、1 只股票、日期从2026-01-02到2026-01-05。被拒绝的那行保留invalid_ohlc_range，说明最高价低于开盘价时不会被悄悄写入后续回测。

本篇实战任务

拉取第 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 线上计算因子。