优化利润的安全策略 - 调整止损+提高盈亏比+趋势过滤
Timeframe
15m
Direction
Long Only
Stoploss
-6.0%
Trailing Stop
Yes
ROI
0m: 12.0%, 60m: 10.0%, 120m: 8.0%, 180m: 7.0%
Interface Version
3
Startup Candles
N/A
Indicators
7
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove这些 imports ---
import numpy as np
import pandas as pd
from datetime import datetime, timedelta, timezone
from pandas import DataFrame
from typing import Dict, Optional, Union, Tuple
from freqtrade.strategy import (
IStrategy,
Trade,
Order,
PairLocks,
informative, # @informative decorator
# Hyperopt Parameters
BooleanParameter,
CategoricalParameter,
DecimalParameter,
IntParameter,
RealParameter,
# timeframe helpers
timeframe_to_minutes,
timeframe_to_next_date,
timeframe_to_prev_date,
timeframe_to_seconds,
timeframe_to_msecs,
)
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
class ProfitSafeStrategy(IStrategy):
"""
优化利润的安全策略 - 调整止损+提高盈亏比+趋势过滤
主要改进:
1. 止损从4%放宽到6%,减少过早止损
2. ROI从2%提高到3-4%,提高盈亏比
3. 加入MA趋势过滤,只做多头趋势
4. 分批止盈,拉高整体收益
5. 增加持仓时间,避免过早止损
"""
INTERFACE_VERSION = 3
# 策略时间框架
timeframe = "15m"
# 是否可以做空
can_short: bool = False
# ROI设置 - 提高盈亏比
minimal_roi = {
"1440": 0.02, # 24小时后2%收益
"720": 0.03, # 12小时后3%收益
"480": 0.04, # 8小时后4%收益
"360": 0.05, # 6小时后5%收益
"240": 0.06, # 4小时后6%收益
"180": 0.07, # 3小时后7%收益
"120": 0.08, # 2小时后8%收益
"60": 0.10, # 1小时后10%收益
"0": 0.12 # 立即12%收益
}
# 止损设置 - 从4%放宽到6%
stoploss = -0.06 # 6%止损(放宽)
# 追踪止损 - 更积极的利润保护
trailing_stop = True
trailing_stop_positive = 0.03 # 3%开始追踪
trailing_stop_positive_offset = 0.04 # 4%偏移
trailing_only_offset_is_reached = True
# 只处理新K线
process_only_new_candles = True
# 策略参数
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
# 策略启动所需的K线数量
startup_candle_count: int = 50
# 订单类型
order_types = {
"entry": "limit",
"exit": "limit",
"stoploss": "market",
"stoploss_on_exchange": False
}
# 订单时间
order_time_in_force = {
"entry": "GTC",
"exit": "GTC"
}
def informative_pairs(self):
return []
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
计算15分钟时间框架的指标,增加趋势过滤
"""
# RSI
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
# 多条MA - 用于趋势过滤
dataframe["ema_20"] = ta.EMA(dataframe, timeperiod=20)
dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
dataframe["sma_100"] = ta.SMA(dataframe, timeperiod=100) # 长期趋势
# MACD
macd = ta.MACD(dataframe)
dataframe["macd"] = macd["macd"]
dataframe["macdsignal"] = macd["macdsignal"]
dataframe["macdhist"] = macd["macdhist"]
# ADX - 趋势强度指标
dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
# 布林带
bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
dataframe["bb_lowerband"] = bollinger["lower"]
dataframe["bb_middleband"] = bollinger["mid"]
dataframe["bb_upperband"] = bollinger["upper"]
dataframe["bb_percent"] = (
(dataframe["close"] - dataframe["bb_lowerband"]) /
(dataframe["bb_upperband"] - dataframe["bb_lowerband"])
)
# ATR - 用于动态止损
dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
dataframe["atr_percent"] = dataframe["atr"] / dataframe["close"]
# 成交量指标
dataframe["volume_sma"] = dataframe["volume"].rolling(window=20).mean()
dataframe["volume_ratio"] = dataframe["volume"] / dataframe["volume_sma"]
# 价格动量
dataframe["momentum_3"] = dataframe["close"] / dataframe["close"].shift(3) - 1
dataframe["momentum_5"] = dataframe["close"] / dataframe["close"].shift(5) - 1
# 趋势强度指标
dataframe["trend_strength"] = (
(dataframe["ema_20"] > dataframe["ema_50"]).astype(int) +
(dataframe["ema_50"] > dataframe["sma_100"]).astype(int) +
(dataframe["close"] > dataframe["ema_20"]).astype(int) +
(dataframe["macd"] > dataframe["macdsignal"]).astype(int)
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
入场条件 - 趋势过滤+确保能开单
"""
# 基础条件(宽松)
base_conditions = (
(dataframe["volume"] > 0) &
(dataframe["rsi"].notna()) &
(dataframe["volume_ratio"] >= 0.3) # 成交量不低于平均的30%
)
# MA趋势过滤 - 只做多头趋势
trend_filter = (
(dataframe["ema_20"] > dataframe["ema_50"]) & # EMA20在EMA50之上
(dataframe["ema_50"] > dataframe["sma_100"]) # EMA50在SMA100之上
)
# RSI信号(宽松)
rsi_signal = (
(dataframe["rsi"] < 50) & # RSI低于50
(dataframe["rsi"] > dataframe["rsi"].shift(1)) # RSI上升
)
# 趋势确认(宽松)
trend_confirm = (
(dataframe["close"] > dataframe["ema_20"] * 0.95) # 价格在EMA20附近(5%容忍度)
)
# 动量确认(宽松)
momentum_confirm = (
(dataframe["momentum_3"] > -0.03) # 3周期内跌幅不超过3%
)
# 布林带位置(宽松)
bb_position = (
(dataframe["bb_percent"] > 0.05) # 不在布林带下轨附近
)
# 综合入场条件
dataframe.loc[
base_conditions &
trend_filter &
rsi_signal &
trend_confirm &
momentum_confirm &
bb_position,
"enter_long"
] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
出场条件 - 基于趋势转弱
"""
# 基础条件
base_conditions = (
(dataframe["volume"] > 0) &
(dataframe["rsi"].notna())
)
# 趋势转弱
trend_weak = (
(dataframe["ema_20"] < dataframe["ema_50"]) | # EMA20跌破EMA50
(dataframe["ema_50"] < dataframe["sma_100"]) # EMA50跌破SMA100
)
# RSI超买
rsi_overbought = (
(dataframe["rsi"] > 75) & # RSI超买
(dataframe["rsi"] < dataframe["rsi"].shift(1)) # RSI下降
)
# 价格跌破关键支撑
price_breakdown = (
(dataframe["close"] < dataframe["ema_20"] * 0.95) # 价格跌破EMA20(5%容忍度)
)
# 综合出场条件
dataframe.loc[
base_conditions &
(trend_weak | rsi_overbought | price_breakdown),
"exit_long"
] = 1
return dataframe
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
动态止损 - 从6%开始,根据盈利情况调整
"""
# 基于盈利情况的止损调整
if current_profit > 0.08: # 盈利超过8%
return -0.03 # 3%止损(保护大部分利润)
elif current_profit > 0.06: # 盈利超过6%
return -0.04 # 4%止损
elif current_profit > 0.04: # 盈利超过4%
return -0.05 # 5%止损
elif current_profit > 0.02: # 盈利超过2%
return -0.055 # 5.5%止损
# 基础止损
return self.stoploss # -0.06 (6%止损)
def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float,
current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
"""
自定义退出逻辑 - 分批止盈
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if len(dataframe) == 0:
return None
last_candle = dataframe.iloc[-1].squeeze()
# 分批止盈逻辑
if current_profit > 0.08: # 8%全部止盈
return "full_profit_exit"
elif current_profit > 0.06: # 6%部分止盈
# 检查是否应该部分止盈
if (last_candle.get("rsi", 50) > 70 or
last_candle.get("close", 0) < last_candle.get("ema_20", 0)):
return "partial_profit_exit"
# 如果趋势转弱且有利润,立即退出
if (current_profit > 0.03 and
(last_candle.get("ema_20", 0) < last_candle.get("ema_50", 0) or
last_candle.get("ema_50", 0) < last_candle.get("sma_100", 0))):
return "trend_weak_exit"
# 如果RSI极度超买且有足够利润,立即退出
if (last_candle.get("rsi", 50) > 80 and
current_profit > 0.04):
return "rsi_extreme_exit"
# 如果持仓时间过长且有利润,退出
if (current_profit > 0.02 and
(current_time - trade.open_date_utc).total_seconds() > 43200): # 12小时
return "time_profit_exit"
return None