多源价格冲突时先写一个 Quote Evidence Adapter-育师

一、问题：两个价格，该信谁

你给开源项目接了两个行情数据源。同一个 symbol，同一时间附近，一个返回 100.12，另一个返回 100.18。

接口都返回了 200。价格都看起来正常。你的第一反应可能是：其中有一个错了。第二反应可能是：我熟的那个应该更准。

这两个反应都合理。但它们都在把你带向同一个坑——在没有证据的情况下选边站。这次你选对了，下次呢？一旦你选了其中一个作为“信任源”，另一个数据源的所有异常都会被你解释为“它本来就不准”。这是一个自我强化的偏见循环，不是数据验收流程。

多源冲突时，真正要查的不是谁更准，而是谁的证据链更完整。价格本身不会告诉你它从哪来、什么时间生成的、市场当时是什么状态。但这些信息，才是判断一条数据能不能用的真正依据。

二、为什么“谁更准”是一个坏问题

两个价格冲突时，直接问“谁更准”，你会被三个东西困住。

第一，你不知道这两个价格的生成条件是否一致。一个可能是盘中实时成交价，另一个可能是几秒前的快照。一个可能是交易所原始价格，另一个可能是经过复权处理的价格。拿这两个数字直接比较，相当于拿不同时间、不同口径的测量结果比大小——结论没有意义。

第二，“谁更准”隐含一个假设——其中一个数据源整体优于另一个。但数据源的表现不是均匀的。一个数据源可能在美股上延迟更低，另一个可能在港股上状态标记更完整。用整体判断覆盖所有场景，会把真正的数据质量问题掩盖掉。

第三，也是最重要的——“谁更准”不需要你去看原始返回。你只需要比两个数字。但真正的冲突原因，从来不在价格数字本身，而在价格旁边的那些字段里：symbol 是不是被悄悄改了后缀？时间戳是哪个时刻？市场状态是不是一个在盘中、一个在盘后？这些信息不查，你永远不知道为什么冲突。

正确的问题不是“谁更准”，而是：这两条数据，哪一条能让我追溯到它是什么时候、在什么市场状态下、以什么口径生成的？

三、设计一个 Quote Evidence Adapter

解决这个问题的工程手段，不是写一个“选数据源”的规则引擎，而是在数据源和下游消费者之间加一层Quote Evidence Adapter。它的职责只有一条：把每条价格数据连同它的元信息封装成一条可追溯的证据记录。

3.1 模块目录结构

quote_adapter/ normalizer.py # symbol 规范化、时间戳对齐 evidence.py # QuoteEvidence 数据结构定义 conflict_checker.py # 多源冲突检测与仲裁 tests/ test_normalizer.py test_evidence.py test_conflict_checker.py

3.2 QuoteEvidence 数据结构

这是整个模块的核心。每一条从数据源获取的价格，都包装成这个结构再往下游传递。不丢失任何可以用于仲裁的信息。

from dataclasses import dataclass, field from typing import Optional, Dict, Any @dataclass class QuoteEvidence: """一条行情报价的证据记录。 不判断价格是否正确，只保留所有可用于仲裁的元信息。 """ source: str # 数据源标识，如 "tickdb"、"source_b" requested_symbol: str # 请求时使用的 symbol returned_symbol: str # 数据源返回的原始 symbol normalized_symbol: str # 规范化后的 symbol，用于跨源比对 price: Optional[str] # 原始价格字符串 timestamp: Optional[int] # 数据源返回的时间戳 timezone: Optional[str] # 时间戳时区，如 "UTC" market_status: Optional[str] # 市场状态，如 "open"、"closed"、"pre_market" raw_snapshot: Dict[str, Any] # 数据源完整原始返回

字段说明：

source：区分这条记录来自哪个数据源，所有后续仲裁都依赖这个标识。
requested_symbol/returned_symbol/normalized_symbol：请求的 symbol、实际返回的 symbol 和规范化后的可比 symbol。这三者不同的情况，就是冲突的第一手证据。
price：原始字符串。不做类型转换，保留数据源原始形态。
timestamp/timezone：时间戳及其时区。没有时区标注的时间戳，在跨市场对比中不可用。
market_status：如果数据源提供市场状态字段，这是仲裁的第一优先级——盘中和盘后价格不可直接比较。
raw_snapshot：完整原始返回。冲突无法解释时，这是最后一条可追溯的证据。

3.3 normalizer.py：symbol 规范化

不同数据源对同一个品种的 symbol 表示可能不同——有的带后缀，有的不带，有的加了前导零。symbol 规范化把这些差异统一为一个可比较的标识。

def normalize_symbol(raw_symbol: str, source: str) -> str: """根据数据源的 symbol 规则，规范化为统一格式。 各数据源的规范化规则需要按实际返回结构单独实现。 当前为示例骨架，需根据接入的数据源补充具体映射逻辑。 """ # 示例：去除前后空格、统一大写 normalized = raw_symbol.strip().upper() return normalized

3.4 evidence.py：构造证据记录

from typing import Dict, Any, Optional from .normalizer import normalize_symbol def build_evidence( source: str, requested_symbol: str, raw_response: Dict[str, Any], extract_symbol_fn, extract_price_fn, extract_timestamp_fn, extract_market_status_fn=None, timezone: str = "UTC", ) -> QuoteEvidence: """从数据源原始返回构造一条 QuoteEvidence。 各数据源的字段提取函数需按实际返回结构实现。 """ returned_symbol = extract_symbol_fn(raw_response) normalized = normalize_symbol(returned_symbol, source) price = extract_price_fn(raw_response) timestamp = extract_timestamp_fn(raw_response) market_status = None if extract_market_status_fn: market_status = extract_market_status_fn(raw_response) return QuoteEvidence( source=source, requested_symbol=requested_symbol, returned_symbol=returned_symbol, normalized_symbol=normalized, price=price, timestamp=timestamp, timezone=timezone, market_status=market_status, raw_snapshot=raw_response, )

3.5 conflict_checker.py：冲突检测与仲裁

冲突检测器的职责不是选出“正确的价格”，而是判断两条证据记录是否具备可比性。不可比较时，给出具体原因。

from typing import List, Optional from .evidence import QuoteEvidence class ConflictResult: """两条证据记录的对比结果。""" def __init__(self): self.comparable: bool = False self.conflicts: List[str] = [] self.reason: Optional[str] = None def check_conflict(a: QuoteEvidence, b: QuoteEvidence) -> ConflictResult: """检查两条 QuoteEvidence 是否可比较，以及是否存在冲突。 不判断哪条数据更准确，只提供冲突检查结果。 """ result = ConflictResult() # 检查 normalized_symbol 是否一致 if a.normalized_symbol != b.normalized_symbol: result.reason = ( f"symbol 不一致: {a.normalized_symbol} vs {b.normalized_symbol}" ) return result # 检查市场状态是否一致 if a.market_status and b.market_status: if a.market_status != b.market_status: result.reason = ( f"市场状态不一致: {a.market_status} vs {b.market_status}，" f"无法直接比较价格" ) return result # 检查时间戳是否在同一可接受窗口内 if a.timestamp and b.timestamp and a.timezone and b.timezone: if a.timezone != b.timezone: result.reason = ( f"时区不一致: {a.timezone} vs {b.timezone}，" f"时间戳不可直接比较" ) return result # 标记为可比较 result.comparable = True # 如果价格不同，记录冲突 if a.price and b.price and a.price != b.price: result.conflicts.append( f"价格冲突: {a.source}={a.price}, {b.source}={b.price}" ) return result

3.6 Smoke Test 思路

不追求完整测试覆盖率。只验证模块的核心链路能否跑通。

def test_smoke_two_sources(): """Smoke test：两个数据源输入，检查是否可比较。 不可比较时输出 reason，不做价格判断。 """ # 构造两条来自不同数据源的证据记录 evidence_a = QuoteEvidence( source="source_a", requested_symbol="AAPL.US", returned_symbol="AAPL.US", normalized_symbol="AAPL.US", price="100.12", timestamp=1719500000000, timezone="UTC", market_status="open", raw_snapshot={"symbol": "AAPL.US", "last_price": "100.12"}, ) evidence_b = QuoteEvidence( source="tickdb", requested_symbol="AAPL.US", returned_symbol="AAPL.US", normalized_symbol="AAPL.US", price="100.18", timestamp=1719500001000, timezone="UTC", market_status="open", raw_snapshot={"symbol": "AAPL.US", "last_price": "100.18"}, ) result = check_conflict(evidence_a, evidence_b) if result.comparable: print(f"可比较，冲突: {result.conflicts}") else: print(f"不可比较，原因: {result.reason}")

四、TickDB 在这类场景里的合理位置

上面这套 Quote Evidence Adapter，是一套多源冲突时的排查框架。不管你用哪个行情数据源，这套框架都能用。问题在于：如果数据源本身的字段定义不清晰、异常情况没有约定、原始返回不可追溯，你光是把这套框架跑通，就要花大量精力在猜数据和补日志上。

TickDB 在这里能做的，是让你在排查多源冲突时，至少有一端的数据是有清晰证据链的。它适合被放进数据质量验收流程里，作为一个字段契约清晰、便于逐项核对的候选行情入口。多源冲突时，你拿 TickDB 的返回和其他数据源对比，至少 TickDB 这一端的 symbol、时间戳、状态标记和异常返回是明确可查的——你可以把精力花在分析冲突原因上，而不是花在猜“这一端的数据到底是什么意思”上。

验证方法：用你自己的 symbol 跑一次请求。保存请求参数、原始返回、检查时间。对照 symbol、时间戳、市场状态、字段口径、异常返回逐项核对。如果要做多源对比，不要只比价格——比证据链。

不适合什么：不替代投资判断。不替代生产监控和异常回放。不用来证明某个数据源永远更准。数据验收的责任，最终还是在你自己手上。