别再手动下载了！用Python+国信QMT自动拉取股票历史Tick数据（附完整代码与避坑点）

Python+QMT全自动获取股票Tick数据实战指南

在量化交易领域，获取高质量的Tick级数据是构建有效策略的基础。传统手动下载方式不仅效率低下，还容易出错。本文将手把手教你如何用Python调用国信QMT的get_market_data_ex接口，实现股票历史Tick数据的全自动获取与处理。

1. 环境准备与QMT配置

1.1 安装必要依赖

确保你的Python环境已安装以下基础库：

pip install pandas numpy pytz

注意：QMT客户端通常自带Python环境，建议使用其内置解释器以避免兼容性问题

1.2 QMT权限配置

1. 登录QMT交易终端
2. 在"系统设置"中开启"量化交易权限"
3. 申请"历史数据下载"权限（需券商审核）

不同券商权限开放程度可能不同，建议提前联系客户经理确认

2. 核心API详解与参数设置

2.1 get_market_data_ex接口解析

国信QMT提供的关键数据获取接口：

data = C.get_market_data_ex( fields=['open','high','low','close','volume','amount'], stock_code=['600000.SH'], period='tick', start_time='20240520093000', end_time='20240520150000', count=-1, dividend_type='follow', fill_data=True, subscribe=False )

关键参数说明：

2.2 时间格式避坑指南

常见错误：直接使用datetime.now()生成的时间格式不兼容

正确做法：

from datetime import datetime def format_qmt_time(dt): return dt.strftime('%Y%m%d%H%M%S') # 获取今天9:30-11:30的数据 start = datetime.now().replace(hour=9, minute=30, second=0) end = datetime.now().replace(hour=11, minute=30, second=0) formatted_start = format_qmt_time(start) formatted_end = format_qmt_time(end)

3. 完整数据获取流程实现

3.1 基础数据获取框架

# encoding: utf-8 import pandas as pd from pytz import timezone def init(C): # 设置显示选项 pd.set_option('display.max_rows', 500) pd.set_option('display.max_columns', 10) pd.set_option('display.width', 1000) # 获取平安银行tick数据 stock_code = '000001.SZ' trade_date = '20240520' tick_data = get_history_ticks( C, stock_code=stock_code, date_str=trade_date ) # 计算逐笔成交额变化 tick_data['delta_amount'] = tick_data['amount'].diff() print(tick_data.head(10)) def get_history_ticks(C, stock_code, date_str): """获取单日全部tick数据""" start_time = f"{date_str}093000" end_time = f"{date_str}150000" data = C.get_market_data_ex( ['price', 'volume', 'amount', 'bid', 'ask'], [stock_code], period='tick', start_time=start_time, end_time=end_time, count=-1, fill_data=False ) return data[stock_code]

3.2 多日数据批量获取

def batch_get_ticks(C, stock_code, start_date, end_date): """获取多日tick数据""" date_range = pd.date_range(start_date, end_date) all_data = [] for date in date_range: if date.weekday() >= 5: # 跳过周末 continue date_str = date.strftime('%Y%m%d') try: daily_data = get_history_ticks(C, stock_code, date_str) daily_data['trade_date'] = date_str all_data.append(daily_data) except Exception as e: print(f"获取{date_str}数据失败: {str(e)}") return pd.concat(all_data) if all_data else None

4. 高级数据处理技巧

4.1 Tick数据清洗与校验

常见数据问题处理方案：

1. 异常值过滤：

   def filter_abnormal_ticks(df): # 价格异常 df = df[(df['price'] > 0) & (df['price'] < df['price'].quantile(0.999))] # 成交量异常 df = df[df['volume'] < df['volume'].quantile(0.99)] return df

2. 时间连续性检查：

   def check_time_gaps(df): df['datetime'] = pd.to_datetime(df.index) time_diff = df['datetime'].diff().dt.total_seconds() gaps = time_diff[time_diff > 5] # 超过5秒间隔 return gaps

4.2 成交快照重构

从逐笔数据重构盘口状态：

def reconstruct_orderbook(tick_df): book = { 'ask1': None, 'ask_vol1': None, 'bid1': None, 'bid_vol1': None } snapshots = [] for idx, row in tick_df.iterrows(): if not pd.isna(row['bid']): book['bid1'] = row['bid'][0][0] book['bid_vol1'] = row['bid'][0][1] if not pd.isna(row['ask']): book['ask1'] = row['ask'][0][0] book['ask_vol1'] = row['ask'][0][1] snapshot = book.copy() snapshot['price'] = row['price'] snapshot['volume'] = row['volume'] snapshot['time'] = idx snapshots.append(snapshot) return pd.DataFrame(snapshots).set_index('time')

5. 性能优化与实战建议

5.1 数据获取加速技巧

• 并行下载：对多只股票使用线程池

  from concurrent.futures import ThreadPoolExecutor def parallel_download(C, stock_list, date_str): with ThreadPoolExecutor(max_workers=5) as executor: futures = { stock: executor.submit(get_history_ticks, C, stock, date_str) for stock in stock_list } return {k: v.result() for k, v in futures.items()}

• 增量更新：只获取最新数据

  def get_incremental_ticks(C, stock_code, last_time): now = datetime.now(timezone('Asia/Shanghai')) start_time = format_qmt_time(last_time) end_time = format_qmt_time(now) return get_history_ticks(C, stock_code, start_time, end_time)

5.2 数据存储方案对比

推荐方案：

# 使用PyArrow保存Parquet tick_data.to_parquet('ticks.parquet', engine='pyarrow') # 带压缩版本 tick_data.to_parquet( 'ticks_compressed.parquet', engine='pyarrow', compression='snappy' )

在实际项目中，这套自动化方案将数据获取时间从原来的小时级缩短到分钟级，且完全避免了人工操作可能带来的错误。一个特别实用的技巧是建立数据校验机制，在每天收盘后自动验证当日数据的完整性和准确性。