Example strategy showing how the user connects their own IFreqaiModel to the strategy.
Timeframe
N/A
Direction
Long & Short
Stoploss
-1.0%
Trailing Stop
No
ROI
0m: 1.0%, 20m: 0.8%, 40m: 0.4%, 60m: 0.1%
Interface Version
N/A
Startup Candles
N/A
Indicators
8
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
import logging
from functools import reduce
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib
import numpy as np
from freqtrade.strategy import IStrategy
import pandas as pd
logger = logging.getLogger(__name__)
from numpy.lib.stride_tricks import sliding_window_view
def triple_barrier_labeling(
close: np.ndarray,
high: np.ndarray,
low: np.ndarray,
up_pct: float,
down_pct: float,
horizon: int,
vol: np.ndarray = None,
vol_mult: float = 1.0,
):
"""
完整版 Triple-Barrier(机构级写法)
参数:
- close/high/low: ndarray
- up_pct / down_pct: 固定障碍百分比
- horizon: 时间障碍(未来 N 根 K)
- vol: 波动率 (如1h滚动 std),如果提供,则 up/down 障碍将基于 volatility
- vol_mult: 波动率倍数(默认 1)
返回:
- labels: up / down / NaN
- t_events: 第几根触发障碍(从 1 开始)
- hit_prices: 触发时的价格
"""
n = len(close)
valid_len = n - horizon
# ================================
# 0) future window vectorization
# ================================
future_high = sliding_window_view(high, window_shape=horizon)[1:]
future_low = sliding_window_view(low, window_shape=horizon)[1:]
c = close[:valid_len]
# ================================
# 1) 构建上下障碍(可选波动率)
# ================================
if vol is not None:
up_barrier = c * (1 + vol_mult * vol[:valid_len])
down_barrier = c * (1 - vol_mult * vol[:valid_len])
else:
up_barrier = c * (1 + up_pct)
down_barrier = c * (1 - down_pct)
# ================================
# 2) 判断未来在哪一根触发障碍
# ================================
hit_up = future_high >= up_barrier[:, None]
hit_down = future_low <= down_barrier[:, None]
# first occurrence index
up_first_idx = np.argmax(hit_up, axis=1)
down_first_idx = np.argmax(hit_down, axis=1)
# 若未触发,argmax 返回0,但 hit_up 全 False → 需要修正
up_first_idx[~hit_up.any(axis=1)] = -1
down_first_idx[~hit_down.any(axis=1)] = -1
# ================================
# 3) 确定先触发者
# ================================
labels = np.full(n, np.nan, dtype=object)
t_events = np.full(n, np.nan)
hit_prices = np.full(n, np.nan)
for i in range(valid_len):
up_i = up_first_idx[i]
dn_i = down_first_idx[i]
if up_i == -1 and dn_i == -1:
continue # NaN
# 谁先触发?
if up_i != -1 and (dn_i == -1 or up_i < dn_i):
labels[i] = "up"
t_events[i] = up_i + 1
hit_prices[i] = future_high[i, up_i]
elif dn_i != -1:
labels[i] = "down"
t_events[i] = dn_i + 1
hit_prices[i] = future_low[i, dn_i]
return labels, t_events, hit_prices
def evaluate_labels(labels):
"""评分指标:越大越好"""
counts = pd.Series(labels).value_counts()
up = counts.get('up', 0)
down = counts.get('down', 0)
nan = counts.get(np.nan, 0)
if up + down == 0:
return -1 # worthless
# 平衡度
balance = min(up, down) / max(up, down)
# 有效标签比例
valid_ratio = (up + down) / (up + down + nan)
score = balance * 0.7 + valid_ratio * 0.3
return score
def grid_search_tb(close, high, low):
best_score = -1
best_params = None
best_labels = None
up_pcts = [0.003, 0.005, 0.01, 0.015, 0.02]
horizons = [3, 5, 8, 10]
for up in up_pcts:
for h in horizons:
labels, _, _ = triple_barrier_labeling(
close, high, low,
up_pct=up, down_pct=up, # 对称
horizon=h
)
score = evaluate_labels(labels)
print(f"up={up:.3%}, horizon={h}, score={score:.4f}")
if score > best_score:
best_score = score
best_params = (up, h)
best_labels = labels
return best_params, best_labels
class MyXGB_gpt_multi(IStrategy):
"""
Example strategy showing how the user connects their own
IFreqaiModel to the strategy.
Warning! This is a showcase of functionality,
which means that it is designed to show various functions of FreqAI
and it runs on all computers. We use this showcase to help users
understand how to build a strategy, and we use it as a benchmark
to help debug possible problems.
This means this is *not* meant to be run live in production.
"""
minimal_roi = {"0": 0.01, "20":0.008, "40":0.004, "60":0.001, "240": -1}
plot_config = {
"main_plot": {},
"subplots": {
"&-s_close": {"&-s_close": {"color": "blue"}},
"do_predict": {
"do_predict": {"color": "brown"},
},
},
}
process_only_new_candles = True
stoploss = -0.01
use_exit_signal = False
# this is the maximum period fed to talib (timeframe independent)
startup_candle_count: int = 40
can_short = True
def feature_engineering_expand_all(
self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
) -> DataFrame:
# === 动量 & 超买超卖 ===
rsi = ta.RSI(dataframe, timeperiod=period)
dataframe[f"%-rsi-{period}"] = rsi
dataframe[f"%-rsi_gt_70-{period}"] = (rsi > 70).astype(int)
dataframe[f"%-rsi_lt_30-{period}"] = (rsi < 30).astype(int)
dataframe[f"%-rsi_middle-{period}"] = ((rsi >= 30) & (rsi <= 70)).astype(int)
mfi = ta.MFI(dataframe, timeperiod=period)
dataframe[f"%-mfi-{period}"] = mfi
dataframe[f"%-mfi_gt_80-{period}"] = (mfi > 80).astype(int)
dataframe[f"%-mfi_lt_20-{period}"] = (mfi < 20).astype(int)
# === 趋势强度 ===
adx = ta.ADX(dataframe, timeperiod=period)
dataframe[f"%-adx-{period}"] = adx
dataframe[f"%-adx_gt_25-{period}"] = (adx > 25).astype(int) # 趋势强弱分界
# === 均线系统 ===
sma = ta.SMA(dataframe, timeperiod=period)
ema = ta.EMA(dataframe, timeperiod=period)
dataframe[f"%-sma-{period}"] = sma
dataframe[f"%-ema-{period}"] = ema
# 价格相对于均线的位置(归一化)
atr = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=period)
dataframe[f"%-price_above_sma_norm-{period}"] = (dataframe['close'] - sma) / (atr + 1e-8)
dataframe[f"%-price_above_ema_norm-{period}"] = (dataframe['close'] - ema) / (atr + 1e-8)
# 布尔状态:是否在均线上方
dataframe[f"%-close_gt_sma-{period}"] = (dataframe['close'] > sma).astype(int)
dataframe[f"%-close_gt_ema-{period}"] = (dataframe['close'] > ema).astype(int)
# === 布林带 ===
bb = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=period, stds=2.2)
bb_width = (bb["upper"] - bb["lower"]) / (bb["mid"] + 1e-8)
dataframe[f"%-bb_width-{period}"] = bb_width
dataframe[f"%-bb_width_high-{period}"] = (bb_width > bb_width.rolling(50).quantile(0.8)).astype(int)
dataframe[f"%-bb_width_low-{period}"] = (bb_width < bb_width.rolling(50).quantile(0.2)).astype(int)
# 价格在布林带中的相对位置(0=下轨, 1=上轨)
bb_position = (dataframe['close'] - bb["lower"]) / (bb["upper"] - bb["lower"] + 1e-8)
dataframe[f"%-bb_position-{period}"] = bb_position.clip(0, 1)
# === 动量变化 ===
macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
dataframe[f"%-macd-{period}"] = macd["macd"]
dataframe[f"%-macd_signal-{period}"] = macd["macdsignal"]
dataframe[f"%-macd_diff-{period}"] = macd["macdhist"]
roc = ta.ROC(dataframe, timeperiod=period)
dataframe[f"%-roc-{period}"] = roc
dataframe[f"%-roc_positive-{period}"] = (roc > 0).astype(int)
# === 成交量异常 ===
vol_mean = dataframe["volume"].rolling(period).mean()
vol_std = dataframe["volume"].rolling(period).std()
dataframe[f"%-relative_volume-{period}"] = dataframe["volume"] / (vol_mean + 1e-8)
dataframe[f"%-volume_spike-{period}"] = (
dataframe["volume"] > (vol_mean + 2 * vol_std)
).astype(int)
return dataframe
def feature_engineering_expand_basic(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
# === 价格变化率(多尺度)===
dataframe["%-pct-change-1"] = dataframe["close"].pct_change(1)
dataframe["%-pct-change-3"] = dataframe["close"].pct_change(3)
dataframe["%-pct-change-6"] = dataframe["close"].pct_change(6)
# === 长期趋势 ===
ema_200 = ta.EMA(dataframe, timeperiod=200)
dataframe["%-ema-200"] = ema_200
dataframe["%-close_gt_ema_200"] = (dataframe["close"] > ema_200).astype(int)
# === 波动率(ATR 归一化)===
atr_14 = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14)
dataframe["%-atr_norm"] = atr_14 / dataframe["close"]
# ⚠️ 不再传递 raw_price / raw_volume(XGBoost 不需要原始尺度)
# 如果确实需要,建议用 rank 或 z-score
return dataframe
def feature_engineering_standard(
self, dataframe: DataFrame, metadata: dict, **kwargs
) -> DataFrame:
# === 时间特征(周期性编码)===
# 使用 sin/cos 编码避免 0 和 23 小时距离过大
hour = dataframe["date"].dt.hour
day_of_week = dataframe["date"].dt.dayofweek
dataframe["%-hour_sin"] = np.sin(2 * np.pi * hour / 24)
dataframe["%-hour_cos"] = np.cos(2 * np.pi * hour / 24)
dataframe["%-dow_sin"] = np.sin(2 * np.pi * day_of_week / 7)
dataframe["%-dow_cos"] = np.cos(2 * np.pi * day_of_week / 7)
# === 价格极值(近期高低点)===
dataframe["%-high_24h"] = dataframe["high"].rolling(24).max()
dataframe["%-low_24h"] = dataframe["low"].rolling(24).min()
dataframe["%-close_to_high_24h"] = (dataframe["close"] - dataframe["%-low_24h"]) / \
(dataframe["%-high_24h"] - dataframe["%-low_24h"] + 1e-8)
return dataframe
# ====== set_freqai_targets: 目标改为回归型(future max/min return) ======
def set_freqai_targets(self, df, metadata: dict, **kwargs):
"""
生成回归目标:
&t-max_ret : 未来 horizon 内的最大上涨率 (max(high)/close -1)
&t-min_ret : 未来 horizon 内的最小下跌率 (min(low)/close -1) (负数)
保持末尾 horizon 行为 NaN。
"""
df = df.copy()
horizon = 4 # 推荐 8 根 = 24 分钟(可 grid-search)
n = len(df)
if n < horizon + 2:
df["&t-max_ret"] = np.nan
df["&t-min_ret"] = np.nan
return df
# future windows using sliding_window_view
highs = df["high"].values
lows = df["low"].values
closes = df["close"].values
from numpy.lib.stride_tricks import sliding_window_view
sw_high = sliding_window_view(highs, window_shape=horizon) # shape (n-h+1, h)
sw_low = sliding_window_view(lows, window_shape=horizon)
# we want future windows starting at i+1, so slice [1:valid_len+1]
valid_len = len(sw_high) - 1
future_high = sw_high[1: valid_len + 1] # shape (valid_len, horizon)
future_low = sw_low[1: valid_len + 1]
cur = closes[:valid_len]
max_ret = (future_high.max(axis=1) - cur) / (cur + 1e-12)
min_ret = (future_low.min(axis=1) - cur) / (cur + 1e-12) # 负数 if drop
# pad to original length
pad = np.array([np.nan] * horizon)
df["&t-max_ret"] = np.concatenate([max_ret, pad])
df["&t-min_ret"] = np.concatenate([min_ret, pad])
return df
# ====== populate_indicators: 添加 EMA200 / ATR_norm / ADX ======
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算额外的过滤指标以用于进出场决策
"""
dataframe = self.freqai.start(dataframe, metadata, self)
# keep other previous engineered features unchanged
return dataframe
# ====== populate_entry_trend: 用预测回归结果并加过滤器 ======
def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
"""
进场:只在多头趋势且波动率充足时,且模型预期有边际收益时入场
逻辑(多头示例):
- do_predict == 1
- close > EMA200
- ATR_norm > atr_th
- edge = pred_up + pred_dn > edge_th
- pred_up > min_up
类似地可扩展空头(你 can_short=False 当前)
"""
df = df.copy()
# EMA200 on 3min (200 candles)
df["%-ema-200"] = ta.EMA(df, timeperiod=20)
# ATR 14 -> normalized by price
atr = ta.ATR(df["high"], df["low"], df["close"], timeperiod=7)
df["%-atr"] = atr
df["%-atr_norm"] = atr / (df["close"] + 1e-12)
# ADX for trend strength
adx = ta.ADX(df, timeperiod=7)
df["%-adx"] = adx
# 超参(可 tune)
atr_th = 0.0015 # ATR_norm threshold (0.15%)
edge_th = 0.0015 # minimal expected edge (0.15%)
min_up = 0.003 # minimal predicted upside (0.1%)
# 预测列(freqai 输出),若不存在则回退到真值(仅离线分析/回测时)
pred_up = df.get("&t-max_ret_pred")
pred_dn = df.get("&t-min_ret_pred")
if pred_up is None or pred_dn is None:
# 没有 predictor 输出时,使用未来真值作为回测目标(仅用于离线调参)
pred_up = df.get("&t-max_ret")
pred_dn = df.get("&t-min_ret")
# ensure numpy arrays
pred_up = np.asarray(pred_up, dtype=float)
pred_dn = np.asarray(pred_dn, dtype=float)
trend_up_ok = df["close"] > df["%-ema-200"]
trend_dn_ok = df["close"] < df["%-ema-200"]
vol_ok = df["%-atr_norm"] > atr_th
# edge measured as predicted upside plus predicted down (down is negative) => net edge
edge = pred_up + pred_dn
long_cond = (
(df["do_predict"] == 1) &
(trend_up_ok) &
(vol_ok) &
(edge > edge_th) &
(pred_up > min_up)
)
df.loc[long_cond, ["enter_long", "enter_tag"]] = (1, "long")
if self.can_short:
short_cond = (
(df["do_predict"] == 1) &
(trend_dn_ok) &
(vol_ok) &
(edge < -edge_th) &
(pred_dn < -min_up)
)
df.loc[short_cond, ["enter_short", "enter_tag"]] = (1, "short")
return df
# ====== populate_exit_trend: 动态退出基于预测与价格 ======
def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
"""
退出逻辑:
- 若模型预测未来下跌风险扩大(pred_min < exit_dn_th) -> 提前退出
- 若模型预测 future upside 大幅缩小 -> 提前退出
- 可结合价格突破短期均线 / 反向信号做退出
"""
df = df.copy()
exit_dn_th = -0.003 # 若预测未来最小回撤 < -0.3%,提前退出
min_up_when_keep = 0.0008 # 如果预测上涨低于 0.08%,也退出
pred_up = df.get("&t-max_ret_pred")
pred_dn = df.get("&t-min_ret_pred")
if pred_up is None or pred_dn is None:
pred_up = df.get("&t-max_ret")
pred_dn = df.get("&t-min_ret")
pred_up = np.asarray(pred_up, dtype=float)
pred_dn = np.asarray(pred_dn, dtype=float)
# 当模型预测下跌变大或上涨微乎其微 -> 退出多头
exit_long_cond = (
(df["do_predict"] == 1) &
(pred_dn < exit_dn_th) |
((pred_up < min_up_when_keep) & (df["enter_tag"] == "long"))
)
df.loc[exit_long_cond, "exit_long"] = 1
# 空头退出(若启用空头,可用相反逻辑)
exit_short_cond = False
if self.can_short:
exit_short_cond = (
(df["do_predict"] == 1) &
(pred_up < -exit_dn_th) # 若上行预测转差则出空
)
df.loc[exit_short_cond, "exit_short"] = 1
return df
# ====== confirm_trade_entry: 强化成交价格保护(避免滑点小亏) ======
def confirm_trade_entry(
self,
pair: str,
order_type: str,
amount: float,
rate: float,
time_in_force: str,
current_time,
entry_tag,
side: str,
**kwargs,
) -> bool:
"""
价格保护:若是多头,拒绝以超过当前 close 的太高价格成交(防止追高)
这里阈值设置为 0.25%(可调整)
"""
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = df.iloc[-1].squeeze()
guard = 0.0025 # 0.25%
if side == "long":
# 防止以高于 last close * (1 + guard) 的价格成交(追高)
if rate > (last_candle["close"] * (1 + guard)):
return False
else:
if rate < (last_candle["close"] * (1 - guard)):
return False
return True