143 lines
5.2 KiB
Python
143 lines
5.2 KiB
Python
import logging
|
||
import numpy as np
|
||
from functools import reduce
|
||
import talib.abstract as ta
|
||
from pandas import DataFrame
|
||
from technical import qtpylib
|
||
from freqtrade.strategy import IStrategy
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class FreqaiPrimer(IStrategy):
|
||
"""
|
||
策略说明：
|
||
- 所有交易信号由 FreqAI 动态预测
|
||
- 不使用 Hyperopt 优化任何参数
|
||
- 使用 FreqAI 提供的趋势和波动率信号进行交易决策
|
||
"""
|
||
|
||
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 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)
|
||
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
|
||
|
||
# 5. 市场状态识别：判断当前是震荡、趋势、超买还是超卖
|
||
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:
|
||
logger.info(f"[{metadata['pair']}] 当前可用列: {list(dataframe.columns)}")
|
||
dataframe = self.freqai.start(dataframe, metadata, self)
|
||
|
||
# FreqAI 提供的预测列
|
||
if "&-trend" in dataframe.columns:
|
||
dataframe["in_uptrend"] = dataframe["&-trend"] > 0.5
|
||
|
||
return dataframe
|
||
|
||
def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
||
conditions = []
|
||
|
||
if "&-buy_signal" in df.columns:
|
||
conditions.append(df["&-buy_signal"] > 0.5)
|
||
else:
|
||
logger.warning("⚠️ &-buy_signal 列缺失，跳过该条件")
|
||
|
||
if len(conditions) == 0:
|
||
return df
|
||
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1
|
||
return df
|
||
|
||
def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
|
||
conditions = [
|
||
df["&-sell_signal"] > 0.5,
|
||
]
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'exit_long'] = 1
|
||
return df
|