简介面向金融数据分析与爬虫开发场景这份基于Python的东方财富A股交易数据采集方案完整演示了从环境搭建、Selenium模拟浏览器请求、HTML页面解析、数据清洗到JSON持久化存储的自动化流程。文档以单独docx文件呈现压缩包共1个文件、约24KB正文围绕规避反爬、动态页面等待、数据表格定位、翻页抓取和异常处理展开适合有一定Python基础且关注真实行情数据源的开发者直接参考实践。内容不仅给出依赖安装与ChromeDriver配置说明还提供可运行的股票数据爬取核心代码覆盖股票代码搜索、详情页跳转、多页数据合并等关键环节并提醒合理控制抓取频率、遵守网站规则。目前已有216人浏览学习对需要为量化分析、金融建模或日常盯盘积累数据样本的读者是一份轻量且可快速落地的入门参考。1. 东方财富A股数据抓取为什么我放弃现成库自己写了一套Python流程做股票策略的人迟早会撞上同一个坎回测需要用本地存的行情数据平台API下载量有限制手工导出Excel又完全撑不起批量回测。很多人第一反应是去买数据但在A股这个场景里东方财富的公开行情接口其实已经能覆盖日线、分钟线和实时快照而且不需要注册、不需要Token只靠Python加两个第三方库就能把数据落成本地文件。这就是标题里“东方财富A股股票交易数据爬取与处理”真正在做的事把网页端看到的那套行情用脚本按需拉下来清洗成规整表格再做自动化存储。我最初接手一个模拟项目X时只想要十来只股票的日线后来扩展到全市场靠的也是这套思路——requests请求JSON数据Pandas做清洗Parquet加CSV做存储每天定时增量更新。适合什么人来读有Python基础、不想被数据供应商绑定、愿意自己维护一条数据管线的从业者。这篇文章没有魔法只有接口规律、字段映射和真正踩过的坑。2. 先搞清楚数据从哪来东方财富行情接口的规律与请求构造2.1 行情接口其实是一个公开JSON接口不需要Token我最早犯的错是去搜“东方财富爬虫教程”结果搜出来一堆模拟浏览器点击的方案。那些方案又慢又脆页面结构一改就得重写。实际上去掉网页外壳东方财富的行情数据走的是一个再普通不过的JSON接口curl https://push2.eastmoney.com/api/qt/stock/kline/get?secid1.600519klt101fqt1lmt120end20500101把这段URL直接丢到浏览器地址栏能返回一段JSON里面就是贵州茅台的日K线、成交量、换手率等字段。这个接口和网页端展示的数据是同一套只是省去了渲染步骤。secid参数里的“1”代表沪市“0”代表深市后面跟着股票代码klt101表示日K线102表示周K线103表示月K线fqt1表示前复权fqt2表示后复权fqt0表示不复权end20500101的意思简单粗暴——拉到这个日期为止。整个请求不需要携带Cookie也不需要额外的Header伪装默认User-Agent就能通。用requests库拿到这段JSON之后数据在data.klines字段里格式是字符串数组每个元素形如2025-01-10,1425.00,1436.50,1410.00,1428.50,31600,490000000.00,0.50,1.20,...。看到这个结构就明白了所谓“爬取”更大一部分工作量其实在解析这个逗号分隔字符串把它拆成Pandas DataFrame。这里有个值得说的细节有人喜欢用akshare这类现成爬虫库封装得确实舒服但问题在于它的数据源依赖上游接口变化接口参数一改库作者得紧急发版。自己写请求的好处是接口规律固定、URL参数可查出了问题能直接定位是哪个参数报错不隔着一层黑匣子。2.2 requests请求封装让单只股票的历史数据一次拿全有了上面的接口逻辑写一个通用的历史数据抓取函数就很简单。我平时会这么写import requests import pandas as pd import time def fetch_kline(secid: str, klt: int 101, fqt: int 1, beg: str 19900101, end: str 20500101, max_retries: int 3) - pd.DataFrame: url https://push2.eastmoney.com/api/qt/stock/kline/get params { secid: secid, klt: klt, fqt: fqt, beg: beg, end: end, lmt: 1000000, fields1: f1,f2,f3,f4,f5,f6, fields2: f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61, } for attempt in range(max_retries): try: resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() data resp.json()[data] if not data or klines not in data: return pd.DataFrame() rows [line.split(,) for line in data[klines]] df pd.DataFrame(rows, columns[ date, open, close, high, low, volume, amount, amplitude, pct_chg, change, turnover ]) for col in [open, close, high, low, volume, amount, amplitude, pct_chg, change, turnover]: df[col] pd.to_numeric(df[col], errorscoerce) df[date] pd.to_datetime(df[date]) return df except Exception as e: if attempt max_retries - 1: print(f{secid} fetch failed: {e}) return pd.DataFrame() time.sleep(2 * (attempt 1))这段代码里有三个关键设计。第一fields2只取了我们真正需要的字段f51是日期、f52是开盘价、f53是收盘价、f54是最高价、f55是最低价、f56是成交量、f57是成交额再往后的f58到f61是振幅、涨跌幅、涨跌额和换手率少了字段响应体更小拉全市场时流量能省一大截。第二lmt1000000配合beg19900101可以直接拿一只股票从上市到现在的所有日线不需要自己做分页。第三重试机制里time.sleep(2 * (attempt 1))是做指数退避第一次失败等2秒、第二次等4秒、第三次等6秒避免连续快速请求把接口打到限流。2.3 股票列表与历史K线从“单只股票”扩展到“全市场”只拉一只股票意义不大真正有用的场景是一键更新全市场。全市场股票列表同样走东方财富的另一个接口叫clist接口。它返回分页形式的股票基本信息包括代码、名称、最新价、涨跌幅、总市值等。这里有一个新手最容易掉进去的坑列表接口名称虽然叫“沪股通十大成交股”里的东西但实际参数完全不是同一个风格。def fetch_stock_list(page: int 1, page_size: int 100) - pd.DataFrame: url https://push2.eastmoney.com/api/qt/clist/get params { pn: page, pz: page_size, po: 1, np: 1, fltt: 2, invt: 2, fid: f3, fs: m:0 t:6,m:0 t:80,m:1 t:2,m:1 t:23, fields: f12,f14,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13,f14,f15,f16,f17,f18,f20,f21,f22,f23,f24,f25,f26 } resp requests.get(url, paramsparams, timeout10) data resp.json()[data] if not data or diff not in data: return pd.DataFrame() df pd.DataFrame(data[diff]) return df[[f12, f14]].rename(columns{f12: code, f14: name})这段里的fs参数指定了A股市场范围m:0 t:6是深圳主板、m:0 t:80是创业板、m:1 t:2是上海主板、m:1 t:23是科创板。fidf3表示按涨跌幅排序pn和pz控制分页。一次请求最多拉100条A股全市场五千多只股票分页拉到第60页左右就齐了。拿回代码列表之后配合上一节的fetch_kline就能写循环批量拉取。常见的做法是先拉列表、再循环拉K线中间加time.sleep(0.5)做节流避免请求太密集触发风控。这里要特别说明列表接口里的diff字段是一个字典列表字段名都以f开头映射规则是f12代码、f14名称、f2最新价、f3涨跌幅、f5成交量、f6成交额。这些字段名不像K线接口那样有规律可循最好的办法是打印一次完整响应把字段和含义对照记录到注释里防止后面混淆。3. 把数据洗成能用的格式字段映射、复权因子与本地存储选型3.1 数据清洗的四个必备动作类型转换、字段改名、去重、排序拿到原始DataFrame之后直接落盘会埋很多雷。最典型的问题有三个日期字段不是时间类型导致排序错误某些接口字段用字符串返回数字后续算收益率会报错同一个交易日的数据在增量更新中可能被重复写入。所以清洗模块我不会省通常固定走四个步骤。def clean_kline(df: pd.DataFrame) - pd.DataFrame: if df.empty: return df df df.copy() df[date] pd.to_datetime(df[date]) numeric_cols [open, close, high, low, volume, amount, amplitude, pct_chg, change, turnover] for col in numeric_cols: df[col] pd.to_numeric(df[col], errorscoerce) df df.drop_duplicates(subset[date], keeplast) df df.sort_values(date).reset_index(dropTrue) return df先给日期列做pd.to_datetime让5000开头和20250110这样的日期格式统一成Timestamp再做全列数值化errorscoerce会把无法解析的值变成NaN而不是中断程序这很重要——一条脏数据不该让整批导入失败接着按日期去重keeplast是为了增量更新时新数据能覆盖旧数据最后排序重建索引保证后续时间序列计算不会因为乱序而出现负收益率这种诡异结果。这套清洗逻辑在所有股票数据管道里通用不管数据是从哪个接口来的落盘前过一遍总是对的。3.2 资金流水与分红除权复权因子别只靠fqt参数K线接口里的fqt1表示前复权绝大多数回测场景直接用前复权价就够了。但前复权有一个隐藏缺陷它是以最新交易日为基准往回调整历史的每次都删掉复权因子历史的收盘价会随着新除权事件不断变化。如果你今天拉了数据、三个月后增量更新同一只股票之前的复权价会被重新计算导致历史数据前后不一致回测时某些信号会被“数据修订”影响。我的做法是双轨存储保留一套不复权数据fqt0再保留一套前复权数据fqt1。不复权数据是“原始事实”用于校验和复核前复权数据是“分析视图”直接喂给策略。这样即使复权因子变化原始数据也不变随时可以重算。顺带说明fqt2是后复权它是以上市首日为基准往前调整好处是历史价格固定不变、不会因新除权而改写但后复权价格在图表上和真实成交价差异很大不太适合用来做带价格阈值的策略逻辑。如果你需要精确的复权因子本身比如算持仓收益、做资金流水对账常见做法是定期抓一次“每日复权因子”数据然后自己用因子表把不复权价换算成复权价。这个方案更重但对数据一致性要求高的场景值得。3.3 存储选型Parquet和CSV怎么分工数据存哪里我见过有人把全部K线塞进一个SQLite文件数据量小时没毛病到全市场日线累积上百万行之后查询开始卡顿。也有人执着于MySQL但对个人量化团队来说运维成本偏高。我更推荐的是Parquet CSV 双写方案存储格式适用场景优点缺点Parquet按日增量写入、跑回测前全量读取列式存储压缩率高读取速度快二进制格式肉眼不可读CSV人工核对、Excel打开查看、交付给同事通用性最强任何工具都能打开文件大、读取慢、无压缩def save_kline(df: pd.DataFrame, symbol: str, base_dir: str ./data): clean_df clean_kline(df) if clean_df.empty: return pq_path f{base_dir}/parquet/{symbol}.parquet csv_path f{base_dir}/csv/{symbol}.csv if os.path.exists(pq_path): old_df pd.read_parquet(pq_path) merged pd.concat([old_df, clean_df], ignore_indexTrue) merged merged.drop_duplicates(subset[date], keeplast) merged merged.sort_values(date).reset_index(dropTrue) merged.to_parquet(pq_path, indexFalse) merged.to_csv(csv_path, indexFalse) else: clean_df.to_parquet(pq_path, indexFalse) clean_df.to_csv(csv_path, indexFalse)增量更新的核心逻辑就藏在这段里读旧文件、拼上新数据、去重、排序、覆写回去。Parquet写二进制、CSV写文本两条路径共用同一个merged保证两套文件内容一致。注意to_parquet需要安装pyarrow或fastparquetto_csv的indexFalse必须写否则日期索引会被当成一列写进文件里。这套方案在单机跑全市场五年日线单只股票一个Parquet文件通常只有几十到几百KB五千只股票也不过几百MB完全在个人电脑的承受范围内。4. 把手动拉数据变成自动化管线增量更新、定时任务与失败重试4.1 增量更新的核心判断拿“本地上限日期”和“接口最新日期”比较全量更新在首次部署时会做一次之后的日常维护几乎都必须走增量——每天收盘后拉一次当天数据和本地已有数据合并。增量逻辑其实不复杂先读本地Parquet文件里最大的日期然后把这个日期作为beg参数去请求接口拉回来的数据天然包含本地最后一天的数据合并且去重后自然把旧数据覆盖或跳过不会追加重复行。def incremental_update(symbol: str, base_dir: str ./data, fqt: int 1) - bool: pq_path f{base_dir}/parquet/{symbol}.parquet if not os.path.exists(pq_path): df fetch_kline(symbol, fqtfqt) save_kline(df, symbol, base_dir) return True old_df pd.read_parquet(pq_path) last_date old_df[date].max().strftime(%Y%m%d) new_df fetch_kline(symbol, beglast_date, fqtfqt) if new_df.empty: return False save_kline(new_df, symbol, base_dir) return True这段有一个容易被忽略的细节last_date直接作为beg参数传给接口而不是“最后一天1”。原因是某些股票在最后一天可能有盘中数据被修订比如收盘竞价后的成交量修正用最后一天作为起点重新拉取能把这些修订也覆盖进去。代价只是多拉一天的数据完全可接受。如果你发现某种情况下数据更新后旧值没变新值通常就是因为beg参数写成了次日漏掉了当天的收盘修正。定时执行的话Linux环境用crontabWindows环境用计划任务。最常见的做法是收盘后跑一次因为A股15:00收盘但当天数据要留出清算时间盘中快照接口在收盘后还会更新部分字段。我一般定在每天17:30执行留两个半小时缓冲既不会太早拿到不完整的收盘数据也不会拖到深夜影响其他任务。4.2 调度脚本的骨架全市场循环、进度反馈、失败留痕批量更新全市场时最容易翻车的是跑到一半被限流或者某只股票停牌导致当天数据为空。这两种情况如果不在脚本里做隔离整个循环会直接中断前面更新的全部白费。常见的做法是在外层套一个任务清单循环每个任务独立try-except失败只记录不中断。import glob import traceback def update_all(base_dir: str ./data): tasks [] code_files glob.glob(f{base_dir}/parquet/*.parquet) for path in code_files: symbol os.path.basename(path).replace(.parquet, ) tasks.append(symbol) log_path f{base_dir}/update_log.txt with open(log_path, a, encodingutf-8) as log: for i, symbol in enumerate(tasks): try: ok incremental_update(symbol, base_dir) status OK if ok else EMPTY log.write(f{pd.Timestamp.now()} {symbol} {status}\n) except Exception: log.write(f{pd.Timestamp.now()} {symbol} ERROR\n) traceback.print_exc(filelog) if (i 1) % 100 0: print(fprogress: {i 1}/{len(tasks)})注意几个细节用glob扫Parquet目录而非调用股票列表接口来构建任务清单好处是停牌股、退市整理股也仍然在本地有文件更新时不会因为接口列表里查不到代码而报错。写日志时用追加模式而不是覆写模式这样历史日志能留底方便确认哪天开始某只股票持续失败。每100只打印一次进度这样在命令行运行时能知道脚本还在工作而不是卡死。traceback.print_exc(filelog)会把完整堆栈写进日志文件后面排查时能看到具体在哪个URL、哪个字段出的问题。4.3 自动化的边界接口变化了怎么办接口参数是硬编码在脚本里的一旦上游修改了URL或字段编号脚本就会大面积报错。这个问题没有办法从代码层面根治只能从运维上减轻。我给自己的管线留了几条退路一是维护一个“接口配置”的JSON文件把URL、字段名、市场范围都抽出来放到配置文件里接口变的时候改配置不改代码二是每天写日志时顺带记录请求URL和响应状态一旦某天日志里大量出现非200状态能立刻定位是哪一类接口出了问题三是保留一套全量重拉脚本增量更新连续失败超过3天时直接全量重跑一遍虽然是笨办法但数据的完整性和可用性优先。5. 数据抓取与存储中最常见的5个坑现象、原因与排查方法5.1 批次请求触发风控接口返回空data字段现象循环跑到第几百只股票时返回的JSON里data字段变成NoneK线DataFrame为空。原因请求频率过高服务端做了临时限制。解决在每次请求之间加固定延时或者把全市场拆成多个批次、每批之间休息30秒。我用的延时是0.3到0.8秒之间随机固定时间间隔容易被识别为脚本特征。5.2 数据单位不统一成交量到底是手还是股现象拉回来的volum字段某天是50000某天变成5000000幅度不稳定或者和同花顺软件上的成交量对不上。原因东方财富的成交量单位在不同接口、不同周期下不一样。日K线接口里通常是“手”但分钟线接口可能是“股”复权和不复权也可能影响数值。解决拿到数据后先用已知的一只股票和一个已知交易日做交叉验证确认单位后再落库。交叉验证的常见做法是把接口返回的成交量乘以100手转股之后和某个资讯平台披露的当日总量对比误差不超过1%就算正常。5.3 增量更新丢数据最后一天重复值没覆盖现象Parquet文件里最后一天的记录还是旧值明明增量更新已经跑完。原因beg参数传的是last_date 1天导致接口从次日才开始返回数据最后一天的修订数据被漏掉了。解决把beg直接设为last_date让接口从最后一天开始返回合并去重时keeplast自然用新数据覆盖旧数据。这一条我吃过两次亏一次是日线一次是分钟线教训就是别自作聪明去“跳过”最后一天。5.4 停牌和无数据股票导致程序中断现象某只股票停牌一个月增量更新返回空DataFrame另一个极端是退市整理期股票历史数据缺失。原因脚本没有处理“接口正常返回但无数据”的情况默认有数据才继续空数据直接走进了异常分支。解决增量更新函数里对new_df.empty做显式判断把“空数据”当成一种正常状态写日志而不是抛异常。同时任务清单不要只依赖股票列表接口要维护本地已有的代码列表否则退市股会从清单里消失历史数据再也更新不到。5.5 本地文件被其他进程占用导致写入失败现象Windows下每天定时任务偶尔报权限错误文件无法写入排查发现是Excel或WPS打开了同一个CSV文件Pandas想覆写时被操作系统拒绝。原因CSV方案本来是为了人工方便查看但查看时文件被锁定写入自然失败。解决to_csv前先检测文件是否可写不可写则只更新Parquet、CSV留到第二天补写或者干脆把CSV加入temporary清单只在人工核对的当天临时生成。6. 验证这套自动化数据管线是否靠谱三条自查命令与一个数据自检清单这一章不写新功能讲怎么确认前面的代码真的在可靠运转。我这里长年保留三个验证动作每次更新完跑一遍基本能把管线问题挡在策略之前。第一个动作检查文件数量与时间戳。find ./data/parquet -name *.parquet | wc -l find ./data/parquet -mtime -2 -name *.parquet | wc -l第一行输出本地总股票数第二行输出最近两天更新过的文件数。正常情况下这两个数字的比值应该在95%以上。如果最近两天更新数量明显偏少说明增量更新脚本在某一步中断了需要回头去看update_log.txt里最近一次的报错。第二个动作对任意一只股票抽查最后三天的数据。import pandas as pd df pd.read_parquet(./data/parquet/600519.parquet) print(df.tail(1).to_string()) print(df[close].pct_change().tail(3))先确认最后一条记录的日期是不是最近一个交易日再计算最近三天的收益率和行情软件上的数字做对比。误差超过0.1%基本可以断定数据有问题——要么复权方式不对要么除权除息那天没有更新到位。别小看这个简单的抽查它能把字段映射错误、单位错误这类最隐蔽的问题揪出来。第三个动作统计历史数据的“空窗期”。df pd.read_parquet(./data/parquet/000001.parquet) daily df.set_index(date)[close].resample(D).count() gaps daily[daily 0] print(gaps.tail(10))用日期的重采样检查有没有缺失交易日。A股的交易日是周一到周五剔除节假日出现空窗可能是停牌也可能就是漏数据。停牌没问题但如果所有股票在同一个日期都缺数据那一定是接口在那个时间段变了需要全量重拉。以上三个动作我用一个小脚本封装了起来每次增量更新完成后自动执行一次结果输出到日志文件。数据管线这东西最怕的不是代码写得不好而是出了问题没人知道。宁可每天多花十秒钟做自检也不要等策略跑完三个月回测之后才发现数据在某个节点悄悄断了。最后讲一个习惯我每次改动抓取脚本之前会先复制一份当前能跑通的版本放在备份目录里。接口变更这种玄学问题不是每次都能当场定位留一条能退回的旧路比自己临时改几行代码硬扛踏实得多。这套方法论帮我度过了很多次“数据突然拉不下来”的至暗时刻希望也能帮到你。本文还有配套的精品资源点击获取
