企业微信

批量导出多只股票历史交易数据的方法

在量化交易和投资研究中，获取多只股票的历史交易数据是常见需求。手动下载不仅繁琐，且难以保证数据的一致性和完整性。利用Python编程语言和tushare等金融数据接口，可以轻松实现自动化批量获取。本文将详细介绍从环境配置到数据存储的完整流程，帮助你高效建立自己的股票数据库。

1. 准备环境与数据源

需要安装Python环境，并安装必要的库：pandas、tushare。tushare是一个提供中国A股数据的开源库，其数据丰富，接口稳定。安装命令如下：


pip install pandas tushare

注册tushare账号并在个人主页获取token（API密钥），该token用于身份验证。tushare有积分限制，不同积分对应不同的数据权限，获取历史日线数据通常需要120积分以上，积分可通过完善资料或参与社区活动获得。

2. 获取股票代码列表

批量导出的前提是确定股票范围。可以从沪深交易所的全部A股、某个指数成分股或自定义股票池中选取。tushare提供了获取股票列表的接口pro.stock_basic()，可以一次性获取所有上市公司的基本信息。

如何批量导出多只股票的历史交易数据


import tushare as ts

import pandas as pd

# 设置token

ts.set_token('你的token')

pro = ts.pro_api()

# 获取全部A股列表

df_stock = pro.stock_basic(exchange='', list_status='L', fields='ts_code,symbol,name,area,industry,list_date')

print(df_stock.head())

ts_code是股票代码（如000001.SZ），list_status='L'表示上市状态为正常上市。若需要排除ST股票或仅保留主板，可后续筛选。

3. 批量获取历史日线数据

核心步骤是循环遍历股票代码，调用pro.daily()接口获取每个股票的日线数据。daily接口的常用参数包括ts_code、trade_date等，但若需获取全部历史数据，需使用pro.daily()并传入start_date和end_date。为提高效率，可先获取交易日历pro.trade_cal()，明确起止日期。


# 获取交易日历，设定起止日期

cal = pro.trade_cal(exchange='SSE', start_date='20200101', end_date='20231231')

trade_days = cal[cal['is_open']==1]['cal_date']

start = trade_days.iloc[0]  # 实际开始日期

end = trade_days.iloc[-1]   # 实际结束日期

# 定义获取数据的函数

def fetch_daily(ts_code, start, end):

    df = pro.daily(ts_code=ts_code, start_date=start, end_date=end)

    return df

# 循环获取数据并存储

all_data = {}

for ts_code in df_stock['ts_code'].head(10):  # 以10只股票为例，实际可全部

    try:

        df = fetch_daily(ts_code, start, end)

        if not df.empty:

            all_data[ts_code] = df

            print(f"{ts_code} 获取成功，行数：{len(df)}")

    except Exception as e:

        print(f"{ts_code} 获取失败: {e}")

pro.daily()返回的字段包括ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount等，需注意返回的数据按日期降序排列，建议调整顺序。

4. 数据合并与存储

获取所有股票的数据后，通常需要合并到一个DataFrame中，便于后续分析。每只股票的数据结构相同，可使用pd.concat纵向合并。为区分不同股票，需保留ts_code列。


# 合并所有数据

all_df = pd.concat(all_data.values(), ignore_index=True)

# 按股票代码和交易日期排序

all_df = all_df.sort_values(['ts_code', 'trade_date'])

# 重置索引

all_df.reset_index(drop=True, inplace=True)

数据存储的常见方式有CSV文件和数据库。若数据量不大，可保存为CSV；若数据量庞大，建议存入SQLite或MySQL。使用pandas的to_csv或to_sql方法即可。


# 保存为CSV文件

all_df.to_csv('all_stock_history.csv', index=False)

若需分股票保存，可循环每只股票存储到单独文件。

5. 异常处理与效率优化

批量获取数据时，网络请求可能失败或超时。tushare对单次调用频率有限制，需在循环中添加延迟。time.sleep(0.1)可降低请求速率，避免被封禁。


import time

for ts_code in df_stock['ts_code'].head(10):

    try:

        df = fetch_daily(ts_code, start, end)

        # ... 处理数据

    except Exception as e:

        print(f"{ts_code} 获取失败: {e}")

    time.sleep(0.1)  # 控制频率

tushare的pro.daily()接口一次最多返回6000条记录，若股票历史超过此数，需分批次获取。可通过分页参数offset和limit实现。


# 示例：分页获取

def fetch_daily_paged(ts_code, start, end):

    all_pages = []

    offset = 0

    limit = 5000  # 每页5000条，避免超出限制

    while True:

        df = pro.daily(ts_code=ts_code, start_date=start, end_date=end, limit=limit, offset=offset)

        if df.empty:

            break

        all_pages.append(df)

        offset += limit

        if len(df) < limit:

            break

    if all_pages:

        result = pd.concat(all_pages, ignore_index=True)

        return result

    else:

        return pd.DataFrame()

6. 增量更新与维护

建立历史数据库后，每日新增数据需增量更新。可记录最新日期，调用pro.daily()时指定起始日期为最新日期+1交易日，仅获取新数据。


# 假设已有数据库max_date列

max_date = '20231231'

new_start = get_next_trade_day(max_date)  # 需自行实现

new_df = pro.daily(ts_code=ts_code, start_date=new_start, end_date=trade_days.end)

确保数据完整性，定期检查缺失交易日，可使用pro.trade_cal()验证。

7. 完整代码示例

以下是一个完整的批量导出脚本，包含了上述所有步骤：


import tushare as ts

import pandas as pd

import time

# 初始化

ts.set_token('你的token')

pro = ts.pro_api()

# 获取股票列表

df_stock = pro.stock_basic(exchange='', list_status='L', fields='ts_code,name')

# 获取交易日历

cal = pro.trade_cal(exchange='SSE', start_date='20100101', end_date='20231231')

trade_days = cal[cal['is_open']==1]['cal_date']

start = '20100101'

end = '20231231'

all_data = []

for ts_code in df_stock['ts_code'].head(10):  # 实际可全部

    try:

        df = pro.daily(ts_code=ts_code, start_date=start, end_date=end)

        if not df.empty:

            df['ts_code'] = ts_code

            all_data.append(df)

            print(f"{ts_code} 获取成功，行数：{len(df)}")

    except Exception as e:

        print(f"{ts_code} 失败: {e}")

    time.sleep(0.12)

if all_data:

    final_df = pd.concat(all_data, ignore_index=True)

    final_df = final_df.sort_values(['ts_code', 'trade_date'])

    final_df.to_csv('stock_history.csv', index=False)

    print("数据已保存")

else:

    print("无数据")

8. 注意事项

  • 确保tushare账号积分足够，否则无法获取日线数据。

  • 在循环中合理使用延迟，避免请求过于频繁。

  • 数据量较大时，内存可能不足，可分批处理或使用数据库。

  • 股票退市或暂停上市时，数据可能缺失，需处理。

批量导出多只股票的历史交易数据并非难事，关键在于合理设计循环和异常处理。掌握此方法后，你可以轻松获取上千只股票数据，为策略回测和量化分析打下坚实基础。