基于你 plot.py 中 StatisticalArbitrageStrategy 的 Z-score 统计套利策略, 迁移到 freqtrade 策略格式。
Timeframe
1m
Direction
Long Only
Stoploss
-10.0%
Trailing Stop
No
ROI
0m: 20.0%
Interface Version
3
Startup Candles
N/A
Indicators
2
freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these imports ---
import numpy as np
import pandas as pd
from datetime import datetime, timedelta, timezone
from pandas import DataFrame
from typing import Dict, Optional, Union, Tuple
from freqtrade.strategy import (
IStrategy,
Trade,
Order,
PairLocks,
informative, # @informative decorator
# Hyperopt Parameters
BooleanParameter,
CategoricalParameter,
DecimalParameter,
IntParameter,
RealParameter,
# timeframe helpers
timeframe_to_minutes,
timeframe_to_next_date,
timeframe_to_prev_date,
# Strategy helper functions
merge_informative_pair,
stoploss_from_absolute,
stoploss_from_open,
AnnotationType,
)
# --------------------------------
# Add your lib to import here
import talib.abstract as ta
from technical import qtpylib
class StatArbZScoreMasterStrategy(IStrategy):
"""
基于你 plot.py 中 StatisticalArbitrageStrategy 的 Z-score 统计套利策略,
迁移到 freqtrade 策略格式。
核心逻辑:
- 对收盘价做滚动均值 / 标准差,计算 z-score
- z-score 下穿 -threshold 开多(价格相对均值偏低)
- z-score 上穿 +threshold 开空(需要在期货模式并开启 can_short)
- 当 |z-score| < exit_threshold 时,平掉对应方向的仓位
"""
INTERFACE_VERSION = 3
# 和你原脚本一致,用 15m
timeframe = "1m"
# 允许做空(如果你只做多,可以改成 False,并注释掉 enter_short / exit_short)
can_short: bool = True
# ROI 和止损可以先用比较宽松的参数,后面你再根据回测调整
# minimal_roi = {
# "0": 0.20
# }
# stoploss = -0.10
minimal_roi = {
"0": 2.0
}
stoploss = -1.0
trailing_stop = False
process_only_new_candles = True
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
# 至少需要这么多 K 线才有稳定的 z-score
startup_candle_count: int = 100
max_open_trades=10
# -----------------------------
# 超参(可以用于 hyperopt)
# -----------------------------
# 回看周期 = 你原脚本的 lookback_period
lookback = IntParameter(20, 288, default=50, space="buy")
# 进场 z-score 阈值 = 你原脚本的 zscore_threshold
entry_long_zscore = RealParameter(1.0, 4.0, default=1.0, space="buy")
entry_short_zscore = RealParameter(1.0, 4.0, default=1.0, space="buy")
# 退出 z-score 阈值 = 你原脚本的 exit_threshold
exit_long_zscore = RealParameter(0.1, 1.0, default=0.3, space="sell")
exit_short_zscore = RealParameter(0.1, 1.0, default=0.3, space="sell")
# 是否使用 EMA 代替 SMA
use_ema = CategoricalParameter([False, True], default=False, space="buy")
# 绘图配置(freqtrade plot 命令用)
@property
def plot_config(self):
return {
"main_plot": {
"close": {},
"mean_price": {"color": "orange"},
"upper_band": {"color": "grey"},
"lower_band": {"color": "grey"},
},
"subplots": {
"Z-Score": {
"zscore": {"color": "blue"},
"entry_pos": {"color": "green"},
"entry_neg": {"color": "red"},
}
}
}
def leverage(self, pair, current_time, current_rate, proposed_leverage, max_leverage, entry_tag, side, **kwargs):
return self.config["leverage"]
# -----------------------------
# 指标计算:把 plot.py 的 calculate_zscore 搬进来
# -----------------------------
def _calculate_zscore(
self, series: pd.Series, lookback: int, use_ema: bool
) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""
计算 Z-score、均值、标准差
对应你 plot.py 里的 calculate_zscore。
"""
if use_ema:
# 参考你原来的 half-life 算法
half_life = lookback / 2
alpha = 1 - np.exp(np.log(0.5) / half_life)
mean = series.ewm(alpha=alpha, adjust=False).mean()
std = series.ewm(alpha=alpha, adjust=False).std()
else:
mean = series.rolling(window=lookback, min_periods=1).mean()
std = series.rolling(window=lookback, min_periods=1).std()
std = std.replace(0, np.nan).ffill().fillna(0.0001)
zscore = (series - mean) / std
return zscore, mean, std
# -----------------------------
# Freqtrade 必须实现的三个函数
# -----------------------------
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
在这里计算所有需要的指标(zscore / 均值 / 上下轨等)
"""
if dataframe.empty:
return dataframe
lb = int(self.lookback.value)
uz_long = float(self.entry_long_zscore.value)
uz_short = float(self.entry_short_zscore.value)
use_ema = bool(self.use_ema.value)
# 计算 Z-score
z, mean, std = self._calculate_zscore(
dataframe["close"], lookback=lb, use_ema=use_ema
)
dataframe["zscore"] = z
dataframe["mean_price"] = mean
dataframe["std_price"] = std
# 上下轨(和你原策略一致:mean ± zscore_threshold * std)
dataframe["upper_band"] = dataframe["mean_price"] + uz_short * dataframe["std_price"]
dataframe["lower_band"] = dataframe["mean_price"] - uz_long * dataframe["std_price"]
# 方便画图看看正负区域
dataframe["entry_pos"] = uz_short
dataframe["entry_neg"] = -uz_long
# 保证 volume 存在(有些数据源 volume 可能为 NaN)
if "volume" not in dataframe.columns:
dataframe["volume"] = 1.0
else:
dataframe["volume"] = dataframe["volume"].fillna(0)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
对应你 plot.py 中 generate_signals 里的开仓逻辑:
- zscore 从上方穿越 -threshold:开多
- zscore 从下方穿越 +threshold:开空
"""
uz_long = float(self.entry_long_zscore.value)
uz_short = float(self.entry_short_zscore.value)
# Long:zscore 从上方 >= -threshold 到 当前 < -threshold
long_cond = (
# (dataframe["zscore"].shift(1) >= -uz_long) &
(dataframe["zscore"] < -uz_long) &
(dataframe["volume"] > 0)
)
dataframe.loc[long_cond, "enter_long"] = 1
if self.can_short:
# Short:zscore 从下方 <= threshold 到 当前 > threshold
short_cond = (
# (dataframe["zscore"].shift(1) <= uz) &
(dataframe["zscore"] > uz_short) &
(dataframe["volume"] > 0)
)
dataframe.loc[short_cond, "enter_short"] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
对应你 plot.py 中“Z-score 回归到 exit_threshold 内时平仓”的逻辑:
- 当 |zscore| < exit_threshold 时,平多 / 平空。
在 freqtrade 里,我们用同一条件触发 exit_long / exit_short。
"""
ez_long = float(self.exit_long_zscore.value)
ez_short = float(self.exit_short_zscore.value)
exit_long_cond = (
# (dataframe["zscore"].shift(1) >= -uz) &
(dataframe["zscore"] > ez_long) &
(dataframe["volume"] > 0)
)
dataframe.loc[exit_long_cond, "exit_long"] = 1
if self.can_short:
exit_short_cond = (
# (dataframe["zscore"].shift(1) >= -uz) &
(dataframe["zscore"] < -ez_short) &
(dataframe["volume"] > 0)
)
dataframe.loc[exit_short_cond, "exit_short"] = 1
return dataframe