Z-Score Mean Reversion with Threshold, Stop-Loss, and Time Exit
Summary
This example implements a daily mean-reversion strategy for a Shanghai Futures Exchange gold contract. It calculates a Z-score from recent closing prices, enters long when the score falls below a negative entry threshold and short when it rises above a positive threshold, then targets a return toward the mean. Exits occur when the score returns near zero, crosses a wider adverse threshold, or the position reaches a time limit. Position size, lookback window, entry and exit levels, stop threshold, and backtest dates are set as parameters.
The code demonstrates order targeting and state tracking in a futures backtest, but reports no performance results. It uses a short historical period and fixed thresholds, with no stated transaction costs or slippage analysis. The comments and implementation differ on the lookback description, and the example does not explain parameter selection or address whether price behavior is suitable for mean reversion.
Key ideas
- The strategy standardizes recent daily closes with a rolling mean and standard deviation.
- It enters long below a negative Z-score threshold and short above a positive threshold.
- Positions close near the mean, at an adverse Z-score level, or after a maximum holding period.
- The example specifies a fixed contract size and backtest interval but presents no results.
- Threshold selection, costs, slippage, and suitability for mean reversion are not evaluated.
Tags
Full text
# z_score.py
```py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
__author__ = "Chaos"
from datetime import date
import numpy as np
from tqsdk import TqApi, TqAuth, TqBacktest, TargetPosTask, BacktestFinished
from tqsdk.tafunc import time_to_str
# ===== 全局参数设置 =====
SYMBOL = "SHFE.au2106"
POSITION_SIZE = 50 # 每次交易手数
START_DATE = date(2020, 11, 1) # 回测开始日期
END_DATE = date(2020, 12, 15) # 回测结束日期
# Z-Score参数
WINDOW_SIZE = 14 # Z-Score计算窗口期
ENTRY_THRESHOLD = 1.8 # 开仓阈值
EXIT_THRESHOLD = 0.4 # 平仓阈值
STOP_LOSS_THRESHOLD = 2.5 # 止损阈值
# 风控参数
TIME_STOP_DAYS = 8 # 时间止损天数
# ===== 全局变量 =====
current_direction = 0 # 当前持仓方向：1=多头，-1=空头，0=空仓
entry_price = 0 # 开仓价格
entry_date = None # 开仓日期
# ===== 策略开始 =====
print("开始运行Z-Score均值回归策略...")
# 创建API实例
api = TqApi(backtest=TqBacktest(start_dt=START_DATE, end_dt=END_DATE),
auth=TqAuth("快期账户", "快期密码"))
# 订阅合约的K线数据
klines = api.get_kline_serial(SYMBOL, 60 * 60 * 24) # 日线数据
# 创建目标持仓任务
target_pos = TargetPosTask(api, SYMBOL)
try:
while True:
# 等待更新
api.wait_update()
# 如果K线有更新
if api.is_changing(klines.iloc[-1], "datetime"):
# 确保有足够的数据计算指标
if len(klines) < WINDOW_SIZE + 10:
continue
# 计算Z-Score
prices = klines.close.iloc[-WINDOW_SIZE:] # 获取最近20天的收盘价
mean = np.mean(prices) # 计算均值
std = np.std(prices) # 计算标准差
current_price = float(klines.close.iloc[-1]) # 当前价格
# 处理标准差为0的情况
if std == 0:
z_score = 0 # 如果标准差为0，说明所有价格都相同，Z-Score设为0
else:
z_score = (current_price - mean) / std # 计算Z-Score
# 获取最新数据
current_timestamp = klines.datetime.iloc[-1]
current_datetime = time_to_str(current_timestamp)
# 打印当前状态
print(f"日期: {current_datetime}, 价格: {current_price:.2f}, Z-Score: {z_score:.2f}")
# ===== 交易逻辑 =====
# 空仓状态 - 寻找开仓机会
if current_direction == 0:
# 多头开仓条件：Z-Score显著低于均值
if z_score < -ENTRY_THRESHOLD:
current_direction = 1
target_pos.set_target_volume(POSITION_SIZE)
entry_price = current_price
entry_date = current_timestamp
print(f"多头开仓: 价格={entry_price:.2f}, Z-Score={z_score:.2f}")
# 空头开仓条件：Z-Score显著高于均值
elif z_score > ENTRY_THRESHOLD:
current_direction = -1
target_pos.set_target_volume(-POSITION_SIZE)
entry_price = current_price
entry_date = current_timestamp
print(f"空头开仓: 价格={entry_price:.2f}, Z-Score={z_score:.2f}")
# 多头持仓 - 检查平仓条件
elif current_direction == 1:
# 止损条件：Z-Score继续大幅下跌
if z_score < -STOP_LOSS_THRESHOLD:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止损平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 止盈条件：Z-Score回归到均值附近
elif -EXIT_THRESHOLD <= z_score <= EXIT_THRESHOLD:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头止盈平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 时间止损
elif (current_timestamp - entry_date) / (60 * 60 * 24) >= TIME_STOP_DAYS:
profit_pct = (current_price - entry_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"多头时间止损: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 空头持仓 - 检查平仓条件
elif current_direction == -1:
# 止损条件：Z-Score继续大幅上涨
if z_score > STOP_LOSS_THRESHOLD:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止损平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 止盈条件：Z-Score回归到均值附近
elif -EXIT_THRESHOLD <= z_score <= EXIT_THRESHOLD:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头止盈平仓: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
# 时间止损
elif (current_timestamp - entry_date) / (60 * 60 * 24) >= TIME_STOP_DAYS:
profit_pct = (entry_price - current_price) / entry_price * 100
target_pos.set_target_volume(0)
current_direction = 0
print(f"空头时间止损: 价格={current_price:.2f}, 盈亏={profit_pct:.2f}%")
except BacktestFinished as e:
print("回测结束")
api.close()
```Shown in full with attribution under the source's licence. Licence: Apache-2.0
This summary was written by Stratmill's research agent from the original; it is not a copy of the source.