企业微信

tushare批量下载的实现途径

tushare的Pro接口提供了pro_bar和query等函数，但实际下载大量数据时，需要组合使用pro.query或ts.pro_api，配合循环和分页处理。标准做法是基于交易日历遍历，每次获取一个交易日的全部股票数据。


import tushare as ts

pro = ts.pro_api('你的token')

# 获取交易日历

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

# 过滤交易日

calendar = cal[cal['is_open']==1]['cal_date'].tolist()

for date in calendar:

    df = pro.daily(trade_date=date)

    # 保存或追加

这种方法的缺点在于每次请求量小，但网络往返频繁，若数据覆盖多年，累计时间很长。升级方式是利用ts.pro_bar的freq参数和asset参数，一次获取单只股票多年的日线数据。但若需要全市场股票，需先获取股票列表，再逐一循环。

优化批量下载的策略

分批处理是核心。避免一次性获取所有数据，因为接口单次最大返回行数有限制（如6000行）。正确做法是结合时间范围分块。例如，获取全部A股日线数据，可按年份分块：

如何用tushare一次性高效下载大量数据？


for year in range(2010, 2024):

    start = f'{year}0101'

    end = f'{year}1231'

    # 获取该年份所有交易日

    ...

每块内部再按股票代码循环。更高效的方式是利用多线程或异步请求，但需注意tushare的访问频率限制（每分钟调用次数）。


import pandas as pd

import tushare as ts

from concurrent.futures import ThreadPoolExecutor

def fetch_stock(code):

    df = ts.pro_bar(ts_code=code, adj='qfq', start_date='20200101', end_date='20211231')

    return df

stock_list = pro.stock_basic(exchange='', list_status='L', fields='ts_code').ts_code.tolist()

with ThreadPoolExecutor(max_workers=5) as executor:

    results = executor.map(fetch_stock, stock_list)

并发数需控制在tushare允许的范围内，否则会被封禁。更稳妥的做法是使用time.sleep控制速率。

存储方案决定了速度和容量

下载的数据量巨大，使用CSV文件保存简单但效率低。推荐使用Parquet或HDF5格式，压缩率高，读写快。


df.to_parquet('data.parquet')

# 或

store = pd.HDFStore('data.h5')

df.to_hdf(store, 'data')

若数据量超过内存，可使用数据库，如SQLite或PostgreSQL，将数据分块写入。


import sqlite3

conn = sqlite3.connect('stock.db')

for chunk in pd.read_csv('large_file.csv', chunksize=10000):

    chunk.to_sql('daily', conn, if_exists='append', index=False)

规避接口限制的技巧

tushare积分等级决定权限和数据获取频率。低积分用户可尝试使用ts.get_hist_data等老接口，但数据量受限。更实用的是利用交易日历生成日期列表，并将日期作为参数分批请求。注意，获取全部历史数据时，一次请求一个交易日的全市场数据可能超出单次返回上限，需确认。


# 检查每次返回的行数

if len(df) >= 6000:

    # 拆分为更小日期范围

使用ts.pro_bar的freq参数可以获取分钟数据，但分钟数据量更大，务必分股票和日期。

实际案例：一次下载五年日线数据

以获取上证50成分股过去五年的日线数据为例。首先获取指定股票列表：


index_weight = pro.index_weight(index_code='000016.SH', start_date='20180101', end_date='20231231')

stocks = index_weight['con_code'].unique().tolist()

然后对每只股票，使用pro_bar获取数据，并存储到本地。


import os

import time

save_dir = 'data/'

if not os.path.exists(save_dir):

    os.makedirs(save_dir)

for code in stocks:

    df = ts.pro_bar(ts_code=code, adj='qfq', start_date='20180101', end_date='20231231', freq='D')

    if df is not None and not df.empty:

        df.to_csv(f'{save_dir}{code}.csv', index=False)

        print(f'{code} saved')

        time.sleep(0.5)  # 控制速率

若股票数量多，此过程可能持续数小时。改进措施：在循环内使用异常捕获，记录失败的股票，便于重试。


failed = []

for code in stocks:

    try:

        df = ts.pro_bar(ts_code=code, adj='qfq', start_date='20180101', end_date='20231231')

        if df is None:

            failed.append(code)

            continue

        df.to_csv(f'{save_dir}{code}.csv', index=False)

    except Exception as e:

        print(e)

        failed.append(code)

        time.sleep(1)

减少重复请求的缓存机制

下载大量数据时，重复请求不可避免。使用本地缓存可显著降低重复加载时间。例如，保存时检查文件是否存在，存在则跳过。


if not os.path.exists(f'{save_dir}{code}.csv'):

    # 请求并保存

else:

    print('已存在')

当需要更新数据时，可只增量下载最新交易日的数据。

内存管理技巧

大量数据一次性载入内存会导致内存不足。使用pandas的read_csv等函数时，指定chunksize参数分批读取；处理时使用dtype指定列类型，降低内存占用。


import pandas as pd

reader = pd.read_csv('big.csv', chunksize=10000, dtype={'ts_code':'str', 'trade_date':'str'})

for chunk in reader:

    # 处理chunk

命令行工具或脚本化

将下载脚本写成函数，支持传入开始日期、结束日期、股票列表，方便重复使用。


def download_data(ts_codes, start, end, freq):

    for ts_code in ts_codes:

        df = ts.pro_bar(ts_code=ts_code, start_date=start, end_date=end, freq=freq)

        # 存储

对于长期更新的任务，可使用cron或计划任务每天运行增量更新。

常见问题与解法

遇到网络超时或错误，增加重试逻辑。


for attempt in range(3):

    try:

        df = pro.query(...)

        break

    except Exception as e:

        if attempt == 2:

            raise

        time.sleep(2)

如果是数据缺失，检查是否因为停牌导致无数据，这种情况下返回的DataFrame为空，可以跳过。

使用tushare大规模下载数据并非难事，关键在于合理设计请求顺序、使用分块和并发、选择高效存储格式。以上方法可提升下载效率，并保证数据完整性。

记住，tushare的权限和限制需要遵守，不要过量请求，否则会被封IP。合理利用时间片，分批获取，是最稳妥的方式。