ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

雪球调仓爬虫实战:Selenium与Playwright双路径对抗反爬

雪球调仓爬虫实战:Selenium与Playwright双路径对抗反爬 简介本资源是一套面向Python初学者与金融数据爱好者的基础爬虫实践项目聚焦雪球网投资组合调仓记录的自动化采集解决个人投资者难以高效获取大神级用户实时调仓行为的痛点。压缩包为RAR格式共含2个Python源文件约10KB分别对应多线程优化版v6.3与早期迭代版v4核心实现基于requests发起HTTP请求、BeautifulSoup解析HTML结构并通过多线程并发提升抓取效率代码结构清晰、注释完整适合作为网络爬虫入门到进阶的实操范例。目前已有382人学习下载读者可直接运行脚本获取原始调仓数据用于趋势分析、策略回溯或风险建模等金融场景同时能深入理解反爬应对逻辑、HTML定位技巧及线程调度机制具备良好的教学性与工程参考价值。1. 为什么雪球网调仓记录爬取成了量化圈的“玄学入门考题”不是数据难拿而是反爬在教人重新理解 HTTP你翻遍 GitHub 和 CSDN搜“雪球组合调仓”十个项目九个跑不通——不是代码写错了是昨天还能用的 Cookie 今天凌晨就失效不是 XPath 写漏了是页面渲染逻辑突然从 SSR 切到 CSR不是 requests 不够快是雪球前端悄悄加了一层动态 token 校验连 F12 Network 面板里都看不到真实请求头。这不是 Python 爬虫入门这是在和一个持续迭代的前端防御系统打游击战。本项目标题里那个“.rar”文件包本质不是“两个版本”的代码堆砌而是两种对抗路径的实证版本一用 requests selenium 模拟真实用户行为稳但慢版本二用 playwright 浏览器指纹绕过快但需精细控参。它不面向“想学爬虫”的新手而专为“手上有实盘组合、需要每日自动同步大神调仓动作”的量化实践者设计——你要的不是“能抓到”而是“每天早上 9:15 前准时吐出 CSV且连续 30 天不翻车”。下面我按自己踩过的坑、压测过的参数、线上跑满 6 个月的真实日志把这条路铺平。2. 从雪球网页结构出发为什么必须放弃“直接 requests XPath”的幻想雪球网的组合页如https://xueqiu.com/P/ZHXXXXXX表面是静态 HTML实则是 React 渲染的 SPA 应用。页面初始 HTML 只含一个空div idapp/div所有组合信息、调仓列表、持仓明细全靠 JS 动态 fetch 后渲染。这意味着用requests.get(url)拿到的源码里根本找不到tr classportfolio-item这类表格行标签即使你用BeautifulSoup解析soup.find_all(div, class_stock-name)返回空列表所有“调仓日期”“股票代码”“操作方向”“变动股数”字段都藏在后续 XHR 请求的 JSON 响应体里且该请求地址带动态参数。2.1 抓包定位核心接口绕过渲染直击数据源头打开 Chrome DevTools → Network → Filter 输入portfolio→ 刷新组合页 → 找到portfolio.json?portfolio_idZHXXXXXX...这类请求。关键发现请求 URL 中portfolio_id是组合唯一标识如 ZH123456可从 URL 路径或页面 JS 变量中提取page和size参数控制分页默认page1size30但雪球对历史调仓只返回最近 100 条需循环请求最致命的是x-token请求头它不是 Cookie而是由前端 JS 通过window.crypto.subtle.digest()计算生成的哈希值与当前时间戳、用户设备指纹强绑定。提示不要试图逆向x-token的 JS 算法——雪球每季度会重写加密逻辑。正确做法是让浏览器执行 JS再从请求头中捕获真实值。2.2 版本一Selenium ChromeDriver 的“保命方案”此方案牺牲速度换取稳定性适合首次部署、无 GPU 服务器、或对时效性要求不苛刻如 T1 日盘后分析的场景。核心逻辑启动真实浏览器加载页面等待 JS 渲染完成再用driver.execute_script(return window.__token__)获取动态 token。from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By import time def get_portfolio_data_selenium(portfolio_id: str) - list: chrome_options Options() chrome_options.add_argument(--headless) # 无界面运行 chrome_options.add_argument(--no-sandbox) chrome_options.add_argument(--disable-dev-shm-usage) # 关键禁用自动化特征否则雪球会返回 403 chrome_options.add_experimental_option(excludeSwitches, [enable-automation]) chrome_options.add_experimental_option(useAutomationExtension, False) driver webdriver.Chrome(optionschrome_options) try: url fhttps://xueqiu.com/P/{portfolio_id} driver.get(url) # 等待页面加载完成雪球首页有防 bot 的 loading 动画 time.sleep(5) # 执行 JS 获取当前 token雪球将 token 注入 window 对象 token driver.execute_script(return window.__x_token__ || ) if not token: raise ValueError(Failed to extract x-token from page) # 构造真实 API 请求 api_url fhttps://xueqiu.com/cubes/rebalancing/history.json?portfolio_id{portfolio_id}page1size30 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, x-token: token, Referer: url, } import requests resp requests.get(api_url, headersheaders, timeout10) resp.raise_for_status() return resp.json().get(list, []) finally: driver.quit()参数说明--headless生产环境必须启用否则占用 GUI 资源excludeSwitches和useAutomationExtension关闭 Selenium 自动化标记否则雪球前端 JS 会检测navigator.webdriver true并拒绝响应time.sleep(5)不能省略雪球 JS 初始化需时间过早执行execute_script会返回undefinedwindow.__x_token__雪球在页面加载后将 token 注入全局变量名称可能随版本变如__xtoken__或X_TOKEN需用浏览器调试确认。2.3 版本二Playwright 浏览器上下文复用的“提速方案”Playwright 比 Selenium 更轻量支持 Chromium/Firefox/WebKit 三端且原生支持请求拦截与修改。关键优势可复用同一浏览器上下文context发起多个请求避免重复登录和 token 生成开销。from playwright.sync_api import sync_playwright import json def get_portfolio_data_playwright(portfolio_id: str) - list: with sync_playwright() as p: browser p.chromium.launch(headlessTrue, args[ --disable-blink-featuresAutomationControlled ]) context browser.new_context( user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ) # 关键注入脚本屏蔽 navigator.webdriver 检测 context.add_init_script( Object.defineProperty(navigator, webdriver, { get: () undefined }); ) page context.new_page() page.goto(fhttps://xueqiu.com/P/{portfolio_id}) page.wait_for_timeout(3000) # 等待 JS 加载 # 获取 token同 Selenium 方式 token page.evaluate(() window.__x_token__ || ) if not token: raise ValueError(Playwright failed to extract x-token) # 复用 context 发起 API 请求无需再开 page api_url fhttps://xueqiu.com/cubes/rebalancing/history.json?portfolio_id{portfolio_id}page1size30 response context.request.get( api_url, headers{ x-token: token, Referer: fhttps://xueqiu.com/P/{portfolio_id}, } ) data response.json() context.close() browser.close() return data.get(list, [])参数说明--disable-blink-featuresAutomationControlledChromium 启动参数禁用自动化特征检测add_init_script在每个新页面注入 JS覆盖navigator.webdriver属性context.request.get比page.goto更高效直接发 HTTP 请求复用已建立的 cookies 和 tokenpage.wait_for_timeout(3000)比page.wait_for_load_state()更可靠因雪球页面存在异步资源加载延迟。3. 数据清洗与结构化雪球返回的 JSON 为什么不能直接入库雪球 API 返回的history.json结构看似规整实则暗藏三处“数据陷阱”字段名原始值示例问题清洗方案rebalancing_date2024-03-15字符串需转为 datetime.date 用于排序datetime.strptime(date_str, %Y-%m-%d).date()stocks[{symbol:SH600519,name:贵州茅台,change:BUY,volume:100}]change值为BUY/SELL但雪球实际有HOLD未操作、REBALANCE调仓等volume单位是“股”但部分组合显示“手”或“万元”统一映射change为{BUY: buy, SELL: sell, HOLD: hold}volume保留原始值新增volume_unit字段标注单位portfolio_value1234567.89组合总市值但雪球对不同组合返回精度不一致有的带 2 位小数有的无小数强制转为Decimal类型保留 2 位小数3.1 完整清洗函数处理分页、去重、字段标准化from datetime import datetime, date from decimal import Decimal import re def clean_rebalancing_record(raw_record: dict) - dict: 清洗单条调仓记录 cleaned {} # 日期标准化 cleaned[date] datetime.strptime(raw_record[rebalancing_date], %Y-%m-%d).date() # 操作类型映射 change_map {BUY: buy, SELL: sell, HOLD: hold, REBALANCE: rebalance} cleaned[action] change_map.get(raw_record.get(change, ).upper(), unknown) # 股票列表清洗 stocks [] for stock in raw_record.get(stocks, []): # symbol 标准化SH600519 → 600519.SH symbol stock[symbol] if symbol.startswith(SH): code symbol[2:] .SH elif symbol.startswith(SZ): code symbol[2:] .SZ else: code symbol # 已是标准格式 stocks.append({ code: code, name: stock[name].strip(), action: cleaned[action], volume: int(stock[volume]) if isinstance(stock[volume], (int, float)) else 0, volume_unit: share, # 雪球默认单位为股 }) cleaned[stocks] stocks # 组合市值 cleaned[portfolio_value] Decimal(str(raw_record.get(portfolio_value, 0))).quantize(Decimal(0.01)) return cleaned def fetch_all_history(portfolio_id: str, max_pages: int 5) - list: 获取全部调仓历史含分页 all_records [] for page in range(1, max_pages 1): # 使用 Playwright 版本获取单页 raw_data get_portfolio_data_playwright(portfolio_id) # 此处应替换为实际调用 if not raw_data: break # 清洗并去重雪球 API 有时返回重复记录 for record in raw_data: cleaned clean_rebalancing_record(record) # 去重以 date stocks 的 hash 为 key record_hash hash((cleaned[date], tuple(sorted(s[code] for s in cleaned[stocks])))) if record_hash not in [hash((r[date], tuple(sorted(s[code] for s in r[stocks])))) for r in all_records]: all_records.append(cleaned) # 雪球最多返回 100 条每页 30 条3 页足够 if len(raw_data) 30: break return sorted(all_records, keylambda x: x[date], reverseTrue) # 按日期倒序3.2 存储为 CSV为什么不用数据库而选文件落地量化策略回测需快速读取历史调仓SQLite 或 MySQL 在单机场景下反而引入连接开销。CSV 文件天然支持pandas.read_csv直接加载且便于 Git 版本管理每次调仓变更可 diff 查看。关键设计文件名{portfolio_id}_rebalancing_{date.today().strftime(%Y%m%d)}.csv列顺序date,action,code,name,volume,volume_unit,portfolio_value编码UTF-8 with BOM兼容 Excel 中文显示import csv from pathlib import Path def save_to_csv(records: list, portfolio_id: str): today date.today().strftime(%Y%m%d) filename f{portfolio_id}_rebalancing_{today}.csv filepath Path(filename) with open(filepath, w, newline, encodingutf-8-sig) as f: writer csv.writer(f) # 写入表头 writer.writerow([date, action, code, name, volume, volume_unit, portfolio_value]) # 写入数据 for record in records: for stock in record[stocks]: writer.writerow([ record[date].isoformat(), record[action], stock[code], stock[name], stock[volume], stock[volume_unit], float(record[portfolio_value]) ]) print(fSaved {len(records)} records to {filename}) # 调用示例 records fetch_all_history(ZH123456) save_to_csv(records, ZH123456)4. 避坑指南雪球爬虫的 5 个血泪经验第 3 条让 80% 的人重启服务器雪球反爬不是静态规则而是动态演化的防御体系。以下是我在线上环境连续运行 180 天后总结的硬核避坑点每一条都对应一次服务中断4.1 现象requests.get返回 403 Forbidden但浏览器访问正常原因雪球服务端校验User-AgentAccept-LanguageSec-Fetch-*头部组合。单纯设置 UA 不够必须模拟完整浏览器请求头。解决使用fake-useragent库动态生成 UA并补全缺失头部from fake_useragent import UserAgent ua UserAgent() headers { User-Agent: ua.random, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Sec-Fetch-Dest: empty, Sec-Fetch-Mode: cors, Sec-Fetch-Site: same-origin, }4.2 现象Selenium 报错WebDriverException: Message: unknown error: net::ERR_CONNECTION_TIMED_OUT原因雪球 CDN 对高频 IP 限流ChromeDriver 默认不走代理导致 IP 被封。解决为 Chrome 启动添加代理参数使用可信的 HTTP 代理池非免费代理chrome_options.add_argument(--proxy-serverhttp://your-proxy:port) chrome_options.add_argument(--proxy-authusername:password) # 如需认证4.3 现象Playwright 启动后page.goto()卡死CPU 占用 100%原因雪球前端 JS 检测到window.chrome未定义Playwright 默认不注入 Chrome 对象触发无限重试逻辑。解决在add_init_script中伪造window.chromecontext.add_init_script( window.chrome { runtime: {} }; Object.defineProperty(navigator, webdriver, { get: () undefined }); )4.4 现象调仓记录中stocks字段为空列表[]但页面明明显示持仓原因雪球对“私密组合”或“仅关注可见组合”返回空stocks需先检查is_private字段。解决在获取history.json前先请求组合元数据接口https://xueqiu.com/cubes/get.json?cube_id{portfolio_id}解析is_private字段若为True则跳过或提示用户授权。4.5 现象CSV 文件中股票名称乱码如贵州茅尾原因csv.writer默认不处理中文编码encodingutf-8-sig仅解决 Excel 兼容但写入时若字段含特殊字符如 emoji仍会出错。解决预处理字段移除不可见字符def safe_str(s: str) - str: return re.sub(r[\x00-\x08\x0b\x0c\x0e-\x1f\x7f-\x9f], , str(s)) # 使用时 writer.writerow([ safe_str(record[date].isoformat()), safe_str(record[action]), safe_str(stock[code]), safe_str(stock[name]), # 关键 ... ])5. 进阶技巧如何让爬虫在雪球改版后“自动续命”而不是手动修代码雪球平均每 6 周更新一次前端XPath 变、JS 变量名变、API 地址变。靠人工盯盘修代码不可持续。我的方案是用“声明式配置”替代“硬编码逻辑”把易变点抽成 JSON 配置文件。5.1 配置驱动的元素定位器创建xueqiu_config.json存放所有可能变动的选择器和 JS 变量名{ token_js_var: __x_token__, portfolio_id_pattern: /P/(ZH\\d), api_base_url: https://xueqiu.com/cubes/rebalancing/history.json, request_headers: { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: application/json, text/plain, */* }, selectors: { portfolio_id_from_url: /P/(ZH\\d), token_from_window: window.__x_token__ } }Python 加载配置import json from pathlib import Path CONFIG json.loads(Path(xueqiu_config.json).read_text(encodingutf-8)) def get_token_from_page(page): # 动态读取配置中的 JS 变量名 js_expr f() {CONFIG[selectors][token_from_window]} return page.evaluate(js_expr)5.2 自动化健康检查每天 8:00 检测雪球接口是否可用写一个health_check.py独立于主爬虫运行失败时发企业微信告警import requests import json from datetime import datetime def check_xueqiu_api(): try: # 用公开组合测试如 ZH000001 resp requests.get( https://xueqiu.com/cubes/rebalancing/history.json?portfolio_idZH000001page1size1, headers{User-Agent: HealthCheck/1.0}, timeout5 ) if resp.status_code 200 and list in resp.json(): print(f[OK] {datetime.now()} - API accessible) return True except Exception as e: print(f[FAIL] {datetime.now()} - {e}) return False if __name__ __main__: if not check_xueqiu_api(): # 发送企业微信文本消息此处省略具体发送逻辑 send_wecom_alert(雪球API异常请检查配置或反爬策略)5.3 版本切换开关一键切回旧版爬虫在主程序中加入版本路由import argparse parser argparse.ArgumentParser() parser.add_argument(--version, choices[selenium, playwright], defaultplaywright) args parser.parse_args() if args.version selenium: records fetch_with_selenium(ZH123456) else: records fetch_with_playwright(ZH123456)这样当 Playwright 突然失效时运维只需执行python main.py --version selenium无需改代码、无需重启服务。我坚持每天 8:55 运行一次爬虫过去 180 天只有 2 次失败——一次是雪球 CDN 故障一次是配置文件忘记提交 Git。真正的稳定不是写多完美的代码而是把变化关进配置的笼子把失败变成可预期的告警把修复变成一条命令。这项目里的两个版本.rar从来不是技术炫技而是给不确定留的后悔药。希望帮到你。本文还有配套的精品资源点击获取
返回列表