1. 项目概述一个纯粹、高效的股票行情查询工具最近在折腾一个叫 OpenClaw 的开源项目它本质上是一个帮你连接各种服务和数据的“智能助理”。在它的生态里一个核心概念叫“技能”Skill你可以理解为一个个功能插件。我琢磨着作为一个经常需要快速看一眼市场动态的人一个能随时查询股票价格的基础技能应该是刚需。市面上现成的方案要么太重依赖一堆库要么太不稳定接口说挂就挂要么就是功能太单一。所以我动手写了一个名为stock-price-query的 OpenClaw 技能。它的目标非常明确用最轻量、最可靠的方式查询全球多个主流市场的实时股票行情数据并且能无缝集成到你的自动化工作流或日常查询中。这个工具的核心是“纯粹”它不依赖任何第三方 Python 包完全基于 Python 3 的标准库这意味着你几乎可以在任何环境包括一些受限的服务器环境中零配置运行它。目前它覆盖了 A 股沪市、深市、港股和美股市场支持个股和主流指数还能一次性批量查询最多 20 只标的。如果你也在寻找一个命令行可调用、能返回结构化 JSON 数据、并且足够稳定的行情查询方案那么这个项目或许能给你提供一个清晰的实现思路和可直接复用的代码。2. 核心设计思路与架构解析2.1 为什么选择“零依赖”架构在金融数据工具的开发中依赖管理常常是个头疼的问题。很多优秀的库比如pandas,requests,aiohttp功能强大但体积也不小。对于一个核心功能只是发起 HTTP 请求并解析特定格式返回数据的工具来说引入它们显得有些“杀鸡用牛刀”。更重要的是在一些生产环境或嵌入式环境中安装额外的包可能受到严格限制或者引发版本冲突。因此我决定采用“零依赖”设计仅使用 Python 内置的urllib.request和json库。这样做的好处显而易见极致轻量整个技能就一个 Python 脚本复制即用无需pip install任何东西。环境兼容性极强只要系统有 Python 3现在主流环境基本都满足就能运行避免了因依赖库缺失或版本问题导致的“跑不起来”。启动速度快没有外部库的导入开销脚本执行几乎瞬间开始。当然这带来了挑战所有网络请求、异常处理、数据解析逻辑都需要自己用标准库手写。但这恰恰是可控性的体现你可以完全掌控请求超时、重试策略、错误响应格式等细节。2.2 市场识别与数据源选型逻辑支持多市场查询首先要解决的是如何根据用户输入的一个代码如600519,AAPL,00700自动判断它属于哪个市场。我设计了一套基于代码格式的规则匹配逻辑A 股沪市sh6 位数字且以6或9开头例如600519贵州茅台900901B股。A 股深市sz6 位数字且以0,2,3或4开头例如000001平安银行300750宁德时代。港股hk1 到 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)。经过长期观察和测试选择它主要基于以下几点考量免费且稳定无需注册、无需 API Key、没有调用频率的严格限制在合理个人使用范围内这对于开源项目和个人工具至关重要。数据全面除了实时价格、涨跌幅、成交量等基础数据还提供市盈率PE、市值等扩展信息且覆盖了 A 股、港股、美股。接口简单请求 URL 构造直观返回的数据是特定分隔符格式的字符串虽然不如 JSON 友好但解析起来确定性强。延迟可接受对于非高频交易的需求其数据延迟通常在可接受范围内。注意公开数据源的服务条款和稳定性可能发生变化。在实际生产应用中如果对数据的准确性、实时性和稳定性有更高要求建议考虑接入官方或授权的商业数据源。本项目的主要价值在于提供一个轻量化的、可工作的多市场查询架构。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格式qts_{市场}{代码} # 例如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 fs_{market}{code} params.append(param) query_str ,.join(params) # 2. 构造URL url fhttp://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, timeout10) 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_sh6005191~贵州茅台~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_sh6005191~贵州茅台~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, default0.0): 安全转换为浮点数 try: val safe_get(arr, idx) return float(val) if val else default except (ValueError, TypeError): return default def safe_int(arr, idx, default0): 安全转换为整数 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/qs_usAAPL查看~分隔的各个位置分别代表什么。数据类型安全转换原始数据可能是空字符串、None或包含逗号的数字如1,234,567。使用safe_float和safe_int这样的辅助函数可以避免解析崩溃。时间戳格式API 返回的时间戳通常是YYYYMMDDHHMMSS格式的字符串需要根据需求格式化为更易读的形式。美股数据差异美股的成交额字段通常无效或为0市盈率和市值字段的位置也可能与 A 股不同。这部分逻辑可能需要根据实际数据微调。4. 集成 OpenClaw 与实战应用4.1 如何将技能部署到 OpenClawOpenClaw 的技能管理非常直观。你只需要将整个stock-price-query目录放到 OpenClaw 的技能搜索路径下即可。克隆或下载项目git clone https://github.com/tjefferson/stock-price-query.git # 或者直接下载ZIP包并解压放置技能目录 OpenClaw 会从两个位置查找技能用户目录和项目目录。用户级安装推荐所有项目可用cp -r stock-price-query ~/.openclaw/skills/项目级安装仅当前项目可用cp -r stock-price-query /path/to/your/openclaw-project/skills/验证安装 启动你的 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 STOCKS600519,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_outputTrue, textTrue) data json.loads(result.stdout) # 转换为 DataFrame 并绘图 df pd.DataFrame(data) df.set_index(name, inplaceTrue) df[[current_price, change_percent]].plot(kindbar, subplotsTrue) 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_outputTrue, textTrue, timeout10) 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(host0.0.0.0, port5000)5. 常见问题、故障排查与优化建议在实际使用和开发过程中你可能会遇到以下问题。这里记录了我的排查经验和解决方案。5.1 数据查询失败或返回空数据现象可能原因排查步骤与解决方案返回status: error 提示网络错误1. 本地网络不通。2. 腾讯财经 API 临时故障或被屏蔽。3. 服务器防火墙策略限制。1. 用curl或浏览器直接访问http://qt.gtimg.cn/qs_sh000001看能否返回数据。2. 检查脚本中的User-Agent是否被目标服务器拒绝可尝试更换。3. 如果是服务器环境检查出站 HTTP 请求是否被允许。返回status: success但所有字段都是 0 或 null1. 股票代码错误或已退市。2. 市场代码指定错误。3. API 字段索引已变化解析失败。1.首先验证代码和市场用python3 stock_query.py 000001 sh和python3 stock_query.py 000001 sz分别测试看哪个有数据。2.手动验证 API在浏览器打开对应的 API URL查看原始返回的字符串是否包含有效数据。如果数据有但解析为0说明字段映射错了。3.检查港股补零港股代码700必须补零为00700再请求。批量查询时部分成功部分失败1. 单个代码错误导致整个请求被 API 侧忽略或返回异常格式。2. 请求 URL 过长。1. 实现“单条失败不影响其他”的逻辑在代码层面可以尝试将批量查询拆分为多个单次查询然后合并结果。或者在解析 API 返回的多行数据时对每一行进行独立的异常捕获。2. 腾讯 API 对单次查询的代码数量可能有限制实测20个以内较稳超过可能返回不完整数据。建议分批查询。5.2 性能与稳定性优化建议增加缓存机制对于频繁查询的股票比如你自选股列表可以在内存或本地文件中缓存结果并设置一个短暂的过期时间如 5-10 秒。这能大幅减少对上游 API 的请求压力并提升响应速度。注意行情数据实时性要求高缓存时间不宜过长。import time from functools import lru_cache lru_cache(maxsize32) def get_cached_price(code_market, expire_seconds10): 带缓存的查询code_market 如 600519.sh # 检查缓存逻辑... pass实现请求重试网络请求偶尔失败是正常的。可以封装一个带有指数退避策略的重试函数。def fetch_with_retry(url, max_retries3): for i in range(max_retries): try: return urllib.request.urlopen(url, timeout10).read() except (urllib.error.URLError, socket.timeout): if i max_retries - 1: raise wait_time 2 ** i # 指数退避1, 2, 4秒... time.sleep(wait_time)使用连接池高级虽然urllib本身简单但对于极高并发的场景可以考虑使用http.client并手动管理连接或者在放弃零依赖的前提下使用requests库它内置了连接池。对于这个工具的目标场景通常不需要。日志记录在生产环境使用建议加入简单的日志功能记录查询的代码、时间、结果状态和耗时便于后期监控和问题追溯。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) # 在 fetch_from_tencent 函数中记录 logger.info(fQuerying: {query_str}, Time: {response_time:.2f}s)5.3 扩展性思考如何支持更多市场或数据源这个项目的架构是易于扩展的。添加新市场如日本、英国在detect_market函数中添加新的代码格式识别规则。在fetch_from_tencent函数中判断如果是新市场则构造对应的 API 参数前提是数据源支持。如果腾讯 API 不支持就需要为这个市场实现一个新的数据获取函数例如fetch_from_yahoo。在parse_qt_data或新的解析函数中实现新市场的字段映射。切换或融合多个数据源抽象一个DataSource基类定义fetch_data(codes, markets)接口。为腾讯财经、雅虎财经如有、新浪财经等分别实现具体的类。在主查询函数中可以根据市场、代码或配置决定使用哪个数据源实例。甚至可以设计一个后备策略主数据源失败时自动尝试备用数据源。增加更多数据指标如果数据源提供了更多字段如分时数据、五档盘口、财务指标只需要在解析函数中增加对应的字段映射和转换逻辑即可输出 JSON 的结构可以向后兼容。这个stock-price-query项目就像搭好了一个轻便稳固的脚手架。它证明了用最小依赖完成一个实用工具是可行的。你可以直接用它来满足快速的命令行查询需求也可以把它作为一块基石融入到你更庞大的数据分析或自动化系统中。在金融数据的领域里稳定和简洁往往比繁复的功能更重要。