277 lines
11 KiB
Python
277 lines
11 KiB
Python
from functools import reduce
|
|
|
|
import talib.abstract as ta
|
|
from pandas import DataFrame
|
|
from freqtrade.strategy import IStrategy, merge_informative_pair
|
|
from freqtrade.persistence import Trade
|
|
from datetime import datetime
|
|
|
|
|
|
class TrendStructureExecutor(IStrategy):
|
|
"""
|
|
TrendStructureExecutor — Trend-continuation strategy (spot/futures).
|
|
|
|
Core concept:
|
|
Identify established trends on the 1h chart (EMA52 + MACD + EMA200),
|
|
then trade 5m continuation entries when the MACD histogram pulls back
|
|
to zero and resumes in the trend direction. Skip low-volatility
|
|
ranging markets. Partial take-profit on momentum weakening.
|
|
"""
|
|
|
|
INTERFACE_VERSION = 3
|
|
|
|
# =========================================================================
|
|
# CONFIGURATION
|
|
# =========================================================================
|
|
|
|
timeframe = "5m"
|
|
informative_timeframe = "1h"
|
|
|
|
# Futures support (long + short)
|
|
# Set can_short = True and switch config to futures mode (BTC/USDT:USDT)
|
|
# to enable short trading.
|
|
can_short = False
|
|
# trading_mode = "futures"
|
|
# margin_mode = "isolated"
|
|
|
|
# Risk management — fixed 0.8% stoploss (tighter than the 1% ROI target)
|
|
stoploss = -0.008
|
|
|
|
# Trailing stop to protect profits
|
|
trailing_stop = True
|
|
trailing_stop_positive = 0.004
|
|
trailing_stop_positive_offset = 0.012
|
|
trailing_only_offset_is_reached = True
|
|
|
|
# Position adjustment for partial take-profits
|
|
position_adjustment_enable = True
|
|
|
|
# ROI disabled — exits managed by trailing stop + partial TP + EMA52 breach
|
|
minimal_roi = {"0": 0.99}
|
|
|
|
# General settings
|
|
use_exit_signal = True
|
|
exit_profit_only = False
|
|
startup_candle_count = 200
|
|
process_only_new_candles = True
|
|
|
|
order_types = {
|
|
"entry": "limit",
|
|
"exit": "limit",
|
|
"stoploss": "market",
|
|
"stoploss_on_exchange": False,
|
|
}
|
|
|
|
# =========================================================================
|
|
# INFORMATIVE PAIRS
|
|
# =========================================================================
|
|
|
|
def informative_pairs(self):
|
|
pairs = self.dp.current_whitelist()
|
|
return [(pair, self.informative_timeframe) for pair in pairs]
|
|
|
|
# =========================================================================
|
|
# INDICATORS
|
|
# =========================================================================
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
1h: EMA52, EMA200, MACD, slope, range/consolidation, trend flags.
|
|
5m: MACD, histogram direction helpers.
|
|
"""
|
|
if self.dp:
|
|
informative = self.dp.get_pair_dataframe(
|
|
pair=metadata["pair"], timeframe=self.informative_timeframe
|
|
)
|
|
|
|
# --- EMA 52 ---
|
|
informative["ema_52"] = ta.EMA(informative, timeperiod=52)
|
|
|
|
# --- EMA 200 (super-trend filter) ---
|
|
informative["ema_200"] = ta.EMA(informative, timeperiod=200)
|
|
|
|
# EMA 52 slope (3-period ROC for noise reduction)
|
|
informative["ema_52_slope"] = (
|
|
informative["ema_52"] - informative["ema_52"].shift(3)
|
|
)
|
|
|
|
# --- MACD (12, 26, 9) ---
|
|
macd_1h = ta.MACD(informative)
|
|
informative["macd_hist_1h"] = macd_1h["macdhist"]
|
|
informative["macd_hist_1h_delta"] = (
|
|
informative["macd_hist_1h"] - informative["macd_hist_1h"].shift(1)
|
|
)
|
|
|
|
# --- Range / consolidation filter ---
|
|
# If the 20-candle price range is less than 1.5 %, the market is
|
|
# considered to be ranging and no entries are allowed.
|
|
informative["range_high_20"] = informative["high"].rolling(20).max()
|
|
informative["range_low_20"] = informative["low"].rolling(20).min()
|
|
informative["range_pct"] = (
|
|
(informative["range_high_20"] - informative["range_low_20"])
|
|
/ informative["range_low_20"]
|
|
)
|
|
informative["is_ranging"] = (informative["range_pct"] < 0.015).astype(int)
|
|
|
|
# --- LONG trend confirmation ---
|
|
# Price above EMA52 + EMA52 sloping up + MACD histogram positive
|
|
# + histogram not shrinking significantly (delta > -0.5 * rolling std)
|
|
informative["trend_bull"] = (
|
|
(informative["close"] > informative["ema_52"])
|
|
& (informative["close"] > informative["ema_200"])
|
|
& (informative["ema_52_slope"] > 0)
|
|
& (informative["macd_hist_1h"] > 0)
|
|
& (
|
|
informative["macd_hist_1h_delta"]
|
|
> -informative["macd_hist_1h"].rolling(20).std() * 0.5
|
|
)
|
|
).astype(int)
|
|
|
|
# --- SHORT trend confirmation ---
|
|
# Price below EMA52 + EMA52 sloping down + MACD histogram negative
|
|
# + histogram not expanding upward (delta < +0.5 * rolling std)
|
|
informative["trend_bear"] = (
|
|
(informative["close"] < informative["ema_52"])
|
|
& (informative["close"] < informative["ema_200"])
|
|
& (informative["ema_52_slope"] < 0)
|
|
& (informative["macd_hist_1h"] < 0)
|
|
& (
|
|
informative["macd_hist_1h_delta"]
|
|
< informative["macd_hist_1h"].rolling(20).std() * 0.5
|
|
)
|
|
).astype(int)
|
|
|
|
# Merge 1h → 5m (merge_informative_pair handles lookahead protection
|
|
# by shifting the higher-timeframe data by one candle)
|
|
dataframe = merge_informative_pair(
|
|
dataframe,
|
|
informative,
|
|
self.timeframe,
|
|
self.informative_timeframe,
|
|
ffill=True,
|
|
)
|
|
|
|
# --- 5m MACD ---
|
|
macd_5m = ta.MACD(dataframe)
|
|
dataframe["macd_hist_5m"] = macd_5m["macdhist"]
|
|
|
|
# Direction helpers (avoids repeating shift logic in entry/exit methods)
|
|
dataframe["macd_hist_5m_up"] = (
|
|
dataframe["macd_hist_5m"] > dataframe["macd_hist_5m"].shift(1)
|
|
)
|
|
dataframe["macd_hist_5m_down"] = (
|
|
dataframe["macd_hist_5m"] < dataframe["macd_hist_5m"].shift(1)
|
|
)
|
|
|
|
return dataframe
|
|
|
|
# =========================================================================
|
|
# ENTRY LOGIC
|
|
# =========================================================================
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
LONG: 1h bullish + 5m MACD hist pullback-then-resumption + recent reset.
|
|
SHORT: 1h bearish + 5m MACD hist pullback-then-resumption + recent reset.
|
|
Both skip ranging markets.
|
|
"""
|
|
# Columns from merge_informative_pair carry the _1h suffix
|
|
trend_bull = dataframe["trend_bull_1h"]
|
|
trend_bear = dataframe["trend_bear_1h"]
|
|
is_ranging = dataframe["is_ranging_1h"]
|
|
|
|
# ── LONG ──────────────────────────────────────────────────────────────
|
|
|
|
long_conditions = [
|
|
trend_bull == 1,
|
|
is_ranging == 0,
|
|
dataframe["macd_hist_5m_down"].shift(1) == True,
|
|
dataframe["macd_hist_5m_up"] == True,
|
|
dataframe["macd_hist_5m"] > 0,
|
|
dataframe["macd_hist_5m"].rolling(3).min() < 0,
|
|
]
|
|
|
|
dataframe.loc[
|
|
reduce(lambda a, b: a & b, long_conditions),
|
|
["enter_long", "enter_tag"],
|
|
] = (1, "long_continuation")
|
|
|
|
# ── SHORT ─────────────────────────────────────────────────────────────
|
|
|
|
short_conditions = [
|
|
trend_bear == 1,
|
|
is_ranging == 0,
|
|
dataframe["macd_hist_5m_up"].shift(1) == True,
|
|
dataframe["macd_hist_5m_down"] == True,
|
|
dataframe["macd_hist_5m"] < 0,
|
|
dataframe["macd_hist_5m"].rolling(3).max() > 0,
|
|
]
|
|
|
|
dataframe.loc[
|
|
reduce(lambda a, b: a & b, short_conditions),
|
|
["enter_short", "enter_tag"],
|
|
] = (1, "short_continuation")
|
|
|
|
return dataframe
|
|
|
|
# =========================================================================
|
|
# EXIT LOGIC
|
|
# =========================================================================
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
LONG: exit on 1h EMA52 breach (trend reversal).
|
|
SHORT (futures only): exit on 1h EMA52 breach.
|
|
"""
|
|
long_cond = dataframe["close"] < dataframe["ema_52_1h"]
|
|
dataframe.loc[long_cond, "exit_long"] = 1
|
|
dataframe.loc[long_cond, "exit_tag"] = "long_exit"
|
|
|
|
if self.can_short:
|
|
short_cond = dataframe["close"] > dataframe["ema_52_1h"]
|
|
dataframe.loc[short_cond, "exit_short"] = 1
|
|
dataframe.loc[short_cond, "exit_tag"] = "short_exit"
|
|
|
|
return dataframe
|
|
|
|
# =========================================================================
|
|
# POSITION ADJUSTMENT (Partial Take-Profit)
|
|
# =========================================================================
|
|
|
|
def adjust_trade_position(self, trade: Trade, current_time: datetime,
|
|
current_rate: float, current_profit: float,
|
|
min_stake: float | None, max_stake: float,
|
|
current_entry_rate: float, current_exit_rate: float,
|
|
current_entry_profit: float, current_exit_profit: float,
|
|
**kwargs) -> float | None:
|
|
"""
|
|
Sell 50% when MACD momentum weakens while in profit.
|
|
|
|
Fires once per trade (guarded by filled_exits). Exits half the
|
|
position when the 5m MACD histogram starts declining toward zero
|
|
while we are still above +0.5% profit.
|
|
"""
|
|
if current_profit <= 0.005:
|
|
return None
|
|
|
|
# Only one partial exit per trade
|
|
filled_exits = trade.select_filled_orders(trade.exit_side)
|
|
if filled_exits:
|
|
return None
|
|
|
|
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
|
|
if dataframe is None or len(dataframe) < 2:
|
|
return None
|
|
|
|
last = dataframe.iloc[-1]
|
|
prev = dataframe.iloc[-2]
|
|
|
|
if trade.is_short:
|
|
if last["macd_hist_5m"] < 0 and last["macd_hist_5m"] > prev["macd_hist_5m"]:
|
|
return -(trade.stake_amount / 2)
|
|
else:
|
|
if last["macd_hist_5m"] > 0 and last["macd_hist_5m"] < prev["macd_hist_5m"]:
|
|
return -(trade.stake_amount / 2)
|
|
|
|
return None
|