很多量化投资者都有这样的经历：

每次回测一个策略，都要等十几分钟甚至几十分钟下载数据，而且如果网络不好，还会下载失败。

这不仅浪费时间，还会打断你的开发思路。

其实，QMT 支持全市场历史数据批量下载，你可以一次性把所有需要的数据下载到本地，以后回测时直接读取本地数据，速度可以提升 10 倍以上。

而且本地数据还可以进行预处理和清洗，进一步提高回测效率。

为什么要本地存储历史数据？

1. 回测速度快：本地读取数据的速度比网络下载快得多，尤其是分钟线和 Tick 级数据，差异非常明显。
2. 数据一致性：每次回测使用相同的本地数据，避免了网络数据差异导致的回测结果不一致。
3. 离线使用：没有网络的时候也可以进行回测和策略开发。
4. 数据预处理：可以提前对数据进行清洗、去重、补全，提高回测的准确性。
5. 自定义数据：可以将自己计算的因子、指标等数据和行情数据一起存储，方便回测时使用。

QMT 支持的历史数据类型

QMT 提供了非常丰富的历史数据接口，支持以下数据类型的批量下载：

• 日线数据：包括开盘价、收盘价、最高价、最低价、成交量、成交额等，支持 2005 年至今的全市场数据。
• 分钟线数据：包括 1 分钟、5 分钟、15 分钟、30 分钟、60 分钟线，支持 2015 年至今的全市场数据。
• Tick 级数据：包括逐笔成交、逐笔委托、委托队列等，支持最近 3 年的全市场数据。
• 财务数据：包括每股收益、净资产、净利润等，支持 2005 年至今的全市场数据。
• 指数数据：包括沪深 300、中证 500 等所有指数的行情数据。

实战：批量下载全市场日线数据

以下是一个完整的 Python 脚本，可以批量下载全市场所有 A 股的日线数据，并存储到本地 CSV 文件中。

# -*- coding: gbk -*-
import os
import pandas as pd
from xtquant import xtdata

# 数据存储目录
DATA_DIR = "D:/QMT_Data/daily"
if not os.path.exists(DATA_DIR):
    os.makedirs(DATA_DIR)

def download_all_stock_daily(start_date="20100101", end_date="20260525"):
    """
    下载全市场所有A股的日线数据
    :param start_date: 开始日期，格式YYYYMMDD
    :param end_date: 结束日期，格式YYYYMMDD
    """
    # 获取全市场A股代码列表
    stock_list = xtdata.get_stock_list_in_sector("沪深A股")
    print(f"共找到{len(stock_list)}只A股，开始下载数据...")
    
    for i, stock in enumerate(stock_list):
        try:
            # 下载数据
            df = xtdata.get_market_data_ex(
                field_list=["open", "high", "low", "close", "volume", "amount"],
                stock_list=[stock],
                start_time=start_date,
                end_time=end_date,
                period="1d",
                dividend_type="front",  # 前复权
                fill_data=True  # 填充停牌数据
            )
            
            # 转换为DataFrame
            df = df[stock].T
            df.index.name = "date"
            
            # 保存到CSV文件
            file_path = os.path.join(DATA_DIR, f"{stock}.csv")
            df.to_csv(file_path, encoding="utf-8-sig")
            
            print(f"[{i+1}/{len(stock_list)}] 下载完成：{stock}，共{len(df)}条数据")
            
        except Exception as e:
            print(f"[{i+1}/{len(stock_list)}] 下载失败：{stock}，错误：{e}")
    
    print("所有数据下载完成！")

def update_daily_data(end_date="20260525"):
    """
    增量更新日线数据
    :param end_date: 更新到的日期
    """
    stock_list = xtdata.get_stock_list_in_sector("沪深A股")
    print(f"共找到{len(stock_list)}只A股，开始更新数据...")
    
    for i, stock in enumerate(stock_list):
        try:
            file_path = os.path.join(DATA_DIR, f"{stock}.csv")
            
            # 如果文件不存在，下载全部数据
            if not os.path.exists(file_path):
                download_all_stock_daily(start_date="20100101", end_date=end_date)
                continue
            
            # 读取本地数据，获取最后日期
            df_local = pd.read_csv(file_path, index_col="date", parse_dates=True)
            last_date = df_local.index.max().strftime("%Y%m%d")
            
            # 如果已经是最新数据，跳过
            if last_date >= end_date:
                print(f"[{i+1}/{len(stock_list)}] 数据已是最新：{stock}，最后日期{last_date}")
                continue
            
            # 下载增量数据
            df_new = xtdata.get_market_data_ex(
                field_list=["open", "high", "low", "close", "volume", "amount"],
                stock_list=[stock],
                start_time=last_date,
                end_time=end_date,
                period="1d",
                dividend_type="front",
                fill_data=True
            )
            
            df_new = df_new[stock].T
            df_new.index.name = "date"
            
            # 合并数据
            df_combined = pd.concat([df_local, df_new])
            df_combined = df_combined[~df_combined.index.duplicated(keep="last")]
            df_combined.sort_index(inplace=True)
            
            # 保存到文件
            df_combined.to_csv(file_path, encoding="utf-8-sig")
            
            print(f"[{i+1}/{len(stock_list)}] 更新完成：{stock}，新增{len(df_new)}条数据")
            
        except Exception as e:
            print(f"[{i+1}/{len(stock_list)}] 更新失败：{stock}，错误：{e}")
    
    print("所有数据更新完成！")

if __name__ == "__main__":
    # 首次下载全部数据
    download_all_stock_daily(start_date="20100101", end_date="20260525")
    
    # 以后每天运行这个函数增量更新
    # update_daily_data(end_date="20260525")

如何在回测中使用本地数据

下载完本地数据后，你可以在回测时直接读取本地 CSV 文件，而不用再从网络下载：

python

运行

# -*- coding: gbk -*-
import os
import pandas as pd

DATA_DIR = "D:/QMT_Data/daily"

def init(context):
    context.stock = "000001.SZ"
    # 读取本地数据
    file_path = os.path.join(DATA_DIR, f"{context.stock}.csv")
    context.df = pd.read_csv(file_path, index_col="date", parse_dates=True)

def handlebar(context):
    # 获取当前日期
    current_date = context.get_bar_timetag(context.barpos).strftime("%Y-%m-%d")
    # 获取当前日期的数据
    row = context.df.loc[current_date]
    # 你的策略逻辑
    # ...

这样，你的回测速度会大大提升，而且再也不用担心网络问题了。

如果你想学习更多数据处理的技巧，或者需要批量下载分钟线、Tick 数据的脚本，欢迎点我头像私信。我会免费为你开通 QMT 量化权限，提供一对一的数据处理指导，以及完整的数据下载和存储方案。

风险提示：本地数据可能存在延迟或错误，回测结果不代表未来收益。量化交易存在市场风险、策略失效风险等。本内容仅为投资者教育目的，不构成任何投资建议。

免责声明：本栏目刊载的信息力求准确可靠，但对信息的准确性或完整性不作任何保证，亦不对因使用该等信息而引发的损失承担责任。