做数据的人十个里有八个都动过爬东方财富网的心思。这个系列第二篇我不再抓静态页面而是直接抓行情数据接口。上一篇跑通了最基础的页面解析但那只是开胃菜页面源码里看不到的几个关键字段——最新价、涨跌幅、成交量、成交额、换手率全都藏在浏览器动态加载的 XHR 响应里。这篇文章把整个过程拆开讲适合已经把 Python 环境配好、想从“能跑”跨到“能稳定每天跑”的人。完整代码我会贴出来你用的时候记得把频率控制留好。文章不会只给脚本也会把每一个参数为什么要这样写讲明白因为接口一变动参数不对就是白跑懂原理比复制代码重要得多。1. 为什么第二篇要换一种抓法从抠HTML转向抓数据接口先说我自己的经历。最早抓东财数据的时候我也按老思路做requests.get 请求行情页面然后用正则或者页面解析库去抠表格结果折腾了两天最大的问题不是解析不了而是页面里根本没有完整的数据。打开行情页右键查看网页源码搜“最新价”搜不到但你在浏览器里明明看到价格、涨跌幅、换手率整整齐齐摆在那。这种页面就是典型的异步渲染数据是浏览器加载完页面之后再通过 XHR 请求从后端接口拿到的 JSONJavaScript 负责把 JSON 填进页面。直接抓 HTML等于只拿到了一个空壳。接口级抓取的优势很直接一次请求能返回几百条股票的完整字段JSON 结构清晰不用跟 HTML 标签、class 名较劲对目标站点来说接口请求比完整下载一个几十 KB 的页面更轻量所以抓接口反而不容易触发限流对写脚本的人来说数据结构稳定存储、清洗、增量更新都方便。我后来把项目里所有页面级抓取全部换成接口级代码维护成本低了一大截。1.1 页面源码里找不到你要的数据很多人第一次抓到东财的 HTML 后都会疑惑明明表格就在网页上为什么 Python 拿到手的 HTML 里没有原因是数据不在首次响应的 HTML 里。浏览器打开网页服务器只返回骨架和 JavaScript 脚本脚本运行后发出异步请求例如/api/qt/clist/get、/api/qt/ulist.np/get这类接口服务器返回 JSONJavaScript 再把 JSON 渲染成表格。所以你用 requests 拿到的只是没有数据的壳真正有用的数据都在接口响应里。如果非要用页面级方案就得想办法模拟浏览器执行 JavaScript比如引入自动化工具有头浏览器。但这样做的代价很高每一个页面都要启动渲染引擎内存占用大、速度慢、容易被识别。更麻烦的是页面结构和脚本逻辑一旦改版解析器大概率直接失效。相比之下接口请求就是一次 HTTP GET轻量、快、参数稳定代价是第一次需要花十几分钟用开发者工具定位接口。1.2 三步定位到真正的行情接口定位接口这件事不需要任何逆向知识会按 F12 就行。第一步用浏览器打开东财的行情页面按下 F12 打开开发者工具切到 Network网络标签。第二步刷新页面在筛选框里输入 XHR 或者 Fetch这样能过滤掉图片、CSS、JS 等无关请求。这时你会看到页面加载过程中发出的一堆接口请求逐个点开看 Response响应凡是返回 JSON 并且里面包含股票代码、价格的就是目标接口。第三步找到返回数据最多的那个请求复制它的完整 URL 和关键参数。通常行情页会有两种接口一种是分页列表接口参数里带pn、pz、fs用来获取全市场股票列表另一种是批量快照接口参数里带secids用逗号拼接多个股票代码用来按代码批量获取实时行情。第二篇的脚本就是围绕这两种接口展开的一个负责拿“有哪些股票”一个负责拿“这些股票现在的价格和成交数据”。1.3 拿到 JSON 后关键字段先对准打开接口响应你看到的 JSON 大致长这样{ data: { total: 5312, diff: [ { f12: 600519, f14: 贵州茅台, f2: 1715.28, f3: 1.2018, f5: 33106, f6: 5688082560.0, f8: 0.26 } ] } }字段名都是短字母得先做一次映射才能看懂。f12是股票代码f14是股票名称f2是最新价f3是涨跌幅百分数接口直接返回小数形式比如 1.2018 表示 1.2018%f5是成交量单位是手f6是成交额单位是元f8是换手率。f13在部分接口里会出现代表市场标识一般 0 是深圳、1 是上海后面拼secids时要用到它。第一次对接接口先手工复制一条 JSON 出来对照这些字段确认单位再开始写代码。这一步省不得很多人脚本跑通了算出的成交额差了一百倍就是没注意单位。2. 用 requests 直连行情接口三小步跑通核心抓取明确了接口剩下的就是用 Python 把它们串起来。整个脚本可以分成三步先拉全市场股票列表再按代码批量取行情快照最后把结果整理成干净的表结构。下面每一步我都会给出代码和设计理由。2.1 环境和请求头先把“浏览器”的壳穿上先交代环境Python 3.8需要安装requests和pandas存储部分用 Python 内置的sqlite3不需要额外引第三方数据库。安装就一行pip install requests pandas请求头需要认真配置。我用requests.Session()而不是直接requests.get()原因有两个Session 会自动复用底层的 TCP 连接请求几百次时能省下大量握手时间Session 可以统一设置 headers不用每次请求都写一遍。具体头信息里User-Agent要设成常见的浏览器 UAReferer设置为行情站的域名很多行情接口会校验来源Referer 不对直接返回空数据。这不算绕过什么反爬机制而是模拟正常浏览器的基本请求环境。2.2 拉全市场列表用 clist 接口翻页“部分股票信息”不等于只抓一两只而是指“不全量拉历史数据只抓当前行情快照”。所以第一步要拿到的就是全市场当前存量的股票列表。列表接口支持分页每页数量可以设到 100通过total字段判断是否翻完。关键参数如下pn页码从 1 开始pz每页条数最大可到 100再大容易被截断po1按排序字段降序返回fidf3排序字段f3是涨跌幅这里会决定列表返回顺序fltt2告诉接口返回数值类型而不是字符串类型这个参数极其重要后面会专门讲invt2返回完整字段集fields指定需要的字段减少网络传输量fs板块范围过滤例如m:0t:6表示深市主板m:0t:80表示创业板m:1t:2表示沪市主板m:1t:23表示科创板这里要按自己的需求筛选板块。如果只做沪深主板去掉创业板和科创板也行如果想覆盖常见 A 股建议四个板块都保留。fs参数像一碗面里的浇头多一块少一块都不会报错但返回的股票集合完全不一样。初次使用不建议自己拼打开 DevTools 从线上请求里直接复制最稳。import random import time import requests import pandas as pd SESSION requests.Session() 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/ }) LIST_URL https://push2.eastmoney.com/api/qt/clist/get def fetch_stock_list(page_size100): all_rows [] page 1 while True: params { pn: page, pz: page_size, po: 1, np: 1, fltt: 2, invt: 2, fid: f3, fs: m:0t:6,m:0t:80,m:1t:2,m:1t:23, fields: f2,f3,f12,f13,f14,f5,f6,f8, } resp SESSION.get(LIST_URL, paramsparams, timeout10) data resp.json().get(data) or {} diff data.get(diff) or [] all_rows.extend(diff) total data.get(total, 0) if page * page_size total: break page 1 time.sleep(random.uniform(0.2, 0.5)) return all_rows这里有个小坑total是接口返回的股票总数但不同市场的股票可能中途新增导致循环次数和实际条数不一致。我建议循环结束后set一下股票代码去重不要盲目相信total。另外每翻一页做 0.2 到 0.5 秒随机休眠是很有必要的能显著降低触发限流的概率。2.3 并发快照把 5000 只股票压缩到十几秒拿到列表之后理论上可以循环对每一只股票调详情接口但 5000 次串行请求就算每次 0.2 秒也要十几分钟太蠢了。东财提供了批量快照接口参数secids用逗号拼接多个“市场前缀 股票代码”一次最多提交几十只几百只也试过能返回但为了稳妥我习惯每批 50 只。市场前缀怎么拼看f13字段值为 1 的对应上海市场前缀1.值为 0 的对应深圳市场前缀0.。拼出来就像1.600519,0.000001,0.300059。如果你的列表接口没取f13可以用一个笨办法6 开头默认上海、0 和 3 开头默认深圳但不建议这样硬编码毕竟代码前缀规则在未来可能调整。批量请求本身用线程池示例开 8 个线程。8 这个数字是实测比较稳的档位再大会把 CPU 和网络带宽跑满也更容易被限流。每批之间强制sleep 1秒把 5000 只股票压到 20 秒以内完全可行。from concurrent.futures import ThreadPoolExecutor, as_completed SNAP_URL https://push2.eastmoney.com/api/qt/ulist.np/get def fetch_snapshot_batch(secids): params { fltt: 2, invt: 2, fields: f2,f3,f12,f13,f14,f5,f6,f8, secids: ,.join(secids), } resp SESSION.get(SNAP_URL, paramsparams, timeout10) data resp.json().get(data) return (data or {}).get(diff) or [] def fetch_all_snapshots(stock_list, batch_size50, max_workers8): secid_list [f{row[f13]}.{row[f12]} for row in stock_list] results [] batches [secid_list[i:i batch_size] for i in range(0, len(secid_list), batch_size)] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {executor.submit(fetch_snapshot_batch, b): b for b in batches} for future in as_completed(future_map): batch future_map[future] try: rows future.result() results.extend(rows) except Exception as exc: print(f批次失败: {batch[:5]} ..., 错误: {exc}) time.sleep(1) return results如果你在测试时发现并发一高就返回空先把max_workers降到 2确认是限流还是代码问题再逐步往上加。2.4 字段标准化别把涨跌幅当字符串存进去接口返回的字段是f2、f3这类短编码直接入库后根本没法看。需要写一个统一映射函数把短字段转成可读列名同时做单位处理。FIELD_MAP { f12: code, f13: market, f14: name, f2: price, f3: pct_chg, f5: volume_hand, f6: amount, f8: turnover, } def to_standard_dataframe(raw_rows): df pd.DataFrame(raw_rows) df df.rename(columnsFIELD_MAP) df[trade_date] pd.Timestamp.now().strftime(%Y-%m-%d) df[updated_at] pd.Timestamp.now().strftime(%Y-%m-%d %H:%M:%S) keep_cols [code, name, trade_date, price, pct_chg, volume_hand, amount, turnover, updated_at] for col in keep_cols: if col not in df.columns: df[col] None return df[keep_cols]单位问题在这个步骤里暴露得最明显。volume_hand是“手”1 手等于 100 股amount是“元”注意它可能是很大的浮点数。如果你后续要算量比、金额占比建议统一换算成“股”和“亿元”但我一般保留原始值只把列名写清楚避免二次计算时搞混。另外pct_chg接口返回的是带百分号的数值例如 1.2 表示 1.2%不需要除以 100看接口时留意。2.5 组合起来的完整抓取脚本把上面三块拼起来一个最基础的完整抓取脚本就成型了。def main(): stock_list fetch_stock_list() print(f列表股票数: {len(stock_list)}) snap_rows fetch_all_snapshots(stock_list) print(f快照返回数: {len(snap_rows)}) df to_standard_dataframe(snap_rows) df df.drop_duplicates(subset[code, trade_date], keeplast) print(df.head()) print(f总记录数: {len(df)}) return df if __name__ __main__: df main()这个脚本没有写入数据库只打印预览方便你先确认字段映射是否正确。我建议第一次跑的时候不要直接全市场 5000 只而是手工截取前 20 只试跑把打印结果和网页数据对一遍确认无误再全量跑。全量跑的时候加一个统计成功数量、失败数量、总耗时你会对脚本的运行状态心里有数。3. 数据落地清洗、入库、每天自动增量更新能抓到数据只是第一步真正的日常工作是怎么把每天的数据变成一张可持续查询的表。这个环节我踩过的坑比接口部分多得多尤其是“重复数据”和“脏字段”这两个问题几乎每个初学者都会遇到。3.1 清洗数据先去掉脏数据再入库接口偶尔会返回一些不完整的数据。表现是某一行price为空或者turnover为None这在停牌股、新股、刚复牌的股票里很常见。如果直接入库后面做统计时会出现一堆空值图表画到一半断掉。清洗分三步做第一步类型转换。用pd.to_numeric把价格、涨跌幅、成交量这些列统一转成数值转不过去的强制变NaN而不是让 pandas 自动推断成字符串对象。第二步处理空值。对price、amount这种核心列如果为空我倾向直接删除该行对于turnover这种辅助列保留行并把空值填充为 0 也行看你下游怎么用。删除前先看比例如果某次全市场缺失比例超过 5%大概率是接口字段名变了而不是个别股票的问题。第三步去重。用code trade_date作为去重键保留最后一条。因为同一个交易日如果你盘中重复跑旧记录会被新记录覆盖保留最后一条最合理。df[price] pd.to_numeric(df[price], errorscoerce) df[pct_chg] pd.to_numeric(df[pct_chg], errorscoerce) df df.dropna(subset[price, amount]) df df.drop_duplicates(subset[code, trade_date], keeplast)清洗这一步不是写一次就完的。接口字段值可能有“-”这种占位符可能返回空字符串每次看到异常数据都要回头检查是清洗规则没覆盖还是接口行为变了。3.2 存储选型CSV 还是 SQLite很多教程喜欢把结果存成 CSV简单直观Excel 能直接打开。但跑增量更新之后 CSV 的缺点越来越明显每天追加一行旧记录不会自动去重文件会越来越大后面读取一次全量数据要好几秒写文件时如果脚本中途崩溃可能留下半个文件。我用了一个多月之后彻底换成了 SQLite。CSV 和 SQLite 的取舍其实很简单你就是想要一个能打开看的数据CSV 够用你要让脚本每天自动追加数据、按日期查历史、计算某只股票近 30 天均值SQLite 省心得多。Python 自带sqlite3不需要安装服务数据就是一个单文件备份也方便。SQLite 的“单写多读”限制对个人项目完全不是问题因为你一天最多写一次。3.3 增量更新让同一天的数据只出现一次增量更新的核心是“唯一约束”。建表时给code和trade_date建立联合唯一索引这样同一只股票同一天的数据在物理上就不可能重复。写入时用INSERT OR IGNORE如果这条记录已经存在就直接忽略不会报错。CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, name TEXT, trade_date TEXT NOT NULL, price REAL, pct_chg REAL, volume_hand REAL, amount REAL, turnover REAL, updated_at TEXT, UNIQUE (code, trade_date) );写入用 DataFrame 的to_sql最快但它不会自动处理唯一冲突。我一般先把数据整理成 records再用executemany批量插入import sqlite3 def save_to_sqlite(df, db_pathstock.db): conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS stock_daily ( code TEXT NOT NULL, name TEXT, trade_date TEXT NOT NULL, price REAL, pct_chg REAL, volume_hand REAL, amount REAL, turnover REAL, updated_at TEXT, UNIQUE (code, trade_date) ); ) records df.to_dict(records) conn.executemany( INSERT OR IGNORE INTO stock_daily (code, name, trade_date, price, pct_chg, volume_hand, amount, turnover, updated_at) VALUES (:code, :name, :trade_date, :price, :pct_chg, :volume_hand, :amount, :turnover, :updated_at) , records) conn.commit() conn.close()如果你需要盘中多次更新当天的数据把INSERT OR IGNORE换成INSERT ... ON CONFLICT(code, trade_date) DO UPDATE SET priceexcluded.price, ...即可这样当天数据会覆盖为最新值。我个人建议不要为了拿实时分时数据每 5 分钟跑一次交易时段内频繁请求既没必要又容易触发限流收盘后跑一次就够用了。3.4 自动化交易日收盘后自动跑一遍数据每天更新手动执行脚本不是长久之计。我用了两种定时方案。Windows 下打开“任务计划程序”创建基本任务触发器设置每天 15:10 重复执行操作指向python和你脚本的绝对路径起始于脚本所在目录。Linux 下用 crontab10 15 * * 1-5 cd /path/to/project /usr/bin/python3 fetch_stock.py logs/run.log 21注意两点。一是时间设在收盘之后避免盘中数据不稳定15:10 比较合理。二是非交易日问题。周一到周五的 cron 表达式在节假日也会触发但那天接口返回的diff可能是空数组或者只有 0 条脚本只要对空数据做保护就不会写入脏数据。更好的做法是维护一份交易日历或者检查当天是否已有数据有就直接退出日志里写一句“今日已入库”。4. 实战中踩过的坑请求被限流、字段缺失、乱码这么解决下面这些问题是我实际运行中真实遇到过的每一条都对应一次“脚本白跑”或“数据错乱”的教训。4.1 请求 503这是限流信号不是网络问题有一次我把脚本并发调成 16跑了不到 200 个请求接口开始连续返回 503。一开始我以为是网络抖动重试三次还不行才发现是自己的频率太高。东财的接口不是没有限流而是普通低频率请求它不拦一旦单 IP 短时间请求量上去503 就会铺天盖地。处理方式分三层。第一层是源头控制并发不超过 8批次之间固定 sleep。第二层是退避重试遇到 503 不要立刻重试第一次等 30 秒第二次等 60 秒第三次等 120 秒还失败就放弃本批次记录日志人工介入。第三层是打散请求时间非交易日不跑交易日尽量固定在收盘后一次跑完而不是全天定时轮询。def request_with_retry(func, max_retry3): for attempt in range(max_retry): try: resp func() if resp.status_code 503: raise requests.HTTPError(503) return resp except Exception: wait 30 * (attempt 1) print(f第 {attempt 1} 次重试等待 {wait} 秒) time.sleep(wait) raise RuntimeError(请求多次失败)这个重试函数我一开始没写结果半夜定时任务失败后没有日志第二天早上才发现昨天数据缺了。加了日志和重试之后脚本失败至少会留下痕迹不至于无声无息。4.2 diff 为空先查参数再看请求头脚本跑完列表是空的diff返回空数组这是第二常见的问题。排查顺序要固定先看参数再看请求头。参数最容易错的是fs。fs里的m:0表示市场 0深圳t:6表示板块类型 6主板每个字母写错接口不会报错只会返回空数据。我的习惯是回到浏览器 DevTools找到线上真实请求把fs原样复制过来而不是凭记忆手敲。请求头方面重点检查Referer是否设置成行情站的域名很多接口会校验来源不设 Referer 直接返回空。最后检查User-Agent是否被识别成爬虫换成最新 Chrome 的 UA 通常能解决。4.3 数字变字符串多半是漏了 fltt2这个问题非常隐蔽。fltt不传或者传 1 时接口会把数字格式化成带千分位的字符串比如1,234.56甚至停牌股某些字段直接给-。如果这时候用float()强行转换会直接抛 ValueError。如果你发现 DataFrame 里price列是 object 类型返回头检查一下请求参数里有没有fltt2。我建议清洗函数里对这种情况做兜底import numpy as np def safe_float(value): if value is None: return np.nan if isinstance(value, str): value value.replace(,, ).replace(%, ) if value in (, -, --): return np.nan try: return float(value) except (TypeError, ValueError): return np.nan有了这个函数无论接口返回空字符串、带逗号的字符串还是数字类型都能统一转成可计算的浮点数。4.4 增量跑重复没有唯一的表约束用to_sql往 SQLite 写数据如果建表时没有UNIQUE (code, trade_date)同一批数据重复执行两次就会产生两条记录。这个问题的隐蔽之处在于前几次跑数据量小看不出异常跑了一个月后查某只股票的历史收盘价发现同一天出现两行或多行整个序列就废了。解决办法不复杂但要注意顺序先清理已经产生的重复数据再建立唯一索引。如果先建唯一索引再清理SQLite 会因违反唯一约束而拒绝创建索引。所以我会先执行DELETE FROM stock_daily WHERE rowid NOT IN (SELECT MIN(rowid) FROM stock_daily GROUP BY code, trade_date)再创建UNIQUE约束。4.5 抓取频率与合规边界这部分我要说清楚不是为了劝退而是希望大家把脚本控制在合理边界内。东财行情接口本身是公开的、页面加载时会调用的个人学习研究使用没问题但要注意几点不要设过高的并发去压测不要按秒级频率全市场轮询不要拿抓来的数据做大规模商业分发不要试图绕过对方已经明确拦截的频率限制。这个项目定位是“部分股票信息”一天抓一次数据量不大对服务端几乎没有压力。把频率控制写进代码不是妥协是让脚本能长期稳定跑下去的基础。4.6 日志与断点出问题时能定位脚本稳定跑起来后日志的重要性会超过代码本身。我至少会记录三类信息每批请求的成功数量和耗时、每次入库的记录数、任何异常的堆栈。不需要复杂的日志框架logging模块足以。断点续抓是我后来才加的。以前跑到一半挂了只能重新全量跑浪费时间还增加被限流的风险。后来每批次请求后把原始 JSON 落盘到backup/目录下一次启动时先扫描目录把未入库的备份补写进去。这个机制在接口偶尔超时的情况下非常实用。5. 我自己的体会和一些扩展方向整套代码跑了大半年最大的变化不是“会爬数据”了而是越来越想清楚“数据拿到之后做什么”。每天收盘后自动更新一次数据SQLite 文件大概 1MB 左右一年下来不到 300MB个人电脑完全没压力。基于这张stock_daily表后续可以扩展几个很实际的方向。第一个方向是直接用 SQL 做指标计算。比如统计某只股票近 20 个交易日的平均换手率只需要一条SELECT code, AVG(turnover) FROM stock_daily WHERE code? AND trade_date date(now, -20 day) GROUP BY code。在表结构稳定的前提下这些计算几乎不用写 PythonSQLite 顺手就完成了。第二个方向是推送通知。每天入库后筛选涨跌幅排名前列或成交额异动的股票把结果生成一个文本摘要通过聊天机器人 Webhook 推到自己手机上。这样不需要每天打开行情软件收盘后扫一眼消息就知道当天市场结构大概什么样。第三个方向是交易日历。cron 在节假日空跑并没有大问题但日志里一堆“今日暂无新增数据”会干扰判断。后来我引入了一个简单的交易日列表脚本启动时先判断今天是否交易日不是就直接退出。这个小改动让日志干净了很多也避免了很多无意义的数据库查询。我个人在实际操作中的体会是爬虫项目最不值钱的是“抓到数据那一刻的成就感”最值钱的是“连续稳定运行一个月不出问题”的工程能力。接口字段映射、频率控制、增量更新、异常重试每一件事单独看都很小但组合在一起才构成一个可用的数据管道。如果你正在做类似的事我建议先跑一周看数据和行情软件对不对得上再考虑扩展功能。接口结构谁也保证不了永远不变脚本里留好日志和容错你才不会在某个早晨被一条空数据日志搞得手忙脚乱。
