DeepSeek Tool Calls 接实时行情：如何避免模型把“猜测”写成当前价格

仅靠 Prompt，DeepSeek 无法证明当前价格来自实时接口。本文以 TickDB REST 行情快照为例，展示如何通过 Tool Calls 请求查询、由程序执行 REST 并回填成功结果；若工具失败，则由程序直接结束价格回答流程，以降低模型补写未经验证行情的风险。

当用户问“现在 AAPL.US 或 BTCUSDT 的价格是多少”，仅靠对话上下文，模型无法证明答案来自当前行情接口。

解决这个问题的关键，不是追加一句“请实时查询”，而是建立一条可核验的数据链路：

用户提问
  -> DeepSeek 返回 Tool Calls
  -> 程序执行 TickDB REST 请求
  -> 程序检查成功或失败
  -> 成功时回填结果并生成回答
  -> 失败时直接返回失败提示，不生成价格答案

本文以 TickDB REST 行情快照接口为数据源示例，展示 DeepSeek Tool Calls 的教学实现。行情内容仅用于接口演示，不构成投资建议。

问题：模型为什么会给出看似合理、却无法验证的当前价格？

在普通聊天调用中，模型获得的是文本，而不是自动连接的数据源。

因此，即便问题中有“最新”“当前”或“实时”，模型也不一定执行过外部查询。常见问题包括：

对于行情类问题，一个更稳健的原则是：

成功响应必须来自实际工具结果；工具失败时，程序应阻止生成行情价格答案。

原因：Tool Calls 提出调用请求，REST 请求由程序执行

DeepSeek 当前官方文档使用 Tool Calls 表述这类机制。其基本过程是：

1. 程序通过 tools 声明可调用函数及参数结构。
2. 模型判断需要外部数据时，返回 tool_calls。
3. 开发者程序执行真实 HTTP 请求。
4. 程序用 role="tool" 与对应的 tool_call_id 回填结果。
5. 模型基于成功工具结果组织最终说明。

本文采用的行情工具边界如下：

last_price 应作为字符串处理。timestamp 可用于说明行情快照对应时间，但不能据此宣称低延迟、强 SLA 或高频交易适用性。

解决：封装一个失败时关闭回答链路的行情工具函数

1. 环境准备

下面代码是教学示例，展示工具调用、REST 请求、结果回填和失败关闭流程。

运行环境：

Python >= 3.10

安装依赖并配置环境变量：

python -m pip install openai requests

export DEEPSEEK_API_KEY="<YOUR_DEEPSEEK_API_KEY>"
export TICKDB_API_KEY="<YOUR_TICKDB_API_KEY>"

2. 教学示例代码

import json
import os
import re
from typing import Any

import requests
from openai import OpenAI


DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY")
TICKDB_API_KEY = os.environ.get("TICKDB_API_KEY")

if not DEEPSEEK_API_KEY or not TICKDB_API_KEY:
    raise RuntimeError(
        "请先设置 DEEPSEEK_API_KEY 与 TICKDB_API_KEY 环境变量"
    )

client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com",
)

SYMBOL_PATTERN = re.compile(r"^[A-Z0-9.]+$")
ALLOWED_TYPES = {"stock", "indices", "crypto", "forex", "futures"}

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_market_ticker",
            "description": (
                "查询一个或多个交易品种的当前行情快照。"
                "用户询问最新价格、当前价格或行情时间时使用。"
            ),
            "parameters": {
                "type": "object",
                "properties": {
                    "symbols": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "例如 ['AAPL.US', 'BTCUSDT']",
                    },
                    "type": {
                        "type": "string",
                        "enum": [
                            "stock",
                            "indices",
                            "crypto",
                            "forex",
                            "futures",
                        ],
                        "description": "可选；需要消歧时传入资产类型。",
                    },
                },
                "required": ["symbols"],
            },
        },
    }
]


def failure(error: str, detail: str, **extra: Any) -> dict[str, Any]:
    result = {"ok": False, "error": error, "detail": detail}
    result.update(extra)
    return result


def get_market_ticker(
    symbols: list[str],
    asset_type: str | None = None,
) -> dict[str, Any]:
    if not isinstance(symbols, list) or not symbols:
        return failure("invalid_arguments", "symbols 必须是非空数组")

    if len(symbols) > 50:
        return failure("invalid_arguments", "单次最多查询 50 个交易品种")

    cleaned_symbols = []
    for symbol in symbols:
        if not isinstance(symbol, str):
            return failure("invalid_arguments", "symbol 必须是字符串")

        normalized = symbol.strip().upper()
        if not SYMBOL_PATTERN.fullmatch(normalized):
            return failure("invalid_arguments", f"非法 symbol: {symbol}")

        cleaned_symbols.append(normalized)

    if asset_type is not None and asset_type not in ALLOWED_TYPES:
        return failure("invalid_arguments", f"不支持的 type: {asset_type}")

    params = {"symbols": ",".join(cleaned_symbols)}
    if asset_type:
        params["type"] = asset_type

    try:
        response = requests.get(
            "https://api.tickdb.ai/v1/market/ticker",
            headers={"X-API-Key": TICKDB_API_KEY},
            params=params,
            timeout=10,
        )
    except requests.Timeout:
        return failure("timeout", "行情请求超时，本次不生成价格答案")
    except requests.RequestException as exc:
        return failure("request_failed", f"行情请求失败：{exc}")

    try:
        payload = response.json()
    except ValueError:
        return failure(
            "invalid_response",
            "接口未返回可解析 JSON，本次不生成价格答案",
            http_status=response.status_code,
        )

    code = payload.get("code")

    if response.status_code == 429 or code == 3001:
        return failure(
            "rate_limited",
            "请求频率受限，请稍后重试",
            retry_after=response.headers.get("Retry-After"),
        )

    if code == 1001:
        return failure("invalid_api_key", "API Key 无效或已过期，请更换 Key")

    if code == 1002:
        return failure("missing_api_key", "请求缺少 X-API-Key，请补充配置")

    if code == 1004:
        return failure("permission_denied", "当前权限无法访问该数据")

    if code == 2002:
        return failure(
            "symbol_not_found",
            "未找到该交易品种，请先调用 /v1/symbols/available 查询可用 symbol",
        )

    if response.status_code >= 400 or code != 0:
        return failure(
            "api_error",
            payload.get("message", "TickDB 请求失败"),
            code=code,
            http_status=response.status_code,
        )

    rows = payload.get("data")
    if not isinstance(rows, list):
        return failure("invalid_response", "响应中的 data 不是数组")

    result = []
    for row in rows:
        result.append(
            {
                "symbol": row.get("symbol"),
                "last_price": row.get("last_price"),
                "timestamp": row.get("timestamp"),
            }
        )

    return {
        "ok": True,
        "data": result,
        "note": "last_price 为行情快照字段；回答时应附带 timestamp。",
    }


def ask_deepseek(
    messages: list[dict[str, Any]],
    tool_choice: str = "auto",
):
    response = client.chat.completions.create(
        model="deepseek-v4-pro",
        messages=messages,
        tools=TOOLS,
        tool_choice=tool_choice,
    )
    return response.choices[0].message


messages = [
    {
        "role": "system",
        "content": (
            "用户询问当前行情时，必须使用 get_market_ticker。"
            "只有工具成功返回的数据可用于回答行情价格。"
            "回答需附带时间戳，并声明仅用于接口演示。"
        ),
    },
    {
        "role": "user",
        "content": "请查询 AAPL.US 和 BTCUSDT 的当前行情快照，并给出时间戳。",
    },
]

assistant_message = ask_deepseek(messages)
messages.append(assistant_message)

if not assistant_message.tool_calls:
    raise RuntimeError("模型未返回预期的行情工具调用")

tool_failed = False
failure_messages = []

for tool_call in assistant_message.tool_calls:
    if tool_call.function.name != "get_market_ticker":
        result = failure("unsupported_tool", f"不支持的工具：{tool_call.function.name}")
    else:
        try:
            arguments = json.loads(tool_call.function.arguments)
            result = get_market_ticker(
                symbols=arguments.get("symbols"),
                asset_type=arguments.get("type"),
            )
        except (TypeError, ValueError, json.JSONDecodeError) as exc:
            result = failure("invalid_tool_arguments", str(exc))

    if not result.get("ok"):
        tool_failed = True
        failure_messages.append(result["detail"])

    messages.append(
        {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result, ensure_ascii=False),
        }
    )

if tool_failed:
    print("行情查询失败：" + "；".join(failure_messages))
    print("本次未生成行情价格答案。")
else:
    final_message = ask_deepseek(messages, tool_choice="none")
    print(final_message.content)

3. 为什么增加 fail-closed 分支？

如果工具请求失败后，程序仍继续让模型生成行情回答，系统就只能依赖提示词约束模型“不猜价格”。

这是一种行为要求，不是程序保证。

上面的修订代码采用更明确的处理方式：

if tool_failed:
    print("本次未生成行情价格答案。")
else:
    final_message = ask_deepseek(messages, tool_choice="none")

也就是说：

• 工具成功：允许模型根据回填结果生成答案。
• 工具失败：程序直接结束行情回答流程。
• 失败场景下：不再把“是否补写价格”交给模型判断。

这不能替代完整生产安全设计，但可以避免教学示例把提示词约束误写成强保证。

验证：怎样检查回答来自真实行情结果？

验证一：先单独检查 REST 请求

在接入模型前，先确认接口和 Key 可正常工作：

curl --get "https://api.tickdb.ai/v1/market/ticker" \
  --data-urlencode "symbols=AAPL.US,BTCUSDT" \
  -H "X-API-Key: ${TICKDB_API_KEY}"

检查重点：

code == 0
data 是数组
data[] 中包含 symbol、last_price、timestamp
last_price 以字符串处理

脱敏示意结构如下：

{
  "code": 0,
  "data": [
    {
      "symbol": "AAPL.US",
      "last_price": "<接口返回的字符串价格>",
      "timestamp": "<UTC 毫秒时间戳>"
    }
  ]
}

验证二：确认第一轮返回 tool_calls

调试时可打印：

print(assistant_message.tool_calls)

期望结果是模型返回 get_market_ticker 调用请求及查询参数，而不是直接输出价格结论。

验证三：确认失败时不会生成价格答案

至少测试以下场景：

验证四：确认最终回答保留边界

成功路径的最终回答应满足：

包含查询品种
包含工具返回的价格字段
包含对应 timestamp
不扩写成收益预测、交易建议或低延迟承诺

常见坑

1. 把 Tool Calls 写成模型自动请求行情接口

Tool Calls 负责表达调用意图；真实 REST 请求、鉴权、参数验证和失败处理均由开发者程序完成。

2. 把鉴权 Header 写错

本文接口使用：

X-API-Key: <YOUR_API_KEY>

3. 读取错误的响应路径

行情快照结果读取主体是：

payload["data"]

本文使用字段：

row["symbol"]
row["last_price"]
row["timestamp"]

不要把其他接口的字段路径直接套到 ticker 响应上。

4. 把示例支持范围误写成接口全部边界

本文工具 schema 展示了：

stock / indices / crypto / forex / futures

实际开发中，仍应根据查询资产与当前官方接口文档核对参数和 symbol 规则。

5. 把时间戳精度写成性能承诺

timestamp 可用于说明快照对应时间，但不能据此声称：

毫秒级无延迟
适用于高频交易
具备固定 SLA

结论

让 DeepSeek 处理当前行情问题，需要把回答绑定到一条可检查的调用链路：

DeepSeek 提出 Tool Calls
  -> 程序执行 TickDB REST 查询
  -> 成功结果回填模型
  -> 失败时程序关闭行情回答流程

这种做法不能替代完整生产系统建设，但能显著降低“接口没有返回数据，模型却继续补写当前价格”的风险。

对于需要查询行情快照并生成自然语言说明的应用，它提供了一个清晰起点。对于持续推送、自动交易、延迟承诺或生产可用性要求更高的场景，应进行独立设计与核验。

参考资料：

• DeepSeek Tool Calls 官方文档
• DeepSeek Create Chat Completion 官方文档
• TickDB 行情快照 REST API
• TickDB 错误码说明

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

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