186 lines
8.0 KiB
Python
186 lines
8.0 KiB
Python
from freqtrade.strategy import IStrategy
|
||
import talib.abstract as ta
|
||
import pandas as pd
|
||
import numpy as np
|
||
import logging
|
||
from technical import qtpylib
|
||
from functools import reduce
|
||
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter
|
||
|
||
class MyDynamicStrategy(IStrategy):
|
||
|
||
# --- 参数空间 ---
|
||
buy_rsi = IntParameter(10, 50, default=30)
|
||
sell_rsi = IntParameter(50, 90, default=70)
|
||
roi_0 = DecimalParameter(0.01, 0.2, default=0.03)
|
||
roi_1 = DecimalParameter(0.005, 0.1, default=0.015)
|
||
stoploss_param = DecimalParameter(-0.3, -0.1, default=-0.15)
|
||
|
||
trailing_stop = True
|
||
trailing_stop_positive = 0.05
|
||
trailing_stop_positive_offset = 0.1
|
||
can_short = False
|
||
|
||
process_only_new_candles = True
|
||
use_exit_signal = True
|
||
stoploss = -0.15
|
||
minimal_roi = {"0": 0.03, "30": 0.01, "60": 0}
|
||
|
||
# --- Plotting config ---
|
||
plot_config = {
|
||
"main_plot": {},
|
||
"subplots": {
|
||
"RSI Buy Threshold": {
|
||
"&-buy_rsi": {"color": "green"}
|
||
},
|
||
"ROI and Stoploss": {
|
||
"&-roi_0": {"color": "orange"},
|
||
"&-stoploss": {"color": "red"}
|
||
}
|
||
}
|
||
}
|
||
|
||
# --- FreqAI 配置 ---
|
||
freqai_info = {
|
||
"model": "CatboostClassifier",
|
||
"feature_parameters": {
|
||
"include_timeframes": ["5m", "1h"],
|
||
"indicator_periods_candles": [10, 20, 50],
|
||
"include_corr_pairlist": ["BTC/USDT"],
|
||
"target_classifier": "value",
|
||
"label_period_candles": 20,
|
||
},
|
||
"training_settings": {
|
||
"train_period_days": 30,
|
||
"startup_candle_count": 200
|
||
}
|
||
}
|
||
|
||
def feature_engineering_expand_all(self, dataframe, period, **kwargs):
|
||
df = dataframe.copy()
|
||
df[f'rsi_{period}'] = ta.RSI(df, timeperiod=period)
|
||
df[f'sma_diff_{period}'] = df['close'] - ta.SMA(df, timeperiod=period)
|
||
df[f'macd_{period}'], _, _ = ta.MACD(df, fastperiod=12, slowperiod=26, signalperiod=9)
|
||
df[f'stoch_rsi_{period}'] = ta.STOCHRSI(df, timeperiod=period)
|
||
df[f'cci_{period}'] = ta.CCI(df, timeperiod=period)
|
||
df[f'willr_{period}'] = ta.WILLR(df, timeperiod=period)
|
||
df[f'atr_{period}'] = ta.ATR(df, timeperiod=period)
|
||
df[f'price_change_rate_{period}'] = df['close'].pct_change(period)
|
||
df[f'volatility_{period}'] = df['close'].pct_change().rolling(window=period).std()
|
||
return df
|
||
|
||
def set_freqai_targets(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
||
# 使用短期和长期均线交叉作为目标标签
|
||
short_ma = ta.SMA(dataframe, timeperiod=10)
|
||
long_ma = ta.SMA(dataframe, timeperiod=50)
|
||
dataframe['target'] = np.where(short_ma > long_ma, 2,
|
||
np.where(short_ma < long_ma, 0, 1))
|
||
return dataframe
|
||
|
||
def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
||
# 示例：使用简单未来N周期涨跌作为目标变量
|
||
# 使用短期均线趋势代替未来价格
|
||
# 计算短期和长期均线
|
||
short_ma = ta.SMA(dataframe, timeperiod=10)
|
||
long_ma = ta.SMA(dataframe, timeperiod=50)
|
||
dataframe['short_ma'] = short_ma
|
||
dataframe['long_ma'] = long_ma
|
||
|
||
# 计算 RSI 和其他动态参数
|
||
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
|
||
dataframe['&-buy_rsi'] = self.buy_rsi.value
|
||
dataframe['&-sell_rsi'] = self.sell_rsi.value
|
||
dataframe['&-roi_0'] = self.roi_0.value
|
||
dataframe['&-stoploss'] = self.stoploss_param.value
|
||
|
||
# 添加调试日志
|
||
logging.info(f"Feature columns after feature engineering: {list(dataframe.columns)}")
|
||
|
||
# 使用短期和长期均线交叉作为目标标签
|
||
dataframe['target'] = np.where(short_ma > long_ma, 2,
|
||
np.where(short_ma < long_ma, 1, 0))
|
||
|
||
# 动态设置 minimal_roi
|
||
# 平滑处理 ROI 参数
|
||
# 基于波动率动态调整 ROI 参数
|
||
# 使用指数加权移动平均 (EWMA) 计算波动率
|
||
volatility = dataframe['close'].pct_change().ewm(span=20, adjust=False).std().mean()
|
||
roi_0_dynamic = max(0.01, min(0.2, self.roi_0.value * (1 + volatility)))
|
||
roi_1_dynamic = max(0.005, min(0.1, self.roi_1.value * (1 + volatility)))
|
||
self.minimal_roi = {
|
||
0: roi_0_dynamic,
|
||
30: roi_1_dynamic,
|
||
60: 0
|
||
}
|
||
# 动态调整止损距离
|
||
volatility_multiplier = max(1.5, min(3.0, 2.0 + volatility)) # 波动率倍数
|
||
self.stoploss = -0.15 * volatility_multiplier
|
||
|
||
# 计算 Bollinger Bands 并添加到 dataframe
|
||
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
|
||
dataframe['bollinger_upper'] = bollinger['upper']
|
||
dataframe['bollinger_mid'] = bollinger['mid']
|
||
dataframe['bollinger_lower'] = bollinger['lower']
|
||
|
||
# 计算 MACD 并添加到 dataframe
|
||
macd, macdsignal, _ = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
|
||
dataframe['macd'] = macd
|
||
dataframe['macdsignal'] = macdsignal
|
||
|
||
# 添加调试日志
|
||
logging.info(f"RSI condition: {(dataframe['rsi'] < dataframe['&-buy_rsi']).sum()}")
|
||
logging.info(f"Volume condition: {(dataframe['volume'] > dataframe['volume'].rolling(window=20).mean() * 1.05).sum()}")
|
||
logging.info(f"MACD condition: {((dataframe['close'] <= dataframe['bollinger_lower'] * 1.01) & (dataframe['macd'] > dataframe['macdsignal'])).sum()}")
|
||
|
||
|
||
# 添加 ADX 趋势过滤器
|
||
dataframe['adx'] = ta.ADX(dataframe, timeperiod=14)
|
||
is_strong_trend = dataframe['adx'].iloc[-1] > 25
|
||
# MACD 穿越信号条件
|
||
(dataframe["close"] < dataframe['bollinger_lower']) & (dataframe['macd'] > dataframe['macdsignal']),
|
||
|
||
# 基于趋势强度动态调整追踪止损
|
||
trend_strength = (dataframe['short_ma'] - dataframe['long_ma']).mean()
|
||
if is_strong_trend:
|
||
self.trailing_stop_positive = max(0.01, min(0.1, abs(trend_strength) * 0.3))
|
||
self.trailing_stop_positive_offset = max(0.01, min(0.2, abs(trend_strength) * 0.6))
|
||
else:
|
||
self.trailing_stop_positive = 0.05
|
||
self.trailing_stop_positive_offset = 0.1
|
||
trend_strength = (dataframe['short_ma'] - dataframe['long_ma']).mean()
|
||
self.trailing_stop_positive = max(0.01, min(0.1, abs(trend_strength) * 0.3))
|
||
|
||
return dataframe
|
||
def populate_entry_trend(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
||
# 增加成交量过滤和 Bollinger Bands 信号
|
||
# 计算 MACD
|
||
macd, macdsignal, _ = ta.MACD(df, fastperiod=12, slowperiod=26, signalperiod=9)
|
||
df['macd'] = macd
|
||
df['macdsignal'] = macdsignal
|
||
|
||
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(df), window=20, stds=2)
|
||
conditions = [
|
||
(df["rsi"] < df["&-buy_rsi"]), # RSI 低于买入阈值
|
||
(df["volume"] > df["volume"].rolling(window=20).mean() * 1.1), # 成交量增长超过 10%
|
||
(df["close"] < df['bollinger_lower']) & (df['macd'] > df['macdsignal']), # MACD 穿越信号
|
||
]
|
||
df.loc[reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1
|
||
return df
|
||
|
||
def populate_exit_trend(self, df: pd.DataFrame, metadata: dict) -> pd.DataFrame:
|
||
# 增加 Bollinger Bands 中轨信号
|
||
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(df), window=20, stds=2)
|
||
exit_long_conditions = [
|
||
(df["rsi"] > df["&-sell_rsi"]),
|
||
(df["close"] > df['bollinger_mid']) # Bollinger Bands 中轨信号
|
||
]
|
||
df.loc[reduce(lambda x, y: x & y, exit_long_conditions), 'exit_long'] = 1
|
||
return df
|
||
def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs):
|
||
df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
|
||
last_candle = df.iloc[-1]
|
||
if rate > last_candle["close"] * 1.0025:
|
||
return False
|
||
return True
|
||
|