DeepSeek Function Calling 接实时行情：从工具定义到多轮查询的完整示例 - 文章 - 开发者社区

30 秒结论：本文用 DeepSeek 官方的 Function Calling 接口，从零搭建了一个可真实调用行情数据的查询系统。DeepSeek 不会主动查股票，但你可以给它一套工具定义（tools 参数），让它自己判断何时调用、传什么参数。本文提供完整 Python 代码、tools 参数完整设计、完整对话轨迹示例、多轮对话任务规划。你将得到一个可复用的工程框架——不只是“让 AI 查股票”，而是“让 AI 自主规划量化任务”。

你或许看到过这样的标题：

“AI 横扫港大美股交易赛，年化回报率超 10%！” “六大 AI 模拟炒股对决，DeepSeek 盈利居首！”

这些报道源于真实实验。曾有学术机构举办 AI 模拟交易竞赛，AI 模型在特定环境下取得了两位数的年化回报率。一些科技媒体也报道了 DeepSeek 在模拟对比中排名靠前的结果。

看起来 AI 在交易上已经无所不能？

等一下。另一组数据鲜有人提：部分量化评估显示，AI 模型的胜率约 40%，但盈亏比偏低，盈利期望有限。更有研究指出，AI 在金融预测中存在“前视偏差”——模型在“回忆”而非“推理”。

所以问题不是“DeepSeek 能不能炒股”。真正的问题是：

当你把真实行情数据喂给 DeepSeek 后，它到底在“分析”还是在“编故事”？

与其看别人评测，不如自己动手。本文用 DeepSeek 官方的 Function Calling API，从零搭建一个真正能调用实时行情数据的系统。跑通之后，你就有了一面“照妖镜”——AI 说的每一句分析，你都能用真实数据验证。

一、Function Calling 原理

Function Calling 的机制很简单：你给模型定义一套“工具”，模型自己判断什么时候该调用哪个工具、传什么参数。

执行流程分四步走：

① 用户提出问题
② 模型分析后返回一个 tool_calls 请求（不是文本回复，是一个结构化的函数调用请求）
③ 你在代码里执行这个函数（比如真的去调行情 API），拿到真实数据
④ 把数据作为 tool role 消息返回给模型，模型基于真实数据生成最终文本回复

关键认知：模型不执行任何代码，它只返回“我想调用哪个函数、传什么参数”。 真正的 API 调用是你写的 Python 代码完成的。这就意味着——数据从哪里来、质量如何、是否及时，完全由你控制。

二、tools 参数完整设计（核心章节）

tools 参数是 Function Calling 的灵魂。设计得太粗糙，模型不知道传什么参数；设计得太复杂，模型容易调用失败。

以下基于 TickDB REST API，设计 3 个核心工具。覆盖实时行情、历史 K 线和股票基本面三大场景。

工具与端点映射

工具名	执行函数	TickDB REST 端点	用途
`get_ticker`	`execute_get_ticker`	`GET /v1/market/ticker`	实时行情快照
`get_kline`	`execute_get_kline`	`GET /v1/market/kline`	历史K线数据
`get_stock_info`	`execute_get_stock_info`	`GET /v1/market/stock-info`	股票基本面

工具 1：get_ticker（实时行情快照）

{
    "type": "function",
    "function": {
        "name": "get_ticker",
        "description": "获取一个或多个交易品种的实时行情快照，包括最新价、涨跌幅、24小时最高最低价、成交量。支持 A 股（.SH/.SZ）、港股（.HK）、美股（.US）、加密货币（如 BTCUSDT）、外汇（如 EURUSD）、贵金属（XAUUSD）、指数（SPX）。当用户想了解某品种的当前价格或涨跌情况时，使用此工具。注意：此工具支持多个品种同时查询（symbols 参数用逗号分隔）。当用户一次查询超过 50 个品种时，需分批调用。",
        "parameters": {
            "type": "object",
            "properties": {
                "symbols": {
                    "type": "string",
                    "description": "交易品种代码，多个用英文逗号分隔，最多 50 个。格式示例：'AAPL.US' 或 'BTCUSDT,700.HK,600519.SH'。A 股格式为 6 位数字+.SH/.SZ（如 600519.SH），港股格式为数字+.HK 无前导零（如 700.HK），美股格式为字母+.US（如 AAPL.US），加密货币直接写币对（如 BTCUSDT），外汇写 6 位货币对（如 EURUSD），贵金属写 XAUUSD/XAGUSD，指数写 3-4 位代码无后缀（如 SPX/NDX/DJI）。"
                }
            },
            "required": ["symbols"],
            "additionalProperties": False
        }
    }
}

工具 2：get_kline（历史 K 线数据）

{
    "type": "function",
    "function": {
        "name": "get_kline",
        "description": "获取某个交易品种的历史 K 线数据（OHLCV），返回开盘价、最高价、最低价、收盘价、成交量。适合用于技术分析、趋势观察、计算技术指标（均线、MACD 等）。当用户提及 K 线、蜡烛图、趋势分析、历史走势或技术指标时，使用此工具。注意：此工具只接受单个品种代码（symbol 参数为单数）。当用户需要同时查询多个品种时，应使用 get_ticker 而非此工具（如需查多个品种的 K 线，需多次调用此工具）。当用户未指定 K 线周期时，默认使用日线（1d）。当用户未指定数量时，默认返回最近 20 根。",
        "parameters": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "单个交易品种代码。格式示例：'700.HK' 或 'AAPL.US' 或 'BTCUSDT'。注意只能传入一个品种代码，不支持逗号分隔。"
                },
                "interval": {
                    "type": "string",
                    "enum": ["1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "1d", "1w", "1M"],
                    "description": "K 线周期。1m=1分钟，3m=3分钟，5m=5分钟，15m=15分钟，30m=30分钟，1h=1小时，2h=2小时，4h=4小时，1d=日线，1w=周线，1M=月线。不支持其他值。"
                },
                "limit": {
                    "type": "integer",
                    "description": "返回 K 线数量。默认 20，最大 200。",
                    "minimum": 1,
                    "maximum": 200
                }
            },
            "required": ["symbol", "interval"],
            "additionalProperties": False
        }
    }
}

工具 3：get_stock_info（股票基本面）

{
    "type": "function",
    "function": {
        "name": "get_stock_info",
        "description": "获取股票的基本面数据，包括公司名称、每手股数、总股本、流通股本、每股盈利(EPS TTM)、每股净资产、股息率等。当用户询问基础股票信息、EPS、BPS、股息率、每手股数时使用。注意：此工具只适用于股票（A 股/港股/美股），不适用于加密货币、外汇、贵金属或指数。如用户询问 PE/PB/市值等估值指标，应另行设计对应的估值查询工具。",
        "parameters": {
            "type": "object",
            "properties": {
                "symbols": {
                    "type": "string",
                    "description": "股票代码，多个用英文逗号分隔，最多 50 个。格式示例：'700.HK,AAPL.US,600519.SH'。只支持股票类品种，不支持加密货币、外汇、贵金属或指数代码。"
                }
            },
            "required": ["symbols"],
            "additionalProperties": False
        }
    }
}

设计要点说明

description 决定模型会不会选对工具：不只是写“获取行情”，而是写清楚“当用户想了解某品种的当前价格或涨跌情况时，使用此工具”。description 中还包含了不触发条件（如 get_stock_info 不适用于加密/外汇）和参数选择指导（如未指定周期默认 1d，未指定数量默认 20），帮助模型在模糊场景下做出正确选择。
enum 约束可选值：K 线周期用 enum 限制了合法取值，避免模型传 2m、6h 等不存在的周期。
required 只标真正必需的：limit 有默认值 20，不加 required。加了反而可能因为模型没传而调用失败。
参数 description 里包含格式示例和反例：如 get_kline 的 symbol 描述中明确“只能传入一个品种代码，不支持逗号分隔”，避免模型错误传多品种。

常见错误示范

以下错误用法会导致工具调用失败或返回空数据：

错误用法	错误原因	正确写法
`get_kline(symbols="700.HK,AAPL.US")`	传了复数参数，get_kline 只接受单数	`get_kline(symbol="700.HK")`，多品种需多次调用
`get_stock_info(symbols="BTCUSDT")`	传了加密货币代码，该工具只适用股票	用 `get_ticker(symbols="BTCUSDT")` 查加密行情
`get_ticker(symbols="0700.HK")`	港股代码加了前导零（不推荐依赖兼容行为）	统一使用无前导零格式 `get_ticker(symbols="700.HK")`

三、完整可运行 Python 代码

环境准备

pip install openai requests

你需要两个 Key，建议使用环境变量存储：

export DEEPSEEK_API_KEY="your-deepseek-api-key"
export TICKDB_API_KEY="your-tickdb-api-key"

DeepSeek API Key：在 DeepSeek 开放平台生成
TickDB API Key：在 TickDB 控制面板生成，有免费体验额度

完整代码

import json
import os
import time
import requests
from decimal import Decimal, InvalidOperation
from openai import OpenAI

# ========== 配置区 ==========
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
TICKDB_API_KEY = os.getenv("TICKDB_API_KEY")
DEEPSEEK_MODEL = os.getenv("DEEPSEEK_MODEL", "deepseek-v4-pro")  # 模型名以 DeepSeek 官方文档为准

if not DEEPSEEK_API_KEY or not TICKDB_API_KEY:
    raise ValueError("请设置环境变量 DEEPSEEK_API_KEY 和 TICKDB_API_KEY")

# DeepSeek 客户端（兼容 OpenAI SDK 格式）
client = OpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url="https://api.deepseek.com",
)

# ========== tools 定义 ==========
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_ticker",
            "description": "获取一个或多个交易品种的实时行情快照，包括最新价、涨跌幅、24小时最高最低价、成交量。支持 A 股（.SH/.SZ）、港股（.HK）、美股（.US）、加密货币（如 BTCUSDT）、外汇（如 EURUSD）、贵金属（XAUUSD）、指数（SPX）。当用户想了解某品种的当前价格或涨跌情况时，使用此工具。注意：此工具支持多个品种同时查询（symbols 参数用逗号分隔）。当用户一次查询超过 50 个品种时，需分批调用。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbols": {
                        "type": "string",
                        "description": "交易品种代码，多个用英文逗号分隔，最多 50 个。格式示例：'AAPL.US' 或 'BTCUSDT,700.HK,600519.SH'。A 股格式为 6 位数字+.SH/.SZ（如 600519.SH），港股格式为数字+.HK 无前导零（如 700.HK），美股格式为字母+.US（如 AAPL.US），加密货币直接写币对（如 BTCUSDT），外汇写 6 位货币对（如 EURUSD），贵金属写 XAUUSD/XAGUSD，指数写 3-4 位代码无后缀（如 SPX/NDX/DJI）。"
                    }
                },
                "required": ["symbols"],
                "additionalProperties": False
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_kline",
            "description": "获取某个交易品种的历史 K 线数据（OHLCV），返回开盘价、最高价、最低价、收盘价、成交量。适合用于技术分析、趋势观察、计算技术指标（均线、MACD 等）。当用户提及 K 线、蜡烛图、趋势分析、历史走势或技术指标时，使用此工具。注意：此工具只接受单个品种代码（symbol 参数为单数）。当用户需要同时查询多个品种时，应使用 get_ticker 而非此工具（如需查多个品种的 K 线，需多次调用此工具）。当用户未指定 K 线周期时，默认使用日线（1d）。当用户未指定数量时，默认返回最近 20 根。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "单个交易品种代码。格式示例：'700.HK' 或 'AAPL.US' 或 'BTCUSDT'。注意只能传入一个品种代码，不支持逗号分隔。"
                    },
                    "interval": {
                        "type": "string",
                        "enum": ["1m", "3m", "5m", "15m", "30m", "1h", "2h", "4h", "1d", "1w", "1M"],
                        "description": "K 线周期。1m=1分钟，3m=3分钟，5m=5分钟，15m=15分钟，30m=30分钟，1h=1小时，2h=2小时，4h=4小时，1d=日线，1w=周线，1M=月线。不支持其他值。"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回 K 线数量。默认 20，最大 200。",
                        "minimum": 1,
                        "maximum": 200
                    }
                },
                "required": ["symbol", "interval"],
                "additionalProperties": False
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_stock_info",
            "description": "获取股票的基本面数据，包括公司名称、每手股数、总股本、流通股本、每股盈利(EPS TTM)、每股净资产、股息率等。当用户询问基础股票信息、EPS、BPS、股息率、每手股数时使用。注意：此工具只适用于股票（A 股/港股/美股），不适用于加密货币、外汇、贵金属或指数。如用户询问 PE/PB/市值等估值指标，应另行设计对应的估值查询工具。",
            "parameters": {
                "type": "object",
                "properties": {
                    "symbols": {
                        "type": "string",
                        "description": "股票代码，多个用英文逗号分隔，最多 50 个。格式示例：'700.HK,AAPL.US,600519.SH'。只支持股票类品种，不支持加密货币、外汇、贵金属或指数代码。"
                    }
                },
                "required": ["symbols"],
                "additionalProperties": False
            }
        }
    }
]

# ========== 工具执行函数 ==========
def execute_get_ticker(symbols: str) -> str:
    """调用 TickDB REST API 获取实时行情"""
    url = "https://api.tickdb.ai/v1/market/ticker"
    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbols": symbols}

    for attempt in range(3):
        try:
            resp = requests.get(url, headers=headers, params=params, timeout=10)

            # 处理 HTTP 429 限流
            if resp.status_code == 429:
                retry_after = resp.headers.get("Retry-After")
                wait = int(retry_after) if retry_after else (2 ** attempt)
                if attempt < 2:
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited", "message": "请求频率超限，请稍后重试"})

            data = resp.json()

            if data["code"] == 0:
                result = []
                for item in data["data"]:
                    result.append({
                        "symbol": item["symbol"],
                        "last_price": item["last_price"],
                        "price_change_percent_24h": item.get("price_change_percent_24h", "N/A"),
                        "volume_24h": item.get("volume_24h", "N/A"),
                        "high_24h": item.get("high_24h", "N/A"),
                        "low_24h": item.get("low_24h", "N/A"),
                    })
                return json.dumps(result, ensure_ascii=False)

            elif data["code"] == 3001:
                if attempt < 2:
                    retry_after = resp.headers.get("Retry-After")
                    wait = int(retry_after) if retry_after else (2 ** attempt)
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited", "message": "请求频率超限，请稍后重试"})

            elif data["code"] == 2002:
                return json.dumps({"error": "symbol_not_found", "message": "品种代码不存在，请检查格式"})

            elif data["code"] in (1001, 1002, 1004):
                return json.dumps({"error": "auth_or_permission_error", "code": data["code"], "message": data.get("message", "鉴权或权限错误")})

            else:
                return json.dumps({"error": "api_error", "code": data["code"], "message": data.get("message", "")})

        except requests.exceptions.Timeout:
            if attempt < 2:
                continue
            return json.dumps({"error": "timeout", "message": "数据源响应超时"})
        except Exception as e:
            return json.dumps({"error": "exception", "message": str(e)})

    return json.dumps({"error": "unknown", "message": "未知错误"})


def execute_get_kline(symbol: str, interval: str, limit: int = 20) -> str:
    """调用 TickDB REST API 获取历史 K 线"""
    url = "https://api.tickdb.ai/v1/market/kline"
    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbol": symbol, "interval": interval, "limit": limit}

    for attempt in range(3):
        try:
            resp = requests.get(url, headers=headers, params=params, timeout=10)

            if resp.status_code == 429:
                retry_after = resp.headers.get("Retry-After")
                wait = int(retry_after) if retry_after else (2 ** attempt)
                if attempt < 2:
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited"})

            data = resp.json()

            if data["code"] == 0:
                klines = data["data"]["klines"]
                result = []
                for k in klines[-limit:]:
                    result.append({
                        "time": k["time"],
                        "open": k["open"],
                        "high": k["high"],
                        "low": k["low"],
                        "close": k["close"],
                        "volume": k["volume"]
                    })
                return json.dumps(result, ensure_ascii=False)

            elif data["code"] == 3001:
                if attempt < 2:
                    retry_after = resp.headers.get("Retry-After")
                    wait = int(retry_after) if retry_after else (2 ** attempt)
                    time.sleep(wait)
                    continue
                return json.dumps({"error": "rate_limited"})

            elif data["code"] in (1001, 1002, 1004):
                return json.dumps({"error": "auth_or_permission_error", "code": data["code"]})

            else:
                return json.dumps({"error": "api_error", "code": data["code"]})

        except requests.exceptions.Timeout:
            if attempt < 2:
                continue
            return json.dumps({"error": "timeout"})
        except Exception as e:
            return json.dumps({"error": "exception", "message": str(e)})

    return json.dumps({"error": "unknown"})


def execute_get_stock_info(symbols: str) -> str:
    """调用 TickDB REST API 获取股票基本面
    注：此函数为简化示例，生产环境建议按 get_ticker 的错误处理模板补齐重试逻辑。
    """
    url = "https://api.tickdb.ai/v1/market/stock-info"
    headers = {"X-API-Key": TICKDB_API_KEY}
    params = {"symbols": symbols}

    try:
        resp = requests.get(url, headers=headers, params=params, timeout=10)
        data = resp.json()

        if data["code"] == 0:
            result = []
            for item in data["data"]:
                result.append({
                    "symbol": item["symbol"],
                    "name_cn": item.get("name_cn", "N/A"),
                    "eps_ttm": item.get("eps_ttm", "N/A"),
                    "bps": item.get("bps", "N/A"),
                    "dividend_yield": item.get("dividend_yield", "N/A"),
                    "lot_size": item.get("lot_size", "N/A"),
                    "total_shares": item.get("total_shares", "N/A"),
                    "circulating_shares": item.get("circulating_shares", "N/A"),
                })
            return json.dumps(result, ensure_ascii=False)
        elif data["code"] in (1001, 1002, 1004):
            return json.dumps({"error": "auth_or_permission_error", "code": data["code"]})
        else:
            return json.dumps({"error": "api_error", "code": data["code"]})

    except Exception as e:
        return json.dumps({"error": "exception", "message": str(e)})


# ========== 工具调度器 ==========
TOOL_EXECUTORS = {
    "get_ticker": execute_get_ticker,
    "get_kline": execute_get_kline,
    "get_stock_info": execute_get_stock_info,
}


def run_conversation(user_query: str, messages: list):
    """执行一轮对话（可能触发多轮工具调用）"""
    messages.append({"role": "user", "content": user_query})

    for _ in range(5):
        response = client.chat.completions.create(
            model=DEEPSEEK_MODEL,
            messages=messages,
            tools=TOOLS
        )

        assistant_message = response.choices[0].message

        if not assistant_message.tool_calls:
            messages.append({"role": "assistant", "content": assistant_message.content})
            return assistant_message.content

        messages.append(assistant_message)

        for tool_call in assistant_message.tool_calls:
            func_name = tool_call.function.name
            try:
                func_args = json.loads(tool_call.function.arguments)
            except json.JSONDecodeError as e:
                result = json.dumps({"error": "invalid_arguments", "message": f"模型生成的参数不是合法 JSON: {str(e)}"})
            else:
                print(f"[Tool Call] {func_name}({func_args})")
                executor = TOOL_EXECUTORS.get(func_name)
                if executor:
                    result = executor(**func_args)
                else:
                    result = json.dumps({"error": "unknown_tool", "message": f"未知工具: {func_name}"})

            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

    return "达到最大工具调用轮数，请简化查询。"


# ========== main ==========
if __name__ == "__main__":
    # 示例 1：查实时行情（独立上下文）
    print("=" * 60)
    print("示例 1：查实时行情")
    result = run_conversation(
        "帮我查一下腾讯(700.HK)和苹果(AAPL.US)的最新股价和涨跌幅。",
        messages=[]
    )
    print(result)

    # 示例 2：拉 K 线并计算均线（独立上下文）
    print("\n" + "=" * 60)
    print("示例 2：拉 K 线并计算均线")
    result = run_conversation(
        "帮我拉取比特币(BTCUSDT)最近 10 根日 K 线的收盘价，计算 5 日均线。注意收盘价是字符串，需要先转为 Decimal 再计算。",
        messages=[]
    )
    print(result)

    # 示例 3：跨市场基本面对比（独立上下文）
    print("\n" + "=" * 60)
    print("示例 3：跨市场基本面对比")
    result = run_conversation(
        "帮我对比腾讯(700.HK)和茅台(600519.SH)的基本面：每手股数、EPS、股息率。",
        messages=[]
    )
    print(result)

四、工具调用速查表

以下速查表帮助模型（以及读者）快速建立“用户意图 → 工具选择 → 端点 → 返回路径”的映射，是整篇文章的核心知识沉淀。

用户意图	工具选择	参数示例	TickDB 端点	返回路径
查当前价格/涨跌幅	`get_ticker`	`symbols="700.HK,AAPL.US"`	`GET /v1/market/ticker`	`data[]` → `last_price`, `price_change_percent_24h`
拉 K 线/技术分析	`get_kline`	`symbol="700.HK"`, `interval="1d"`, `limit=20`	`GET /v1/market/kline`	`data.klines[]` → `time`, `open`, `high`, `low`, `close`, `volume`
查基本面/股息率	`get_stock_info`	`symbols="700.HK,600519.SH"`	`GET /v1/market/stock-info`	`data[]` → `eps_ttm`, `bps`, `dividend_yield`, `lot_size`

五、完整对话轨迹示例

以下是“示例 1”在实际运行中的完整消息流，展示了 Function Calling 的四步循环：

【用户输入】

"帮我查一下腾讯(700.HK)和苹果(AAPL.US)的最新股价和涨跌幅。"

【DeepSeek 返回的 tool_calls】

{
  "id": "chatcmpl-xxxxxxxxxxxxxxxxxxxx",
  "object": "chat.completion",
  "choices": [{
    "index": 0,
    "message": {
      "role": "assistant",
      "content": null,
      "tool_calls": [{
        "id": "call_xxxxxxxxxxxxxxxxxxxx",
        "type": "function",
        "function": {
          "name": "get_ticker",
          "arguments": "{\"symbols\":\"700.HK,AAPL.US\"}"
        }
      }]
    },
    "finish_reason": "tool_calls"
  }]
}

【工具返回结果】（TickDB REST API 真实返回结构）

{
  "role": "tool",
  "tool_call_id": "call_xxxxxxxxxxxxxxxxxxxx",
  "content": "[{\"symbol\":\"700.HK\",\"last_price\":\"XXX\",\"price_change_percent_24h\":\"0.55\",\"volume_24h\":\"XXXXXX\",\"high_24h\":\"XXX\",\"low_24h\":\"XXX\"},{\"symbol\":\"AAPL.US\",\"last_price\":\"XXX\",\"price_change_percent_24h\":\"0.32\",\"volume_24h\":\"XXXXXX\",\"high_24h\":\"XXX\",\"low_24h\":\"XXX\"}]"
}

【DeepSeek 最终回复】

腾讯控股(700.HK)最新价：XXX 港元，涨跌幅 +0.55%
苹果(AAPL.US)最新价：XXX 美元，涨跌幅 +0.32%
📡 数据由 TickDB.ai 提供

这个完整轨迹清晰展示了 Function Calling 的消息流：

user → 用户提问
assistant(tool_calls) → 模型请求调用 get_ticker，参数 symbols="700.HK,AAPL.US"
tool → 代码执行 TickDB REST API 调用，返回结构化行情数据
assistant(text) → 模型基于真实数据生成最终回复

六、多轮对话进阶：让 DeepSeek 自己规划任务

单次工具调用只是入门。Function Calling 的真正威力在于多轮任务规划。同一个对话窗口内，模型可以连续调用多个不同工具。

示例场景：

用户：帮我分析一下腾讯（700.HK）。先查实时行情，再拉最近 10 根日 K 线，最后查一下它的基本面（EPS、股息率、每手股数）。

DeepSeek 的规划过程：

轮次	模型行为	调用的工具
第 1 轮	识别到“实时行情”需求	`get_ticker(symbols="700.HK")`
第 2 轮	识别到“日 K 线”需求	`get_kline(symbol="700.HK", interval="1d", limit=10)`
第 3 轮	识别到“基本面”需求	`get_stock_info(symbols="700.HK")`
第 4 轮	所有数据就绪	生成综合回复（文本）

整个过程，你不需要告诉它“应该先查谁、再查谁”——模型自己判断调用顺序和数据依赖关系。

七、错误处理与工程化建议

常见错误处理（已在代码中实现）：

错误场景	处理策略
HTTP 429 / 业务码 3001 限流	读取 `Retry-After` 响应头，指数退避重试（2^attempt 秒），最多 3 次
品种代码不存在（2002）	返回错误信息，模型能理解并提示用户检查代码
鉴权/权限错误	根据错误码区分：1001（Key无效或过期）、1002（未提供 Key）、1004（权限不足）
网络超时	设置 `timeout=10`，最多重试 3 次
模型生成的参数不是合法 JSON	捕获 `json.JSONDecodeError`，返回结构化错误让模型修正

工程化建议：

tools 定义与执行函数分离：tools 定义放在 JSON 文件中，执行函数放在独立模块。新增工具时改配置不加代码。
tool_call_id 精确关联：每个 tool_call 都有唯一 ID，返回结果时必须带上 tool_call_id。
对话历史长度控制：多轮对话会导致 messages 超出模型上下文限制。建议保留最近 20 轮，早期消息做摘要压缩。

八、它能做什么，不能做什么

✅ 适合的场景：

场景	说明
快速查询跨市场行情	A 股、港股、美股、加密、外汇、贵金属统一查询
拉 K 线做技术分析	真实数据验证技术指标（均线、MACD 等）
股票基本面查询	EPS、每股净资产、股息率等财务数据
AI Agent 自主规划查询	多轮对话中模型自主决定调用顺序和数据依赖

❌ 不推荐使用的场景：

场景	原因
毫秒级高频交易	标准 HTTP 调用链路，不是交易所直连网关
投资决策的唯一依据	行情数据不构成投资建议
直接生成买卖信号	工具调用输出是数据，不是交易策略

九、常见问题

Q1：模型不调用工具，直接回复文本怎么办？

检查三点：

tools 参数是否正确传入 client.chat.completions.create()
tool 的 description 是否清晰描述了“何时该用这个工具”
如果仍然不触发，可以尝试 tool_choice="required" 强制模型调用工具

Q2：调用工具后返回的数据太长，模型“记不住”怎么办？

在工具执行函数中截取关键字段返回。例如 get_kline 只保留最近 N 根 K 线的 OHLC，get_stock_info 只保留核心财务指标。

Q3：DeepSeek Function Calling 和 ChatGPT Function Calling 有什么区别？

两者都兼容 OpenAI SDK 格式，tools 参数结构相似。如果你已有 OpenAI Function Calling 的开发经验，切换到 DeepSeek 几乎不需要改代码结构——只需要改 base_url 和 api_key。

收尾

回到开头那个问题：当真实数据喂给 DeepSeek 后，它到底在“分析”还是在“编故事”？

跑完这套代码，你就有了答案。Function Calling 不是魔法，而是一套精确的工程规范：你定义工具，模型规划调用，你执行查询，模型基于真实数据生成回复。这个流程里的每一个环节——数据从哪里来、工具怎么定义、错误怎么处理——都是透明的、可替换的、可优化的。

TickDB 提供了这套数据通道：通过 Skill、MCP、REST API、CLI 等多组入口，覆盖 4 大市场、40,000+ 品种，实现统一字段、同一套 API Key 体系。配合 DeepSeek Function Calling，你得到的不只是一个查价工具，而是一个可复用的 AI Agent 行情查询框架。

📡 本文行情数据服务由 TickDB.ai 提供。GitHub 开源，文档可查，代码可跑。

本文仅讨论技术接入和工具配置方式，不构成任何投资建议。文中所有代码和示例输出仅用于教学目的，不构成交易信号或投资策略推荐。