基于动态网格的交易策略,每个网格仅触发一次信号,带权重仓位分配。 支持 Freqtrade 的超参数优化。
Timeframe
1h
Direction
Long Only
Stoploss
-2.0%
Trailing Stop
No
ROI
0m: 4.0%
Interface Version
N/A
Startup Candles
N/A
Indicators
3
from freqtrade.strategy.interface import IStrategy
from pandas import DataFrame
from technical.indicators import ema, sma, atr
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, CategoricalParameter
class DynamicGridWeightStrategy(IStrategy):
"""
基于动态网格的交易策略,每个网格仅触发一次信号,带权重仓位分配。
支持 Freqtrade 的超参数优化。
"""
# 配置策略参数
minimal_roi = {"0": 0.04} # 固定收益
stoploss = -0.02 # 最大止损 10%
timeframe = '1h' # 时间周期为 1 小时
# 定义超参数范围(供 Hyperopt 优化使用)
grid_size = IntParameter(3, 10, default=5, space="buy", optimize=True) # 网格数量(上下各 3~10 个)
atr_multiplier = DecimalParameter(0.5, 3.0, default=1.5, space="buy", optimize=True) # ATR 倍数
atr_period = DecimalParameter(5, 100, default=14, space="atr", optimize=True) # ATR 计算周期
ma_type = CategoricalParameter(["ema", "sma"], default="ema", space="buy", optimize=True) # 均线类型
ma_period = IntParameter(20, 100, default=50, space="buy", optimize=True) # 均线周期
weight_mode = CategoricalParameter(["linear", "exponential"], default="linear", space="buy", optimize=True) # 权重模式
# 初始化时设置绘图配置
def __init__(self, config: dict) -> None:
super().__init__(config)
self.plot_config = {
'main_plot': {
'mid_line': {'color': 'blue'}, # 中间均线
},
'subplots': {
"ATR": {
'atr': {'color': 'orange'}, # ATR 值曲线
},
},
}
# 确保网格线绘制配置在初始化时写入
for i in range(1, self.grid_size.value + 1): # 注意这里使用 `grid_size.upper`获取最大网格值
self.plot_config['main_plot'][f"grid_upper_{i}"] = {'color': 'green'}
self.plot_config['main_plot'][f"grid_lower_{i}"] = {'color': 'red'}
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算所需指标:移动均线、ATR,以及动态网格。
"""
# 选择移动均线类型
if self.ma_type.value == "ema":
dataframe["mid_line"] = ema(dataframe["close"], self.ma_period.value)
else:
dataframe["mid_line"] = sma(dataframe["close"], self.ma_period.value)
# 计算 ATR,作为动态间距的基础
dataframe["atr"] = atr(dataframe, self.atr_period.value)
# 生成上下网格
for i in range(1, self.grid_size.value + 1):
dataframe[f"grid_upper_{i}"] = dataframe["mid_line"] + i * self.atr_multiplier.value * dataframe["atr"]
dataframe[f"grid_lower_{i}"] = dataframe["mid_line"] - i * self.atr_multiplier.value * dataframe["atr"]
# 初始化触发状态为 False
dataframe[f"triggered_upper_{i}"] = False
dataframe[f"triggered_lower_{i}"] = False
# 添加到 `plot_config` 中,确保线条显示在主图
self.plot_config['main_plot'][f"grid_upper_{i}"] = {'color': 'green'}
self.plot_config['main_plot'][f"grid_lower_{i}"] = {'color': 'red'}
# 计算网格权重
if self.weight_mode.value == "linear":
# 线性权重分配
dataframe[f"grid_weight_{i}"] = i
elif self.weight_mode.value == "exponential":
# 指数权重分配
dataframe[f"grid_weight_{i}"] = 2 ** i
return dataframe
def reset_trigger_status(self, dataframe: DataFrame) -> DataFrame:
"""
重置网格触发状态,当价格远离网格时。
"""
# 如果价格超过最高的上网格,重置所有上网格状态
all_upper_triggered = (
dataframe['close'] > dataframe[f'grid_upper_{self.grid_size.value}']
)
for i in range(1, self.grid_size.value + 1):
dataframe.loc[all_upper_triggered, f'triggered_upper_{i}'] = False
# 如果价格低于最低的下网格,重置所有下网格状态
all_lower_triggered = (
dataframe['close'] < dataframe[f'grid_lower_{self.grid_size.value}']
)
for i in range(1, self.grid_size.value + 1):
dataframe.loc[all_lower_triggered, f'triggered_lower_{i}'] = False
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
开仓逻辑:每个网格只触发一次,并基于权重进行仓位分配。
"""
# 先重置触发状态
dataframe = self.reset_trigger_status(dataframe)
# 初始化信号
dataframe['enter_long'] = False
dataframe['enter_short'] = False
dataframe['position_size'] = 0 # 初始化仓位大小为 0
for i in range(1, self.grid_size.value + 1):
# 检查下网格(开多信号)
cond_long = (
(dataframe['close'] <= dataframe[f'grid_lower_{i}']) & # 当前价格小于等于下网格
(~dataframe[f'triggered_lower_{i}']) # 并且网格未触发过
)
dataframe[f'long_grid_{i}'] = cond_long
dataframe['enter_long'] |= cond_long
dataframe.loc[cond_long, f'triggered_lower_{i}'] = True # 标记网格已触发
# 权重对应仓位
dataframe.loc[cond_long, "position_size"] = dataframe[f"grid_weight_{i}"]
# 检查上网格(开空信号)
cond_short = (
(dataframe['close'] >= dataframe[f'grid_upper_{i}']) & # 当前价格大于等于上网格
(~dataframe[f'triggered_upper_{i}']) # 并且网格未触发过
)
dataframe[f'short_grid_{i}'] = cond_short
dataframe['enter_short'] |= cond_short
dataframe.loc[cond_short, f'triggered_upper_{i}'] = True # 标记网格已触发
# 权重对应仓位
dataframe.loc[cond_short, "position_size"] = dataframe[f"grid_weight_{i}"]
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
平仓逻辑:在网格触发信号的另一侧平仓。
"""
dataframe['exit_long'] = False
dataframe['exit_short'] = False
# 检查平多信号(价格触碰上网格)
for i in range(1, self.grid_size.value + 1):
cond_exit_long = (dataframe['close'] >= dataframe[f'grid_upper_{i}'])
dataframe['exit_long'] |= cond_exit_long
# 检查平空信号(价格触碰下网格)
for i in range(1, self.grid_size.value + 1):
cond_exit_short = (dataframe['close'] <= dataframe[f'grid_lower_{i}'])
dataframe['exit_short'] |= cond_exit_short
return dataframe