3步搞定A股实时监控脚本:用 pysnowball 雪球数据接口快速上手
【免费下载链接】pysnowball雪球股票数据接口 python edition项目地址: https://gitcode.com/gh_mirrors/py/pysnowball
每天开盘前,你是不是也要手动打开好几个页面:先看自选股涨跌,再翻一遍持仓基金净值,最后还要核对几只重点股票的成交量?我最初也是这么干的,直到我把这套流程写成一个脚本,用 pysnowball(雪球股票数据接口的 Python 版)一次性拉回所有数据。这篇文章就把这套"从配置到跑通"的完整路径拆给你看,全程只需要三件套:装库、配 token、写调用。
自己爬网页 VS 现成数据接口
在动手之前,先回答一个绕不开的问题:为什么不自建爬虫抓行情?我自己踩过坑,把两种方案的关键差异整理成一张表:
| 对比项 | 手写爬虫 | pysnowball 数据接口 |
|---|---|---|
| 接口稳定性 | 网页改版就失效,要持续维护 | 封装雪球接口,结构稳定 |
| 数据格式 | HTML 里自己抠字段 | 直接返回 JSON,字段成体系 |
| 覆盖面 | 通常只能拿到行情 | 行情、财务、基金、资金流、F10 全覆盖 |
| 上手成本 | 要处理请求头、Cookie、反爬 | 一个函数一个接口 |
一句话总结:爬虫是把时间花在"怎么拿到数据"上,而用 pysnowball 是把时间花在"用数据做什么"上。下面是它的实际能力清单,代码在仓库的pysnowball/目录下按模块划分,__init__.py里统一导出了全部函数:
- 实时行情:
quotec、quote_detail、pankou、kline(日/周/月/60分/30分/1分) - 财务数据:
income、balance、cash_flow、indicator、business三大报表加核心指标 - 资金流向:
capital_flow当日分时、capital_history历史日线、capital_assort成交分布 - 基金数据:
fund_info、fund_nav_history、fund_growth、fund_asset等 9 个接口 - F10 专题:
top_holders十大股东、bonus分红、main_indicator主要指标等
首次配置三步走:安装、token、连通性验证
第一步:安装并导入
安装命令和普通 Python 库没有区别:
pip install pysnowball然后以ball为别名导入,后面所有接口都从这个命名空间调用:
import pysnowball as ball第二步:配置雪球 token
这是整个上手过程中唯一容易卡住的环节。原理很简单:pysnowball 底层请求雪球接口时,需要携带你登录雪球的 Cookie 凭证,核心是两个值——xq_a_token和u。获取方式:浏览器登录雪球网页版,按 F12 打开开发者工具,在 Network 面板任意请求的 Cookie 里就能找到它们。
拿到后调用set_token写入环境变量(pysnowball/token.py的实现就是把值存进XUEQIUTOKEN):
ball.set_token("xq_a_token=你的token值;u=你的用户id")注意两点:一是这个 token 有有效期,过期后脚本会报错,届时重新抓取即可;二是不要把它写进公开仓库,建议放进本地配置文件。
第三步:用一次性接口验证连通性
有个小技巧值得先记住:行情快照quotec走的是fetch_without_token(见pysnowball/utls.py),不校验 token,很适合拿来当"网没网、库装没装好"的冒烟测试:
# 上证指数快照,无需token即可返回 print(ball.quotec("SH000001"))这一步能跑通,说明环境没问题。接下来需要 token 的接口就都能放心使用了。
实战一:把雪球自选股拉成一张涨跌表
很多人不知道watch_list可以直接读取你雪球账号下的自选股列表,配合quotec批量取行情,十几行代码就是一个"今日自选监控":
import pysnowball as ball # 1. 读取自选股列表(返回的分组信息里,category=1 是股票分组) watch = ball.watch_list()['data']['stocks'] my_group = [g for g in watch if g['id'] == -1][0] # id=-1 代表"全部"分组 # 2. 取该分组下的股票代码 stocks = ball.watch_stock(my_group['id'])['data']['stocks'] symbols = ",".join(s['symbol'] for s in stocks) # 逗号拼接,一次批量请求 # 3. 批量获取行情,只保留关心的字段 quotes = ball.quotec(symbols)['data'] for q in quotes: print(f"{q['name']:<8} 现价 {q['current']:>8.2f} 涨跌幅 {q['percent']:>6.2f}% 成交量 {q['volume']}")这段代码的讲解价值在于:自选列表和行情是两个独立接口,先取名单、再批量取价,中间用逗号把代码拼成一个字符串,一次请求搞定全部股票,比逐个调用高效得多。
实战二:把基金净值走势变成可视化报告
股票看完,再看基金。基金类接口走的是蛋卷数据源(utls.py中的fetch_danjuan_fund),不需要 token,非常省心。下面这段代码抓取一只基金的近期净值,并顺手算出区间收益:
import pysnowball as ball from datetime import datetime def fund_snapshot(fund_code, pages=3): # 1. 基金基本信息:名称、最新净值、各周期涨跌幅都在 fund_derived 里 info = ball.fund_info(fund_code)['data'] derived = info['fund_derived'] print(f"基金:{info['fd_name']} 最新净值:{derived['unit_nav']}") print(f"近1月 {derived['nav_grl1m']}% 近3月 {derived['nav_grl3m']}% 近1年 {derived['nav_grl1y']}%") # 2. 分页拉取历史净值(默认每页10条,这里取3页共30天) history = [] for page in range(1, pages + 1): data = ball.fund_nav_history(fund_code, page=page, size=10) history.extend(data['data']['items']) # 3. 转成日期-净值序列并计算区间收益率 series = sorted( (datetime.fromtimestamp(int(i['nav_date']) / 1000), float(i['unit_nav'])) for i in history if i.get('unit_nav') ) start_d, start_v = series[0] end_d, end_v = series[-1] print(f"区间 {start_d:%Y-%m-%d} ~ {end_d:%Y-%m-%d}") print(f"累计收益 {(end_v / start_v - 1) * 100:.2f}%") fund_snapshot("008975")这段代码的关键点有三个:分页参数page/size控制拉取量;时间戳毫秒转日期需要// 1000;排序保证净值按时间递增,收益率计算才不出错。fund_nav_history的完整参数说明可以在仓库的APIs/fund_nav_history.md文档里找到。
进阶:批量场景下的限流与报错排查清单
脚本跑通之后,如果你开始批量拉全市场数据,会遇到两类问题:请求太频繁被限流、token 过期导致报错。这里给出一份可照抄的排查清单:
报错 1:未设置TOKEN
pysnowball/cons.py里定义了这个错误信息,触发原因是set_token没调用或环境变量丢失。检查代码里是否在导入后立即设置了 token,且拼写与 Cookie 中的字段名完全一致。
报错 2:接口返回异常或非 200
utls.py的fetch在非 200 状态时会直接抛出异常。最常见的原因是 token 失效——重新登录雪球抓取最新 Cookie 即可。其次检查网络代理是否拦截了stock.xueqiu.com域名。
报错 3:请求频繁被限流
批量场景建议自己加一层限流器,控制每分钟请求数,避免触发风控:
import time from collections import deque class RateLimiter: def __init__(self, max_per_minute=30): self.max = max_per_minute self.stamps = deque() def wait(self): now = time.time() while self.stamps and now - self.stamps[0] > 60: self.stamps.popleft() if len(self.stamps) >= self.max: sleep = 60 - (now - self.stamps[0]) + 1 time.sleep(max(sleep, 0)) self.stamps.append(time.time()) # 用法:每次调用前 limiter.wait() limiter = RateLimiter(max_per_minute=30) for code in stock_list: limiter.wait() data = ball.quote_detail(code)另外,重试机制也很实用:遇到偶发网络抖动,包一层带退避的重试即可,不必每次手动重跑。
把脚本变成你的每日开盘前习惯
回头看,这套流程的全部价值就是把"手动刷新十几个页面"压缩成了"运行一个脚本":自选股行情、持仓基金净值、区间收益一次性输出。pysnowball 值得一试的理由很实在——安装一条命令、配置一个 token、调用就是一个个函数,数据从行情到财务到资金流全覆盖,足够支撑个人量化研究和投资跟踪的大部分需求。
想进一步探索,仓库里还有几份现成的接口文档可以直接查阅:基金类接口见APIs/目录下的fund_*.md,K 线返回结构见APIs/realtime_kline.md,全部函数签名以pysnowball/__init__.py的导出为准。如果你也想加入量化交流,可以扫描项目根目录下的wx.png二维码(搜索"宽客笔记")找到组织。
现在就动手吧:git clone https://gitcode.com/gh_mirrors/py/pysnowball,装上依赖、配好 token,把上面两个实战脚本改成你自己的自选列表,明天开盘前,让数据替你打工。
【免费下载链接】pysnowball雪球股票数据接口 python edition项目地址: https://gitcode.com/gh_mirrors/py/pysnowball
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考