做T好难啊，还是网格策略简单点。这个策略还需要晚上修改。
1、先做大小网格策略，大网格负责行情，小网格负责做T。
CODE
# -*- coding: gbk -*-
"""
QMT版VWAP均值回归参数扫描回测
策略逻辑：
1. 使用QMT本地1分钟行情计算每日累计VWAP。
2. 收盘价低于VWAP指定阈值时产生做多信号。
3. 收盘价高于VWAP指定阈值时产生做空信号。
4. 价格回归VWAP时平仓，14:50后强制平仓。
5. 信号在当前分钟收盘后确认，并在下一分钟开盘成交，避免未来函数。
6. 每个方向每天最多开仓一次，同一时刻最多持有一个方向。
7. 扫描全部偏离阈值，输出收益、胜率、盈亏比、回撤和夏普比率。
重要说明：
- 本程序是QMT策略中的独立参数扫描器，不使用QMT内置撮合。
- 运行前必须通过QMT“数据管理”补充对应标的的1分钟历史行情。
- 普通ETF无法直接做空，SHORT结果仅代表理论双向回测结果。
"""
import os
import warnings
from datetime import datetime
import numpy as np
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
warnings.filterwarnings('ignore')
matplotlib.rcParams['font.sans-serif'] = ['SimHei']
matplotlib.rcParams['axes.unicode_minus'] = False
# ================================================================
# 第一部分：回测配置
# ================================================================
CONFIG = {
'stock_code': '159915.SZ',
'start_date': '2021-09-01',
'end_date': '2026-09-01',
'initial_capital': 100000,
'position_ratio': 0.5,
'fee_rate': 0.00015,
'slippage': 0.001,
'min_trade_volume': 100,
'force_exit_time': '14:50:00',
'entry_start_time': '09:30:00',
'lookback_years': 5,
'output_dir': r'C:\Users\Administrator\Desktop\py\数据输出'
}
# 参数扫描范围
THRESHOLDS = []
# 0.3%至1.5%，步长0.1%
THRESHOLDS.extend([round(i / 10.0, 2) for i in range(3, 16)])
# 1.75%至3.0%，步长0.25%
THRESHOLDS.extend([round(i / 100.0, 2) for i in range(175, 325, 25)])
# 3.5%和4.0%
THRESHOLDS.extend([3.5, 4.0])
THRESHOLDS = sorted(set(THRESHOLDS))
# ================================================================
# 第二部分：通用辅助函数
# ================================================================
def safe_float(value, default_value=0.0):
"""将数据安全转换为浮点数。"""
try:
result = float(value)
if np.isfinite(result):
return result
except Exception:
pass
return default_value
def parse_datetime_index(index_values):
"""
将QMT行情DataFrame的索引转换为DatetimeIndex。
兼容：
- YYYYMMDD
- YYYYMMDDHHMMSS
- 毫秒时间戳
- pandas时间对象
"""
raw_series = pd.Series(list(index_values))
if len(raw_series) == 0:
return pd.DatetimeIndex([])
numeric_series = pd.to_numeric(raw_series, errors='coerce')
numeric_ratio = numeric_series.notnull().sum() / float(len(raw_series))
if numeric_ratio > 0.95:
valid_numeric = numeric_series.dropna()
if len(valid_numeric) > 0 and valid_numeric.abs().median() > 100000000000:
return pd.DatetimeIndex(
pd.to_datetime(numeric_series, unit='ms', errors='coerce')
)
string_series = raw_series.astype(str).str.strip()
string_series = string_series.str.replace(r'\.0$', '', regex=True)
result = pd.Series(pd.NaT, index=string_series.index)
mask_8 = string_series.str.len() == 8
mask_14 = string_series.str.len() == 14
if mask_8.any():
result.loc[mask_8] = pd.to_datetime(
string_series.loc[mask_8],
format='%Y%m%d',
errors='coerce'
)
if mask_14.any():
result.loc[mask_14] = pd.to_datetime(
string_series.loc[mask_14],
format='%Y%m%d%H%M%S',
errors='coerce'
)
other_mask = ~(mask_8 | mask_14)
if other_mask.any():
result.loc[other_mask] = pd.to_datetime(
string_series.loc[other_mask],
errors='coerce'
)
return pd.DatetimeIndex(result)
def prepare_output_directory():
"""创建输出目录；原目录不可用时退回QMT当前工作目录。"""
output_dir = CONFIG['output_dir']
try:
if not os.path.exists(output_dir):
os.makedirs(output_dir)
return output_dir
except Exception as error:
fallback_dir = os.path.join(os.getcwd(), 'vwap_backtest_output')
if not os.path.exists(fallback_dir):
os.makedirs(fallback_dir)
print('指定输出目录创建失败：{}'.format(error))
print('改用输出目录：{}'.format(fallback_dir))
return fallback_dir
# ================================================================
# 第三部分：QMT行情读取
# ================================================================
def load_qmt_minute_data(ContextInfo):
"""
使用QMT get_market_data_ex读取本地1分钟行情。
subscribe=False表示只读取已经补充到本地的数据，不发起实时订阅。
"""
stock_code = CONFIG['stock_code']
start_time = CONFIG['start_date'].replace('-', '') + '000000'
end_time = CONFIG['end_date'].replace('-', '') + '235959'
print('正在读取QMT本地1分钟行情...')
print('标的：{}'.format(stock_code))
print('请求区间：{} 至 {}'.format(
CONFIG['start_date'],
CONFIG['end_date']
))
try:
market_data = ContextInfo.get_market_data_ex(
fields=['open', 'high', 'low', 'close', 'volume', 'amount'],
stock_code=[stock_code],
period='1m',
start_time=start_time,
end_time=end_time,
count=-1,
dividend_type='none',
fill_data=True,
subscribe=False
)
except Exception as error:
print('QMT行情读取失败：{}'.format(error))
print('请先在QMT数据管理中补充该标的的1分钟历史行情。')
return None
if not isinstance(market_data, dict):
print('QMT行情接口返回值不是字典。')
return None
if stock_code not in market_data:
print('行情结果中不存在标的：{}'.format(stock_code))
print('接口返回代码：{}'.format(list(market_data.keys())))
return None
raw_df = market_data.get(stock_code)
if raw_df is None or not isinstance(raw_df, pd.DataFrame) or raw_df.empty:
print('未取得有效的1分钟行情。')
return None
df = raw_df.copy()
df['datetime'] = parse_datetime_index(df.index)
df = df.reset_index(drop=True)
required_columns = ['open', 'high', 'low', 'close', 'volume']
missing_columns = [
column for column in required_columns if column not in df.columns
]
if missing_columns:
print('行情缺少字段：{}'.format(', '.join(missing_columns)))
return None
numeric_columns = ['open', 'high', 'low', 'close', 'volume', 'amount']
for column in numeric_columns:
if column in df.columns:
df[column] = pd.to_numeric(df[column], errors='coerce')
df = df.dropna(
subset=['datetime', 'open', 'high', 'low', 'close', 'volume']
)
df = df[
(df['open'] > 0) &
(df['high'] > 0) &
(df['low'] > 0) &
(df['close'] > 0) &
(df['volume'] >= 0)
]
df = df.sort_values('datetime')
df = df.drop_duplicates(subset=['datetime'], keep='last')
df = df.reset_index(drop=True)
start_datetime = pd.to_datetime(CONFIG['start_date'])
end_datetime = pd.to_datetime(CONFIG['end_date']) + pd.Timedelta(days=1)
df = df[
(df['datetime'] >= start_datetime) &
(df['datetime'] < end_datetime)
].copy()
# 只保留A股连续竞价时间段
time_text = df['datetime'].dt.strftime('%H:%M:%S')
morning_mask = (
(time_text >= '09:30:00') &
(time_text <= '11:30:00')
)
afternoon_mask = (
(time_text >= '13:00:00') &
(time_text <= '15:00:00')
)
df = df[morning_mask | afternoon_mask].copy()
df = df.reset_index(drop=True)
if df.empty:
print('日期和交易时段过滤后没有有效行情。')
return None
actual_end = df['datetime'].max()
configured_end = pd.to_datetime(CONFIG['end_date'])
if actual_end.normalize() < configured_end.normalize():
print(
'提示：配置结束日期晚于本地最新行情，实际回测截止到 {}。'.format(
actual_end.strftime('%Y-%m-%d %H:%M:%S')
)
)
print('行情读取完成。')
print('实际区间：{} 至 {}'.format(
df['datetime'].min().strftime('%Y-%m-%d %H:%M:%S'),
df['datetime'].max().strftime('%Y-%m-%d %H:%M:%S')
))
print('分钟记录数：{:,}'.format(len(df)))
return df
# ================================================================
# 第四部分：VWAP计算
# ================================================================
def add_vwap_columns(df):
"""
计算每日累计VWAP。
使用典型价格：
typical_price = (high + low + close) / 3
使用成交量加权：
VWAP = 累计(typical_price * volume) / 累计volume
"""
result = df.copy()
result['trade_date'] = result['datetime'].dt.strftime('%Y-%m-%d')
result['trade_time'] = result['datetime'].dt.strftime('%H:%M:%S')
result['typical_price'] = (
result['high'] +
result['low'] +
result['close']
) / 3.0
result['typical_volume'] = (
result['typical_price'] * result['volume']
)
result['cum_volume'] = result.groupby(
'trade_date'
)['volume'].cumsum()
result['cum_typical_volume'] = result.groupby(
'trade_date'
)['typical_volume'].cumsum()
result['vwap'] = (
result['cum_typical_volume'] /
result['cum_volume'].replace(0, np.nan)
)
result['deviation_pct'] = (
(result['close'] / result['vwap']) - 1.0
) * 100.0
result = result.dropna(subset=['vwap', 'deviation_pct'])
result = result[result['vwap'] > 0]
result = result.reset_index(drop=True)
return result
# ================================================================
# 第五部分：单阈值事件驱动回测
# ================================================================
def calculate_trade_result(position, exit_bar, exit_reason, equity):
"""根据持仓及退出K线计算单笔净盈亏。"""
direction = position['direction']
shares = position['shares']
fee_rate = CONFIG['fee_rate']
slippage = CONFIG['slippage']
raw_exit_price = safe_float(exit_bar['open'])
if raw_exit_price <= 0:
raw_exit_price = safe_float(exit_bar['close'])
if direction == 'LONG':
exit_price = max(raw_exit_price - slippage, 0.0001)
gross_pnl = (
exit_price - position['entry_price']
) * shares
else:
exit_price = raw_exit_price + slippage
gross_pnl = (
position['entry_price'] - exit_price
) * shares
exit_fee = exit_price * shares * fee_rate
net_pnl = gross_pnl - position['entry_fee'] - exit_fee
entry_notional = position['entry_price'] * shares
if entry_notional > 0:
pnl_pct = net_pnl / entry_notional * 100.0
else:
pnl_pct = 0.0
new_equity = equity + net_pnl
trade = {
'signal_time': position['signal_time'],
'entry_time': position['entry_time'],
'exit_time': exit_bar['datetime'],
'direction': direction,
'signal_close': position['signal_close'],
'signal_vwap': position['signal_vwap'],
'entry_price': position['entry_price'],
'exit_price': exit_price,
'shares': shares,
'entry_notional': entry_notional,
'entry_fee': position['entry_fee'],
'exit_fee': exit_fee,
'gross_pnl': gross_pnl,
'net_pnl': net_pnl,
'pnl_pct': pnl_pct,
'exit_reason': exit_reason,
'threshold': position['threshold'],
'equity_after_trade': new_equity
}
return trade, new_equity
def open_position_from_signal(signal, entry_bar, equity, threshold):
"""使用下一分钟开盘价执行开仓信号。"""
direction = signal['direction']
raw_entry_price = safe_float(entry_bar['open'])
slippage = CONFIG['slippage']
if raw_entry_price <= 0:
return None
if direction == 'LONG':
entry_price = raw_entry_price + slippage
else:
entry_price = max(raw_entry_price - slippage, 0.0001)
capital_used = equity * CONFIG['position_ratio']
lot_size = int(CONFIG['min_trade_volume'])
shares = int(
capital_used / entry_price / lot_size
) * lot_size
if shares < lot_size:
return None
entry_notional = entry_price * shares
entry_fee = entry_notional * CONFIG['fee_rate']
return {
'direction': direction,
'signal_time': signal['signal_time'],
'entry_time': entry_bar['datetime'],
'signal_close': signal['signal_close'],
'signal_vwap': signal['signal_vwap'],
'entry_price': entry_price,
'entry_fee': entry_fee,
'shares': shares,
'threshold': threshold
}
def run_backtest_for_threshold(df, threshold):
"""
对单个偏离阈值执行事件驱动回测。
为避免未来函数：
- 当前分钟收盘确认信号。
- 下一分钟开盘成交。
- VWAP回归信号同样在下一分钟开盘平仓。
"""
initial_capital = float(CONFIG['initial_capital'])
equity = initial_capital
entry_start_time = CONFIG['entry_start_time']
force_exit_time = CONFIG['force_exit_time']
trades = []
daily_equity_records = []
position = None
pending_entry = None
pending_exit_reason = None
grouped_data = df.groupby('trade_date', sort=True)
for trade_date, day_df in grouped_data:
day_df = day_df.sort_values('datetime').reset_index(drop=True)
if day_df.empty:
continue
position = None
pending_entry = None
pending_exit_reason = None
used_directions = set()
for row_index in range(len(day_df)):
bar = day_df.iloc[row_index]
current_time = bar['trade_time']
# ---------- 1. 执行上一分钟产生的平仓信号 ----------
if position is not None and pending_exit_reason is not None:
trade, equity = calculate_trade_result(
position,
bar,
pending_exit_reason,
equity
)
trades.append(trade)
position = None
pending_exit_reason = None
# ---------- 2. 执行上一分钟产生的开仓信号 ----------
if (
position is None and
pending_entry is not None and
current_time < force_exit_time
):
new_position = open_position_from_signal(
pending_entry,
bar,
equity,
threshold
)
if new_position is not None:
position = new_position
used_directions.add(position['direction'])
pending_entry = None
# ---------- 3. 到达强制平仓时间 ----------
if current_time >= force_exit_time:
pending_entry = None
if position is not None:
trade, equity = calculate_trade_result(
position,
bar,
'尾盘强制平仓',
equity
)
trades.append(trade)
position = None
continue
# ---------- 4. 根据当前分钟收盘价产生平仓信号 ----------
if position is not None:
close_price = safe_float(bar['close'])
vwap_price = safe_float(bar['vwap'])
if position['direction'] == 'LONG':
if close_price >= vwap_price:
pending_exit_reason = '回归VWAP平仓'
else:
if close_price <= vwap_price:
pending_exit_reason = '回归VWAP平仓'
# ---------- 5. 空仓时根据当前分钟收盘价产生开仓信号 ----------
if (
position is None and
pending_exit_reason is None and
pending_entry is None and
current_time >= entry_start_time
):
close_price = safe_float(bar['close'])
vwap_price = safe_float(bar['vwap'])
long_trigger_price = vwap_price * (
1.0 - threshold / 100.0
)
short_trigger_price = vwap_price * (
1.0 + threshold / 100.0
)
if (
close_price <= long_trigger_price and
'LONG' not in used_directions
):
pending_entry = {
'direction': 'LONG',
'signal_time': bar['datetime'],
'signal_close': close_price,
'signal_vwap': vwap_price
}
elif (
close_price >= short_trigger_price and
'SHORT' not in used_directions
):
pending_entry = {
'direction': 'SHORT',
'signal_time': bar['datetime'],
'signal_close': close_price,
'signal_vwap': vwap_price
}
# 当天最后仍有持仓时，使用最后一根K线收盘价强制退出
if position is not None:
last_bar = day_df.iloc[-1].copy()
last_bar['open'] = last_bar['close']
trade, equity = calculate_trade_result(
position,
last_bar,
'当日最后K线强制平仓',
equity
)
trades.append(trade)
position = None
daily_equity_records.append({
'trade_date': trade_date,
'equity': equity
})
return build_backtest_statistics(
threshold,
trades,
daily_equity_records
)
# ================================================================
# 第六部分：绩效统计
# ================================================================
def build_backtest_statistics(
threshold,
trades,
daily_equity_records
):
"""根据成交记录和每日权益计算绩效指标。"""
initial_capital = float(CONFIG['initial_capital'])
if len(trades) == 0:
return {
'threshold': threshold,
'total_trades': 0,
'winning_trades': 0,
'losing_trades': 0,
'win_rate': 0.0,
'total_pnl': 0.0,
'total_return_pct': 0.0,
'avg_pnl': 0.0,
'avg_pnl_pct': 0.0,
'profit_factor': 0.0,
'max_drawdown_pct': 0.0,
'sharpe_ratio': 0.0,
'final_equity': initial_capital,
'long_trades': 0,
'short_trades': 0,
'trades': pd.DataFrame(),
'daily_equity': pd.DataFrame(daily_equity_records)
}
trades_df = pd.DataFrame(trades)
trades_df = trades_df.sort_values(
['entry_time', 'exit_time']
).reset_index(drop=True)
total_trades = len(trades_df)
winning_trades = int((trades_df['net_pnl'] > 0).sum())
losing_trades = total_trades - winning_trades
win_rate = (
winning_trades / float(total_trades) * 100.0
if total_trades > 0 else 0.0
)
total_pnl = trades_df['net_pnl'].sum()
final_equity = initial_capital + total_pnl
total_return_pct = (
final_equity / initial_capital - 1.0
) * 100.0
avg_pnl = trades_df['net_pnl'].mean()
avg_pnl_pct = trades_df['pnl_pct'].mean()
gross_profit = trades_df.loc[
trades_df['net_pnl'] > 0,
'net_pnl'
].sum()
gross_loss = abs(
trades_df.loc[
trades_df['net_pnl'] <= 0,
'net_pnl'
].sum()
)
if gross_loss > 0:
profit_factor = gross_profit / gross_loss
elif gross_profit > 0:
profit_factor = np.inf
else:
profit_factor = 0.0
long_trades = int((trades_df['direction'] == 'LONG').sum())
short_trades = int((trades_df['direction'] == 'SHORT').sum())
daily_equity_df = pd.DataFrame(daily_equity_records)
if not daily_equity_df.empty:
daily_equity_df = daily_equity_df.drop_duplicates(
subset=['trade_date'],
keep='last'
)
daily_equity_df = daily_equity_df.sort_values(
'trade_date'
).reset_index(drop=True)
initial_row = pd.DataFrame([{
'trade_date': 'INITIAL',
'equity': initial_capital
}])
equity_curve = pd.concat(
[initial_row, daily_equity_df],
ignore_index=True
)
running_high = equity_curve['equity'].cummax()
drawdown_series = (
equity_curve['equity'] / running_high - 1.0
) * 100.0
max_drawdown_pct = abs(drawdown_series.min())
daily_returns = equity_curve['equity'].pct_change().dropna()
if (
len(daily_returns) > 1 and
daily_returns.std(ddof=1) > 0
):
sharpe_ratio = (
daily_returns.mean() /
daily_returns.std(ddof=1) *
np.sqrt(252.0)
)
else:
sharpe_ratio = 0.0
daily_equity_df['daily_return'] = (
equity_curve['equity'].pct_change().iloc[1:].values
)
else:
max_drawdown_pct = 0.0
sharpe_ratio = 0.0
return {
'threshold': threshold,
'total_trades': total_trades,
'winning_trades': winning_trades,
'losing_trades': losing_trades,
'win_rate': win_rate,
'total_pnl': total_pnl,
'total_return_pct': total_return_pct,
'avg_pnl': avg_pnl,
'avg_pnl_pct': avg_pnl_pct,
'profit_factor': profit_factor,
'max_drawdown_pct': max_drawdown_pct,
'sharpe_ratio': sharpe_ratio,
'final_equity': final_equity,
'long_trades': long_trades,
'short_trades': short_trades,
'trades': trades_df,
'daily_equity': daily_equity_df
}
# ================================================================
# 第七部分：参数扫描
# ================================================================
def run_parameter_scan(df):
"""依次扫描全部VWAP偏离阈值。"""
results = []
print('')
print('=' * 80)
print('开始扫描VWAP偏离阈值')
print('阈值数量：{}'.format(len(THRESHOLDS)))
print('阈值列表：{}'.format(THRESHOLDS))
print('=' * 80)
for index, threshold in enumerate(THRESHOLDS):
print(
'正在回测阈值 {:.2f}%（{}/{}）'.format(
threshold,
index + 1,
len(THRESHOLDS)
)
)
result = run_backtest_for_threshold(df, threshold)
results.append(result)
print(
' 交易次数：{}，胜率：{:.2f}%，收益：{:.2f}%，'
'回撤：{:.2f}%'.format(
result['total_trades'],
result['win_rate'],
result['total_return_pct'],
result['max_drawdown_pct']
)
)
return results
# ================================================================
# 第八部分：报告与图表
# ================================================================
def normalize_score(series, higher_is_better=True):
"""将指标归一化至0至1区间。"""
clean_series = series.replace([np.inf, -np.inf], np.nan).fillna(0.0)
minimum = clean_series.min()
maximum = clean_series.max()
value_range = maximum - minimum
if value_range <= 0:
normalized = pd.Series(
np.ones(len(clean_series)),
index=clean_series.index
)
else:
normalized = (clean_series - minimum) / value_range
if not higher_is_better:
normalized = 1.0 - normalized
return normalized
def build_report_dataframe(results):
"""将参数扫描结果整理为汇总DataFrame。"""
report_rows = []
for result in results:
profit_factor = result['profit_factor']
if np.isinf(profit_factor):
report_profit_factor = 999.0
else:
report_profit_factor = profit_factor
report_rows.append({
'阈值(%)': result['threshold'],
'交易次数': result['total_trades'],
'多头交易': result['long_trades'],
'空头交易': result['short_trades'],
'盈利次数': result['winning_trades'],
'亏损次数': result['losing_trades'],
'胜率(%)': result['win_rate'],
'总盈亏(元)': result['total_pnl'],
'总收益(%)': result['total_return_pct'],
'期末权益(元)': result['final_equity'],
'平均每笔盈亏(元)': result['avg_pnl'],
'平均每笔收益(%)': result['avg_pnl_pct'],
'盈亏比': report_profit_factor,
'最大回撤(%)': result['max_drawdown_pct'],
'夏普比率': result['sharpe_ratio']
})
report_df = pd.DataFrame(report_rows)
numeric_columns = [
'胜率(%)',
'总盈亏(元)',
'总收益(%)',
'期末权益(元)',
'平均每笔盈亏(元)',
'平均每笔收益(%)',
'盈亏比',
'最大回撤(%)',
'夏普比率'
]
for column in numeric_columns:
report_df[column] = pd.to_numeric(
report_df[column],
errors='coerce'
)
return report_df
def select_best_result(report_df):
"""
按原策略的综合权重选择最优参数：
- 胜率：40%
- 总收益：40%
- 最大回撤：20%
"""
score_df = report_df.copy()
score_df['胜率得分'] = normalize_score(
score_df['胜率(%)'],
higher_is_better=True
)
score_df['收益得分'] = normalize_score(
score_df['总收益(%)'],
higher_is_better=True
)
score_df['回撤得分'] = normalize_score(
score_df['最大回撤(%)'],
higher_is_better=False
)
score_df['综合得分'] = (
score_df['胜率得分'] * 0.4 +
score_df['收益得分'] * 0.4 +
score_df['回撤得分'] * 0.2
)
best_index = score_df['综合得分'].idxmax()
return best_index, score_df
def save_parameter_chart(report_df, output_dir):
"""绘制并保存参数敏感性图表。"""
figure, axes = plt.subplots(2, 2, figsize=(14, 10))
figure.suptitle(
'VWAP均值回归策略参数敏感性分析',
fontsize=16
)
# 阈值与胜率
axes[0, 0].plot(
report_df['阈值(%)'],
report_df['胜率(%)'],
marker='o',
color='blue',
linewidth=2
)
axes[0, 0].set_title('阈值与胜率')
axes[0, 0].set_xlabel('偏离阈值(%)')
axes[0, 0].set_ylabel('胜率(%)')
axes[0, 0].grid(True, alpha=0.3)
# 阈值与收益
axes[0, 1].plot(
report_df['阈值(%)'],
report_df['总收益(%)'],
marker='s',
color='green',
linewidth=2
)
axes[0, 1].set_title('阈值与总收益')
axes[0, 1].set_xlabel('偏离阈值(%)')
axes[0, 1].set_ylabel('总收益(%)')
axes[0, 1].grid(True, alpha=0.3)
# 阈值与交易次数
axes[1, 0].plot(
report_df['阈值(%)'],
report_df['交易次数'],
marker='^',
color='orange',
linewidth=2
)
axes[1, 0].set_title('阈值与交易次数')
axes[1, 0].set_xlabel('偏离阈值(%)')
axes[1, 0].set_ylabel('交易次数')
axes[1, 0].grid(True, alpha=0.3)
# 阈值与最大回撤
axes[1, 1].plot(
report_df['阈值(%)'],
report_df['最大回撤(%)'],
marker='D',
color='red',
linewidth=2
)
axes[1, 1].set_title('阈值与最大回撤')
axes[1, 1].set_xlabel('偏离阈值(%)')
axes[1, 1].set_ylabel('最大回撤(%)')
axes[1, 1].grid(True, alpha=0.3)
plt.tight_layout(rect=[0, 0, 1, 0.96])
chart_path = os.path.join(
output_dir,
'vwap_parameter_scan_chart.png'
)
figure.savefig(chart_path, dpi=150, bbox_inches='tight')
plt.close(figure)
return chart_path
def save_best_equity_chart(best_result, output_dir):
"""保存最优阈值对应的每日权益曲线。"""
equity_df = best_result['daily_equity']
if equity_df is None or equity_df.empty:
return None
chart_df = equity_df.copy()
chart_df['trade_date'] = pd.to_datetime(
chart_df['trade_date'],
errors='coerce'
)
chart_df = chart_df.dropna(subset=['trade_date'])
if chart_df.empty:
return None
figure, axis = plt.subplots(figsize=(14, 6))
axis.plot(
chart_df['trade_date'],
chart_df['equity'],
color='blue',
linewidth=1.5
)
axis.axhline(
CONFIG['initial_capital'],
color='gray',
linestyle='--',
linewidth=1
)
axis.set_title(
'最优阈值 {:.2f}% 每日权益曲线'.format(
best_result['threshold']
)
)
axis.set_xlabel('日期')
axis.set_ylabel('账户权益(元)')
axis.grid(True, alpha=0.3)
plt.tight_layout()
equity_chart_path = os.path.join(
output_dir,
'vwap_best_equity_curve.png'
)
figure.savefig(
equity_chart_path,
dpi=150,
bbox_inches='tight'
)
plt.close(figure)
return equity_chart_path
def generate_report(results, df):
"""输出、保存参数扫描结果和最优参数。"""
output_dir = prepare_output_directory()
report_df = build_report_dataframe(results)
if report_df.empty:
print('没有可输出的回测结果。')
return None
display_df = report_df.copy()
decimal_columns = [
'胜率(%)',
'总盈亏(元)',
'总收益(%)',
'期末权益(元)',
'平均每笔盈亏(元)',
'平均每笔收益(%)',
'盈亏比',
'最大回撤(%)',
'夏普比率'
]
for column in decimal_columns:
display_df[column] = display_df[column].round(3)
print('')
print('=' * 120)
print('VWAP均值回归策略参数扫描汇总')
print('=' * 120)
print(display_df.to_string(index=False))
print('=' * 120)
report_path = os.path.join(
output_dir,
'vwap_parameter_scan_results.csv'
)
report_df.to_csv(
report_path,
index=False,
encoding='gbk'
)
best_index, score_df = select_best_result(report_df)
best_row = score_df.loc[best_index]
best_threshold = safe_float(best_row['阈值(%)'])
best_result = None
for result in results:
if abs(result['threshold'] - best_threshold) < 0.000001:
best_result = result
break
if best_result is None:
best_result = results[int(best_index)]
# 保存最优参数的逐笔成交记录
trades_path = os.path.join(
output_dir,
'vwap_best_threshold_trades.csv'
)
if (
best_result['trades'] is not None and
not best_result['trades'].empty
):
best_result['trades'].to_csv(
trades_path,
index=False,
encoding='gbk'
)
chart_path = save_parameter_chart(
report_df,
output_dir
)
equity_chart_path = save_best_equity_chart(
best_result,
output_dir
)
config_path = os.path.join(
output_dir,
'vwap_best_threshold.txt'
)
with open(config_path, 'w', encoding='gbk') as output_file:
output_file.write('VWAP均值回归策略最优参数\n')
output_file.write(
'生成时间：{}\n'.format(
datetime.now().strftime('%Y-%m-%d %H:%M:%S')
)
)
output_file.write(
'标的：{}\n'.format(CONFIG['stock_code'])
)
output_file.write(
'配置回测区间：{} 至 {}\n'.format(
CONFIG['start_date'],
CONFIG['end_date']
)
)
output_file.write(
'实际数据区间：{} 至 {}\n'.format(
df['datetime'].min().strftime('%Y-%m-%d %H:%M:%S'),
df['datetime'].max().strftime('%Y-%m-%d %H:%M:%S')
)
)
output_file.write(
'最优阈值：{:.2f}%\n'.format(best_threshold)
)
output_file.write(
'交易次数：{}\n'.format(
int(best_row['交易次数'])
)
)
output_file.write(
'胜率：{:.2f}%\n'.format(
best_row['胜率(%)']
)
)
output_file.write(
'总收益：{:.2f}%\n'.format(
best_row['总收益(%)']
)
)
output_file.write(
'总盈亏：{:.2f}元\n'.format(
best_row['总盈亏(元)']
)
)
output_file.write(
'期末权益：{:.2f}元\n'.format(
best_row['期末权益(元)']
)
)
output_file.write(
'最大回撤：{:.2f}%\n'.format(
best_row['最大回撤(%)']
)
)
output_file.write(
'盈亏比：{:.3f}\n'.format(
best_row['盈亏比']
)
)
output_file.write(
'夏普比率：{:.3f}\n'.format(
best_row['夏普比率']
)
)
output_file.write(
'综合得分：{:.4f}\n'.format(
best_row['综合得分']
)
)
print('')
print('=' * 80)
print('综合最优参数')
print('=' * 80)
print('最优阈值：{:.2f}%'.format(best_threshold))
print('交易次数：{}'.format(int(best_row['交易次数'])))
print('多头交易：{}'.format(int(best_row['多头交易'])))
print('空头交易：{}'.format(int(best_row['空头交易'])))
print('胜率：{:.2f}%'.format(best_row['胜率(%)']))
print('总收益：{:.2f}%'.format(best_row['总收益(%)']))
print('总盈亏：{:.2f}元'.format(best_row['总盈亏(元)']))
print('期末权益：{:.2f}元'.format(best_row['期末权益(元)']))
print('最大回撤：{:.2f}%'.format(best_row['最大回撤(%)']))
print('盈亏比：{:.3f}'.format(best_row['盈亏比']))
print('夏普比率：{:.3f}'.format(best_row['夏普比率']))
print('综合得分：{:.4f}'.format(best_row['综合得分']))
print('=' * 80)
print('参数汇总：{}'.format(report_path))
print('逐笔成交：{}'.format(trades_path))
print('参数图表：{}'.format(chart_path))
if equity_chart_path is not None:
print('权益曲线：{}'.format(equity_chart_path))
print('最优参数：{}'.format(config_path))
return report_df
# ================================================================
# 第九部分：QMT策略入口
# ================================================================
def init(ContextInfo):
"""QMT策略初始化函数，只执行一次。"""
ContextInfo.set_universe([CONFIG['stock_code']])
# 防止实时行情中最后一根K线的多个tick重复执行扫描
ContextInfo.scan_completed = False
ContextInfo.scan_running = False
print('=' * 80)
print('QMT版VWAP均值回归参数扫描回测')
print('=' * 80)
print('标的：{}'.format(CONFIG['stock_code']))
print('配置区间：{} 至 {}'.format(
CONFIG['start_date'],
CONFIG['end_date']
))
print('初始资金：{:.2f}元'.format(
CONFIG['initial_capital']
))
print('仓位比例：{:.2f}%'.format(
CONFIG['position_ratio'] * 100.0
))
print('手续费率：{:.6f}'.format(
CONFIG['fee_rate']
))
print('单边滑点：{:.4f}元'.format(
CONFIG['slippage']
))
print('阈值数量：{}'.format(len(THRESHOLDS)))
print('=' * 80)
print('请确认已经在QMT数据管理中补充1分钟历史行情。')
def handlebar(ContextInfo):
"""
QMT行情事件函数。
策略只在主图最后一根K线执行一次完整扫描，避免每根历史K线重复读取
全量分钟数据。
"""
if not ContextInfo.is_last_bar():
return
if getattr(ContextInfo, 'scan_completed', False):
return
if getattr(ContextInfo, 'scan_running', False):
return
ContextInfo.scan_running = True
try:
minute_df = load_qmt_minute_data(ContextInfo)
if minute_df is None or minute_df.empty:
print('回测终止：没有取得有效行情。')
return
print('')
print('正在计算每日累计VWAP...')
minute_df = add_vwap_columns(minute_df)
if minute_df.empty:
print('回测终止：VWAP计算后没有有效数据。')
return
print('VWAP计算完成。')
print('有效交易日数：{}'.format(
minute_df['trade_date'].nunique()
))
print('有效分钟数：{:,}'.format(len(minute_df)))
results = run_parameter_scan(minute_df)
report_df = generate_report(results, minute_df)
if report_df is not None:
ContextInfo.vwap_scan_report = report_df
ContextInfo.scan_completed = True
print('')
print('VWAP参数扫描回测全部完成。')
except Exception as error:
print('VWAP参数扫描发生异常：{}'.format(error))
import traceback
print(traceback.format_exc())
finally:
ContextInfo.scan_running = False
def stop(ContextInfo):
"""QMT策略停止回调。"""
print('VWAP参数扫描策略已停止。')
💬 评论 (0)
还没有评论，来抢沙发吧。