Z-Score均值回归策略
一、交易策略解释
核心思想
Z-Score均值回归策略基于金融市场价格具有均值回归特性的假设，即价格在短期内可能偏离其长期均值，但最终会回归。该策略通过计算资产价格相对于其历史移动平均线的标准化偏离程度（Z-Score），识别被显著高估或低估的价格水平，并在预期其回归均值时进行交易。
当价格大幅偏离均值（高Z-Score值）时，策略认为这种偏离难以持续，价格将回归，从而产生逆势交易的机会。价格远低于均值时买入，远高于均值时卖出，当价格回归至均值附近时平仓获利。
理论基础
-
统计学基础：Z-Score（标准分数）是统计学中的基本概念，表示某观测值与均值的差距相对于标准差的比例。在正态分布中，约68%的数据落在均值±1个标准差范围内，95%落在±2个标准差内，99.7%落在±3个标准差内。
-
均值回归理论：经济学家罗伯特·席勒和尤金·法玛等人的研究表明，金融资产价格在短期内会出现过度反应，但长期看往往会回归至其基本面价值。这构成了均值回归交易的学术基础。
-
实证证据：多项学术研究表明，金融市场中确实存在均值回归现象，特别是在较短的时间尺度上。例如，Poterba和Summers(1988)的研究发现，股票收益在短期内表现出负相关性，支持均值回归假设。
策略适用场景
-
横盘整理市场：策略在价格区间波动的市场中表现最佳
-
波动率适中：过低的波动率提供较少交易机会，过高的波动率可能导致虚假信号
-
非强趋势市场：在单边强趋势市场中，均值回归策略可能连续亏损
-
相关性强的资产对：配对交易中使用相关性高的资产效果更佳
二、天勤介绍
天勤平台概述
天勤（TqSdk）是一个由信易科技开发的开源量化交易系统，为期货、期权等衍生品交易提供专业的量化交易解决方案。平台具有以下特 点：
- 丰富的行情数据： 提供所有可交易合约的全部Tick和K线数据，基于内存数据库实现零延迟访问。
- 一站式的解决方案： 从历史数据分析到实盘交易的完整工具链，打通开发、回测、模拟到实盘的全流程。
- 专业的技术支持： 近百个技术指标源码，深度集成pandas和numpy，采用单线程异步模型保证性能。
策略开发流程
- 环境准备
- 安装Python环境（推荐Python 3.6或以上版本）
- 安装tqsdk包：pip install tqsdk
- 注册天勤账户获取访问密钥
- 数据准备
- 订阅近月与远月合约行情
- 获取历史K线或者Tick数据（用于分析与行情推进）
- 策略编写
- 设计信号生成逻辑（基于价差、均值和标准差）
- 编写交易执行模块（开仓、平仓逻辑）
- 实现风险控制措施（止损、资金管理）
- 回测验证
- 设置回测时间区间和初始资金
- 运行策略获取回测结果
- 分析绩效指标（胜率、收益率、夏普率等）
- 策略优化
- 调整参数（标准差倍数、窗口大小等）
- 添加过滤条件（成交量、波动率等）
- 完善风险控制机制
三、天勤实现策略
策略原理
Z-Score计算是该策略的核心，具体步骤如下：
-
确定观察窗口期（本策略使用14天）
-
计算窗口期内的价格均值 μ：取最近14天收盘价的算术平均值
-
计算窗口期内的价格标准差 σ：测量14天内价格的离散程度
-
计算当前价格的Z-Score：Z = (当前价格 - μ) / σ
Z-Score实际上是一个标准化指标，反映了当前价格偏离均值的程度：
-
Z = 0：价格恰好位于均值位置
-
Z > 0：价格高于均值
-
Z < 0：价格低于均值
-
|Z| > 1.8：价格显著偏离均值（本策略的开仓阈值）
-
|Z| > 2.5：价格极端偏离均值（本策略的止损阈值）
在正态分布假设下，约95%的数据点落在Z-Score为±2的范围内，因此Z-Score超过这一范围意味着价格处于统计上的极端区域，有较大概率回归。
交易逻辑
开仓条件：
-
做多信号：当Z-Score < -1.8时，表明价格显著低于均值，开多仓
-
做空信号：当Z-Score > 1.8时，表明价格显著高于均值，开空仓
平仓条件：
-
止盈平仓：当Z-Score回归到-0.4至0.4区间内，表明价格接近均值，平仓获利
-
止损设置：
-
多头止损：当Z-Score < -2.5时，表明下跌趋势可能加强，触发止损
-
空头止损：当Z-Score > 2.5时，表明上涨趋势可能加强，触发止损
-
时间止损：最长持仓时间为8个交易日，若超过则强制平仓
回测
回测初始设置
- 测试周期： 2020 年 11 月 1 日 - 2020 年 12 月 15 日
- 交易品种： SHFE.au2106
- 初始资金： 1000万元
回测结果
上述回测累计收益走势图
完整代码示例
#!/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()