Python批量获取A股全市场历史数据：从Baostock实战到数据库管理

1. 项目缘起为什么需要批量获取股票数据在量化研究、策略回测或者仅仅是个人投资分析中我们常常面临一个基础但繁琐的问题如何高效、稳定地获取全市场股票的历史和实时数据手动从财经网站复制粘贴显然不现实而市面上的数据终端要么价格昂贵要么导出功能受限。这时Python就成为了一个绝佳的工具。它不仅能自动化数据获取流程更能将数据清洗、分析、可视化乃至策略开发整合到一个连贯的工作流中。我最初接触这个需求是为了测试一个简单的多因子选股模型。模型本身不复杂但第一步“准备股票池的全量历史数据”就卡了我好几天。尝试过各种免费接口、爬虫不是数据不全就是速度太慢或者因为网站反爬策略频繁中断。这个过程让我深刻体会到在金融数据分析领域“获取数据”这一步的稳定性和效率直接决定了后续所有工作的天花板。因此今天我想系统性地梳理一下如何利用Python构建一个可靠、高效的A股全市场股票数据获取方案。我们将不局限于单一方法而是对比几种主流途径的优劣并深入其中一个最稳定的方案手把手带你走通从环境搭建到数据落地的全过程。2. 核心方案选型免费数据源深度对比在开始写代码之前选择一个合适的数据源至关重要。网络上相关的库和接口很多但质量参差不齐。这里我结合自己的踩坑经验对几个主流且免费的方案进行对比分析帮你做出最适合的选择。2.1 常见数据接口横评为了更直观地展示差异我将它们的关键特性整理成了下表数据源/库数据范围与质量稳定性与速度使用复杂度主要适用场景潜在风险Tushare Pro非常全面涵盖股票、基金、期货、宏观经济等。历史数据质量高部分实时数据。稳定API调用有频率限制积分制。免费版每日调用次数有限。中等需要注册获取Token接口参数较多但文档清晰。专业的量化研究、策略开发需要多维度金融数据。免费额度可能不够用深度数据需付费。AkShare极其广泛数据源聚合来自东方财富、新浪、腾讯等。覆盖A股、港股、美股、宏观、行业等。依赖第三方网页稳定性随源站策略变化可能存在访问限制或格式变动。较低安装即用函数命名直观但数据格式因源而异。数据探索、教学、对数据源多样性要求高的场景。网页结构变更会导致接口失效需要维护。Baostock专注于A股数据较全日、周、月、5分钟、分钟线。免费提供。非常稳定通过官方数据服务器提供速度有保障。低需简单登录接口简洁。专注于A股历史行情数据获取是批量下载的优选。数据更新可能略有延迟通常T1非实时。Yahoo Finance (yfinance)全球市场美股数据最全。A股数据以沪深港通形式存在如0700.HK。比较稳定但有访问频率限制大量请求可能被临时阻断。极低yfinance.download()一行代码即可。主要针对美股或需要获取港股、A股通过港股通代码的研究。对A股直接支持不友好代码需转换如600036.SS。爬虫如requestspandas取决于目标网站如东方财富、新浪财经。可获取明细数据但完整性难保证。不稳定受反爬策略影响大IP封锁、验证码。速度慢错误处理复杂。高需要解析HTML/JSON维护成本高。获取特定不存在于通用接口的另类数据。法律与合规风险网站结构变动导致代码失效。2.2 为什么本次选择Baostock进行演示从上表可以看出对于“获取A股全部股票数据”这个批量、历史行情导向的需求Baostock是一个平衡性最好的选择。原因如下免费且稳定由官方机构维护数据服务器稳定无需担心突然失效或收费。数据质量可靠数据经过初步处理格式统一规范省去大量数据清洗工作。接口简单高效专门为行情数据设计几个核心函数就能满足日、周、月、分钟线数据的查询特别适合批量操作。对A股支持专注无需处理复杂的市场代码映射直接使用A股股票代码即可。因此下文将围绕Baostock展开从环境准备到批量获取全市场日线数据的完整流程。当然掌握了这个框架后你完全可以替换成Tushare或AkShare的接口逻辑是相通的。3. 环境准备与Baostock初体验工欲善其事必先利其器。稳定的Python环境是第一步。为了避免包依赖冲突强烈建议使用conda或venv创建独立的虚拟环境。3.1 创建并激活Python虚拟环境如果你使用Anaconda或Miniconda打开终端Windows CMD/PowerShell, macOS/Linux Terminal执行以下命令# 创建一个名为 stock_data 的新环境指定Python版本为3.8兼容性较好 conda create -n stock_data python3.8 # 激活该环境 conda activate stock_data如果使用原生Python的venv模块# 切换到你的项目目录 cd your_project_path # 创建虚拟环境文件夹 python -m venv venv # 激活环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate激活后命令行提示符前会出现环境名(stock_data)或(venv)表示你已进入隔离的环境。3.2 安装必备库在激活的虚拟环境中使用pip安装我们所需的库pip install baostock pandas -i https://pypi.tuna.tsinghua.edu.cn/simple这里我们主要安装两个库baostock: 本次的核心数据获取库。pandas: 数据处理和分析的基石baostock返回的数据本身就是DataFrame格式用pandas处理天经地义。安装完成后可以在Python交互环境中测试一下import baostock as bs import pandas as pd print(bs.__version__) print(pd.__version__)如果没有报错输出版本号说明安装成功。3.3 完成首次数据查询理解接口逻辑Baostock的操作遵循一个清晰的流程登录 - 查询 - 登出。让我们通过获取单只股票的历史数据来熟悉这个流程。import baostock as bs import pandas as pd # 1. 登录系统 lg bs.login() # 查看登录返回值如果成功会返回0和success print(login respond error_code:lg.error_code) print(login respond error_msg:lg.error_msg) # 2. 查询个股历史行情 # 参数code-股票代码格式如sh.600036market-证券交易所sh或sz # start_date, end_date-查询起止日期格式YYYY-MM-DD # frequency-数据类型默认为d日线w周线m月线55分钟线... rs bs.query_history_k_data_plus(sh.600036, date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST, start_date2023-01-01, end_date2023-12-31, frequencyd, adjustflag3) # 复权类型3-后复权2-前复权1-不复权 # 3. 将查询结果转换为pandas DataFrame data_list [] while (rs.error_code 0) rs.next(): # 逐条获取记录并合并到data_list data_list.append(rs.get_row_data()) result pd.DataFrame(data_list, columnsrs.fields) # 4. 打印数据前几行和基本信息 print(result.head()) print(f共获取到 {len(result)} 条数据) # 5. 登出系统 bs.logout()运行这段代码你应该能看到招商银行600036在2023年的日线数据被打印出来。这里有几个关键点需要理解adjustflag参数这是处理股票除权除息的关键。‘3’代表后复权保证股价历史可比性是做技术分析的首选。‘2’是前复权‘1’是不复权。通常建议使用后复权数据。返回字段我们查询的是一长串字段包括开盘价open、收盘价close、成交量volume、市盈率peTTM等。你可以根据需求增减。数据格式返回的result是一个Pandas DataFrame这是进行后续分析最方便的格式。注意bs.login()在程序运行期间只需调用一次。如果你在Jupyter Notebook中分单元格运行只需在开始执行一次登录最后执行一次登出即可中间可以多次查询。4. 核心实战批量获取全市场股票数据获取单只股票的数据只是热身我们的目标是全市场。这涉及到两个步骤首先获取当前所有上市股票的代码列表然后遍历这个列表批量获取每只股票的历史数据。4.1 获取全量股票代码列表Baostock提供了查询股票基本信息的接口我们可以借此获取代码列表。import baostock as bs import pandas as pd # 登录 lg bs.login() print(登录状态:, lg.error_code) # 查询当前所有上市公司的证券基本资料 rs bs.query_stock_basic() # 证券代码 证券名称 所属板块sh/sz/... 上市日期 ... # sh.600000 浦发银行 sh 1999-11-10 ... data_list [] while (rs.error_code 0) rs.next(): data_list.append(rs.get_row_data()) stock_basic_df pd.DataFrame(data_list, columnsrs.fields) # 筛选出A股主板、创业板、科创板的股票通常我们关注这些 # 可以根据code字段前缀sh.或sz.以及type字段1股票2指数3其它进行筛选 # 这里我们简单筛选出类型为‘1’的股票 a_stock_list stock_basic_df[stock_basic_df[type] 1][code].tolist() print(f获取到A股股票数量{len(a_stock_list)}) print(前10只股票代码, a_stock_list[:10]) # 登出如果后续不再查询 # bs.logout()stock_basic_df这个DataFrame包含了丰富的字段如code代码、code_name名称、ipoDate上市日期、outDate退市日期等。你可以根据outDate字段是否为None来进一步筛选当前未退市的股票使列表更精确。4.2 构建稳健的批量下载函数直接用一个for循环遍历几千只股票去查询看似简单但存在很大问题网络波动、服务器限流、个别股票数据缺失都可能导致程序中途崩溃前功尽弃。我们必须构建一个容错、可续传的下载函数。核心思路如下分批处理将股票列表分成小批次如每批50或100只降低单次请求压力也便于出错时定位。异常捕获与重试对每只股票的查询进行try-except包装记录失败案例并可设定重试机制。数据持久化每成功下载一批或一只股票立即将数据追加保存到本地文件如CSV而不是全部存在内存里。这样即使程序中断已下载的数据也不会丢失。进度记录记录已下载和失败的股票代码方便后续查漏补缺。下面是一个实现了上述思路的完整函数示例import baostock as bs import pandas as pd import time import os from tqdm import tqdm # 用于显示进度条可选安装pip install tqdm def download_all_stock_history(stock_list, start_date, end_date, save_path./stock_data, batch_size100, retry_times3): 批量下载股票历史日线数据 参数 stock_list: 股票代码列表如 [sh.600000, sz.000001] start_date: 开始日期YYYY-MM-DD end_date: 结束日期YYYY-MM-DD save_path: 数据保存目录 batch_size: 每批处理的股票数量 retry_times: 单只股票查询失败重试次数 # 创建保存目录 os.makedirs(save_path, exist_okTrue) # 定义最终保存的文件名按日期范围 final_csv_path os.path.join(save_path, fall_stocks_{start_date}_{end_date}.csv) # 记录失败和跳过的股票 failed_stocks [] skipped_stocks [] # 登录Baostock lg bs.login() if lg.error_code ! 0: print(f登录失败: {lg.error_msg}) return print(登录成功开始批量下载...) # 使用tqdm创建进度条 for i in tqdm(range(0, len(stock_list), batch_size), desc批次进度): batch stock_list[i:ibatch_size] batch_data_frames [] # 临时存储本批次所有股票的数据 for code in batch: success False for attempt in range(retry_times): try: # 查询历史行情数据 rs bs.query_history_k_data_plus(code, date,code,open,high,low,close,volume,amount,turn,pctChg, start_datestart_date, end_dateend_date, frequencyd, adjustflag3) if rs.error_code 0: # 解析数据 data_list [] while rs.next(): data_list.append(rs.get_row_data()) if data_list: # 如果有数据 df pd.DataFrame(data_list, columnsrs.fields) # 确保数值列是数字类型 for col in [open, high, low, close, volume, amount, turn, pctChg]: df[col] pd.to_numeric(df[col], errorscoerce) batch_data_frames.append(df) success True break # 成功则跳出重试循环 else: # 查询成功但无数据可能是新上市或停牌期 print(f 警告{code} 在指定日期范围内无数据。) skipped_stocks.append(code) success True # 不算失败只是跳过 break else: # 查询请求失败如网络问题 print(f 尝试 {attempt1}/{retry_times}: 查询{code}失败错误: {rs.error_msg}) time.sleep(2) # 失败后等待2秒再重试 except Exception as e: print(f 尝试 {attempt1}/{retry_times}: 查询{code}时发生异常: {e}) time.sleep(2) if not success: print(f **最终失败**: {code}) failed_stocks.append(code) # 将本批次成功的数据合并并保存到文件 if batch_data_frames: batch_combined_df pd.concat(batch_data_frames, ignore_indexTrue) # 如果是第一次写入创建文件并写表头否则追加数据不写表头 header not os.path.exists(final_csv_path) batch_combined_df.to_csv(final_csv_path, modea, indexFalse, headerheader) print(f 批次 {i//batch_size 1} 数据已追加保存。) # 批次间短暂停顿减轻服务器压力避免被封 time.sleep(1) # 登出 bs.logout() # 打印总结报告 print(\n *50) print(批量下载完成) print(f数据已保存至: {final_csv_path}) print(f成功处理股票数: {len(stock_list) - len(failed_stocks) - len(skipped_stocks)}) print(f跳过股票数无数据: {len(skipped_stocks)}) if skipped_stocks: print(f跳过的股票代码: {skipped_stocks[:20]}) # 只打印前20个 print(f失败股票数: {len(failed_stocks)}) if failed_stocks: print(f失败的股票代码: {failed_stocks}) print(*50) # 使用示例 if __name__ __main__: # 假设我们已经有了A股代码列表 a_stock_list # 这里用少量代码做演示 demo_list [sh.600000, sz.000001, sh.600036, sz.000858] download_all_stock_history(demo_list, start_date2023-01-01, end_date2023-12-31, save_path./stock_history, batch_size2)这个函数是批量下载的核心它包含了工业级代码应有的健壮性考虑。tqdm库提供的进度条能让你清晰掌握下载进程。函数会输出详细的日志告诉你哪些股票成功了、跳过了、失败了。4.3 执行全市场数据下载结合4.1和4.2的代码我们就可以启动全市场数据下载了。这是一个耗时较长的过程建议在网络稳定的环境下运行。# 1. 登录并获取全量A股代码列表 lg bs.login() rs bs.query_stock_basic() data_list [] while (rs.error_code 0) rs.next(): data_list.append(rs.get_row_data()) stock_basic_df pd.DataFrame(data_list, columnsrs.fields) # 获取当前未退市的A股股票代码 a_stock_list stock_basic_df[(stock_basic_df[type] 1) (stock_basic_df[outDate] )][code].tolist() bs.logout() # 先登出下载函数内会重新登录 print(f准备下载 {len(a_stock_list)} 只A股的历史数据。) # 2. 调用批量下载函数 # 注意首次下载建议先用小日期范围和小股票列表测试比如最近一个月、前10只股票。 # download_all_stock_history(a_stock_list[:10], 2024-04-01, 2024-04-30, ./test_data) # 正式下载例如下载2023年全年数据 # 这是一个非常耗时的操作可能需数小时甚至更久请确保网络稳定。 download_all_stock_history(a_stock_list, start_date2023-01-01, end_date2023-12-31, save_path./stock_data_2023, batch_size50) # 可以根据网络情况调整批次大小重要提示第一次运行时强烈建议先用极小的范围如a_stock_list[:5]和最近一周测试整个流程是否通畅确认无误后再进行全量下载。全市场数年数据的数据量可能达到GB级别下载时间很长。5. 数据落地后的处理、管理与分析建议当CSV文件生成后工作才刚刚开始。如何高效地管理和使用这些海量数据是下一个挑战。5.1 从巨型CSV到高效数据库将数据保存在一个巨大的CSV文件中对于简单查询尚可但对于复杂的多股票、多时间维度的分析效率极低。我强烈建议将数据导入到数据库中。SQLite是一个零配置、单文件的关系型数据库非常适合个人或小型项目。以下是将下载的CSV数据导入SQLite并建立索引以加速查询的示例import sqlite3 import pandas as pd # 1. 连接到SQLite数据库如果不存在则会创建 db_path ./stock_data.db conn sqlite3.connect(db_path) cursor conn.cursor() # 2. 读取我们刚才下载的巨型CSV文件分块读取以避免内存不足 chunk_size 100000 # 每次读取10万行 csv_file_path ./stock_data_2023/all_stocks_2023-01-01_2023-12-31.csv # 3. 分块读取并写入数据库 for chunk in pd.read_csv(csv_file_path, chunksizechunk_size, dtype{code: str}): # 将数据写入名为stock_daily的表中如果表不存在则创建 chunk.to_sql(stock_daily, conn, if_existsappend, indexFalse) # 4. 创建索引这是提升查询速度的关键。 # 我们最常用的查询模式是“查询某只股票在某段时间的数据”和“查询某一天所有股票的数据” cursor.execute(CREATE INDEX IF NOT EXISTS idx_code_date ON stock_daily (code, date);) cursor.execute(CREATE INDEX IF NOT EXISTS idx_date ON stock_daily (date);) conn.commit() print(数据导入并索引创建完成。) # 5. 示例查询获取招商银行sh.600036在2023年第一季度的数据 query SELECT date, open, high, low, close, volume FROM stock_daily WHERE code sh.600036 AND date BETWEEN 2023-01-01 AND 2023-03-31 ORDER BY date df_query pd.read_sql_query(query, conn) print(df_query.head()) # 6. 关闭连接 conn.close()建立(code, date)的联合索引后上述按股票代码和时间范围的查询速度会提升几个数量级。数据库的另一个好处是便于数据更新你可以定期运行下载脚本只下载新增日期的数据然后INSERT或REPLACE到数据库中。5.2 基础分析示例计算全市场股票年度涨跌幅有了数据库我们就可以轻松进行跨股票的分析。比如一个常见的需求是计算所有股票在2023年的年度涨跌幅基于后复权收盘价。import sqlite3 import pandas as pd import numpy as np conn sqlite3.connect(./stock_data.db) # 我们需要每只股票在年初和年末的复权收盘价 # 思路先找出每只股票在年度内第一个和最后一个交易日的收盘价 query WITH first_last AS ( SELECT code, MIN(date) as first_date, MAX(date) as last_date FROM stock_daily WHERE date BETWEEN 2023-01-01 AND 2023-12-31 GROUP BY code ) SELECT d.code, MAX(CASE WHEN d.date f.first_date THEN d.close END) as price_start, MAX(CASE WHEN d.date f.last_date THEN d.close END) as price_end FROM stock_daily d JOIN first_last f ON d.code f.code AND (d.date f.first_date OR d.date f.last_date) GROUP BY d.code HAVING price_start IS NOT NULL AND price_end IS NOT NULL df_pct pd.read_sql_query(query, conn) conn.close() # 计算涨跌幅 df_pct[price_start] pd.to_numeric(df_pct[price_start]) df_pct[price_end] pd.to_numeric(df_pct[price_end]) df_pct[annual_return] (df_pct[price_end] - df_pct[price_start]) / df_pct[price_start] * 100 # 查看涨跌幅最高的10只股票 top_gainers df_pct.nlargest(10, annual_return)[[code, annual_return]] print(2023年度涨幅前十) print(top_gainers) # 查看涨跌幅最低的10只股票 top_losers df_pct.nsmallest(10, annual_return)[[code, annual_return]] print(\n2023年度跌幅前十) print(top_losers) # 简单的统计描述 print(f\n全市场年度涨跌幅统计) print(f平均涨幅: {df_pct[annual_return].mean():.2f}%) print(f中位数涨幅: {df_pct[annual_return].median():.2f}%) print(f正收益股票占比: {(df_pct[annual_return] 0).sum() / len(df_pct) * 100:.2f}%)这个分析只是一个起点。基于这个本地数据库你可以进行更复杂的操作如计算移动平均线、布林带等技术指标进行因子分析如市盈率、市净率与收益的关系或者构建简单的动量策略进行回测。6. 避坑指南与进阶优化在实际操作中你肯定会遇到各种各样的问题。这里分享几个我踩过的坑和对应的解决方案。6.1 网络超时与服务器限流问题在批量下载时可能会遇到requests.exceptions.ConnectionError或Baostock返回“连接超时”的错误。解决方案增加重试与等待如我们在download_all_stock_history函数中所做加入retry_times和time.sleep()。批次间的等待time.sleep(1)至关重要这是对数据源的尊重也能有效避免被限制。使用代理谨慎如果IP被限制可以考虑使用可靠的代理IP池进行轮换。但这会引入复杂性和额外成本对于免费数据源优先通过降低请求频率增大批次间隔、减少批次大小来解决。分时段下载将全量下载任务拆分成多个小任务在不同时间段如凌晨执行。6.2 数据缺失与异常值处理问题某些股票在特定日期可能没有数据如停牌、新上市或者数据中存在明显的异常值如价格为0或负数。解决方案下载时识别我们的函数已经能识别并记录“无数据”的股票。入库后清洗从数据库读取数据进行分析前务必进行清洗。# 示例清洗从数据库读取的某股票数据 df pd.read_sql_query(SELECT * FROM stock_daily WHERE codesh.600036, conn) # 1. 确保日期为datetime类型并排序 df[date] pd.to_datetime(df[date]) df df.sort_values(date).reset_index(dropTrue) # 2. 处理缺失值对于停牌日通常用前复权收盘价填充或直接删除 # 删除所有关键价格open, high, low, close为0或NaN的行 df_cleaned df[(df[[open, high, low, close]] 0).all(axis1)] df_cleaned df_cleaned.dropna(subset[open, high, low, close, volume]) # 3. 检查数据连续性可选 print(f原始数据 {len(df)} 条清洗后 {len(df_cleaned)} 条。)6.3 数据更新策略问题如何让本地数据库与最新市场数据保持同步解决方案编写一个增量更新脚本。从数据库中查询已有数据的最新日期last_date。以last_date的下一个交易日作为start_date以当前日期作为end_date调用批量下载函数但只下载代码列表。将新数据APPEND到数据库的同一张表中。为了避免重复可以在插入前检查(code, date)是否已存在或者使用INSERT OR REPLACE语句。将此脚本设置为每日收盘后定时任务如使用crontab或Windows任务计划程序自动执行。6.4 性能优化从数据库查询到Pandas当数据量很大时即使有索引不恰当的查询也会很慢。一个常见的错误是先SELECT *把所有数据读到Pandas内存中再用Pandas过滤。正确做法尽量让数据库完成过滤和聚合只返回你需要的最小数据集。# 低效做法 df_all pd.read_sql(SELECT * FROM stock_daily, conn) # 读取GB级数据到内存 df_target df_all[(df_all[code]sh.600036) (df_all[date] 2023-06-01)] # 高效做法 query SELECT * FROM stock_daily WHERE codesh.600036 AND date 2023-06-01 df_target pd.read_sql_query(query, conn) # 数据库利用索引快速过滤只返回KB/MB级数据7. 探索其他数据源与工具链整合Baostock在A股历史行情数据上很出色但你的需求可能更多元。这里简要提及其他工具的整合思路。7.1 使用AkShare补充基本面或实时数据AkShare的数据源更杂可以用来补充一些Baostock没有的另类数据比如龙虎榜、融资融券、大宗交易等。import akshare as ak # 获取某只股票的实时行情新浪接口 stock_zh_a_spot_df ak.stock_zh_a_spot() print(stock_zh_a_spot_df.head()) # 获取新闻联播文本用于情绪分析 news_cctv_df ak.news_cctv(date20240520)需要注意的是AkShare的接口稳定性需要自己维护且数据格式可能变化。7.2 使用Tushare Pro获取更丰富的维度如果你有Tushare Pro的积分注册即有基础积分可以获取更全面的财务数据、宏观经济数据等。import tushare as ts pro ts.pro_api(你的token) # 获取上市公司基本信息 df pro.stock_basic(exchange, list_statusL, fieldsts_code,symbol,name,area,industry,list_date) # 获取日线行情需一定积分 df_daily pro.daily(ts_code600036.SH, start_date20230101, end_date20231231)Tushare的数据结构更接近专业金融数据库但需要处理其独特的代码格式如600036.SH。7.3 构建本地数据管道与Jupyter Notebook分析将上述所有步骤串联起来一个理想的个人量化数据平台工作流是数据获取层用Python脚本基于Baostock/Tushare定时自动下载/更新数据存入SQLite数据库。数据服务层将数据库文件放在固定位置或使用更专业的时序数据库如DolphinDB, InfluxDB。分析层在Jupyter Notebook或VSCode中使用Pandas、NumPy、Matplotlib/Plotly、TA-Lib技术指标库等进行数据分析和可视化。策略层使用backtrader、zipline或qstrader等回测框架基于本地数据测试交易策略。这个流程的核心思想是自动化、模块化、可复现。一旦数据管道搭建完成你的核心精力就可以从“找数据、洗数据”解放出来完全投入到策略研究和模型优化上。