摘要

研究报告里写“截至某日某时，某股票价格为 X”，一周后复查时发现数据源已刷新，当时的快照没保存，数字无法核对。这不是数据源的问题，是取数流程里缺了一环——没有在取值的同时留下原始快照和字段溯源记录。本文给出一套最小留痕模型：用 Python 保存请求参数、原始响应、字段路径和校验记录，让报告里的每一个行情数字都能追溯到“什么时候、从哪个接口、用哪个字段取的、原始返回值是什么”。

研究员用 Python 拉行情数据写报告，最常见的一个坑不是数据源挂了，而是报告交上去之后发现数据对不上，回头查的时候数据源已经刷新了——那条行情的快照早就没了。

问题不在数据源，在于取值的时候没有同步留痕。 你只记了数字本身，没记这个数字是从哪个字段路径拿到的、原始返回体长什么样、取值时间是多少。一周后想复核，只剩报告里一个孤零零的数字，既查不到来源，也复现不了当时的状态。

本文给出一套最小留痕模型，帮你把每一次行情取值都变成一条可追溯的记录。它不是数据库设计教程，也不是生产级数据管道，而是一套能让你的研究数据“经得起复核”的工程方法。

1. 最小留痕模型：一条记录至少包含什么

每次取数，至少保留以下六类信息。这六类合在一起，构成了一个行情数字的“完整身世”。

为什么 raw_snapshot 是必选项：保存原始响应是留痕和日志最大的区别。日志只记录“发生了什么”，原始快照记录的是“当时数据源交回了什么”。一周后如果价格对不上，你可以把原始快照重新灌进校验脚本，复现当时的取值过程。没有快照，只剩猜测。

2. 字段表：研究报告引用的行情数据溯源记录

以下字段表适合作为研究报告的数据引用附录，每引用一个行情数字，填一行。这张表的核心是让读者（或审查者）能从报告里的数字反查到你取数时的原始状态。

注意：上表中的 payload_timestamp 和 value_raw 为字段占位说明，实际值以接口返回为准。request_params_hash 和 raw_snapshot_hash 用于防篡改校验，具体哈希算法可选用 SHA-256。

3. Python 伪代码：取值即留痕

以下代码为教学伪代码，展示取值留痕的核心逻辑流程，非生产级可运行代码。实际使用时需根据数据源的接口文档调整字段路径和鉴权方式。

import json
import hashlib
from datetime import datetime, timezone
from decimal import Decimal


def fetch_market_data(symbol: str, market: str) -> dict:
    """向行情数据源请求快照数据。
    返回原始响应体。实际实现需替换为真实请求逻辑。
    """
    # 教学伪代码：实际应使用 requests.get() 并处理超时、异常
    # response = requests.get(..., params={...}, headers={...}, timeout=10)
    # return response.json()
    pass


def save_raw_snapshot(data: dict) -> tuple:
    """保存原始响应快照，返回 (snapshot_id, snapshot_hash)。"""
    raw_json = json.dumps(data, ensure_ascii=False, sort_keys=True)
    snapshot_id = f"SNAP-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M%S')}"
    snapshot_hash = hashlib.sha256(raw_json.encode()).hexdigest()
    # 实际实现：将 raw_json 写入文件或数据库
    return snapshot_id, snapshot_hash


def extract_field_value(data: dict, field_path: str) -> str:
    """从返回体中按路径提取字段值。
    例如 field_path='data[0].last_price'。
    """
    # 教学伪代码：简化路径解析，实际需处理嵌套和数组索引
    parts = field_path.replace("[", ".").replace("]", "").split(".")
    current = data
    for part in parts:
        if isinstance(current, dict):
            current = current.get(part)
        elif isinstance(current, list) and part.isdigit():
            current = current[int(part)]
        else:
            return None
    return str(current) if current is not None else None


def write_lineage_record(report_id: str, section: str, symbol: str,
                         market: str, data_source: str, query_method: str,
                         requested_at: str, payload_ts: int,
                         field_name: str, field_path: str, field_semantics: str,
                         value_raw: str, snapshot_id: str, snapshot_hash: str,
                         processing_note: str) -> dict:
    """写入一条数据溯源记录。"""
    # 教学伪代码：实际实现应写入数据库
    record = {
        "report_id": report_id,
        "report_section": section,
        "symbol": symbol,
        "market": market,
        "data_source": data_source,
        "query_method": query_method,
        "requested_at": requested_at,
        "payload_timestamp": payload_ts,
        "field_name": field_name,
        "field_path": field_path,
        "field_semantics": field_semantics,
        "value_raw": value_raw,
        "value_normalized": str(Decimal(value_raw)) if value_raw else None,
        "raw_snapshot_id": snapshot_id,
        "raw_snapshot_hash": snapshot_hash,
        "processing_note": processing_note,
        "review_status": "待复核"
    }
    # db.insert(record)
    return record


def build_report_citation_note(record: dict) -> str:
    """根据溯源记录生成报告中可引用的数据来源注释。"""
    return (
        f"数据来源：{record['data_source']}（{record['query_method']}），"
        f"品种 {record['symbol']}，字段 {record['field_path']}，"
        f"请求时间 {record['requested_at']}，"
        f"快照 ID {record['raw_snapshot_id']}，"
        f"复核状态：{record['review_status']}"
    )

4. 失败分支速查

取数留痕流程中，以下八类异常各有对应的处理方式。任何一类失败，这条数据的 review_status 都应标记为“存疑”或直接阻断。

5. TickDB 在取数留痕流程中的合理位置

TickDB 可作为候选结构化行情入口，帮助统一保存请求参数、字段语义、原始快照和检查记录。它在取数留痕流程里的作用不是“保证数据一定正确”，而是让你的取值动作有一个结构化的起点——symbol 和 market 有统一格式，接口返回有明确的字段路径，原始响应体可以完整保存。这些恰好是留痕模型前四类信息的基础。

验证方式：用自己的 symbol 发一次请求，按本文字段表逐列填写。保存原始快照和哈希，计算 value_normalized，写一条完整溯源记录。如果当前接口返回的字段路径和本文示例不同，以官方文档或实测为准。

不适合什么：不替代研究报告的数据审核流程；不承诺数据永远正确；不保证所有字段在所有市场可用。具体端点、Header、字段枚举和错误码，以 TickDB 官方文档和审核为准。

6. 发布前检查清单

📡 本文取数留痕示例以 TickDB.ai 作为候选行情入口，具体字段以官方文档和审核为准
⚠️ 本文为技术教程，不构成任何投资建议