3个坑填平后我手写实现东方财富网站数据抓取全解
昨天调试到凌晨两点，盯着终端里满屏的 403 Forbidden 和 JSONDecodeError，那种复制来的代码跑不通、改参数也没反应的崩溃感，相信做过爬虫的兄弟都懂。我试过换 IP、加 Headers、甚至上云函数，结果还是被反爬机制拦得死死的。最后没办法，我决定放弃那些花里胡哨的第三方库，回归本质，直接基于 HTTP 协议手写实现一个最小化的数据抓取器。这一路下来，不仅搞定了东方财富网站的实时行情获取，还顺带把 Python 网络编程里最底层的几个坑给填平了。
今天不整虚的，直接上干货。咱们不讲大道理，就聊聊怎么从零开始，用不到 100 行核心代码，稳定地拿到你想要的数据。
项目目标与反爬初探
在动手写代码之前，得先搞清楚咱们要干嘛。对于量化交易或数据分析来说，东方财富网站的接口之所以受欢迎，是因为它返回的是结构清晰的 JSON 数据，而不是需要解析复杂 HTML 表格。我们的目标很明确：获取指定股票代码（如 600519 贵州茅台）的实时股价、涨跌幅以及成交量。
很多新手一上来就用 requests 库，发个 GET 请求，结果发现数据不对，或者接口直接失效。为什么？因为东方财富网站的接口并不是对所有人都“敞开怀抱”的。如果你打开浏览器，按 F12 查看 Network 面板，你会发现它请求的是 push2.eastmoney.com 这样的子域名，而且参数里藏着一个叫 ut 的 token。
这就是第一个坑：单纯的 HTTP 请求缺少了必要的鉴权参数。很多人复制网上的旧代码，里面的 ut 值是写死的，或者过期了，导致请求虽然通，但返回的数据是空的或者错误的。这时候，光靠猜是不行的，必须得逆向分析一下。
目录结构与依赖管理
为了保持代码的工程化，咱们别把所有东西都塞进一个 main.py 里。我习惯用这种简洁但清晰的目录结构：
em-crawler/
├── main.py # 入口文件
├── crawler/
│ ├── __init__.py
│ ├── client.py # 封装 HTTP 客户端
│ ├── parser.py # 数据解析逻辑
│ └── config.py # 配置文件
├── requirements.txt
└── README.md在 requirements.txt 里，我们只需要最基础的库：
requests=2.31.0
pandas=2.0.0注意，这里我特意没有引入 selenium 或 scrapy。对于这种轻量级的 JSON 接口，引入重型框架反而会增加调试复杂度。手写实现的核心价值在于“可控”，每一个字节进出，你都心里有数。
核心代码实现：逐行拆解
好，进入正题。这是最关键的 client.py 文件，也就是我们的“手写实现”核心。
import requests
import time
import json
from .config import API_BASE_URL, DEFAULT_UTclass EastMoneyClient:def __init__(self):self.session = requests.Session()# 设置基础头，模拟浏览器环境self.session.headers.update({User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36,Referer: https://quote.eastmoney.com/,Accept: application/json, text/plain, */*,Connection: keep-alive})def get_realtime_quote(self, secid: str) - dict:获取实时行情:param secid: 股票ID，格式为 市场.代码，如 1.600519 (沪) 0.000001 (深):return: 解析后的字典数据# 构造请求参数，注意这里必须包含 ut 字段params = {fltt: 2,invt: 2,secid: secid,fields: f43,f44,f45,f46,f47,f48,f50,f51,f52,f57,f58,f60,f170,ut: DEFAULT_UT # 关键参数}try:# 发起 GET 请求response = self.session.get(f{API_BASE_URL}/api/qt/stock/get, params=params, timeout=5)# 检查 HTTP 状态码if response.status_code != 200:raise Exception(fHTTP Error: {response.status_code})# 解析 JSONdata = response.json()# 检查业务状态码if data.get(rc) != 0:raise Exception(fAPI Error: {data.get('msg')})return data.get(data, {})except requests.exceptions.Timeout:print(请求超时，请检查网络或稍后重试)return {}except json.JSONDecodeError:print(JSON 解析失败，可能触发了反爬机制)return {}except Exception as e:print(f未知错误: {e})return {}这里有几个细节必须注意：Session 对象：使用 requests.Session() 而不是每次新建 requests.get()，可以复用 TCP 连接，减少握手开销，这在高频请求时非常明显。
Headers 伪装：User-Agent 和 Referer 是必须的。虽然东方财富网站对普通行情接口没有极端的反爬，但如果没有正确的 Referer，某些接口会直接拒绝。
ut 参数：这是很多教程忽略的。我在 config.py 里写死了一个通用的 ut 值（通常是一个固定的哈希字符串，如 7eea3edcaed734bea9cbfc24409ed989，具体值需根据当前官方文档或抓包获取，因为偶尔会变）。如果这个值失效了，你需要去浏览器抓包重新复制。
secid 格式：注意沪市（6开头）是 1.600519，深市（0开头）是 0.000001。这是东方财富网站特有的编码规则，很多新手在这里踩坑，传成了 600519 导致查不到数据。运行与测试：验证有效性
代码写完了，跑一下看看。main.py 很简单：
from crawler.client import EastMoneyClient
import timedef main():client = EastMoneyClient()# 测试贵州茅台print(正在获取贵州茅台 (600519) 数据...)data = client.get_realtime_quote(1.600519)if data:print(f当前价格: {data.get('f43')})print(f涨跌幅: {data.get('f170')}%)print(f成交量: {data.get('f47')})else:print(获取失败)time.sleep(2) # 简单限速，避免过于频繁if __name__ == __main__:main()运行结果应该是这样的：
正在获取贵州茅台 (600519) 数据...
当前价格: 1750.50
涨跌幅: 1.23%
成交量: 123456如果在运行过程中遇到 403 错误，别慌。这通常是因为你的 IP 被暂时标记了。这时候，手写实现的优势就体现出来了：你可以在 client.py 里加一个简单的重试机制，或者切换出口 IP。
import randomdef _retry_request(self, url, params, retries=3):for i in range(retries):try:resp = self.session.get(url, params=params, timeout=5)if resp.status_code == 403:wait_time = random.uniform(2, 5)print(f被拦截，等待 {wait_time:.2f} 秒后重试...)time.sleep(wait_time)continuereturn respexcept Exception:if i == retries - 1:raisetime.sleep(1)return None把原来的 self.session.get 替换成这个 _retry_request，稳定性会提升一个档次。
优化扩展：从单点到批量
搞定单只股票后，肯定有人想问：“我怎么批量获取整个板块的数据？”
这时候，简单的循环请求就不够用了，因为并发请求容易触发更严格的反爬。我建议引入 asyncio 和 aiohttp，但这需要重构代码。作为手写实现的进阶，我们可以先用多线程做个折中方案，或者更推荐的方式：利用东方财富网站的批量接口。
其实，东方财富网站的 api/qt/stock/get 接口本身并不支持批量传多个 secid。但是，它有一个 clist 接口，可以分页获取某个板块的所有股票列表及简要数据。
我们可以这样扩展 client.py：def get_stock_list(self, fs: str, page: int = 1, limit: int = 100) - list:获取股票列表:param fs: 过滤条件，如 'm:1 t:6' (沪深A股):param page: 页码:param limit: 每页数量:return: 股票列表params = {pn: page,pz: limit,po: 1,np: 1,fltt: 2,invt: 2,fid: f3,fs: fs,fields: f12,f14,f2,f3, # 代码,名称,现价,涨跌幅ut: DEFAULT_UT}url = f{API_BASE_URL}/api/qt/clist/get# 复用重试逻辑...# 此处省略具体实现，逻辑同上pass通过这种方式，你可以一次性获取几千只股票的快照数据，然后再针对感兴趣的个股去调用 get_realtime_quote 获取详细数据。这种“先广后深”的策略，是处理东方财富网站这类数据源的最佳实践。
小结与避坑指南
回顾整个过程，我们从零手写实现了一个针对东方财富网站的数据抓取器。虽然代码量不多，但涵盖了 HTTP 会话管理、反爬应对、数据解析和错误处理等核心环节。
这里有几个血泪教训，送给正在看这篇文章的你：不要迷信第三方库：很多封装好的库（如 akshare、tushare）底层也是调用的这些接口。当库报错时，如果你不懂底层的 HTTP 交互，你就只能干瞪眼。自己动手，才能丰衣足食。
关注官方文档与变动：虽然东方财富网站没有公开完整的 API 文档，但其前端 JS 文件（通常在 static 目录下）里藏着所有接口的定义。定期检查这些文件，能帮你第一时间发现参数变化。
合规使用：爬取数据仅供个人学习或低频研究使用。不要高频请求，不要用于商业转售，尊重数据源的服务条款。根据《计算机信息网络国际联网安全保护管理办法》，非法获取数据是违法的。技术栈的选择没有绝对的好坏，关键在于是否适合你的场景。对于中小团队或个人开发者，手写实现轻量级爬虫，比引入庞大的框架更灵活、更易维护。
你公司项目里是怎么处理这类实时金融数据源的？是用现成的商业 API，还是自己维护爬虫集群？或者你在逆向东方财富网站接口时遇到过什么更刁钻的反爬手段？欢迎在评论区聊聊，咱们一起交流避坑经验。 3个坑填平后我手写实现东方财富网站数据抓取全解
3个坑填平后我手写实现东方财富网站数据抓取全解
昨天调试到凌晨两点，盯着终端里满屏的 403 Forbidden 和 JSONDecodeError，那种复制来的代码跑不通、改参数也没反应的崩溃感，相信做过爬虫的兄弟都懂。我试过换 IP、加 Headers、甚至上云函数，结果还是被反爬机制拦得死死的。最后没办法，我决定放弃那些花里胡哨的第三方库，回归本质，直接基于 HTTP 协议手写实现一个最小化的数据抓取器。这一路下来，不仅搞定了东方财富网站的实时行情获取，还顺带把 Python 网络编程里最底层的几个坑给填平了。
今天不整虚的，直接上干货。咱们不讲大道理，就聊聊怎么从零开始，用不到 100 行核心代码，稳定地拿到你想要的数据。
项目目标与反爬初探
在动手写代码之前，得先搞清楚咱们要干嘛。对于量化交易或数据分析来说，东方财富网站的接口之所以受欢迎，是因为它返回的是结构清晰的 JSON 数据，而不是需要解析复杂 HTML 表格。我们的目标很明确：获取指定股票代码（如 600519 贵州茅台）的实时股价、涨跌幅以及成交量。
很多新手一上来就用 requests 库，发个 GET 请求，结果发现数据不对，或者接口直接失效。为什么？因为东方财富网站的接口并不是对所有人都“敞开怀抱”的。如果你打开浏览器，按 F12 查看 Network 面板，你会发现它请求的是 push2.eastmoney.com 这样的子域名，而且参数里藏着一个叫 ut 的 token。
这就是第一个坑：单纯的 HTTP 请求缺少了必要的鉴权参数。很多人复制网上的旧代码，里面的 ut 值是写死的，或者过期了，导致请求虽然通，但返回的数据是空的或者错误的。这时候，光靠猜是不行的，必须得逆向分析一下。
目录结构与依赖管理
为了保持代码的工程化，咱们别把所有东西都塞进一个 main.py 里。我习惯用这种简洁但清晰的目录结构：
em-crawler/
├── main.py # 入口文件
├── crawler/
│ ├── __init__.py
│ ├── client.py # 封装 HTTP 客户端
│ ├── parser.py # 数据解析逻辑
│ └── config.py # 配置文件
├── requirements.txt
└── README.md在 requirements.txt 里，我们只需要最基础的库：
requests=2.31.0
pandas=2.0.0注意，这里我特意没有引入 selenium 或 scrapy。对于这种轻量级的 JSON 接口，引入重型框架反而会增加调试复杂度。手写实现的核心价值在于“可控”，每一个字节进出，你都心里有数。
核心代码实现：逐行拆解
好，进入正题。这是最关键的 client.py 文件，也就是我们的“手写实现”核心。
import requests
import time
import json
from .config import API_BASE_URL, DEFAULT_UTclass EastMoneyClient:def __init__(self):self.session = requests.Session()# 设置基础头，模拟浏览器环境self.session.headers.update({User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36,Referer: https://quote.eastmoney.com/,Accept: application/json, text/plain, */*,Connection: keep-alive})def get_realtime_quote(self, secid: str) - dict:获取实时行情:param secid: 股票ID，格式为 市场.代码，如 1.600519 (沪) 0.000001 (深):return: 解析后的字典数据# 构造请求参数，注意这里必须包含 ut 字段params = {fltt: 2,invt: 2,secid: secid,fields: f43,f44,f45,f46,f47,f48,f50,f51,f52,f57,f58,f60,f170,ut: DEFAULT_UT # 关键参数}try:# 发起 GET 请求response = self.session.get(f{API_BASE_URL}/api/qt/stock/get, params=params, timeout=5)# 检查 HTTP 状态码if response.status_code != 200:raise Exception(fHTTP Error: {response.status_code})# 解析 JSONdata = response.json()# 检查业务状态码if data.get(rc) != 0:raise Exception(fAPI Error: {data.get('msg')})return data.get(data, {})except requests.exceptions.Timeout:print(请求超时，请检查网络或稍后重试)return {}except json.JSONDecodeError:print(JSON 解析失败，可能触发了反爬机制)return {}except Exception as e:print(f未知错误: {e})return {}这里有几个细节必须注意：Session 对象：使用 requests.Session() 而不是每次新建 requests.get()，可以复用 TCP 连接，减少握手开销，这在高频请求时非常明显。
Headers 伪装：User-Agent 和 Referer 是必须的。虽然东方财富网站对普通行情接口没有极端的反爬，但如果没有正确的 Referer，某些接口会直接拒绝。
ut 参数：这是很多教程忽略的。我在 config.py 里写死了一个通用的 ut 值（通常是一个固定的哈希字符串，如 7eea3edcaed734bea9cbfc24409ed989，具体值需根据当前官方文档或抓包获取，因为偶尔会变）。如果这个值失效了，你需要去浏览器抓包重新复制。
secid 格式：注意沪市（6开头）是 1.600519，深市（0开头）是 0.000001。这是东方财富网站特有的编码规则，很多新手在这里踩坑，传成了 600519 导致查不到数据。运行与测试：验证有效性
代码写完了，跑一下看看。main.py 很简单：
from crawler.client import EastMoneyClient
import timedef main():client = EastMoneyClient()# 测试贵州茅台print(正在获取贵州茅台 (600519) 数据...)data = client.get_realtime_quote(1.600519)if data:print(f当前价格: {data.get('f43')})print(f涨跌幅: {data.get('f170')}%)print(f成交量: {data.get('f47')})else:print(获取失败)time.sleep(2) # 简单限速，避免过于频繁if __name__ == __main__:main()运行结果应该是这样的：
正在获取贵州茅台 (600519) 数据...
当前价格: 1750.50
涨跌幅: 1.23%
成交量: 123456如果在运行过程中遇到 403 错误，别慌。这通常是因为你的 IP 被暂时标记了。这时候，手写实现的优势就体现出来了：你可以在 client.py 里加一个简单的重试机制，或者切换出口 IP。
import randomdef _retry_request(self, url, params, retries=3):for i in range(retries):try:resp = self.session.get(url, params=params, timeout=5)if resp.status_code == 403:wait_time = random.uniform(2, 5)print(f被拦截，等待 {wait_time:.2f} 秒后重试...)time.sleep(wait_time)continuereturn respexcept Exception:if i == retries - 1:raisetime.sleep(1)return None把原来的 self.session.get 替换成这个 _retry_request，稳定性会提升一个档次。
优化扩展：从单点到批量
搞定单只股票后，肯定有人想问：“我怎么批量获取整个板块的数据？”
这时候，简单的循环请求就不够用了，因为并发请求容易触发更严格的反爬。我建议引入 asyncio 和 aiohttp，但这需要重构代码。作为手写实现的进阶，我们可以先用多线程做个折中方案，或者更推荐的方式：利用东方财富网站的批量接口。
其实，东方财富网站的 api/qt/stock/get 接口本身并不支持批量传多个 secid。但是，它有一个 clist 接口，可以分页获取某个板块的所有股票列表及简要数据。
我们可以这样扩展 client.py：def get_stock_list(self, fs: str, page: int = 1, limit: int = 100) - list:获取股票列表:param fs: 过滤条件，如 'm:1 t:6' (沪深A股):param page: 页码:param limit: 每页数量:return: 股票列表params = {pn: page,pz: limit,po: 1,np: 1,fltt: 2,invt: 2,fid: f3,fs: fs,fields: f12,f14,f2,f3, # 代码,名称,现价,涨跌幅ut: DEFAULT_UT}url = f{API_BASE_URL}/api/qt/clist/get# 复用重试逻辑...# 此处省略具体实现，逻辑同上pass通过这种方式，你可以一次性获取几千只股票的快照数据，然后再针对感兴趣的个股去调用 get_realtime_quote 获取详细数据。这种“先广后深”的策略，是处理东方财富网站这类数据源的最佳实践。
小结与避坑指南
回顾整个过程，我们从零手写实现了一个针对东方财富网站的数据抓取器。虽然代码量不多，但涵盖了 HTTP 会话管理、反爬应对、数据解析和错误处理等核心环节。
这里有几个血泪教训，送给正在看这篇文章的你：不要迷信第三方库：很多封装好的库（如 akshare、tushare）底层也是调用的这些接口。当库报错时，如果你不懂底层的 HTTP 交互，你就只能干瞪眼。自己动手，才能丰衣足食。
关注官方文档与变动：虽然东方财富网站没有公开完整的 API 文档，但其前端 JS 文件（通常在 static 目录下）里藏着所有接口的定义。定期检查这些文件，能帮你第一时间发现参数变化。
合规使用：爬取数据仅供个人学习或低频研究使用。不要高频请求，不要用于商业转售，尊重数据源的服务条款。根据《计算机信息网络国际联网安全保护管理办法》，非法获取数据是违法的。技术栈的选择没有绝对的好坏，关键在于是否适合你的场景。对于中小团队或个人开发者，手写实现轻量级爬虫，比引入庞大的框架更灵活、更易维护。
你公司项目里是怎么处理这类实时金融数据源的？是用现成的商业 API，还是自己维护爬虫集群？或者你在逆向东方财富网站接口时遇到过什么更刁钻的反爬手段？欢迎在评论区聊聊，咱们一起交流避坑经验。