358 lines
15 KiB
Python
358 lines
15 KiB
Python
import logging
|
||
import numpy as np
|
||
import datetime
|
||
from functools import reduce
|
||
from freqtrade.persistence import Trade
|
||
import talib.abstract as ta
|
||
from pandas import DataFrame
|
||
from typing import Dict, Optional
|
||
from technical import qtpylib
|
||
from freqtrade.strategy import IStrategy
|
||
from logging import getLogger
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class FreqaiPrimer(IStrategy):
|
||
"""
|
||
策略说明：
|
||
- 所有交易信号由 FreqAI 动态预测
|
||
- 不使用 Hyperopt 优化任何参数
|
||
- 使用 FreqAI 提供的趋势和波动率信号进行交易决策
|
||
"""
|
||
# 👇 添加在这里
|
||
PAIR_PARAMS = {
|
||
"BTC/USDT": {"volatility_factor": 1.0, "min_stop": -0.005},
|
||
"TON/USDT": {"volatility_factor": 1.3, "min_stop": -0.0035},
|
||
"DOGE/USDT": {"volatility_factor": 1.5, "min_stop": -0.003},
|
||
"DOT/USDT": {"volatility_factor": 1.3, "min_stop": -0.0035},
|
||
"XRP/USDT": {"volatility_factor": 1.35, "min_stop": -0.0032},
|
||
"OKB/USDT": {"volatility_factor": 0.9, "min_stop": -0.006},
|
||
"SOL/USDT": {"volatility_factor": 1.4, "min_stop": -0.0032},
|
||
"WCT/USDT": {"volatility_factor": 1.4, "min_stop": -0.004},
|
||
"TRUMP/USDT": {"volatility_factor": 1.4, "min_stop": -0.0035},
|
||
"SUI/USDT": {"volatility_factor": 1.6, "min_stop": -0.0025},
|
||
"PEPE/USDT": {"volatility_factor": 1.5, "min_stop": -0.0036},
|
||
"TRB/USDT": {"volatility_factor": 1.45, "min_stop": -0.004},
|
||
"MASK/USDT": {"volatility_factor": 1.65, "min_stop": -0.0045},
|
||
"UNI/USDT": {"volatility_factor": 1.45, "min_stop": -0.005},
|
||
"KAITO/USDT": {"volatility_factor": 1.45, "min_stop": -0.004},
|
||
}
|
||
|
||
use_custom_stoploss = True
|
||
trailing_stop = True
|
||
trailing_stop_positive = 0.005
|
||
trailing_stop_positive_offset = 0.01
|
||
trailing_only_offset_is_reached = True
|
||
plot_config = {
|
||
"main_plot": {},
|
||
"subplots": {
|
||
"Signals": {
|
||
"enter_long": {"color": "green"},
|
||
"exit_long": {"color": "red"}
|
||
},
|
||
"FreqAI Predictions": {
|
||
"&-buy_signal": {"color": "blue"},
|
||
"&-sell_signal": {"color": "orange"},
|
||
"&-volatility_forecast": {"color": "purple"}
|
||
}
|
||
}
|
||
}
|
||
|
||
freqai_info = {
|
||
"model": "LightGBMClassifier",
|
||
"feature_parameters": {
|
||
"include_timeframes": ["5m", "15m", "1h"],
|
||
"label_period_candles": 24,
|
||
"include_shifted_candles": 3,
|
||
},
|
||
"data_split_parameters": {
|
||
"test_size": 0.2,
|
||
"shuffle": False,
|
||
},
|
||
"model_training_parameters": {
|
||
"n_estimators": 200,
|
||
"learning_rate": 0.05,
|
||
"num_leaves": 31,
|
||
"verbose": -1,
|
||
},
|
||
}
|
||
def __init__(self, config: dict, *args, **kwargs):
|
||
super().__init__(config, *args, **kwargs)
|
||
|
||
# 设置日志级别为 DEBUG
|
||
logger.setLevel(logging.DEBUG)
|
||
logger.debug("✅ 策略已初始化，日志级别设置为 DEBUG")
|
||
def feature_engineering_expand_all(self, dataframe: DataFrame, period: int, metadata: dict, **kwargs) -> DataFrame:
|
||
# 保留原有指标
|
||
dataframe["%-rsi-period"] = ta.RSI(dataframe, timeperiod=period)
|
||
dataframe["%-sma-period"] = ta.SMA(dataframe, timeperiod=period)
|
||
dataframe["%-ema-period"] = ta.EMA(dataframe, timeperiod=period)
|
||
|
||
real = ta.TYPPRICE(dataframe)
|
||
upperband, middleband, lowerband = ta.BBANDS(real, timeperiod=period, nbdevup=2.0, nbdevdn=2.0)
|
||
dataframe["bb_lowerband-period"] = lowerband
|
||
dataframe["bb_upperband-period"] = upperband
|
||
dataframe["bb_middleband-period"] = middleband
|
||
dataframe["%-bb_width-period"] = (dataframe["bb_upperband-period"] - dataframe["bb_lowerband-period"]) / dataframe["bb_middleband-period"]
|
||
dataframe["%-roc-period"] = ta.ROC(dataframe, timeperiod=period)
|
||
|
||
# 添加策略 B 的指标
|
||
dataframe["%-mfi-period"] = ta.MFI(dataframe, timeperiod=period) # 资金流指标
|
||
dataframe["%-adx-period"] = ta.ADX(dataframe, timeperiod=period) # 趋势强度
|
||
dataframe["%-tema-period"] = ta.TEMA(dataframe, timeperiod=period) # TEMA 指标
|
||
dataframe["%-relative_volume-period"] = dataframe["volume"] / dataframe["volume"].rolling(period).mean() # 相对成交量
|
||
dataframe["%-close-bb_lower-period"] = dataframe["close"] / dataframe["bb_lowerband-period"] # 收盘价/布林带下轨
|
||
|
||
# 数据清理
|
||
columns_to_clean = [
|
||
"%-rsi-period", "%-mfi-period", "%-sma-period", "%-ema-period", "%-adx-period",
|
||
"bb_lowerband-period", "bb_middleband-period", "bb_upperband-period",
|
||
"%-bb_width-period", "%-roc-period", "%-relative_volume-period", "%-close-bb_lower-period", "%-tema-period"
|
||
]
|
||
for col in columns_to_clean:
|
||
dataframe[col] = dataframe[col].replace([np.inf, -np.inf], 0).ffill().fillna(0)
|
||
|
||
return dataframe
|
||
|
||
def set_freqai_targets(self, dataframe: DataFrame, metadata: dict, **kwargs) -> DataFrame:
|
||
"""
|
||
使用历史窗口预测未来趋势和波动率作为辅助信号，
|
||
避免使用任何未来数据（如 shift(-N)）
|
||
"""
|
||
|
||
label_period = self.freqai_info["feature_parameters"]["label_period_candles"]
|
||
|
||
# 1. 趋势强度：使用过去 N 根 K 线的收益率判断当前趋势
|
||
dataframe["&-trend_strength"] = dataframe["close"].pct_change(label_period)
|
||
|
||
# 2. 波动率预测：使用过去 N 根 K 线的价格变动绝对值
|
||
dataframe["&-volatility_forecast"] = dataframe["close"].pct_change(label_period).abs()
|
||
|
||
# 3. 动态 ROI 目标：基于趋势强度缩放
|
||
dataframe["&-roi_target"] = np.where(
|
||
dataframe["&-trend_strength"] > 0,
|
||
dataframe["&-trend_strength"] * 1.5,
|
||
0.01 # 最小 ROI
|
||
)
|
||
|
||
# 4. 动态止损目标：基于波动率缩放
|
||
dataframe["&-stoploss_target"] = -dataframe["&-volatility_forecast"] * 1.2
|
||
|
||
# 👇 新增：z-score 标准化模型输出
|
||
if self.freqai and hasattr(self.freqai, "data"):
|
||
labels_mean = self.freqai.data.labels_mean
|
||
labels_std = self.freqai.data.labels_std
|
||
|
||
mean_roi = labels_mean.get("&-roi_target", 0.01)
|
||
std_roi = labels_std.get("&-roi_target", 0.01)
|
||
mean_stop = labels_mean.get("&-stoploss_target", -0.01)
|
||
std_stop = labels_std.get("&-stoploss_target", 0.01)
|
||
|
||
# z-score 标准化
|
||
dataframe["&-roi_target_z"] = (dataframe["&-roi_target"] - mean_roi) / std_roi
|
||
dataframe["&-stoploss_target_z"] = (dataframe["&-stoploss_target"] - mean_stop) / std_stop
|
||
|
||
# 市场状态识别：判断当前是震荡、趋势、超买还是超卖
|
||
if "bb_upperband-period" in dataframe.columns and "bb_lowerband-period" in dataframe.columns:
|
||
dataframe["&-market_condition"] = np.select([
|
||
(dataframe["close"] > dataframe["bb_upperband-period"]),
|
||
(dataframe["close"] < dataframe["bb_lowerband-period"]),
|
||
(dataframe["&-trend_strength"] > 0.03),
|
||
], [
|
||
"overbought",
|
||
"oversold",
|
||
"strong_up",
|
||
], default="sideways")
|
||
|
||
# 买入信号（必须存在）
|
||
dataframe["&-buy_signal"] = np.where(dataframe["&-trend_strength"] > 0.01, 1, 0)
|
||
|
||
# 卖出信号（必须存在）
|
||
dataframe["&-sell_signal"] = np.where(dataframe["&-trend_strength"] < -0.01, 1, 0)
|
||
|
||
return dataframe
|
||
|
||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
dataframe = self.freqai.start(dataframe, metadata, self)
|
||
|
||
# 缓存 mean/std 供后续使用
|
||
if self.freqai and hasattr(self.freqai, "data"):
|
||
labels_mean = self.freqai.data.labels_mean
|
||
labels_std = self.freqai.data.labels_std
|
||
|
||
dataframe["mean_trend"] = labels_mean.get("&-trend_strength", 0.0001)
|
||
dataframe["std_trend"] = labels_std.get("&-trend_strength", 0.0031)
|
||
|
||
dataframe["mean_volatility"] = labels_mean.get("&-volatility_forecast", 0.0021)
|
||
dataframe["std_volatility"] = labels_std.get("&-volatility_forecast", 0.0022)
|
||
|
||
dataframe["mean_roi"] = labels_mean.get("&-roi_target", 0.0065)
|
||
dataframe["std_roi"] = labels_std.get("&-roi_target", 0.0041)
|
||
|
||
dataframe["mean_stop"] = labels_mean.get("&-stoploss_target", -0.0026)
|
||
dataframe["std_stop"] = labels_std.get("&-stoploss_target", 0.0027)
|
||
|
||
return dataframe
|
||
|
||
def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
||
"""
|
||
基于 FreqAI 的预测结果生成入场信号。
|
||
使用滚动分位数 + Z-score + 市场状态过滤，提高信号质量。
|
||
"""
|
||
|
||
conditions = []
|
||
|
||
# 确保 &-buy_signal 存在
|
||
if "&-buy_signal" in df.columns:
|
||
|
||
# 使用过去 20 根 K 线中 buy_signal 的 80 分位作为动态阈值
|
||
df["buy_threshold"] = df["&-buy_signal"].rolling(20).quantile(0.8)
|
||
buy_condition = df["&-buy_signal"] >= df["buy_threshold"]
|
||
|
||
# 使用 Z-score 判断趋势显著性（仅保留趋势较强的时间点）
|
||
if "mean_trend" in df.columns and "std_trend" in df.columns:
|
||
mean_trend = df["mean_trend"].iloc[-1]
|
||
std_trend = df["std_trend"].iloc[-1]
|
||
|
||
z_score_trend = (df["&-trend_strength"] - mean_trend) / std_trend
|
||
buy_condition &= (z_score_trend > 1) # 只保留趋势显著的信号
|
||
|
||
# 市场状态过滤：震荡市减少交易频率
|
||
if "&-market_condition" in df.columns:
|
||
buy_condition &= ~(
|
||
(df["&-market_condition"] == "sideways") &
|
||
(df["&-buy_signal"] < df["buy_threshold"] * 1.2)
|
||
)
|
||
|
||
# 在超买市场中进一步过滤买入信号
|
||
buy_condition &= ~(df["&-market_condition"] == "overbought")
|
||
|
||
# EMA 趋势过滤：只在价格高于 EMA50 时开仓
|
||
df["ema50"] = ta.EMA(df, timeperiod=50)
|
||
buy_condition &= (df["close"] > df["ema50"])
|
||
|
||
# 成交量过滤：成交量需大于 20 日均值的 1 倍
|
||
df["vol_mean"] = df["volume"].rolling(20).mean()
|
||
buy_condition &= (df["volume"] > df["vol_mean"] * 1)
|
||
|
||
conditions.append(buy_condition)
|
||
else:
|
||
self.logger.warning("⚠️ &-buy_signal 列缺失，跳过该条件")
|
||
|
||
# 如果有多个条件，使用逻辑与合并
|
||
if len(conditions):
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1
|
||
|
||
return df
|
||
def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
||
# Compute sell_signal_high outside the conditions list
|
||
|
||
pair = metadata['pair']
|
||
trade = Trade.get_trades([("pair", pair)]).first()
|
||
|
||
if trade and trade.is_open:
|
||
open_candle_index = df.index.get_loc(trade.open_date.replace(tzinfo=None))
|
||
current_index = len(df) - 1
|
||
holding_period = current_index - open_candle_index
|
||
|
||
if holding_period < 5:
|
||
df.loc[df.index[-1], 'exit_long'] = 0
|
||
return df
|
||
df["sell_signal_high"] = (df["&-sell_signal"] > 0.3).rolling(3).sum() >= 2
|
||
|
||
conditions = [
|
||
df["sell_signal_high"] # Append the condition, not an assignment
|
||
]
|
||
|
||
if "&-market_condition" in df.columns:
|
||
# 在超卖市场中减少卖出信号
|
||
conditions.append(df["&-market_condition"] != "oversold")
|
||
|
||
# 在超买市场中增强卖出信号
|
||
conditions.append(~((df["&-market_condition"] == "overbought") & (df["&-sell_signal"] < 0.7)))
|
||
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'exit_long'] = 1
|
||
return df
|
||
|
||
def adjust_trade_position(self, trade: Trade, current_time: datetime,
|
||
current_rate: float, current_profit: float,
|
||
min_roi: Dict[float, float], max_profit: float) -> Optional[Dict[float, float]]:
|
||
"""
|
||
动态调整持仓逻辑，基于 FreqAI 预测的 ROI 目标。
|
||
根据 Z-score 判断趋势强度，动态设置止盈时间窗口。
|
||
"""
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
|
||
last_candle = dataframe.iloc[-1]
|
||
|
||
# 获取预测的 ROI 目标 和 统计信息
|
||
roi_target = last_candle.get("&-roi_target", 0.0065)
|
||
mean_roi = last_candle.get("mean_roi", 0.0065)
|
||
std_roi = last_candle.get("std_roi", 0.0041)
|
||
|
||
# 计算 Z-score
|
||
z_score_roi = (roi_target - mean_roi) / std_roi if std_roi != 0 else 0
|
||
|
||
# 根据 Z-score 设置不同的 ROI 表
|
||
if z_score_roi > 1:
|
||
return {
|
||
"0": roi_target * 1.2,
|
||
"10": roi_target * 0.9,
|
||
"30": roi_target * 0.6,
|
||
"60": roi_target * 0.3,
|
||
"120": 0.01
|
||
}
|
||
elif z_score_roi > 0:
|
||
return {
|
||
"0": roi_target * 1.0,
|
||
"20": roi_target * 0.7,
|
||
"60": roi_target * 0.3,
|
||
"120": 0.01
|
||
}
|
||
else:
|
||
return {
|
||
"0": max(0.005, roi_target),
|
||
"60": 0.005,
|
||
"180": 0.005
|
||
}
|
||
|
||
def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
|
||
current_rate: float, current_profit: float, **kwargs) -> float:
|
||
"""
|
||
动态止损回调函数，基于 FreqAI 预测的 &-stoploss_target 值。
|
||
"""
|
||
|
||
# 获取当前币种的 dataframe
|
||
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
last_candle = dataframe.iloc[-1]
|
||
|
||
# 获取预测的止损目标（使用 z-score 版本）
|
||
stoploss_z = last_candle.get("&-stoploss_target_z", -1.0) # 默认值为 -1.0（一个标准差）
|
||
|
||
# 获取模型统计信息用于自适应判断
|
||
mean_stop = -0.01
|
||
std_stop = 0.01
|
||
|
||
if self.freqai and hasattr(self.freqai, "data"):
|
||
labels_mean = self.freqai.data.labels_mean
|
||
labels_std = self.freqai.data.labels_std
|
||
|
||
mean_stop = labels_mean.get("&-stoploss_target", mean_stop)
|
||
std_stop = labels_std.get("&-stoploss_target", std_stop)
|
||
|
||
# 获取币种特定参数
|
||
params = self.PAIR_PARAMS.get(pair, {"volatility_factor": 1.0, "min_stop": -0.005})
|
||
volatility_factor = params["volatility_factor"]
|
||
min_stop = params["min_stop"]
|
||
|
||
# 动态止损下限 = 平均值 - volatility_factor * 标准差
|
||
dynamic_min_stop = min(min_stop, mean_stop - volatility_factor * std_stop)
|
||
|
||
# 如果 z-score < -1，则说明当前止损偏移较大，适当放宽
|
||
if stoploss_z < -1.0:
|
||
dynamic_min_stop = min(dynamic_min_stop * 1.2, -0.003)
|
||
|
||
# 最终止损值 = 模型建议值 和 动态下限中的较小者
|
||
dynamic_stoploss = dynamic_min_stop
|
||
|
||
return dynamic_stoploss
|