K线接口返回 code=0，为什么数据仍不能直接写进数据库

TickDB Research ·

摘要

HTTP 200、code=0、字段存在和可安全入库是四个不同层次。K线数据从API到数据库，至少需要五层成功语义和七道闸门校验。本文给出一套可独立运行的Python校验器，只使用标准库，不发送请求、不连数据库，专门回答同一个问题：响应成功的数据，还需要检查什么才能安全提交。

以下是假设场景。

你写了一个脚本，每天从行情API拉取K线数据。接口返回HTTP 200，body里code=0，data.klines是个数组，长度也对。你把数组直接写入数据库，日志打印"写入成功"。

几天后回测异常。排查发现，某天同一时间戳出现了两条记录，某天high比low还低，另一天close字段缺失，你的代码用默认值0填了进去。接口每次都返回code=0。

快递已经送到门口，不代表箱内商品、收件人、数量和批次都正确。

code=0只证明服务端处理了你的请求，没有发生鉴权失败或参数错误。它不承诺返回数据满足你本地数据库的约束，不承诺字段完整，不承诺数值合理，也不承诺没有重复。HTTP 200、code=0、字段存在和可安全入库，每一层都在回答不同的问题。

五层成功语义

一次K线请求从发出到数据可安全入库，至少经过五层判断。每一层通过只说明上一层没出问题，不代表下一层自动成立。

层级含义通过标准失败意味着什么
传输成功HTTP请求得到响应状态码200，body可解析为JSON网络或服务端不可达；重试或退出
业务成功服务端处理了请求code=0，无业务层错误鉴权、限流或参数错误；根据错误码重试或阻断
schema合法返回结构与文档一致字段存在、类型正确、klines为数组响应格式与契约不符；不入库，需排查上游
语义正确数据满足当前任务契约数值可解析、行内关系成立、无重复时间数据可能损坏或不符合任务要求；不入库
整批可提交当前批次可安全写入整批校验通过，无部分成功任一行失败则整批拒绝，避免部分数据入库

前两层是API的职责边界。后三层是调用方的工程责任。API不会替你做schema校验，更不会理解你的任务契约。

七道闸门

把一次K线响应从JSON变成数据库里的行，需要依次通过七道校验。任何一道失败，整批数据不应入库。每道闸门都回答三个问题：检查什么，失败意味着什么，代码如何关闭。

第一道：响应信封

HTTP状态检查和JSON解析属于调用层职责，不在validate_kline_payload范围内。本闸门只校验已解析的payload：顶层code必须为0，data必须为字典。code不等于0时根据错误码分类处理——1001、1002、1004阻断重试，3001按Retry-After退避，其他业务码非零退出。校验失败抛出EnvelopeError。

失败意味着请求本身没有成功到达业务层，数据不可用。

第二道：请求身份

检查返回的data.symbol和data.interval是否与请求参数一致。如果请求AAPL.US和1d，返回的这两个字段也必须匹配。不匹配抛出IdentityError。

失败意味着响应答非所问，可能上游路由错误或API版本不兼容。整批不入库。

第三道：数组形态

检查data.klines是否存在且为list类型。若为空数组，进入明确的no-data分支——不抛出异常，返回空列表。空数组不能简单等同于"正常"，需由调用方根据交易日历和任务上下文判断是预期内无数据还是异常缺失。

失败意味着返回结构不符合契约，或字段缺失。抛出ArrayShapeError。

第四道：时间身份

检查每根K线的time字段：type(time_ms) is int and time_ms > 0。批内所有time不能重复。重复抛出TimeIdentityError。

失败意味着同一根K线被多次返回，或时间戳格式异常。数据库主键可以阻止重复写入，但不能在入库前发现这个问题。

第五道：数值解析

open、high、low、close、volume、quote_volume必须可解析为Decimal，且is_finite()为True。价格字段常见为字符串类型——API为保持精度刻意如此。字段缺失时bar.get()返回None，类型检查失败，抛出DecimalParseError。零成交量不判错。

失败意味着数值无法参与后续计算，或出现无穷大、NaN等异常值。代码不得用默认0替代缺失字段。

第六道：行内关系

每根K线内部，low <= high，open和close需落在[low, high]范围内。违反时抛出RowRelationError。

失败意味着该行数据存在明显的数值矛盾。这个检查只能发现明显异常，不能证明数据绝对正确——如果四个价格同时错位，单行关系检查无法发现。

第七道：整批提交

以上六道针对单条记录全部通过后，整批数据才通过契约检查。如果任意一行在任何一道失败，整批不提交数据库。部分成功不是成功——数据库里少了一行坏数据，但多了整批"看起来完整"的假象。

Python 校验器

以下校验器只处理已经获取的payload，不发送HTTP请求，不连接数据库。仅使用Python标准库。

from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from typing import List, Dict, Any


@dataclass
class KLine:
    symbol: str
    interval: str
    time: int
    open: Decimal
    high: Decimal
    low: Decimal
    close: Decimal
    volume: Decimal
    quote_volume: Decimal


class KLineValidationError(Exception):
    pass


class EnvelopeError(KLineValidationError):
    pass


class IdentityError(KLineValidationError):
    pass


class ArrayShapeError(KLineValidationError):
    pass


class TimeIdentityError(KLineValidationError):
    pass


class DecimalParseError(KLineValidationError):
    pass


class RowRelationError(KLineValidationError):
    pass


def _require_condition(condition: bool, exc_type, message: str):
    if not condition:
        raise exc_type(message)


def _parse_finite_decimal(value, field_name: str, bar_index: int) -> Decimal:
    if not isinstance(value, str):
        raise DecimalParseError(
            f"bar[{bar_index}].{field_name}: expected str, got {type(value).__name__}"
        )
    try:
        d = Decimal(value)
    except InvalidOperation:
        raise DecimalParseError(
            f"bar[{bar_index}].{field_name}: invalid Decimal: {value!r}"
        )
    if not d.is_finite():
        raise DecimalParseError(
            f"bar[{bar_index}].{field_name}: non-finite: {d}"
        )
    return d


def validate_kline_payload(payload: Dict[str, Any], expected_symbol: str, expected_interval: str) -> List[KLine]:

    # 第一道：响应信封（HTTP状态和JSON解析由调用层负责）
    _require_condition(
        isinstance(payload, dict),
        EnvelopeError,
        "payload must be a dict"
    )
    code = payload.get("code")
    _require_condition(
        code == 0,
        EnvelopeError,
        f"code={code}, expected 0"
    )

    data = payload.get("data")
    _require_condition(
        isinstance(data, dict),
        EnvelopeError,
        "data must be a dict"
    )

    # 第二道：请求身份
    resp_symbol = data.get("symbol")
    _require_condition(
        resp_symbol == expected_symbol,
        IdentityError,
        f"symbol mismatch: expected {expected_symbol}, got {resp_symbol}"
    )
    resp_interval = data.get("interval")
    _require_condition(
        resp_interval == expected_interval,
        IdentityError,
        f"interval mismatch: expected {expected_interval}, got {resp_interval}"
    )

    # 第三道：数组形态
    klines = data.get("klines")
    _require_condition(
        isinstance(klines, list),
        ArrayShapeError,
        "klines must be a list"
    )
    if len(klines) == 0:
        return []

    # 第四至六道：逐行校验
    seen_times: set[int] = set()
    result: list[KLine] = []

    for i, bar in enumerate(klines):
        _require_condition(
            isinstance(bar, dict),
            ArrayShapeError,
            f"bar[{i}] must be a dict"
        )

        # 第四道：时间身份
        time_ms = bar.get("time")
        _require_condition(
            type(time_ms) is int and time_ms > 0,
            TimeIdentityError,
            f"bar[{i}].time must be a positive int (not bool), got {type(time_ms).__name__} {time_ms!r}"
        )
        _require_condition(
            time_ms not in seen_times,
            TimeIdentityError,
            f"bar[{i}].time={time_ms} is duplicate"
        )
        seen_times.add(time_ms)

        # 第五道：数值解析
        open_val = _parse_finite_decimal(bar.get("open"), "open", i)
        high_val = _parse_finite_decimal(bar.get("high"), "high", i)
        low_val = _parse_finite_decimal(bar.get("low"), "low", i)
        close_val = _parse_finite_decimal(bar.get("close"), "close", i)
        volume_val = _parse_finite_decimal(bar.get("volume"), "volume", i)
        quote_volume_val = _parse_finite_decimal(bar.get("quote_volume"), "quote_volume", i)

        # 第六道：行内关系
        _require_condition(
            low_val <= high_val,
            RowRelationError,
            f"bar[{i}]: low={low_val} > high={high_val}"
        )
        _require_condition(
            low_val <= open_val <= high_val,
            RowRelationError,
            f"bar[{i}]: open={open_val} not in [{low_val}, {high_val}]"
        )
        _require_condition(
            low_val <= close_val <= high_val,
            RowRelationError,
            f"bar[{i}]: close={close_val} not in [{low_val}, {high_val}]"
        )

        result.append(KLine(
            symbol=resp_symbol,
            interval=resp_interval,
            time=time_ms,
            open=open_val,
            high=high_val,
            low=low_val,
            close=close_val,
            volume=volume_val,
            quote_volume=quote_volume_val,
        ))

    # 排序是应用行为，不声称接口固定升序或降序
    result.sort(key=lambda k: k.time)
    return result

四项最小单元测试

import unittest

class TestKLineValidation(unittest.TestCase):

    def setUp(self):
        self.valid_bar = {
            "time": 1717891200000,
            "open": "190.50",
            "high": "192.00",
            "low": "189.80",
            "close": "191.20",
            "volume": "50000000",
            "quote_volume": "9550000000.00",
        }

    def build_payload(self, bars):
        return {
            "code": 0,
            "data": {
                "symbol": "AAPL.US",
                "interval": "1d",
                "klines": bars,
            },
        }

    def test_normal_payload(self):
        """正常payload应通过全部校验"""
        bars = [
            dict(self.valid_bar, time=1717891200000),
            dict(self.valid_bar, time=1717977600000),
            dict(self.valid_bar, time=1718064000000),
        ]
        result = validate_kline_payload(
            self.build_payload(bars), "AAPL.US", "1d"
        )
        self.assertEqual(len(result), 3)
        self.assertEqual(result[0].time, 1717891200000)
        self.assertEqual(result[-1].time, 1718064000000)

    def test_duplicate_time(self):
        """重复time应抛出TimeIdentityError"""
        bars = [
            dict(self.valid_bar, time=1717891200000),
            dict(self.valid_bar, time=1717891200000),
        ]
        with self.assertRaises(TimeIdentityError):
            validate_kline_payload(
                self.build_payload(bars), "AAPL.US", "1d"
            )

    def test_missing_close(self):
        """缺失close字段应抛出DecimalParseError"""
        bar = dict(self.valid_bar)
        del bar["close"]
        with self.assertRaises(DecimalParseError):
            validate_kline_payload(
                self.build_payload([bar]), "AAPL.US", "1d"
            )

    def test_time_true(self):
        """time字段为布尔值True应抛出TimeIdentityError"""
        bar = dict(self.valid_bar, time=True)
        with self.assertRaises(TimeIdentityError):
            validate_kline_payload(
                self.build_payload([bar]), "AAPL.US", "1d"
            )

if __name__ == "__main__":
    unittest.main()

PostgreSQL 的第二道防线

数据库关系表约束可以加固应用层校验，但有其明确边界。NOT NULL防止空值写入，CHECK可确保low <= high，多列UNIQUE约束在相关列均为NOT NULL的前提下能阻止重复的(symbol, interval, time)组合写入。

这些约束不验证原始响应信封（如code、data结构、klines数组形态），无法区分空数组与正常数据，也不能发现跨行时间缺口。应用层必须先将数据校验至语义正确，数据库约束仅作为最后一层兜底。本文不展开Upsert或滚动回补策略。

发布前检查卡

在把K线数据写入数据库之前，逐项确认：

  • 响应信封：code为0，data为dict
  • 请求身份：返回symbol和interval与请求一致
  • 数组形态：klines为list；空数组按no-data分支处理，不伪装成功
  • 时间身份：time为正整数毫秒，非布尔值；批内无重复time
  • 数值解析：OHLC、volume、quote_volume可解析为有限Decimal；不用float；缺失字段不补0
  • 行内关系：low <= high，open和close在[low, high]内
  • 整批提交：任一行在任何一道失败，整批不提交

FAQ

code=0，为什么仍不够？

code=0只证明业务层没有拒绝请求，不承诺字段完整、数值合理或数据可用。schema和语义检查是调用方的责任。

空数组一定是错误吗？

不一定。空数组可能表示请求范围内没有已结束的K线。但究竟是因为非交易日、休市时段还是数据缺失，需要结合请求的具体时间范围、对应市场的交易日历和任务预期来判断。空数组应进入专门的no-data分支处理，而不是被静默当作"写入成功"。

主键能发现缺K线吗？

不能。主键只防止重复，不检测时间连续性。缺口需要独立的时间连续性检查。

为什么价格不用float？

float是二进制浮点数，无法精确表示所有十进制小数。Decimal使用十进制表示，精度可控。API返回字符串价格正是为了保持精度，用float解析等于丢掉了这份保护。

参考来源

相关文章

A股美股港股行情看板Demo三天、上线三个月：拆解7个返工点

同一张行情表，寒武纪的成交额自己算是5,414万元，接口给的是54.20亿元，差了100倍，原因只是一个单位。行情看板三天就能接通，上线后的返工，几乎都来自界面上说不清“自己是什么”的数字。 本文依据上交所、港交所的一手文件，加上10月12日盘中对7类资产的实测，给你四样东西： 一张“价格身份证”：每个数字上线前要答的5个问题 7个返工点逐条拆：钱在哪里算错、为什么会错 一份可以直接发给数据…

TickDB Research ·

拆解美股夜盘数据5个口径：12月6日起美股进入23小时交易

同一只苹果股票，同一个周六中午，三个“最新价”，1万股持仓估值差出 9,500 美元。12月6日起，美股将进入近23小时交易：夜盘数据的5个口径不说清，日K线、收益率、风控和对账都会悄悄算错。 本文依据纳斯达克、SEC、DTCC、FINRA 等一手文件，加上30只美股（含9只中概股）的实测数据，给你四样东西： 一张“北京时间 × 美股夜盘”全景图（含11月1日夏令时切换前后） 5个口径逐条拆…

TickDB Research ·

港股数据源选型：踩一个回测就假（富途/IBKR/TickDB）

回测里某个港股策略表现稳定，实盘跑了一个月，收益差了将近一半。 不是完全对不上，是那种“差异不大但一直在”的偏差。团队排查了三周——策略逻辑、滑点模型、手续费、执行延迟，全部查过。最后发现问题在数据层：数据源返回的港股代码是“700”，而团队数据库里存的是“00700”。两个格式查出来的数据，部分重叠，部分不重叠。回测用的样本，比实际可交易的样本少了一批。 这不是某个操作失误，是港股数据接入…

TickDB Research ·

A股休市10天，港股跌了，美联储还要开会，APEC即将落地深圳——8号开盘，你盘清楚了几件事？

后天开盘。10月8日，A股恢复交易。你手里的A股仓位，价格停在9月30日；但如果你同时持有港股、美股、QDII，这10天它们一直在动——港股10月2日就开盘了，美股全程正常交易。 这篇文章要说的不是“A股休市了”。有仓位的人都知道A股休市。它要说的是： 同一段时间里，你的三个市场仓位在动，你的信息却只跟得上一个。 核心概念是 信息时态差 ——港股和美股的信息是连续的，每天有K线；A股的信息是…

TickDB Research ·

自选股跨了A股港股美股，还在打开三个APP翻？盘中扫异动前，先定这五个字段

自选股四十只，A股港股美股都有，每天早上打开三个 APP 翻一遍。有天腾讯盘中跌了 3%，又弹回来，我在外面吃饭，完全没看到。下午再打开，只看到它跌了 2%。 这个 2% 是结果，不是过程。 问题不在我没有实时数据。问题在我看到的那个数字——涨跌幅——它只告诉我一个方向，没告诉我这个方向是怎么走出来的。 同一只股票，只看涨跌幅和看开高低收加成交额，你做出的判断可以完全相反。股票池越大，这个差…

TickDB Research ·