零依赖多市场股票行情查询工具：Python标准库实现与OpenClaw集成

1. 项目概述：一个纯粹、高效的股票行情查询工具

最近在折腾一个叫 OpenClaw 的开源项目，它本质上是一个帮你连接各种服务和数据的“智能助理”。在它的生态里，一个核心概念叫“技能”（Skill），你可以理解为一个个功能插件。我琢磨着，作为一个经常需要快速看一眼市场动态的人，一个能随时查询股票价格的基础技能应该是刚需。市面上现成的方案要么太重（依赖一堆库），要么太不稳定（接口说挂就挂），要么就是功能太单一。

所以，我动手写了一个名为stock-price-query的 OpenClaw 技能。它的目标非常明确：用最轻量、最可靠的方式，查询全球多个主流市场的实时股票行情数据，并且能无缝集成到你的自动化工作流或日常查询中。这个工具的核心是“纯粹”，它不依赖任何第三方 Python 包，完全基于 Python 3 的标准库，这意味着你几乎可以在任何环境（包括一些受限的服务器环境）中零配置运行它。目前，它覆盖了 A 股（沪市、深市）、港股和美股市场，支持个股和主流指数，还能一次性批量查询最多 20 只标的。

如果你也在寻找一个命令行可调用、能返回结构化 JSON 数据、并且足够稳定的行情查询方案，那么这个项目或许能给你提供一个清晰的实现思路和可直接复用的代码。

2. 核心设计思路与架构解析

2.1 为什么选择“零依赖”架构？

在金融数据工具的开发中，依赖管理常常是个头疼的问题。很多优秀的库，比如pandas,requests,aiohttp，功能强大但体积也不小。对于一个核心功能只是发起 HTTP 请求并解析特定格式返回数据的工具来说，引入它们显得有些“杀鸡用牛刀”。更重要的是，在一些生产环境或嵌入式环境中，安装额外的包可能受到严格限制，或者引发版本冲突。

因此，我决定采用“零依赖”设计，仅使用 Python 内置的urllib.request和json库。这样做的好处显而易见：

1. 极致轻量：整个技能就一个 Python 脚本，复制即用，无需pip install任何东西。
2. 环境兼容性极强：只要系统有 Python 3（现在主流环境基本都满足），就能运行，避免了因依赖库缺失或版本问题导致的“跑不起来”。
3. 启动速度快：没有外部库的导入开销，脚本执行几乎瞬间开始。

当然，这带来了挑战：所有网络请求、异常处理、数据解析逻辑都需要自己用标准库手写。但这恰恰是可控性的体现，你可以完全掌控请求超时、重试策略、错误响应格式等细节。

2.2 市场识别与数据源选型逻辑

支持多市场查询，首先要解决的是如何根据用户输入的一个代码（如600519,AAPL,00700）自动判断它属于哪个市场。我设计了一套基于代码格式的规则匹配逻辑：

• A 股（沪市sh）：6 位数字，且以6或9开头（例如600519贵州茅台，900901B股）。
• A 股（深市sz）：6 位数字，且以0,2,3或4开头（例如000001平安银行，300750宁德时代）。
• 港股（hk）：1 到 5 位数字，或者特定的指数代码如HSI（恒生指数）。这里需要注意，港股股票代码通常是 4 或 5 位数字，前面补零，例如腾讯是00700。
• 美股（us）：由字母组成的股票代码，或者以点.开头的指数代码（例如AAPL,.IXIC）。

这个匹配顺序有讲究。例如，输入000001，它既符合深市 6 位数字以0开头的规则，也符合港股 1-5 位数字的规则。我的策略是优先匹配更具体的规则，即先判断是否为 A 股格式，如果不是再判断是否为港股数字代码，最后才落到美股。对于000001（上证指数），虽然代码像深市股票，但通常我们会指定市场为sh，或者在批量查询时，程序会通过查询接口返回的实际名称来辅助确认（实际上，腾讯财经 API 对000001.sh和000001.sz的响应是不同的）。

数据源方面，我选择了腾讯财经的公开 API (qt.gtimg.cn)。经过长期观察和测试，选择它主要基于以下几点考量：

1. 免费且稳定：无需注册、无需 API Key、没有调用频率的严格限制（在合理个人使用范围内），这对于开源项目和个人工具至关重要。
2. 数据全面：除了实时价格、涨跌幅、成交量等基础数据，还提供市盈率（PE）、市值等扩展信息，且覆盖了 A 股、港股、美股。
3. 接口简单：请求 URL 构造直观，返回的数据是特定分隔符格式的字符串，虽然不如 JSON 友好，但解析起来确定性强。
4. 延迟可接受：对于非高频交易的需求，其数据延迟通常在可接受范围内。

注意：公开数据源的服务条款和稳定性可能发生变化。在实际生产应用中，如果对数据的准确性、实时性和稳定性有更高要求，建议考虑接入官方或授权的商业数据源。本项目的主要价值在于提供一个轻量化的、可工作的多市场查询架构。

2.3 结构化输出与错误处理设计

工具的输出必须是机器可读的，这样才能方便地被其他程序（如 OpenClaw、你自己的监控脚本等）调用和处理。我定义了统一的 JSON 输出结构，无论成功失败，都返回一个包含status字段的 JSON 对象。

成功时，结构如下：

{ "code": "AAPL", "name": "苹果", "market": "us", "current_price": 168.82, "change": 1.24, "change_percent": 0.74, "open": 167.50, "high": 169.58, "low": 167.15, "prev_close": 167.58, "volume": 52890123, "amount": 0, // 美股该字段通常无效 "pe_ratio": 28.5, // 可能为 null "market_cap": 2600000000000, // 可能为 null "time": "20231026160000", "status": "success" }

失败时，结构如下：

{ "status": "error", "code": "INVALID_CODE", "message": "无法识别的股票代码格式: XYZ123" }

这种设计保证了调用方可以用一致的方式解析结果。在错误处理上，脚本会捕获网络超时、连接错误、解析异常等情况，并尽可能返回有意义的错误码和信息，而不是让程序直接崩溃。

3. 核心代码实现与关键细节剖析

3.1 市场检测函数的实现

这是整个工具的“大脑”，负责解析输入。代码如下（关键部分）：

def detect_market(code): """ 根据股票代码自动检测市场。 规则优先级：A股 > 港股(数字) > 美股 """ code = str(code).strip().upper() # 1. 检查是否为美股指数 (如 .IXIC, .DJI) if code.startswith('.'): return 'us' # 2. 检查是否为A股 (6位数字，特定开头) if code.isdigit() and len(code) == 6: first_char = code[0] if first_char in ('6', '9'): # 沪市主板/科创板/B股 return 'sh' elif first_char in ('0', '2', '3', '4'): # 深市主板/中小板/创业板/B股 return 'sz' # 3. 检查是否为港股指数代码 (如 HSI, HSCEI) if code in ('HSI', 'HSCEI'): return 'hk' # 4. 检查是否为港股股票 (1-5位数字) if code.isdigit() and 1 <= len(code) <= 5: return 'hk' # 5. 默认视为美股 (字母代码，如 AAPL, GOOGL) # 注意：这里假设剩余的都是美股。更严谨的做法可以加一个美股代码格式校验。 if code.isalpha(): return 'us' # 6. 无法识别 return None

关键细节与避坑指南：

• 字符串处理：首先用.strip().upper()处理输入，消除空格和大小写问题，这是避免后续匹配失败的基础。
• 优先级顺序：规则顺序很重要。000001按数字匹配会先被归为sz，但上证指数实际需要sh。因此，在工具中，对于知名指数代码，我内置了一个映射表来覆盖自动检测，或者允许用户通过参数手动指定市场。
• 港股代码补零：腾讯财经 API 要求港股代码是 5 位数字。所以如果用户输入700（腾讯），在构造请求前需要格式化为00700。这个补零逻辑在detect_market之后、发起请求前完成。
• 美股代码校验：目前的code.isalpha()判断比较简单，实际上美股代码可以是 1-5 个字母。更健壮的实现可以加入长度检查，并排除一些明显无效的字符组合。

3.2 网络请求与数据解析

这是工具的“双手”，负责抓取和清洗数据。我们使用urllib.request。

import urllib.request import urllib.error import json import time def fetch_from_tencent(codes, markets): """ 从腾讯财经API获取数据。 codes: 股票代码列表 markets: 对应的市场列表 返回解析后的数据列表 """ # 1. 构造查询参数 # 腾讯API格式：qt=s_{市场}{代码} # 例如：s_sh600519, s_usAAPL, s_hk00700 params = [] for code, market in zip(codes, markets): if market == 'hk' and code.isdigit(): code = code.zfill(5) # 港股补零至5位 param = f"s_{market}{code}" params.append(param) query_str = ",".join(params) # 2. 构造URL url = f"http://qt.gtimg.cn/q={query_str}" req = urllib.request.Request( url, headers={ 'User-Agent': 'Mozilla/5.0 (compatible; StockQuery/1.0)' # 模拟浏览器头，避免被简单屏蔽 } ) # 3. 发起请求与异常处理 try: with urllib.request.urlopen(req, timeout=10) as response: content = response.read().decode('gbk') # 注意编码是GBK except urllib.error.URLError as e: return {'status': 'error', 'message': f'网络请求失败: {e.reason}'} except socket.timeout: return {'status': 'error', 'message': '请求超时'} except Exception as e: return {'status': 'error', 'message': f'未知错误: {str(e)}'} # 4. 解析返回的数据 # 数据格式：每行一个股票，字段间以波浪线~分隔 # 例如：v_s_sh600519="1~贵州茅台~600519~1466.80~-18.50~-1.25~1521.00~1524.40~1463.60~1485.30~4191300~6198840000~20260224161416~~~28.5~2600000000000~"; lines = content.strip().split(';') results = [] for line in lines: if not line: continue # 解析单行数据... # (具体字段解析逻辑略，下文详述) return results

关键细节与避坑指南：

• 编码问题：腾讯 API 返回的数据是GBK编码，而不是更常见的UTF-8。使用.decode('gbk')是必须的，否则会得到乱码。
• 用户代理：设置一个合理的User-Agent是良好的网络公民行为，也能避免一些简单的反爬策略。
• 超时设置：务必设置timeout参数（这里设为 10 秒）。网络环境复杂，没有超时控制的请求可能会永远挂起。
• 错误处理：使用try...except块捕获URLError（网络问题）和timeout，并返回结构化的错误信息，而不是让脚本崩溃。

3.3 数据字段解析与映射

腾讯 API 返回的是一长串由波浪线~分隔的字符串，不同市场、不同类型的标的（股票 vs 指数），字段顺序和含义略有不同。我们需要根据市场来映射字段。

def parse_qt_data(line, market): """ 解析腾讯财经单行数据。 不同市场的字段索引位置不同，需要分别处理。 """ # 去除前缀和引号，得到纯数据字符串 # line 示例: v_s_sh600519="1~贵州茅台~600519~1466.80~-18.50~-1.25~..." if '=' not in line: return None data_str = line.split('=')[1].strip('"') fields = data_str.split('~') # 基础字段映射（A股、港股、美股股票通用性较高） # 注意：索引可能因API微调而变化，需要定期验证 mapping = { 'name': safe_get(fields, 1), # 名称 'code': safe_get(fields, 2), # 代码 'current_price': safe_float(fields, 3), # 当前价/最新价 'change': safe_float(fields, 4), # 涨跌额 'change_percent': safe_float(fields, 5), # 涨跌幅% 'open': safe_float(fields, 6), # 今开 'high': safe_float(fields, 7), # 最高 'low': safe_float(fields, 8), # 最低 'prev_close': safe_float(fields, 9), # 昨收 } # 处理成交量/成交额：A股和港股是同一套，美股不同 if market in ('sh', 'sz', 'hk'): mapping['volume'] = safe_int(fields, 10) # 成交量（手或股） mapping['amount'] = safe_float(fields, 11) # 成交额（元或港元） # A股和港股的部分数据在更后的位置 mapping['pe_ratio'] = safe_float(fields, 14) # 市盈率 mapping['market_cap'] = safe_float(fields, 15) # 总市值 elif market == 'us': # 美股字段索引有所不同 mapping['volume'] = safe_int(fields, 10) # 成交量（股） # 美股amount字段可能无效或位置不同，这里可能为0 mapping['amount'] = 0.0 mapping['pe_ratio'] = safe_float(fields, 14) # 可能位置不同 mapping['market_cap'] = safe_float(fields, 15) # 时间戳处理 time_str = safe_get(fields, 12) or safe_get(fields, 13) mapping['time'] = format_timestamp(time_str) if time_str else None mapping['market'] = market mapping['status'] = 'success' return mapping def safe_get(arr, idx, default=''): """安全获取列表元素""" return arr[idx] if idx < len(arr) else default def safe_float(arr, idx, default=0.0): """安全转换为浮点数""" try: val = safe_get(arr, idx) return float(val) if val else default except (ValueError, TypeError): return default def safe_int(arr, idx, default=0): """安全转换为整数""" try: val = safe_get(arr, idx) # 处理可能带逗号的数字，如 "1,234,567" val_clean = val.replace(',', '') if val else '0' return int(float(val_clean)) # 先转float再转int，处理科学计数法或小数 except (ValueError, TypeError): return default

关键细节与避坑指南：

• 字段索引偏移：这是最大的“坑”。腾讯 API 的字段顺序并非一成不变，历史上就有过调整。上述索引是基于当前（撰写时）观察到的常见顺序。强烈建议你在使用前，手动用浏览器访问一下 API 链接，核对一下返回数据的字段顺序。例如，访问http://qt.gtimg.cn/q=s_usAAPL，查看~分隔的各个位置分别代表什么。
• 数据类型安全转换：原始数据可能是空字符串、None或包含逗号的数字（如"1,234,567"）。使用safe_float和safe_int这样的辅助函数可以避免解析崩溃。
• 时间戳格式：API 返回的时间戳通常是YYYYMMDDHHMMSS格式的字符串，需要根据需求格式化为更易读的形式。
• 美股数据差异：美股的成交额字段通常无效或为0，市盈率和市值字段的位置也可能与 A 股不同。这部分逻辑可能需要根据实际数据微调。

4. 集成 OpenClaw 与实战应用

4.1 如何将技能部署到 OpenClaw

OpenClaw 的技能管理非常直观。你只需要将整个stock-price-query目录放到 OpenClaw 的技能搜索路径下即可。

1. 克隆或下载项目：

   git clone https://github.com/tjefferson/stock-price-query.git # 或者直接下载ZIP包并解压

2. 放置技能目录： OpenClaw 会从两个位置查找技能：用户目录和项目目录。

   • 用户级安装（推荐，所有项目可用）：

     cp -r stock-price-query ~/.openclaw/skills/

   • 项目级安装（仅当前项目可用）：

     cp -r stock-price-query /path/to/your/openclaw-project/skills/

3. 验证安装： 启动你的 OpenClaw 应用（例如飞书机器人），然后直接发送自然语言查询，如：

   • “贵州茅台股价多少？”
   • “查一下 AAPL 和 MSFT 的价格。”
   • “恒生指数今天怎么样？”

OpenClaw 会自动识别查询意图，并调用stock-price-query技能中的处理函数。技能的核心定义在SKILL.md文件中，它描述了技能的触发关键词、处理函数和输出格式。

4.2 技能定义文件解析

SKILL.md是 OpenClaw 技能的“说明书”。一个典型的定义如下：

# Stock Price Query ## Metadata - **Author**: Your Name - **Version**: 1.0.0 - **Description**: 查询实时股票价格（A股、港股、美股） ## Intents - `query_stock_price`: 查询股票价格 ## Patterns - `query_stock_price`: - `{stock}股价` - `{stock}股票` - `{stock}行情` - `查询{stock}` - `{stock}现在多少钱` - `How much is {stock}` - `price of {stock}` ## Functions ```python def query_stock_price(stock: str): """ 查询单只或多只股票价格。 参数 stock: 股票代码，多个用逗号分隔。 返回: 格式化的查询结果字符串。 """ # 调用 scripts/stock_query.py 的逻辑 # 解析 stock 参数，调用 fetch_from_tencent # 将返回的 JSON 数据格式化为友好的文本（如带表情符号的摘要） # 返回给 OpenClaw 用于展示

关键点：

• Patterns：这里定义了自然语言触发模式。{stock}是一个变量，OpenClaw 会从用户消息中提取出对应的股票代码。支持中英文混合模式，提高了识别率。
• Functions：这里指向真正的处理函数。这个函数接收从 Pattern 中提取的参数，执行查询逻辑，并返回一个字符串。这个字符串最终会被 OpenClaw 渲染到聊天界面（如飞书）。
• 格式化输出：在 OpenClaw 中，返回的文本可以包含简单的 Markdown 或表情符号来提升可读性，就像项目介绍中的示例那样。

4.3 独立脚本的进阶用法

除了作为 OpenClaw 技能，scripts/stock_query.py本身就是一个功能完整的命令行工具，可以集成到各种自动化场景中。

场景一：每日开盘价监控脚本你可以写一个 cron 任务，每天上午开盘后运行，获取你关注股票的开盘价并与昨日收盘价对比。

#!/bin/bash # monitor.sh STOCKS="600519,00700,AAPL" OUTPUT=$(python3 /path/to/stock_query.py $STOCKS) # 解析 JSON 输出，判断涨跌幅是否超过阈值，然后发送邮件或钉钉通知 echo $OUTPUT | jq -r '.[] | select(.change_percent > 5) | "\(.name) 涨幅超过5%: \(.change_percent)%"' # 使用 jq 工具解析 JSON，筛选出涨幅大于5%的股票

场景二：结合其他工具生成可视化报告将查询结果（JSON）导入到 Python 的pandas和matplotlib中，快速生成简单的价格走势对比图（需要先安装这些库）。

import subprocess import json import pandas as pd import matplotlib.pyplot as plt # 调用命令行工具获取数据 result = subprocess.run(['python3', 'stock_query.py', 'AAPL,MSFT,GOOGL'], capture_output=True, text=True) data = json.loads(result.stdout) # 转换为 DataFrame 并绘图 df = pd.DataFrame(data) df.set_index('name', inplace=True) df[['current_price', 'change_percent']].plot(kind='bar', subplots=True) plt.tight_layout() plt.savefig('stock_snapshot.png')

场景三：作为微服务的 API 端点你可以用 Flask 或 FastAPI 快速包装这个脚本，提供一个简单的 HTTP API 供内部系统调用。

from flask import Flask, request, jsonify import subprocess import json app = Flask(__name__) @app.route('/api/stock') def get_stock(): code = request.args.get('code') market = request.args.get('market', '') # 可选参数 if not code: return jsonify({'error': 'Missing code parameter'}), 400 cmd = ['python3', 'stock_query.py', code] if market: cmd.append(market) try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) return jsonify(json.loads(result.stdout)) except subprocess.TimeoutExpired: return jsonify({'error': 'Query timeout'}), 500 except Exception as e: return jsonify({'error': str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

5. 常见问题、故障排查与优化建议

在实际使用和开发过程中，你可能会遇到以下问题。这里记录了我的排查经验和解决方案。

5.1 数据查询失败或返回空数据

5.2 性能与稳定性优化建议

1. 增加缓存机制：对于频繁查询的股票（比如你自选股列表），可以在内存或本地文件中缓存结果，并设置一个短暂的过期时间（如 5-10 秒）。这能大幅减少对上游 API 的请求压力，并提升响应速度。注意，行情数据实时性要求高，缓存时间不宜过长。

   import time from functools import lru_cache @lru_cache(maxsize=32) def get_cached_price(code_market, expire_seconds=10): """带缓存的查询，code_market 如 '600519.sh'""" # 检查缓存逻辑... pass

2. 实现请求重试：网络请求偶尔失败是正常的。可以封装一个带有指数退避策略的重试函数。

   def fetch_with_retry(url, max_retries=3): for i in range(max_retries): try: return urllib.request.urlopen(url, timeout=10).read() except (urllib.error.URLError, socket.timeout): if i == max_retries - 1: raise wait_time = 2 ** i # 指数退避：1, 2, 4秒... time.sleep(wait_time)

3. 使用连接池（高级）：虽然urllib本身简单，但对于极高并发的场景，可以考虑使用http.client并手动管理连接，或者（在放弃零依赖的前提下）使用requests库，它内置了连接池。对于这个工具的目标场景，通常不需要。

4. 日志记录：在生产环境使用，建议加入简单的日志功能，记录查询的代码、时间、结果状态和耗时，便于后期监控和问题追溯。

   import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # 在 fetch_from_tencent 函数中记录 logger.info(f"Querying: {query_str}, Time: {response_time:.2f}s")

5.3 扩展性思考：如何支持更多市场或数据源？

这个项目的架构是易于扩展的。

1. 添加新市场（如日本、英国）：

   • 在detect_market函数中添加新的代码格式识别规则。
   • 在fetch_from_tencent函数中，判断如果是新市场，则构造对应的 API 参数（前提是数据源支持）。如果腾讯 API 不支持，就需要为这个市场实现一个新的数据获取函数，例如fetch_from_yahoo。
   • 在parse_qt_data或新的解析函数中，实现新市场的字段映射。

2. 切换或融合多个数据源：

   • 抽象一个DataSource基类，定义fetch_data(codes, markets)接口。
   • 为腾讯财经、雅虎财经（如有）、新浪财经等分别实现具体的类。
   • 在主查询函数中，可以根据市场、代码或配置，决定使用哪个数据源实例。甚至可以设计一个后备策略：主数据源失败时，自动尝试备用数据源。

3. 增加更多数据指标：如果数据源提供了更多字段（如分时数据、五档盘口、财务指标），只需要在解析函数中增加对应的字段映射和转换逻辑即可，输出 JSON 的结构可以向后兼容。

这个stock-price-query项目就像搭好了一个轻便稳固的脚手架。它证明了用最小依赖完成一个实用工具是可行的。你可以直接用它来满足快速的命令行查询需求，也可以把它作为一块基石，融入到你更庞大的数据分析或自动化系统中。在金融数据的领域里，稳定和简洁往往比繁复的功能更重要。