一、调用成功 ≠ 数据可提交一个常见的工程误区是把 HTTP 200 和code0直接当成“数据没问题”。实际链路中它们只证明服务端接受了请求并返回了响应。K 线数据在入库前至少还需要过五道检查身份返回的symbol、interval是否与请求一致结构必填字段是否齐全klines是否为数组时间时间戳是否为合法正毫秒整数批内是否有重复数值所有 OHLC 和成交量字段能否解析为有限 Decimal行内关系low high且open、close在高低区间内。四层成功定义传输成功HTTP 200响应体可解析为 JSON。业务成功code0服务端表示请求已受理。结构成功data为 dictsymbol/interval与请求一致klines为 list每行必填字段齐全。可提交结构成功 整批通过时间类型、Decimal finite、重复检测和 OHLC 关系校验。validate_kline_payload只处理已解析的 payload不检查 HTTP 状态或 JSON 解析——这些属于调用层职责。本文也不涉及 HTTP 请求和数据库落库只聚焦这一段独立功能已拿到 K 线 payload如何用显式规则判定它能否通过合同检查。二、校验器设计2.1 异常类型所有校验失败均抛出PayloadValidationError携带具体错误描述。调用方只需捕获这一种异常即可统一处理坏批次不暴露底层实现细节。2.2 空数组策略API 在某些条件下可能返回code0且klines[]。空数组是否可接受取决于业务场景回测可能视其为数据缺口展示则可能视为“暂无数据”。校验器不替调用方做决定通过allow_empty参数强制显式选择allow_emptyFalse空数组抛出PayloadValidationErrorallow_emptyTrue空数组返回空列表。未传入该参数将导致函数拒绝执行。2.3 时间字段严格拒绝布尔型Python 中isinstance(True, int)为True若只做isinstance(value, int)检查True和False会分别变成1和0并静默通过。校验器使用type(value) is int严格匹配且要求value 0。True、False、0、负数均触发PayloadValidationError。2.4 数值字段必须为字符串 Decimal is_finite所有价格和成交量字段必须先检查为str类型非字符串值立即抛错。之后使用Decimal(value)解析并检查is_finite()。不满足则抛错。缺失字段不会默认为Decimal(0)——缺失就是缺失零是零二者不可混淆。2.5 行内关系每根 K 线必须满足low highlow open highlow close high违反任意一项即抛PayloadValidationError。三、完整代码以下代码为 Python 3.11 独立模块只处理已获取的 payload不发送 HTTP 请求不连接数据库。 kline_contract_check.py K 线 JSON 入库前的 Contract Check。 Python 3.11 from dataclasses import dataclass from decimal import Decimal, InvalidOperation from typing import List class PayloadValidationError(Exception): K 线 payload 校验失败。 pass dataclass class KLine: 单根 K 线所有数值字段为 Decimal。 time: int open: Decimal high: Decimal low: Decimal close: Decimal volume: Decimal quote_volume: Decimal def validate_kline_payload( payload: dict, requested_symbol: str, requested_interval: str, allow_empty: bool ) - List[KLine]: 校验 K 线 payload返回按 time 升序排列的 KLine 列表。 参数: payload: API 返回的完整 JSON 字典。 requested_symbol: 请求时使用的 symbol如 AAPL.US。 requested_interval: 请求时使用的 interval如 1d。 allow_empty: True 时允许 klines 为空数组False 时空数组抛错。 调用方必须显式选择策略。 返回: List[KLine]: 按 time 升序排列的 K 线列表。 异常: PayloadValidationError: 任意校验规则不通过。 # 1. 顶层结构 if not isinstance(payload, dict): raise PayloadValidationError(payload 不是 dict) if payload.get(code) ! 0: raise PayloadValidationError(fcode 不为 0: {payload.get(code)}) # 2. data 结构 data payload.get(data) if not isinstance(data, dict): raise PayloadValidationError(data 不是 dict) # 3. symbol / interval 身份 actual_symbol data.get(symbol) if actual_symbol ! requested_symbol: raise PayloadValidationError( fsymbol 不一致: 期望 {requested_symbol}实际 {actual_symbol} ) actual_interval data.get(interval) if actual_interval ! requested_interval: raise PayloadValidationError( finterval 不一致: 期望 {requested_interval}实际 {actual_interval} ) # 4. klines 数组 klines data.get(klines) if not isinstance(klines, list): raise PayloadValidationError(klines 不是 list) if len(klines) 0: if allow_empty: return [] else: raise PayloadValidationError(klines 为空数组且 allow_emptyFalse) # 5. 逐行校验 REQUIRED {time, open, high, low, close, volume, quote_volume} result: List[KLine] [] seen_times: set set() for i, row in enumerate(klines): if not isinstance(row, dict): raise PayloadValidationError(fklines[{i}] 不是 dict) # 必填字段存在性 missing REQUIRED - row.keys() if missing: raise PayloadValidationError(fklines[{i}] 缺少字段: {, .join(sorted(missing))}) # 时间字段严格拒绝 bool raw_time row[time] if type(raw_time) is not int or raw_time 0: raise PayloadValidationError( fklines[{i}].time 必须为正整数 int实际: {type(raw_time).__name__}{raw_time} ) # 批内重复时间 if raw_time in seen_times: raise PayloadValidationError(fklines[{i}].time 重复: {raw_time}) seen_times.add(raw_time) # 数值字段必须先为 str再解析为 Decimal is_finite for field_name in (open, high, low, close, volume, quote_volume): raw_val row[field_name] if not isinstance(raw_val, str): raise PayloadValidationError( fklines[{i}].{field_name} 必须为字符串实际: {type(raw_val).__name__} ) try: open_val Decimal(row[open]) high_val Decimal(row[high]) low_val Decimal(row[low]) close_val Decimal(row[close]) volume_val Decimal(row[volume]) quote_volume_val Decimal(row[quote_volume]) except (InvalidOperation, ValueError) as e: raise PayloadValidationError(fklines[{i}] 数值字段解析失败: {e}) for field_name, val in [ (open, open_val), (high, high_val), (low, low_val), (close, close_val), (volume, volume_val), (quote_volume, quote_volume_val) ]: if not val.is_finite(): raise PayloadValidationError( fklines[{i}].{field_name} 为非有限数值: {val} ) # 行内关系 if low_val high_val: raise PayloadValidationError( fklines[{i}] low ({low_val}) high ({high_val}) ) if not (low_val open_val high_val): raise PayloadValidationError( fklines[{i}] open ({open_val}) 不在 [{low_val}, {high_val}] 区间 ) if not (low_val close_val high_val): raise PayloadValidationError( fklines[{i}] close ({close_val}) 不在 [{low_val}, {high_val}] 区间 ) result.append(KLine( timeraw_time, openopen_val, highhigh_val, lowlow_val, closeclose_val, volumevolume_val, quote_volumequote_volume_val, )) # 6. 按 time 排序 result.sort(keylambda k: k.time) return result四、四项单元测试以下测试使用标准库unittest所有 payload 为最小构造数据不冒充实时行情。 test_kline_contract_check.py K 线校验器单元测试——四项最小覆盖。 Python 3.11 import unittest from kline_contract_check import validate_kline_payload, PayloadValidationError, KLine from decimal import Decimal class TestValidateKlinePayload(unittest.TestCase): def _base_payload(self, klines): 构造基础 payload 骨架。 return { code: 0, message: success, data: { symbol: AAPL.US, interval: 1d, klines: klines } } # 1. 正常 payload def test_valid_single_kline(self): 正常 payload 应返回一根 KLine。 payload self._base_payload([{ time: 1749398400000, open: 200.00, high: 205.00, low: 199.00, close: 204.00, volume: 1000, quote_volume: 204000.00 }]) result validate_kline_payload(payload, AAPL.US, 1d, allow_emptyFalse) self.assertEqual(len(result), 1) self.assertIsInstance(result[0], KLine) self.assertEqual(result[0].time, 1749398400000) self.assertEqual(result[0].open, Decimal(200.00)) # 2. 重复 time def test_duplicate_time_raises(self): 两条 K 线 time 相同时必须抛错。 payload self._base_payload([ {time: 1749398400000, open: 200, high: 205, low: 199, close: 204, volume: 1000, quote_volume: 204000}, {time: 1749398400000, open: 204, high: 206, low: 203, close: 205, volume: 1200, quote_volume: 246000} ]) with self.assertRaises(PayloadValidationError) as ctx: validate_kline_payload(payload, AAPL.US, 1d, allow_emptyFalse) self.assertIn(重复, str(ctx.exception)) # 3. 缺失 close def test_missing_close_raises(self): 缺少必填字段 close 时必须抛错。 payload self._base_payload([{ time: 1749398400000, open: 200, high: 205, low: 199, volume: 1000, quote_volume: 204000 }]) with self.assertRaises(PayloadValidationError) as ctx: validate_kline_payload(payload, AAPL.US, 1d, allow_emptyFalse) self.assertIn(缺少字段, str(ctx.exception)) self.assertIn(close, str(ctx.exception)) # 4. timeTrue def test_time_true_raises(self): timeTrue 时必须抛错布尔型不能静默通过。 payload self._base_payload([{ time: True, open: 200, high: 205, low: 199, close: 204, volume: 1000, quote_volume: 204000 }]) with self.assertRaises(PayloadValidationError) as ctx: validate_kline_payload(payload, AAPL.US, 1d, allow_emptyFalse) self.assertIn(time, str(ctx.exception).lower()) if __name__ __main__: unittest.main()正常 payload 用例预期返回 KLine重复 time、缺失 close、timeTrue 三项预期抛出 PayloadValidationError并检查相应错误信息。五、应用层与数据库层职责分工本文使用的校验规则按职责划分为应用层验证和数据库层约束。校验规则应用层Python数据库层PostgreSQL 示例symbol/interval 与请求一致✓ 逐次校验—必填字段齐全✓ 逐行校验NOT NULL数值为合法 Decimal 且有限✓ Decimal is_finite有限值需单独设计显式 CHECK本文不展开具体 SQL批内 time 无重复✓ 集合去重UNIQUE(symbol, interval, time) 要求相关列为 NOT NULLlow high, open/close 在区间内✓ 逐行校验CHECK空数组策略✓ allow_empty 显式选择—本文表格列出的 NOT NULL、UNIQUE 和行级 CHECK 约束不负责发现跨行时间缺口不检测调用方请求的 symbol 是否与返回一致不判断空数组的业务合理性。这些示例不代表 PostgreSQL 全部约束能力。六、部署前检查清单allow_empty参数是否根据业务场景显式传递时间字段是否使用type(value) is int严格拒绝bool数值字段是否先检查为str再经过Decimal(value)is_finite()字段缺失是否触发错误而非默认为零数据库表是否有 NOT NULL、CHECK 和 UNIQUE 约束作为第二道防线校验失败后的降级路径是否明确丢弃批次 / 死信队列 / 告警七、FAQQ1空数组到底该报错还是放行取决于调用方。回测场景中空数组可能代表数据缺口应报错展示场景中可能是合理的“暂无数据”。因此allow_empty不设默认值调用方必须显式选择。Q2为什么时间字段要严格拒绝 boolPython 中isinstance(True, int)为True只用isinstance检查会使True变成1、False变成0静默通过。type(value) is int精确排除了布尔型。Q3为什么价格用 Decimal 而不是 floatfloat是二进制浮点数0.1 0.2不等于0.3。Decimal 以十进制存储适合金融数据的精确校验。Q4数据库有 UNIQUE 约束了为什么应用层还要检测批内重复时间数据库约束在 INSERT 时触发此时坏数据已试图落库。应用层提前检测可更早定位错误行避免无
