简介：一套基于 Python 的股票数据爬取实现源码，面向金融数据分析、量化策略研究及投资研究初学者，可解决从公开财经站点批量获取历史行情与实时报价的问题。方案覆盖 yfinance 接口调用与 requests+BeautifulSoup 手工解析 HTML 两条技术路线，并谈及开盘价、收盘价、最高价、最低价与成交量等关键字段，附带数据清洗、缺失值处理、时间戳转换及多线程/异步加速等扩展思路，源码结构简洁，适合对照学习与二次改造。资源共 10 个文件，包含 1 个 Python 主脚本、1 个已编译 pyc 文件以及若干 xml/iml 工程配置文件，另有 gitignore 和 DS_Store 等辅助文件，压缩包仅 7KB，体量轻量但逻辑完整。目前已有 997 人学习下载，对于想快速上手股票数据采集并能理解爬虫编码与数据整理的读者，是一份高性价比的参考实现。
1. 为什么“爬股票数据”不是写个requests.get那么简单——目标、边界与一份源码最该回答的问题
打开行情软件的导出菜单，最快也只能拿到近三年的日线，想要一分钟频次的历史K线、想要把全市场收盘快照存进本地库、想让数据跟上自己的研究节奏，手动导出就成了体力活。Python 爬取股票数据源码实现方案，讲的就是把这套体力活变成可重复运行的代码工程：定时抓取、断点续传、增量合并、双源核对。这篇笔记面向想给自己建离线行情库的人，可能是全市场标的存档，也可能是自选股五年历史；需要的数据越琐碎，越能体现一套完整源码的价值。先明确两个边界：只讨论公开行情接口的合规访问，不碰登录态与验证码对抗；实时性做到 5 分钟级足够，更细的分笔数据另选方案。
2. 数据源选型：自建爬虫、免费结构化行情库与付费快照接口的取舍
很多人一上来就问“能不能爬某网站”，这个问题的前提就错了。爬股票数据的第一步不是写代码，而是搞清楚你手里的数据源有多少种、各自的持久性如何。选错源，后面的反爬、字段解析、补数工作会成倍放大，最终把整条链路拖垮。
2.1 三种数据源的本质区别与适用人群
自建网页爬虫，指的是直接解析财经网站的页面或公开 JSON 接口。优点是想抓什么抓什么，字段最全，冷门品种也只有网页上才有；缺点是要跟着页面改版走，字段命名说变就变，请求频率稍高就可能被网关断连。这类源适合做兜底，不适合当主力。
开源结构化行情库是社区维护的封装层，把行情接口、字段格式、交易日判断都统一了，一条pip install就能装，调用后直接得到干净的表格数据。对个人投研来说，80% 的主流行情需求用它都能解决，但它的能力边界在接口限频和字段覆盖上，遇到高频写入或全市场分钟线补全时会吃力。
付费快照接口稳定性和字段完整度最高，复权因子、分钟线、盘口快照一应俱全，但成本摆在那里，个人研究者先用免费源验证策略，等确实需要再升级不迟。三类源的对比如下。
| 数据源类型 | 初始成本 | 字段完整性 | 反爬风险 | 适合人群 |
|---|---|---|---|---|
| 自建网页爬虫 | 低资金，高人力 | 取决于目标页面 | 高，需持续维护 | 冷门品种、定制需求 |
| 开源结构化行情库 | 几乎为零 | 较高，覆盖主流品种 | 中，受社区维护节奏影响 | 个人投研、原型验证 |
| 付费快照接口 | 高 | 高，含复权因子与分钟线 | 低 | 量化回测、机构生产环境 |
2.2 如何识别一个“能落地”的行情接口：频率、字段与稳定性
判断一个行情接口能不能长期用，不要看它给不给免费 token，而是看四件事。第一，文档是否明文给出每秒请求上限，没给限频说明的接口，往往是“用到一定次数就断你”，而不是“不让你用”。第二，字段是否有一致的列名规范，并在一段时间内保持不变，字段名三天两头改版的接口，脚本维护成本极高。第三，是否同时覆盖股票、指数、ETF，且能用同一套格式返回数据，只支持单一品种的接口会把你困死在索引表里。第四，是否支持按日期区间拉历史，而不是只给最近 N 条记录，后者没法做增量补全。
我自己在选型时会额外要求 JSON 响应而不是 HTML。JSON 结构再乱也能用递归解析稳住，HTML 的改版则是灾难级的。用一个接口之前，先手工请求一次，把返回结构打印出来看两个点：日期字段的格式是否统一，成交量字段在停牌日是否为 0，这两个小细节决定了你清洗数据的复杂程度。
2.3 我推荐的默认组合：结构化行情库加公开页面兜底
我给大多数个人方案的建议是一主一副双源。主源用社区常见的结构化行情库做日线和分钟线批量下载，它免费、无需 token、字段风格接近行情软件，适合快速构建初始数据集。副源留一个财经网站的公开行情接口，用于交叉校验和冷门字段补全。
两个源的字段口径往往有细微差异，比如有的源把集合竞价成交量算进开盘那一笔，有的源单独列成一条。因此主源和副源之间不是谁覆盖谁的关系，而是在差异超过阈值时报警，让脚本把有问题的交易日单独标记出来，而不是静默覆盖入库。后面第六篇章会专门给出这个校验脚本。选型阶段还有一条血泪经验：不要把一个接口用到死，在代码里把数据访问层单独隔离，换源时只改适配器，不动存储和校验逻辑，这就是“可落地的源码实现方案”与一次性脚本的最大区别。
3. 核心源码骨架：限频、重试与增量更新的最小可运行实现
选定数据源之后，真正要写的代码其实就四块：配置、下载、交易日判断、入口调度。很多人把爬股票数据想成一个大爬虫，其实生产环境里最重要的是那些不起眼的控制逻辑：多久请求一次、失败重试几次、今天该拉哪一天的数据、拉完怎么标记。这章直接给最小可运行骨架，代码量少，但是每一行都在解决实际问题。
3.1 最小仓库结构：配置、下载、存储、入口四个文件
先把目录摆出来，让没接触过工程化爬虫的人有个整体印象。
stock_data_pipeline/ ├─ config.py # 全局配置：股票池、日期范围、限频与重试参数 ├─ downloader.py # 限频、请求、重试的核心类 ├─ trading_calendar.py # 简易交易日判断，周末加节假日表 ├─ main.py # 入口：增量判断、按交易日拉取、落盘标记 └─ data/ └─ raw/ # 原始数据按日期分目录存放配置文件的完整内容如下，参数都写在注释里，稍后逐一解释。
# config.py TICKERS = ["600000", "000001", "300750"] # 示例占位，替换为实际六位代码 START_DATE = "2020-01-01" # 历史数据起始日 END_DATE = "2024-12-31" # 历史数据结束日 OUTPUT_DIR = "data/raw" # 原始数据根目录 REQUEST_INTERVAL = 0.5 # 两次请求之间的基础间隔，单位秒 RANDOM_JITTER = (0.2, 0.8) # 额外随机抖动区间，单位秒 MAX_RETRIES = 3 # 单次请求失败的最大重试次数 RETRY_BACKOFF = 2 # 重试等待基数，单位秒 TIMEOUT = 10 # 单次请求超时，单位秒 USER_AGENT = "Mozilla/5.0 (compatible;># downloader.py import time import random import logging import requests logger = logging.getLogger(__name__) class StockDownloader: def __init__(self, cfg): self.cfg = cfg self.session = requests.Session() self.session.headers.update({"User-Agent": cfg["USER_AGENT"]}) self.last_request_ts = 0.0 def _throttle(self): """在每次请求前计算需要的等待时间，并执行等待。""" interval = self.cfg["REQUEST_INTERVAL"] + random.uniform(*self.cfg["RANDOM_JITTER"]) elapsed = time.time() - self.last_request_ts if elapsed < interval: time.sleep(interval - elapsed) self.last_request_ts = time.time() def get_json(self, url, params, retries=None): """带限频、超时、重试的请求入口；失败返回 None，由调用方决定是否写入失败队列。""" retries = retries if retries is not None else self.cfg["MAX_RETRIES"] for attempt in range(retries): self._throttle() try: resp = self.session.get( url, params=params, timeout=self.cfg["TIMEOUT"] ) resp.raise_for_status() data = resp.json() if not data: raise ValueError("empty payload") return data except (requests.RequestException, ValueError) as exc: logger.warning("[%s] attempt %d failed: %s", params, attempt + 1, exc) if attempt < retries - 1: time.sleep(self.cfg["RETRY_BACKOFF"] * (attempt + 1)) return None这段代码的逻辑链路是：每次请求前先_throttle做频率控制，然后发请求并强制校验 HTTP 状态码，再检查返回内容是否为空。三个失败出口分别对应三种常见情况：网络抖动、服务端返回 4xx 或 5xx、接口返回空数据。重试用的是倍增退避，第一次失败等 2 秒，第二次等 4 秒，第三次等 8 秒后放弃，并把该请求参数记录到调用方的失败列表里，方便断点续传时补拉。
3.3 增量判断与交易日历：怎么知道“今天该拉哪些标的”
爬虫跑起来以后，最大的问题不是请求失败，而是不知道自己该拉哪一天。全年无休地请求，会把非交易日白跑一遍，还会在数据源侧留下大量无意义请求。交易日判断的逻辑不复杂，周末加节假日表就够用。
# trading_calendar.py from datetime import date, timedelta def is_trading_day(day, holidays): """判断某一天是否为交易日：周末直接排除，节假日按表排除。""" if day.weekday() >= 5: return False return day not in holidays def previous_trading_day(day, holidays): """返回给定日期之前的最近一个交易日，用于增量判断。""" day = day - timedelta(days=1) while not is_trading_day(day, holidays): day -= timedelta(days=1) return day节假日表用 Python 原生set存字符串日期即可，例如{"2024-10-01", "2024-10-02", ...}，每年年初维护一次。入口脚本里的增量逻辑如下：读取上次成功拉取的日期标记，从标记日期的下一交易日开始拉，而不是从程序启动日往回推；每拉完一个交易日，把日期写入done.list文件。这样即使脚本中途被杀，重启后也能从断点继续，不需要每次全量重跑。
# main.py 中的增量判断片段 from datetime import date from trading_calendar import previous_trading_day, is_trading_day last_done = load_done_marker() # 从 data/raw/done.list 读取字符串日期 today = date.today() target_day = previous_trading_day(today, holidays) if last_done is None: start = date.fromisoformat(START_DATE) else: start = date.fromisoformat(last_done) + timedelta(days=1) while not is_trading_day(start, holidays): start += timedelta(days=1) # 循环内调用 downloader 拉取 [start, target_day] 的行情并落盘这里的关键参数是target_day，永远取“今天之前的最近一个交易日”。因为行情数据在收盘后还要经过数据源校验，盘中或收盘后立刻拉容易拿到不完整数据。我一般把脚本放在交易日晚间 18:30 之后跑，既能确保当天数据定型，又避开了晚间数据更新的高峰期。
4. 数据存储与历史补全：CSV、SQLite 与复权因子怎么搭才不回测翻车
很多人在爬完第一批数据后，随手存成 Excel 就开始看行情，等到要做回测时才发现数据格式、缺失值、复权口径一片混乱。存储方案的取舍，决定了你后面能不能做增量更新，以及回测结果是否可信。
4.1 CSV 与 SQLite 的差异：读写、增量、并发与类型
对个人级数据量，CSV 和 SQLite 是两种最靠谱的存储介质。它们的差异不在性能，而在使用方式。
| 维度 | CSV | SQLite |
|---|---|---|
| 文件形式 | 每标的一个文件，可直接用文本工具查看 | 单文件数据库，内部多表 |
| 增量更新 | 追加行方便，但修改历史行要重写文件 | 支持按主键INSERT OR REPLACE |
| 类型约束 | 无，日期容易存成字符串 | 有类型亲和性，可约束日期、数值字段 |
| 并发写入 | 多个进程同时写会互相覆盖 | 支持简单锁，同一时刻一个写者 |
| 后续查询 | 需自己写合并逻辑 | 按日期、代码、字段作条件查询方便 |
我的建议是：原始行情落地用 CSV，按日期分目录管理；加工后的宽表建 SQLite 索引表，供回测和统计查询使用。这样做的好处是原始数据永远可回溯，加工出错时能随时重跑，不会把脏数据污染到源文件里。
4.2 目录与表结构设计：按日期聚合还是按标的聚合
存储结构决定了补数和排查的难度。我常用的目录组织方式是按日期聚合，也就是data/raw/2024-05-20/600000.csv，每个交易日一个子目录，目录里是该日全市场标的的快照。这样增量写入的语义非常清晰：某一天的文件夹存在，就说明这一天已经拉取完成；不存在则是缺失。
另有一种按标的聚合的组织方式，也就是data/raw/600000.csv，每个文件里按日期追加行。这种方式适合只关注少数几只股票的轻量场景，但对全市场回测来说，增量更新时需要在每个文件尾部追加重启标记，一旦文件损坏，恢复成本很高。两者可以结合：日线原文件按日期分目录，研究和回测时再用脚本把它合并为按标的的宽表。
4.3 历史数据补全策略：从近到远分批拉、断点续传
历史数据补全最忌讳一次性从 2010 年拉到今天，中途断网或触限频就得从头再来。正确的做法是分段拉取，每拉完一个周期落一个标记文件。伪代码如下。
# 历史补全分段逻辑 batch_days = 20 cursor = END_DATE while cursor >= START_DATE: batch_start = cursor - timedelta(days=batch_days) for day in iter_trading_days(batch_start, cursor): download_day(day) mark_done(day) cursor = batch_start - timedelta(days=1)这个方案的第一个关键是mark_done，它写在每个交易日下完数据之后，而不是整个批次之后。第二个关键是复权口径。历史行情里最坑的是除权除息日的前后价格不连续，比如一只股票分红 10 元，股价从 30 元跳到 20 元，不做复权处理的收益率曲线会莫名出现大坑。我统一采用“不复权原始价存储，复权因子单独存一张表”的方案。这样回测时可以根据策略需要自由切换前复权、后复权或原始价，而不必重新下载数据。
提示：别把数据源的“前复权价”直接当成原始价入库，否则日后想调整复权基准日时没有任何后悔药，只能全部重拉。
5. 爬取股票数据常见问题排查：请求被断、字段漂移与复权缺失
再稳的数据源也有抽风的时候，再稳的脚本也会遇到奇奇怪怪的脏数据积累。这章是排错顺序，每一条都是踩坑现场，按“现象、原因、解决”三步讲清楚。
5.1 问题一：请求频率稍高就被断连
现象是脚本跑了几百次请求后突然连续失败，报 403 或连接被重置，换 IP 后恢复，过一会又被断。
原因是请求频率已经超过数据源侧的单 IP 阈值，服务端不是封你，而是临时限流。解决方法是严格执行限频参数，并给RANDOM_JITTER足够的浮动空间。把REQUEST_INTERVAL从 0.3 改到 0.8，抖动区间保持 0.3 到 1.0，往往就能从小时级断连降到几天一次。还有一条细节：对同一个 Session 对象复用 TCP 连接，比每次新建 Session 更不容易触发风控，因为连接频率对服务端可见，新建连接本身就是一种请求特征。
5.2 问题二：字段顺序漂移导致数据错位
现象是某个月份的数据里，收盘价和开盘价对不上，成交量偶尔变成负数，但文件结构看起来正常。
原因是数据源在某个时间点调整了返回字段顺序，而你的解析代码仍然按位置取值，把成交量读成了成交额，或者把复权因子读成了收盘价。这类问题只会在跨越版本更新的时间点时出现，排查难度极高。解决方法是解析时永远按列名取值，不按索引位置取值，并在入库前做字段级别的校验：开盘价、最高价、最低价三者必须满足high >= max(open, close)且low <= min(open, close)，不满足的行直接标记为异常并跳过入库。
5.3 问题三：拿到的是复权价还是原始价
现象是某几只股票在分红除息日前后，K 线突然出现巨大跳空，回测年化收益率莫名其妙高出十几个点。
原因是数据源默认返回前复权价，而你在算收益率时又做了一次复权处理，两个复权因子叠加导致数据失真。解决方法是入库前先验证价格连续性：拉取同一标的分红除息日前后 20 个交易日的数据，计算单日涨跌幅，如果某日跳空幅度超过 15% 且无对应重大公告，基本可以判定是复权口径混用。统一策略就是第 4 章说的，存原始价加复权因子，回测时再用脚本统一应用。
5.4 问题四：增量缺失与“多空日”混淆
现象是明明每天都跑了定时任务，补数据时发现中间缺了两天，翻日志发现那两天脚本正常执行了。
原因是那两天虽然不是周末，但属于调休导致的非交易日，或者数据源当天数据发布延迟，凌晨前没更新完毕，脚本拉了个空响应却当作成功。解决方法是交易日判断增加节假日表，并在空响应处理上做严格判断：数据源返回 0 行时要区分“非交易日无数据”和“交易日数据未发布”，后者需要延迟重试而不是直接跳过。我习惯在下载器里加一个规则，对目标交易日请求结果为空时，延迟一小时再补最多三次，仍失败则写入missing.list，次日启动时优先补拉。
5.5 问题五：本地文件被其他程序占用导致写入失败
现象是脚本偶发报PermissionError，或者某个 CSV 文件写了一半被截断。
原因是定时任务和手工分析同时打开了同一个文件，Windows 下尤其明显，编辑器锁住了文件句柄。解决方法是写入先写临时文件，落盘后原子重命名；读取时也要容忍文件不存在的瞬间。这个处理对后续做增量更新特别重要，能避免半文件状态污染整个数据集。
6. 定时增量更新与双源交叉校验：把爬虫从跑通变成可信
源码跑通只是开始，真正让数据可信的是两件事：每日定时增量更新，以及独立数据源的交叉校验。
6.1 每日定时任务：避开交易高峰与快速失败
定时任务的时间放在交易日晚间 18:30 以后，而不是收盘瞬间。原因是数据源对当日数据有一个校正窗口，过早拉取容易拿到未定型的快照，过晚又容易撞上其他批量任务。脚本内部加上快速失败机制：单个标的连续重试 3 次失败后，立刻把它写入失败队列，不阻塞其他标的的更新流程，第二天启动时优先补拉失败队列。
6.2 双源交叉校验：同日 OHLC 误差阈值与覆盖度比对
校验脚本的逻辑很简单：对同一交易日同一标的，比较主副源的四个价格字段，误差阈值设为 0.2%，超过则标记异常。这个阈值要足够小，能发现字段错位和数值漂移；又要足够大，能容忍两个数据源在四舍五入规则上的合理差异。
# 双源交叉校验主流程示例 def cross_validate(day, ticker, primary_df, backup_df): merged = primary_df.merge( backup_df, on="date", suffixes=("_p", "_b"), how="outer", ) # 返回两个源都存在的交易日 merged = merged.dropna(subset=["close_p", "close_b"]) ratio = (merged["close_p"] - merged["close_b"]).abs() / merged["close_b"] bad_days = merged.loc[ratio > 0.002, "date"].tolist() if bad_days: raise ValueError(f"{ticker} 在 {bad_days} 出现主副源收盘价偏差超过 0.2%")上线这套校验之后，我自己的习惯是每周检查一次missing.list和abnormal.list，而不是等回测结果异样才回头查数据。数据采集这种体力活，最怕的不是慢，而是“看起来正常”的脏数据积累。先把数据可信度做高，再谈策略和模型，这是一条反复被验证的顺序。希望帮到你。
本文还有配套的精品资源，点击获取
