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

摘要

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

前两层是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解析等于丢掉了这份保护。

参考来源

• TickDB REST API文档，K线端点：https://docs.tickdb.ai/zh-Hans/rest/api_kline
• 2026-06-09 TickDB MCP get_kline调用记录：AAPL.US/1d、700.HK/1d、BTCUSDT/1d，limit=3
• PostgreSQL约束文档：https://www.postgresql.org/docs/current/ddl-constraints.html
• Python Decimal文档：https://docs.python.org/3/library/decimal.html

通过 TickDB API 获取实时行情数据

一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。