多空双开对冲策略 V1 (Dual Position Hedge Strategy)
Timeframe
5m
Direction
Long & Short
Stoploss
-8.0%
Trailing Stop
No
ROI
0m: 4.0%, 15m: 2.5%, 30m: 2.0%, 60m: 1.5%
Interface Version
3
Startup Candles
200
Indicators
6
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
import numpy as np
import talib.abstract as ta
import pandas as pd
from pandas import DataFrame
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, merge_informative_pair
from datetime import datetime
from typing import Optional
import logging
logger = logging.getLogger(__name__)
class DualPositionHedgeV1(IStrategy):
"""
多空双开对冲策略 V1 (Dual Position Hedge Strategy)
核心思路:
1. 在关键位置同时开多空单,锁定风险
2. 根据趋势方向,逐步平掉反向单子
3. 保留顺势单子,获取趋势利润
4. 通过对冲机制,实现永不爆仓
适用场景:
- 震荡行情中的突破
- 重要支撑阻力位
- 不确定方向但预期有大波动的时候
"""
INTERFACE_VERSION = 3
timeframe = '5m'
can_short = True
# 策略参数 - 回到第一轮优化的最佳配置
rsi_overbought = IntParameter(70, 85, default=78, space='sell') # 回到最优设置
rsi_oversold = IntParameter(15, 30, default=22, space='buy') # 回到最优设置
ema_period = IntParameter(20, 50, default=25, space='buy') # 保持不变
volume_threshold = DecimalParameter(1.5, 2.5, default=2.0, space='buy') # 回到最优设置
# 新增动态止损参数
dynamic_stoploss = DecimalParameter(0.05, 0.12, default=0.07, space='buy') # 动态止损
# 风险控制 - 最优版本设置
stoploss = -0.08 # 8%止损,最优版本
minimal_roi = {
"0": 0.04, # 4% 目标利润(降低以增加获利机会)
"15": 0.025, # 15分钟 2.5%
"30": 0.02, # 30分钟 2%
"60": 0.015, # 1小时 1.5%
"120": 0.01 # 2小时 1%
}
# 不使用追踪止损,手动管理仓位
trailing_stop = False
process_only_new_candles = True
startup_candle_count = 200
custom_leverage = 6.0 # 最优版本杠杆
def leverage(self, pair: str, current_time: datetime, current_rate: float,
proposed_leverage: float, max_leverage: float, entry_tag: str, side: str,
**kwargs) -> float:
return self.custom_leverage
def custom_stoploss(self, pair: str, trade, current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
动态止损:根据持仓时间和盈利情况调整止损
"""
# 获取持仓时间(分钟)
trade_duration = (current_time - trade.open_date_utc).total_seconds() / 60
# 如果已经盈利,使用更紧的止损保护利润
if current_profit > 0.02: # 盈利超过2%
return -0.015 # 1.5%止损保护利润
elif current_profit > 0.01: # 盈利超过1%
return -0.03 # 3%止损
elif trade_duration > 60: # 持仓超过1小时
return -0.06 # 6%止损,避免长时间亏损
else:
return self.stoploss # 使用默认7%止损
def informative_pairs(self) -> list:
pairs = self.dp.current_whitelist()
return [(pair, '1h') for pair in pairs] # 改用1小时数据
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# === 5分钟指标 ===
dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_period.value)
dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_period.value * 2)
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20)
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
# ATR for volatility
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
dataframe['atr_ratio'] = dataframe['atr'] / dataframe['close']
# Bollinger Bands for exit signals
bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0)
dataframe['bb_upper'] = bollinger['upperband']
dataframe['bb_middle'] = bollinger['middleband']
dataframe['bb_lower'] = bollinger['lowerband']
# === 1小时趋势指标 ===
inf_tf = '1h'
informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=inf_tf)
informative['ema_trend'] = ta.EMA(informative, timeperiod=50)
informative['rsi_trend'] = ta.RSI(informative, timeperiod=14)
informative['macd'], informative['macdsignal'], informative['macdhist'] = ta.MACD(informative)
# 趋势强度判断
informative['trend_strength'] = np.where(
informative['close'] > informative['ema_trend'], 1, # 上涨趋势
np.where(informative['close'] < informative['ema_trend'], -1, 0) # 下跌趋势
)
dataframe = merge_informative_pair(dataframe, informative, self.timeframe, inf_tf, ffill=True)
# 重命名列
dataframe['trend_strength'] = dataframe[f'trend_strength_{inf_tf}']
dataframe['rsi_trend'] = dataframe[f'rsi_trend_{inf_tf}']
dataframe['macd_trend'] = dataframe[f'macd_{inf_tf}']
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# === 双开条件:回到第一轮优化的最佳配置 ===
# 使用已验证有效的参数组合
dual_open_condition = (
# RSI极值 - 最优阈值
((dataframe['rsi'] >= self.rsi_overbought.value) |
(dataframe['rsi'] <= self.rsi_oversold.value)) &
# 成交量放大 - 最优要求
(dataframe['volume_ratio'] > self.volume_threshold.value) &
# 波动率较高 - 最优阈值
(dataframe['atr_ratio'] > 0.025) &
# 布林带位置确认 - 最优设置
((dataframe['close'] > dataframe['bb_upper'] * 0.98) | # 接近上轨
(dataframe['close'] < dataframe['bb_lower'] * 1.02)) # 接近下轨
)
# === 多单入场 ===
# 在双开条件下,如果趋势偏多,优先开多单
dataframe.loc[
(
dual_open_condition &
(
# 5分钟超跌反弹
(dataframe['rsi'] <= self.rsi_oversold.value) |
# 1小时趋势向上
(dataframe['trend_strength'] > 0) |
# EMA金叉
((dataframe['ema_fast'] > dataframe['ema_slow']) &
(dataframe['ema_fast'].shift(1) <= dataframe['ema_slow'].shift(1)))
)
),
'enter_long'] = 1
# === 空单入场 ===
# 在双开条件下,如果趋势偏空,优先开空单
dataframe.loc[
(
dual_open_condition &
(
# 5分钟超买回调
(dataframe['rsi'] >= self.rsi_overbought.value) |
# 1小时趋势向下
(dataframe['trend_strength'] < 0) |
# EMA死叉
((dataframe['ema_fast'] < dataframe['ema_slow']) &
(dataframe['ema_fast'].shift(1) >= dataframe['ema_slow'].shift(1)))
)
),
'enter_short'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
# === 多单离场 - 更积极的获利策略 ===
dataframe.loc[
(
# RSI超买但阈值更低,更早离场
(dataframe['rsi'] > self.rsi_overbought.value - 3) |
# 趋势转弱
(dataframe['trend_strength'] < 0) |
# 价格接近布林带中轨(获利回吐信号)
(dataframe['close'] < dataframe['bb_middle']) |
# EMA死叉信号
((dataframe['ema_fast'] < dataframe['ema_slow']) &
(dataframe['ema_fast'].shift(1) >= dataframe['ema_slow'].shift(1)))
),
'exit_long'] = 1
# === 空单离场 - 更积极的获利策略 ===
dataframe.loc[
(
# RSI超卖但阈值更高,更早离场
(dataframe['rsi'] < self.rsi_oversold.value + 3) |
# 趋势转强
(dataframe['trend_strength'] > 0) |
# 价格接近布林带中轨(获利回吐信号)
(dataframe['close'] > dataframe['bb_middle']) |
# EMA金叉信号
((dataframe['ema_fast'] > dataframe['ema_slow']) &
(dataframe['ema_fast'].shift(1) <= dataframe['ema_slow'].shift(1)))
),
'exit_short'] = 1
return dataframe
def custom_entry_price(self, pair: str, current_time: datetime, proposed_rate: float,
entry_tag: Optional[str], side: str, **kwargs) -> float:
"""
自定义入场价格,确保在关键位置入场
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty:
return proposed_rate
last_candle = dataframe.iloc[-1]
if side == 'long':
# 多单稍微激进一点入场
return proposed_rate * 1.001
else:
# 空单稍微激进一点入场
return proposed_rate * 0.999
def custom_exit_price(self, pair: str, trade, current_time: datetime,
proposed_rate: float, current_profit: float, exit_tag: Optional[str],
**kwargs) -> float:
"""
自定义离场价格,确保及时离场
"""
if current_profit > 0.02: # 利润超过2%时,稍微保守离场
if trade.is_short:
return proposed_rate * 1.001 # 空单稍高价格平仓
else:
return proposed_rate * 0.999 # 多单稍低价格平仓
return proposed_rate
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: Optional[str],
side: str, **kwargs) -> bool:
"""
确认交易入场,可以在这里实现更复杂的双开逻辑
"""
# 这里可以添加额外的风控逻辑
# 比如检查当前持仓情况,确保不会过度开仓
return True