QMT本地数据缓存全解析：get_market_data、get_market_data_ex、get_local_data到底该用哪个？

QMT本地数据缓存全解析：get_market_data、get_market_data_ex、get_local_data到底该用哪个？

在量化交易的世界里，数据就是一切。对于使用迅投QMT平台的开发者来说，如何高效、准确地获取市场数据是策略成功的关键。xtquant模块提供了三个核心函数：get_market_data、get_market_data_ex和get_local_data，它们都能从本地缓存中获取历史数据，但在数据结构、性能表现和适用场景上却各有千秋。本文将深入剖析这三个函数的差异，帮助你在不同场景下做出最优选择。

1. 数据获取基础：理解QMT的数据架构

在深入比较三个函数之前，我们需要先了解QMT平台的数据架构。QMT采用"先下载后使用"的模式，所有历史数据都需要先通过download_history_data或download_history_data2函数下载到本地缓存中，然后才能通过获取函数读取。

数据下载的关键参数：

xtdata.download_history_data( stock_code='600519.SH', # 股票代码 period='1d', # 数据周期：1m/5m/1d等 start_time='20230101', # 开始时间 end_time='20231231', # 结束时间 incrementally=True # 是否增量下载 )

注意：增量下载模式可以显著减少重复数据下载的时间，特别适合定期更新数据的场景。

数据下载完成后，会被存储在本地特定目录中，形成结构化的缓存文件。这三个获取函数本质上都是从这些缓存文件中读取数据，但在处理方式和返回结果上有所不同。

2. 函数深度对比：从数据结构到性能表现

2.1 get_market_data：双层字典结构

get_market_data返回的是一个双层嵌套的字典结构，这种设计在需要单独处理不同字段时特别方便。

典型使用示例：

data = xtdata.get_market_data( field_list=['open', 'high', 'low', 'close', 'volume'], stock_list=['600519.SH', '000858.SZ'], period='1d', start_time='20230101', end_time='20231231' )

返回数据结构解析：

{ '600519.SH': { 'open': DataFrame(...), 'high': DataFrame(...), 'low': DataFrame(...), 'close': DataFrame(...), 'volume': DataFrame(...) }, '000858.SZ': { 'open': DataFrame(...), 'high': DataFrame(...), ... } }

适用场景：

• 需要对不同字段进行差异化处理的策略
• 仅需提取部分字段进行分析的场景
• 字段间需要独立运算的情况

2.2 get_market_data_ex：单层字典结构

get_market_data_ex返回的是一个单层字典，将所有字段整合在一个DataFrame中，更适合整体分析。

典型使用示例：

data = xtdata.get_market_data_ex( field_list=['open', 'high', 'low', 'close', 'volume'], stock_list=['600519.SH', '000858.SZ'], period='1d', start_time='20230101', end_time='20231231' )

返回数据结构解析：

{ '600519.SH': DataFrame(columns=['open', 'high', 'low', 'close', 'volume']), '000858.SZ': DataFrame(columns=['open', 'high', 'low', 'close', 'volume']) }

性能对比表：

2.3 get_local_data：纯本地数据获取

get_local_data与前两者最大的区别在于它只返回本地缓存中的数据，不包含任何实时行情。

关键差异点：

• 不依赖实时订阅
• 数据获取速度最快
• 适合回测等不需要实时数据的场景

local_data = xtdata.get_local_data( field_list=['open', 'high', 'low', 'close'], stock_list=['600519.SH'], period='1d', start_time='20230101', end_time='20231231' )

3. 实战场景选择指南

3.1 高频交易场景

在高频交易中，每毫秒都至关重要。这时应该：

1. 使用get_market_data_ex获取数据
2. 预先订阅所需股票的实时行情
3. 设置合理的缓存策略

# 高频交易数据获取示例 def get_realtime_data(stock_list): # 订阅实时行情 xtdata.subscribe_quote(stock_list, period='tick') # 获取数据 return xtdata.get_market_data_ex( field_list=['last_price', 'volume'], stock_list=stock_list, period='tick', count=100 # 获取最近100个tick )

3.2 量化回测场景

回测对数据完整性要求高，但对实时性要求低：

1. 使用get_local_data确保数据一致性
2. 提前下载完整历史数据
3. 批量处理提高效率

# 回测数据准备最佳实践 def prepare_backtest_data(stock_list, start_date, end_date): # 确保数据已下载 xtdata.download_history_data2( stock_list=stock_list, period='1d', start_time=start_date, end_time=end_date ) # 获取本地数据 return xtdata.get_local_data( field_list=['open', 'high', 'low', 'close', 'volume'], stock_list=stock_list, period='1d', start_time=start_date, end_time=end_date )

3.3 混合使用策略

对于既需要历史数据又需要实时行情的场景，可以采用混合策略：

1. 使用get_local_data获取历史数据
2. 使用get_market_data_ex补充实时数据
3. 在内存中合并两部分数据

def get_combined_data(stock_code): # 获取历史数据 hist_data = xtdata.get_local_data( field_list=['close'], stock_list=[stock_code], period='1d', start_time='20230101', end_time='20231231' ) # 获取实时数据 xtdata.subscribe_quote([stock_code], period='1d') realtime_data = xtdata.get_market_data_ex( field_list=['close'], stock_list=[stock_code], period='1d', count=1 ) # 合并数据 combined = pd.concat([ hist_data[stock_code], realtime_data[stock_code] ]) return combined

4. 高级技巧与性能优化

4.1 数据预加载策略

合理的数据预加载可以显著提高程序响应速度：

1. 在策略初始化阶段预加载常用数据
2. 使用多线程异步加载
3. 建立数据缓存机制

from threading import Thread class DataPreloader: def __init__(self): self.cache = {} def preload(self, stock_list, period): def download_task(): data = xtdata.get_market_data_ex( field_list=['open', 'high', 'low', 'close'], stock_list=stock_list, period=period ) self.cache[(tuple(stock_list), period)] = data Thread(target=download_task).start()

4.2 内存管理技巧

大量数据操作时需要注意内存管理：

• 及时释放不再使用的数据
• 使用生成器处理大数据集
• 考虑分块处理数据

4.3 错误处理与数据校验

健壮的数据获取需要完善的错误处理：

def safe_get_data(stock_code, retry=3): for i in range(retry): try: data = xtdata.get_market_data_ex( field_list=['close'], stock_list=[stock_code], period='1d' ) if data and not data[stock_code].empty: return data except Exception as e: print(f"Attempt {i+1} failed: {str(e)}") time.sleep(1) return None

在实际项目中，根据不同的数据需求特点选择合适的获取函数，往往能让策略性能提升数倍。记住，没有最好的函数，只有最适合场景的选择。