趋势跟踪盈利策略 - 专门在趋势市场中获利
Timeframe
15m
Direction
Long Only
Stoploss
-8.0%
Trailing Stop
Yes
ROI
0m: 1000.0%
Interface Version
3
Startup Candles
N/A
Indicators
3
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
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from pandas import DataFrame
from typing import Optional, Union
from collections import deque
from freqtrade.strategy import (
IStrategy,
Trade,
Order,
DecimalParameter,
IntParameter,
BooleanParameter,
)
import talib.abstract as ta
from technical import qtpylib
class TrendFollowProfitStrategy(IStrategy):
"""
趋势跟踪盈利策略 - 专门在趋势市场中获利
核心思想:
1. 识别趋势市场环境
2. 在趋势确认后进场
3. 金字塔加仓,长期持有
4. 追踪止盈,保护利润
"""
INTERFACE_VERSION = 3
can_short: bool = False
# 趋势跟踪设置
minimal_roi = {"0": 10} # 不使用固定ROI
stoploss = -0.08 # 8%止损
trailing_stop = True
trailing_stop_positive = 0.02 # 2%启动追踪
trailing_stop_positive_offset = 0.03 # 3%追踪距离
timeframe = '15m'
use_custom_stoploss = True
# 趋势识别参数
ema_fast = IntParameter(12, 21, default=15, space="buy", optimize=True)
ema_slow = IntParameter(26, 55, default=35, space="buy", optimize=True)
adx_period = IntParameter(10, 21, default=14, space="buy", optimize=True)
adx_threshold = IntParameter(25, 40, default=30, space="buy", optimize=True)
# 趋势确认参数
trend_confirmation_period = IntParameter(3, 8, default=5, space="buy", optimize=True)
momentum_threshold = DecimalParameter(0.003, 0.015, default=0.008, space="buy", optimize=True)
# 加仓参数
enable_pyramid = BooleanParameter(default=True, space="buy", optimize=True)
pyramid_distance = DecimalParameter(0.02, 0.05, default=0.03, space="buy", optimize=True)
max_pyramid_levels = IntParameter(2, 4, default=3, space="buy", optimize=True)
# 利润目标
profit_target_1 = DecimalParameter(0.05, 0.10, default=0.08, space="sell", optimize=True)
profit_target_2 = DecimalParameter(0.15, 0.25, default=0.20, space="sell", optimize=True)
profit_target_3 = DecimalParameter(0.30, 0.50, default=0.40, space="sell", optimize=True)
startup_candle_count: int = 200
def __init__(self, config: dict):
super().__init__(config)
self.pyramid_levels = {}
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
趋势跟踪指标
"""
# EMA系统
dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
dataframe['ema_trend'] = dataframe['ema_fast'] > dataframe['ema_slow']
# ADX趋势强度
dataframe['adx'] = ta.ADX(dataframe, timeperiod=self.adx_period.value)
dataframe['strong_trend'] = dataframe['adx'] > self.adx_threshold.value
# MACD动量确认
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macd_positive'] = dataframe['macd'] > dataframe['macdsignal']
# 动量计算
dataframe['momentum'] = dataframe['close'].pct_change(self.trend_confirmation_period.value)
dataframe['strong_momentum'] = dataframe['momentum'] > self.momentum_threshold.value
# 成交量确认
dataframe['volume_sma'] = dataframe['volume'].rolling(20).mean()
dataframe['volume_surge'] = dataframe['volume'] > dataframe['volume_sma'] * 1.2
# 综合趋势信号
dataframe['trend_score'] = (
dataframe['ema_trend'].astype(int) +
dataframe['strong_trend'].astype(int) +
dataframe['macd_positive'].astype(int) +
dataframe['strong_momentum'].astype(int) +
dataframe['volume_surge'].astype(int)
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
趋势跟踪入场
"""
dataframe['enter_long'] = (
# 强趋势信号:至少4个确认
(dataframe['trend_score'] >= 4) &
# EMA排列确认
(dataframe['ema_fast'] > dataframe['ema_slow']) &
# 价格在EMA之上
(dataframe['close'] > dataframe['ema_fast']) &
# ADX确认强趋势
(dataframe['strong_trend'] == True) &
# 动量确认
(dataframe['strong_momentum'] == True)
)
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
趋势跟踪出场
"""
dataframe['exit_long'] = (
# 趋势转弱
(dataframe['ema_fast'] < dataframe['ema_slow']) |
# ADX下降
(dataframe['adx'] < self.adx_threshold.value * 0.8) |
# MACD转负
(dataframe['macd'] < dataframe['macdsignal']) |
# 动量转负
(dataframe['momentum'] < -self.momentum_threshold.value)
)
return dataframe
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
动态追踪止损
"""
# 分阶段追踪止损
if current_profit >= self.profit_target_3.value:
return current_profit - 0.15 # 保护85%利润
elif current_profit >= self.profit_target_2.value:
return current_profit - 0.08 # 保护92%利润
elif current_profit >= self.profit_target_1.value:
return current_profit - 0.04 # 保护96%利润
elif current_profit >= 0.03:
return current_profit - 0.02 # 保护部分利润
else:
return self.stoploss
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] = None, side: str = 'long', **kwargs) -> bool:
"""
趋势确认入场
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
if dataframe.empty:
return False
latest = dataframe.iloc[-1]
# 确认趋势信号强度
if latest['trend_score'] < 3:
return False
return True