给开源项目接行情 API：多源价格冲突时先写一个 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.pytest_evidence.pytest_conflict_checker.py3.2 QuoteEvidence 数据结构这是整个模块的核心。每一条从数据源获取的价格都包装成这个结构再往下游传递。不丢失任何可以用于仲裁的信息。from dataclasses import dataclass, fieldfrom typing import Optional, Dict, Anydataclassclass 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.pysymbol 规范化不同数据源对同一个品种的 symbol 表示可能不同——有的带后缀有的不带有的加了前导零。symbol 规范化把这些差异统一为一个可比较的标识。def normalize_symbol(raw_symbol: str, source: str) - str:“”根据数据源的 symbol 规则规范化为统一格式。各数据源的规范化规则需要按实际返回结构单独实现。 当前为示例骨架需根据接入的数据源补充具体映射逻辑。 # 示例去除前后空格、统一大写 normalized raw_symbol.strip().upper() return normalized3.4 evidence.py构造证据记录from typing import Dict, Any, Optionalfrom .normalizer import normalize_symboldef build_evidence(source: str,requested_symbol: str,raw_response: Dict[str, Any],extract_symbol_fn,extract_price_fn,extract_timestamp_fn,extract_market_status_fnNone,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( sourcesource, requested_symbolrequested_symbol, returned_symbolreturned_symbol, normalized_symbolnormalized, priceprice, timestamptimestamp, timezonetimezone, market_statusmarket_status, raw_snapshotraw_response, )3.5 conflict_checker.py冲突检测与仲裁冲突检测器的职责不是选出“正确的价格”而是判断两条证据记录是否具备可比性。不可比较时给出具体原因。from typing import List, Optionalfrom .evidence import QuoteEvidenceclass ConflictResult:“”“两条证据记录的对比结果。”“”definit(self):self.comparable: bool Falseself.conflicts: List[str] []self.reason: Optional[str] Nonedef check_conflict(a: QuoteEvidence, b: QuoteEvidence) - ConflictResult:“”检查两条 QuoteEvidence 是否可比较以及是否存在冲突。不判断哪条数据更准确只提供冲突检查结果。 result ConflictResult() # 检查 normalized_symbol 是否一致 if a.normalized_symbol ! b.normalized_symbol: result.reason ( fsymbol 不一致: {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 result3.6 Smoke Test 思路不追求完整测试覆盖率。只验证模块的核心链路能否跑通。def test_smoke_two_sources():“”Smoke test两个数据源输入检查是否可比较。不可比较时输出 reason不做价格判断。 # 构造两条来自不同数据源的证据记录 evidence_a QuoteEvidence( sourcesource_a, requested_symbolAAPL.US, returned_symbolAAPL.US, normalized_symbolAAPL.US, price100.12, timestamp1719500000000, timezoneUTC, market_statusopen, raw_snapshot{symbol: AAPL.US, last_price: 100.12}, ) evidence_b QuoteEvidence( sourcetickdb, requested_symbolAAPL.US, returned_symbolAAPL.US, normalized_symbolAAPL.US, price100.18, timestamp1719500001000, timezoneUTC, market_statusopen, 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、时间戳、市场状态、字段口径、异常返回逐项核对。如果要做多源对比不要只比价格——比证据链。不适合什么不替代投资判断。不替代生产监控和异常回放。不用来证明某个数据源永远更准。数据验收的责任最终还是在你自己手上。