量化入门必备：一键下载全市场A股历史数据的脚本

想做量化或者复盘，第一步就是要有数据。 于是我自己写了一个小工具，能一次性抓取 1999年至2025年 全市场 A 股的历史日线数据，并保存成 CSV 文件。

脚本功能覆盖 沪深京三市 所有上市公司数据区间：1999-01-01 至 2025-08-30支持 前复权数据保存为 CSV 格式，按股票单独存放实现思路使用 BaoStock API 获取股票列表。循环遍历股票列表，逐一下载日线数据。每只股票生成一个 CSV 文件存储。打印进度和统计信息，方便检查成功与失败数量。总结

有了这些数据，不管是做量化回测、策略研究，还是单纯想看一只股票的历史走势，都非常方便。关键是免费、可控、不依赖第三方软件。

附源码# -*- coding: utf-8 -*-import baostock as bsimport pandas as pdimport osimport sysimport ioimport timefrom datetime import datetime# 设置标准输出编码为UTF-8sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')# 定义全局常量START_DATE = '1999-01-01'END_DATE = '2025-08-30'STOCK_LIST_DATE = '2023-06-30'# 用于获取股票列表的日期OUTPUT_DIR = 'stock_data'ADJUST_FLAG = '2'# 前复权def login_baostock():    '''    登录BaoStock系统        Returns:        bs.Baostock: 登录后的BaoStock对象    '''    lg = bs.login()        if lg.error_code != '0':        print(f'登录失败: {lg.error_msg}')        sys.exit(1)    else:        print('登录成功')        return lgdef logout_baostock(lg):    '''    登出BaoStock系统        Args:        lg (bs.Baostock): BaoStock对象    '''    bs.logout()    print('登出成功')def get_stock_list():    '''    获取所有A股股票列表        Returns:        pandas.DataFrame: 包含股票代码和股票名称的DataFrame    '''    print('正在获取股票列表...')        # 获取证券基本资料    rs = bs.query_all_stock(day=STOCK_LIST_DATE)        # 打印结果集    data_list = []    while (rs.error_code == '0') & rs.next():        # 获取一条记录，将记录合并在一起        data_list.append(rs.get_row_data())        # 转换为DataFrame    result = pd.DataFrame(data_list, columns=rs.fields)        print(f'获取到 {len(result)} 条原始股票记录')        # 过滤掉非A股股票（code以sh.6、sz.0、sz.3、bj.开头的为A股）    # 先查看有哪些类型的股票代码    print('股票代码前缀统计:')    ifnot result.empty:        print(result['code'].str[:3].value_counts())        # 过滤A股股票，确保包含北京证券交易所股票    result = result[        (result['code'].str.startswith('sh.6')) |  # 上海证券交易所A股        (result['code'].str.startswith('sz.0')) |  # 深圳证券交易所A股        (result['code'].str.startswith('sz.3')) |  # 深圳证券交易所创业板        (result['code'].str.startswith('bj.'))     # 北京证券交易所股票    ]        print(f'过滤后共获取到 {len(result)} 只A股股票')    return resultdef download_stock_data(lg, stock_code, stock_name):    '''    下载单只股票的历史日线数据        Args:        lg (bs.Baostock): 登录后的BaoStock对象        stock_code (str): 股票代码        stock_name (str): 股票名称            Returns:        pandas.DataFrame: 股票历史数据    '''    print(f'正在下载 {stock_code} {stock_name} 的历史数据...')        # 获取历史K线数据    rs = bs.query_history_k_data_plus(        stock_code,        'date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,isST',        start_date=START_DATE,        end_date=END_DATE,        frequency='d',        adjustflag=ADJUST_FLAG  # 前复权    )        # 打印结果集    data_list = []    while (rs.error_code == '0') & rs.next():        # 获取一条记录，将记录合并在一起        data_list.append(rs.get_row_data())        # 转换为DataFrame    result = pd.DataFrame(data_list, columns=rs.fields)        print(f'{stock_code} {stock_name} 数据下载完成，共 {len(result)} 条记录')    return resultdef save_to_csv(df, stock_code, stock_name):    '''    将股票数据保存为CSV文件        Args:        df (pandas.DataFrame): 股票数据        stock_code (str): 股票代码        stock_name (str): 股票名称    '''    # 清理股票代码，去掉市场前缀(sh./sz./bj.)    clean_code = stock_code.split('.')[-1]        # 创建文件名    filename = f'{clean_code}_{stock_name}.csv'    filepath = os.path.join(OUTPUT_DIR, filename)        # 保存到CSV文件    df.to_csv(filepath, index=False, encoding='utf-8-sig')    print(f'已保存至 {filepath}')def main():    '''    主函数    '''    print('=' * 50)    print('A股历史日线数据下载程序')    print('=' * 50)    print(f'数据时间范围: {START_DATE} 至 {END_DATE}')    print(f'复权类型: 前复权')    print(f'输出目录: {OUTPUT_DIR}')    print('-' * 50)        # 记录开始时间    start_time = time.time()    start_datetime = datetime.now()    print(f'开始时间: {start_datetime.strftime('%Y-%m-%d %H:%M:%S')}')        # 登录BaoStock    lg = login_baostock()        try:        # 获取股票列表        stock_list = get_stock_list()                # 显示前几只股票作为示例        print('前5只股票示例:')        print(stock_list.head())                # 创建输出目录        ifnot os.path.exists(OUTPUT_DIR):            os.makedirs(OUTPUT_DIR)            print(f'已创建输出目录: {OUTPUT_DIR}')                # 统计变量        success_count = 0        fail_count = 0        total_stocks = len(stock_list)                print(f'\n开始下载 {total_stocks} 只股票的数据...')        print('-' * 50)                # 遍历股票列表下载数据        for index, row in stock_list.iterrows():            stock_code = row['code']            stock_name = row['code_name']                        # 显示进度            progress = (index + 1) / total_stocks * 100            print(f'[{progress:.1f}%] 正在处理第 {index + 1}/{total_stocks} 只股票: {stock_code} {stock_name}')                        try:                # 下载股票数据                stock_data = download_stock_data(lg, stock_code, stock_name)                                # 检查是否有数据                ifnot stock_data.empty:                    # 保存到CSV文件                    save_to_csv(stock_data, stock_code, stock_name)                    success_count += 1                else:                    print(f'{stock_code} {stock_name} 无数据，跳过')                    fail_count += 1                                except Exception as e:                print(f'下载 {stock_code} {stock_name} 数据时发生错误: {e}')                fail_count += 1                continue                # 记录结束时间        end_time = time.time()        end_datetime = datetime.now()        elapsed_time = end_time - start_time                # 打印总结信息        print('-' * 50)        print('下载完成!')        print(f'开始时间: {start_datetime.strftime('%Y-%m-%d %H:%M:%S')}')        print(f'结束时间: {end_datetime.strftime('%Y-%m-%d %H:%M:%S')}')        print(f'总耗时: {elapsed_time:.2f} 秒')        print(f'成功下载: {success_count} 只股票')        print(f'下载失败: {fail_count} 只股票')        print(f'总计处理: {success_count + fail_count} 只股票')        print('=' * 50)            finally:        # 登出BaoStock        logout_baostock(lg)if __name__ == '__main__':    main() 本站仅提供存储服务，所有内容均由用户发布，如发现有害或侵权内容，请点击举报。