175 lines
6.3 KiB
Python
175 lines
6.3 KiB
Python
"""
|
||
SOL5mStrategy - 基于 EMA26_EMA52_Cross 的改进版
|
||
|
||
核心改进（相比原版）：
|
||
★ 去掉了反向交叉退出信号（原版中这是最大亏损来源，206笔亏-3021 USDT）
|
||
★ 加入 trailing stop 保护利润
|
||
★ 只靠 ROI + trailing stop + 硬止损 管理退出
|
||
|
||
逻辑：
|
||
- 底层使用 1m K线（由 config 中 timeframe: "1m" 控制）
|
||
- resample 到 30m 计算 EMA26/EMA52 交叉
|
||
- 交叉后延迟 30 根 1m K线入场（等待确认）
|
||
- ROI 从 15% 逐步递减
|
||
- Trailing stop：盈利 6% 后激活，回撤 3% 退出
|
||
- 硬止损 -15%（安全网）
|
||
|
||
使用命令：
|
||
freqtrade backtesting -c ./user_data/Chan/config/Local_Test.json \
|
||
--strategy SOL5mStrategy --strategy-path ./user_data/Chan/strategies \
|
||
--timerange=20250301-
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
import talib.abstract as ta
|
||
from pandas import DataFrame
|
||
from technical.util import resample_to_interval, resampled_merge
|
||
|
||
from freqtrade.strategy import IStrategy
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SOL5mStrategy(IStrategy):
|
||
INTERFACE_VERSION: int = 3
|
||
|
||
# === 基础配置 ===
|
||
# 注意：实际 timeframe 由 config 文件中的 "timeframe": "1m" 控制
|
||
# 这里不设置 timeframe，让 config 覆盖
|
||
can_short = True
|
||
startup_candle_count: int = 1600
|
||
|
||
# 硬止损 -3%（超短线合理止损，配合更严格的入场过滤）
|
||
stoploss = -0.03
|
||
use_custom_stoploss = False
|
||
|
||
# Trailing stop：盈利 3% 后激活，回撤 1.5% 退出
|
||
trailing_stop = True
|
||
trailing_stop_positive = 0.015 # 回撤 1.5% 触发退出
|
||
trailing_stop_positive_offset = 0.03 # 盈利 3% 后才开始追踪
|
||
trailing_only_offset_is_reached = True
|
||
|
||
# ROI：从 6% 逐步递减（给盈利交易更多空间）
|
||
minimal_roi = {
|
||
"0": 0.06, # 6% 立即止盈
|
||
"60": 0.04, # 60分钟后 4%
|
||
"120": 0.03, # 120分钟后 3%
|
||
"240": 0.02, # 240分钟后 2%
|
||
"480": 0.01, # 480分钟后 1%
|
||
"720": 0, # 720分钟后不设止盈
|
||
}
|
||
|
||
order_types = {
|
||
"entry": "market",
|
||
"exit": "market",
|
||
"stoploss": "market",
|
||
"stoploss_on_exchange": False,
|
||
}
|
||
|
||
# resample 时间倍数
|
||
time15 = 15
|
||
time30 = 30
|
||
time60 = 60
|
||
|
||
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
"""在 15m / 30m / 60m 级别计算 EMA26/52 交叉信号"""
|
||
ticker = self.get_ticker_indicator()
|
||
|
||
# resample 到更大时间框架
|
||
dataframe_15m = resample_to_interval(dataframe, ticker * self.time15)
|
||
dataframe_30m = resample_to_interval(dataframe, ticker * self.time30)
|
||
dataframe_60m = resample_to_interval(dataframe, ticker * self.time60)
|
||
|
||
# 在每个时间框架上计算指标
|
||
dataframe_15m = self.add_indicators(dataframe_15m)
|
||
dataframe_30m = self.add_indicators(dataframe_30m)
|
||
dataframe_60m = self.add_indicators(dataframe_60m)
|
||
dataframe = self.add_indicators(dataframe)
|
||
|
||
# 合并回 1m dataframe
|
||
dataframe = resampled_merge(dataframe, dataframe_15m)
|
||
dataframe = resampled_merge(dataframe, dataframe_30m)
|
||
dataframe = resampled_merge(dataframe, dataframe_60m)
|
||
|
||
return dataframe
|
||
|
||
def add_indicators(self, dataframe: DataFrame) -> DataFrame:
|
||
"""计算 EMA26/52 及其交叉信号，以及RSI和成交量过滤"""
|
||
dataframe["ema26"] = ta.EMA(dataframe, timeperiod=26)
|
||
dataframe["ema52"] = ta.EMA(dataframe, timeperiod=52)
|
||
|
||
# RSI用于确认趋势强度
|
||
dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
|
||
|
||
# 成交量均线用于确认成交量
|
||
dataframe["volume_mean"] = dataframe["volume"].rolling(window=20).mean()
|
||
|
||
# 上穿：本根 EMA26 > EMA52，上一根 EMA26 ≤ EMA52
|
||
dataframe["ema26_cross_up_52"] = (
|
||
(dataframe["ema26"] > dataframe["ema52"])
|
||
& (dataframe["ema26"].shift(1) <= dataframe["ema52"].shift(1))
|
||
)
|
||
|
||
# 下穿：本根 EMA26 < EMA52，上一根 EMA26 ≥ EMA52
|
||
dataframe["ema26_cross_down_52"] = (
|
||
(dataframe["ema26"] < dataframe["ema52"])
|
||
& (dataframe["ema26"].shift(1) >= dataframe["ema52"].shift(1))
|
||
)
|
||
|
||
return dataframe
|
||
|
||
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
"""30m EMA26/52 交叉入场，延迟 15 根 1m K线，加入RSI和成交量确认"""
|
||
ticker = self.get_ticker_indicator()
|
||
time = self.time30
|
||
delay = 15 # 减少延迟从30到15分钟
|
||
cross_up = f"resample_{ticker * time}_ema26_cross_up_52"
|
||
cross_down = f"resample_{ticker * time}_ema26_cross_down_52"
|
||
|
||
# 获取当前时间框架的RSI和成交量
|
||
rsi_col = "rsi"
|
||
volume_col = "volume"
|
||
volume_mean_col = "volume_mean"
|
||
|
||
# 做多：30m EMA26 上穿 EMA52 + RSI > 50（确认上涨趋势）+ 成交量确认
|
||
dataframe.loc[
|
||
(dataframe[cross_up].shift(delay) == True) &
|
||
(dataframe[rsi_col] > 50) & # RSI确认上涨趋势
|
||
(dataframe[volume_col] > dataframe[volume_mean_col]), # 成交量确认
|
||
["enter_long", "enter_tag"],
|
||
] = (1, "ema26x52_long")
|
||
|
||
# 做空：30m EMA26 下穿 EMA52 + RSI < 50（确认下跌趋势）+ 成交量确认
|
||
dataframe.loc[
|
||
(dataframe[cross_down].shift(delay) == True) &
|
||
(dataframe[rsi_col] < 50) & # RSI确认下跌趋势
|
||
(dataframe[volume_col] > dataframe[volume_mean_col]), # 成交量确认
|
||
["enter_short", "enter_tag"],
|
||
] = (1, "ema26x52_short")
|
||
|
||
return dataframe
|
||
|
||
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
||
"""不使用信号退出，完全依赖 ROI / trailing stop / 硬止损"""
|
||
return dataframe
|
||
|
||
def leverage(
|
||
self,
|
||
pair: str,
|
||
current_time: datetime,
|
||
current_rate: float,
|
||
proposed_leverage: float,
|
||
max_leverage: float,
|
||
entry_tag: Optional[str],
|
||
side: str,
|
||
**kwargs,
|
||
) -> float:
|
||
return 1.0
|
||
|
||
def get_ticker_indicator(self) -> int:
|
||
"""获取 timeframe 的分钟数"""
|
||
return int(self.timeframe[:-1])
|