当前位置：首页 > news >正文

【QMT 量化实战】股市 “高开低走“ 是陷阱还是机会？（下）附完整 Python 代码

news 2026/5/23 16:53:41

摘要

本文为《股市 "高开低走" 是陷阱还是机会？》系列下篇，提供基于QMT 量化交易平台的完整 "高开低走" 策略 Python 实现代码。包含 Xtquant 初始化、全局参数配置、数据获取、趋势判断、周期计算及结果输出等全流程代码，帮助量化投资者快速复现并应用该策略。

关键词： QMT；量化交易；高开低走策略；Python 量化；Xtquant

1. 前言（书接上回）

在上篇文章中，我们详细分析了股市 "高开低走" 现象背后的市场逻辑，区分了哪些是真正的陷阱，哪些是值得把握的机会。本文将提供基于 QMT 平台的完整策略实现代码，帮助大家将理论转化为可执行的量化策略。

以下代码仅供学习参考，实盘交易前请务必进行充分的回测和风险评估。

2. 完整策略代码实现

2.1 导入相关库

import pandas as pd import numpy as np from xtquant import xtconstant from xtquant.xttrader import XtQuantTrader, XtQuantTraderCallback from xtquant.xttype import StockAccount import datetime import time

2.2 Xtquant 初始化

# Xtquant初始化 class MyXtQuantTraderCallback(XtQuantTraderCallback): def on_disconnected(self): print("连接断开") def on_account_status(self, status): print(f"账户状态: {status}") if __name__ == "__main__": # 创建交易对象 xt_trader = XtQuantTrader(r"D:\迅投极速交易终端 睿智融科版\userdata_mini", session_id=123456) # 创建回调对象 callback = MyXtQuantTraderCallback() # 注册回调 xt_trader.register_callback(callback) # 启动交易线程 xt_trader.start() # 连接客户端 connect_result = xt_trader.connect() if connect_result != 0: print("连接失败") exit() # 订阅账户 acc = StockAccount("你的资金账号", "STOCK") subscribe_result = xt_trader.subscribe(acc) if subscribe_result != 0: print("订阅账户失败") exit() print("连接成功，开始运行策略")

2.3 全局参数配置

# 全局参数配置 START_DATE = "20260101" # 回测开始日期 END_DATE = "20260519" # 回测结束日期 STOCK_CODE = "000001.SZ" # 股票代码：平安银行 HIGH_OPEN_RATIO = 0.005 # 高开阈值：0.5% LOW_CLOSE_RATIO = 0.003 # 低收阈值：0.3% CYCLE_DAYS = 5 # 统计周期：5个交易日

2.4 核心功能函数

# 获取指定股票的K线数据 def get_stock_data(xt_trader, stock_code, start_date, end_date): """ 获取股票K线数据 :param xt_trader: XtQuantTrader对象 :param stock_code: 股票代码 :param start_date: 开始日期 :param end_date: 结束日期 :return: 包含日期、开盘价、最高价、最低价、收盘价的DataFrame """ # 获取日线数据 klines = xt_trader.get_history_klines( stock_code, period=xtconstant.PERIOD_DAY, start_time=start_date, end_time=end_date, count=-1, dividend_type=xtconstant.DIVIDEND_TYPE_ADJUST ) if not klines: print(f"获取{stock_code}数据失败") return None # 转换为DataFrame df = pd.DataFrame(klines) df['date'] = pd.to_datetime(df['time'], unit='ms').dt.strftime('%Y%m%d') df = df[['date', 'open', 'high', 'low', 'close']] df = df.sort_values('date').reset_index(drop=True) return df # 数据清洗 def clean_data(df): """ 数据清洗：删除缺失值、异常值 :param df: 原始数据 :return: 清洗后的数据 """ # 删除缺失值 df = df.dropna() # 计算涨跌幅 df['pre_close'] = df['close'].shift(1) df['pct_change'] = (df['close'] - df['pre_close']) / df['pre_close'] # 删除涨跌幅超过10%的异常值（A股主板涨跌幅限制） df = df[(df['pct_change'] >= -0.1) & (df['pct_change'] <= 0.1)] return df.reset_index(drop=True) # 识别高开低走形态 def classify_trend(df): """ 识别高开低走形态 :param df: 清洗后的数据 :return: 包含趋势标签的DataFrame """ # 计算高开幅度 df['high_open'] = (df['open'] - df['pre_close']) / df['pre_close'] # 计算低收幅度 df['low_close'] = (df['close'] - df['open']) / df['open'] # 标记高开低走 df['is_high_open_low_close'] = np.where( (df['high_open'] > HIGH_OPEN_RATIO) & (df['low_close'] < -LOW_CLOSE_RATIO), 1, 0 ) return df # 计算未来周期收益率 def calc_future_return(df, cycle_days): """ 计算未来N个交易日的收益率 :param df: 包含趋势标签的数据 :param cycle_days: 统计周期 :return: 包含未来收益率的DataFrame """ # 计算未来N日收盘价 df['future_close'] = df['close'].shift(-cycle_days) # 计算未来N日收益率 df['future_return'] = (df['future_close'] - df['close']) / df['close'] # 删除最后N行（无未来数据） df = df.dropna() return df # 统计分析结果 def statistic_result(df): """ 统计高开低走后的收益率分布 :param df: 包含未来收益率的数据 :return: 统计结果 """ # 筛选出高开低走的交易日 high_open_low_close_days = df[df['is_high_open_low_close'] == 1] # 统计基本信息 total_days = len(df) high_open_low_close_count = len(high_open_low_close_days) high_open_low_close_ratio = high_open_low_close_count / total_days * 100 # 统计收益率 avg_return = high_open_low_close_days['future_return'].mean() * 100 win_rate = len(high_open_low_close_days[high_open_low_close_days['future_return'] > 0]) / high_open_low_close_count * 100 max_return = high_open_low_close_days['future_return'].max() * 100 min_return = high_open_low_close_days['future_return'].min() * 100 # 输出结果 print("="*50) print(f"统计区间: {START_DATE} 至 {END_DATE}") print(f"股票代码: {STOCK_CODE}") print(f"总交易日数: {total_days}") print(f"高开低走天数: {high_open_low_close_count}") print(f"高开低走占比: {high_open_low_close_ratio:.2f}%") print("-"*50) print(f"未来{CYCLE_DAYS}个交易日平均收益率: {avg_return:.2f}%") print(f"未来{CYCLE_DAYS}个交易日胜率: {win_rate:.2f}%") print(f"未来{CYCLE_DAYS}个交易日最大收益率: {max_return:.2f}%") print(f"未来{CYCLE_DAYS}个交易日最小收益率: {min_return:.2f}%") print("="*50) return { "total_days": total_days, "high_open_low_close_count": high_open_low_close_count, "high_open_low_close_ratio": high_open_low_close_ratio, "avg_return": avg_return, "win_rate": win_rate, "max_return": max_return, "min_return": min_return }

2.5 主函数

# 主函数 if __name__ == "__main__": # 初始化XtQuantTrader（前面已完成） # 获取数据 raw_df = get_stock_data(xt_trader, STOCK_CODE, START_DATE, END_DATE) if raw_df is None: exit() # 数据清洗 clean_df = clean_data(raw_df) # 趋势分类 trend_df = classify_trend(clean_df) # 计算未来收益率 final_df = calc_future_return(trend_df, CYCLE_DAYS) # 统计结果 result = statistic_result(final_df) print("策略测试完成！") # 停止交易线程 xt_trader.stop()