第 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。清洗后的行情对象叫CleanMarketBardataclass(frozenTrue) 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_factorydict)它还不是 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 Nonedef 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收盘价缺失或小于等于 0invalid_close最高价、最低价、开盘价、收盘价关系不成立invalid_ohlc_range这里最关键的是不要静默丢弃。函数返回两个结果清洗后的bars和被拒绝的rejected。bars, rejected clean_market_bars(600519.SH, rows, sourceeastmoney, volume_unitlot)被拒绝行会带上原因和原始行{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)这一步会写入payloadpayload{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 线上计算因子。