# pragma pylint: disable=missing-docstring, invalid-name """ TrendShort — long+short futures strategy for Kraken Futures perps (1x, isolated). The long side IS TrendRiderRS (Supertrend up + close>EMA + RS>=0 + BTC bull gate). The short side is its mirror (Supertrend down + close 1.0): a fully-collateralized short, the "never get called by surprise" structure. Isolated margin walls off the rest. Env knobs (SH_*, TR_*) for cheap sweeps. Run in futures/isolated mode. """ import os import talib.abstract as ta import pandas_ta as pta import pandas as pd from pandas import DataFrame import numpy as np from freqtrade.strategy import IStrategy, merge_informative_pair, stoploss_from_open from freqtrade.persistence import Trade def _envf(name: str, default: float) -> float: v = os.environ.get(name) return float(v) if v not in (None, "") else default class TrendShort(IStrategy): INTERFACE_VERSION = 3 timeframe = "1h" process_only_new_candles = True can_short = True startup_candle_count = 240 minimal_roi = {"0": 10.0} # ride the trend; Supertrend flip is the exit stoploss = -0.10 use_exit_signal = True use_custom_stoploss = True # IDEA2: ATR stop when SH_ATR_MULT>0; else returns None (= fixed stoploss) # GRID MODE (off unless SH_GRID_ON=1): native scale-in/out position adjustment, enabled in __init__. position_adjustment_enable = False max_entry_position_adjustment = -1 def __init__(self, config: dict) -> None: super().__init__(config) self.stoploss = -_envf("SH_STOP", 0.10) # hard take-profit (off by default): exit any trade up >= SH_ROI. Caps upside; # use for "bank 5% and move on" tests. SH_ROI_T adds a time-decayed second rung. roi = _envf("SH_ROI", 0.0) if roi > 0: self.minimal_roi = {"0": roi} roi_t = _envf("SH_ROI_T", 0.0) # after SH_ROI_MIN candles, accept SH_ROI_T if roi_t > 0: self.minimal_roi[str(int(_envf("SH_ROI_MIN", 24)))] = roi_t # profit-lock trailing (off by default): trail SH_TRAIL once up SH_TRAIL_OFF if _envf("SH_TRAIL", 0.0) >= 1.0: self.trailing_stop = True self.trailing_stop_positive = _envf("SH_TRAIL_POS", 0.04) self.trailing_stop_positive_offset = _envf("SH_TRAIL_OFF", 0.08) self.trailing_only_offset_is_reached = True # GRID MODE: a neutral-band (|btc_dev|<=SH_REGIME_BAND) chop harvester implemented as a # native scale-in/out position. OFF unless SH_GRID_ON=1 -> the deployed trend behavior is # byte-identical when unset. The third regime branch: bull->long, bear->short, neutral->GRID. self.grid_on = _envf("SH_GRID_ON", 0.0) >= 1.0 if self.grid_on: self.position_adjustment_enable = True self.max_entry_position_adjustment = int(_envf("SH_GRID_LEVELS", 8)) # Static stoploss is a HARD floor in freqtrade (custom_stoploss can only TIGHTEN it). # Widen it to the grid's catastrophe backstop; custom_stoploss re-imposes the tight # SH_STOP on trend trades so the trend book is unaffected. self.stoploss = -max(_envf("SH_STOP", 0.10), _envf("SH_GRID_STOP", 0.25)) @property def protections(self): return [ {"method": "CooldownPeriod", "stop_duration_candles": int(_envf("TR_CD", 4))}, {"method": "MaxDrawdown", "lookback_period_candles": int(_envf("TR_DD_LB", 168)), "trade_limit": 10, "stop_duration_candles": int(_envf("TR_DD_STOP", 48)), "max_allowed_drawdown": _envf("TR_DD_MAX", 0.12), # "ratios" (default) sums per-trade %returns => over-states drawdown when positions # are small vs the account; "equity" measures REAL account-balance drawdown. "calculation_mode": os.environ.get("TR_DD_MODE", "ratios")}, {"method": "StoplossGuard", "lookback_period_candles": int(_envf("TR_SG_LB", 72)), "trade_limit": int(_envf("TR_SG_LIM", 6)), "stop_duration_candles": int(_envf("TR_SG_STOP", 48)), "only_per_pair": False}, ] st_len, st_mult, ema_len = 10, 4.0, 200 # mult4 = deployed/validated default (beats mult5 in A/B); SH_ST_MULT overrides rs_lb, mkt_sma = 168, 100 def leverage(self, pair, current_time, current_rate, proposed_leverage, max_leverage, side, **kwargs) -> float: # 1.0 = fully-collateralized (notional == margin), the deployed default. SH_LEV scales # the return/DD profile for a higher-octane risk-capital sleeve (off => 1.0, unchanged). # Side-aware: lever the OFFENSE (longs) and keep the INSURANCE (shorts) at SH_LEV_SHORT # (default 1.0) — leveraged shorts in a down-regime are what craters the bad year. lev = _envf("SH_LEV", 1.0) if side == "long" else _envf("SH_LEV_SHORT", 1.0) return float(min(max(lev, 1.0), max_leverage)) def custom_stoploss(self, pair, trade, current_time, current_rate, current_profit, **kwargs): # GRID trades live or die by the band-flip kill-switch (grid_flat, profitable) + the laddered # level cap, NOT a tight stop. A -10% stop on an AVERAGING position just locks in losses the # mean-reversion would recover (first cut: 136 stop-outs = -$589). Give grids only a WIDE # catastrophe backstop (SH_GRID_STOP), so normal in-band DCA can breathe. if getattr(self, "grid_on", False): if trade.enter_tag == "grid": return None # grid uses the WIDE static stop (set in __init__) as sole backstop if _envf("SH_ATR_MULT", 0.0) <= 0: # trend trade: re-impose the tight SH_STOP the widened static would otherwise loosen return stoploss_from_open(-_envf("SH_STOP", 0.10), current_profit, is_short=trade.is_short, leverage=trade.leverage) # IDEA2: stop a FIXED ATR-multiple from entry (vol-scaled), replacing the flat -SH_STOP. # Off unless SH_ATR_MULT>0. Distance floored at 2% and capped at the -SH_STOP backstop, # so it never risks more than the hard stop. Fixed-at-entry (not trailing) via entry ATR%. mult = _envf("SH_ATR_MULT", 0.0) if mult <= 0: return None df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df is None or len(df) == 0: return None row = df.loc[df["date"] <= trade.open_date_utc, "atrp"] atrp = row.iloc[-1] if len(row) else df["atrp"].iloc[-1] if not (atrp > 0): return None dist = min(max(mult * atrp, 0.02), _envf("SH_STOP", 0.10)) return stoploss_from_open(-dist, current_profit, is_short=trade.is_short, leverage=trade.leverage) def _btc_vol(self, current_time): """Trailing 7d realized vol of BTC daily returns as of current_time (strictly past candles only => no lookahead). Volatility is the one thing that persists (autocorr ~0.23 vs ~0.03 for direction), so it's the robust axis for dynamic exposure.""" try: btc_d = self.dp.get_pair_dataframe("BTC/USD:USD", "1d") if btc_d is None or len(btc_d) < 9: return None b = btc_d[btc_d["date"] < current_time] if len(b) < 9: return None return float(b["close"].pct_change().rolling(7).std().iloc[-1]) except Exception: return None def _btc_dev(self, current_time): """BTC close vs its 100d SMA as a fraction (neg=below), as of current_time (no lookahead).""" try: btc_d = self.dp.get_pair_dataframe("BTC/USD:USD", "1d") if btc_d is None or len(btc_d) < 101: return None b = btc_d[btc_d["date"] < current_time] if len(b) < 101: return None sma = b["close"].rolling(100).mean().iloc[-1] if not (sma > 0): return None return float(b["close"].iloc[-1] / sma - 1.0) except Exception: return None def custom_stake_amount(self, pair, current_time, current_rate, proposed_stake, min_stake, max_stake, leverage, entry_tag, side, **kwargs): stake = proposed_stake # GRID level sizing: each ladder level risks SH_GRID_STAKE of a normal slot (swept in backtest). if entry_tag == "grid": return max(min_stake or 0.0, proposed_stake * _envf("SH_GRID_STAKE", 0.5)) # IDEA1: de-weight the weaker LONG leg (off by default, SH_LONG_STAKE=1.0). # 0.5 = longs risk half a slot; shorts (the engine) unchanged. if side == "long": f = _envf("SH_LONG_STAKE", 1.0) if f != 1.0: stake = stake * f # Vol-scaled exposure (off unless SH_VOL_REF>0): size inversely to BTC realized vol # => smaller bets when the market is dangerous, full size when calm. Clamped so it # never goes crazy. ref = the "normal" vol anchor; m = ref/current, bounded. vol_ref = _envf("SH_VOL_REF", 0.0) if vol_ref > 0: bv = self._btc_vol(current_time) if bv and bv > 0: m = min(max(vol_ref / bv, _envf("SH_VOL_FLOOR", 0.5)), _envf("SH_VOL_CAP", 1.5)) stake = stake * m # REGIME-CONVICTION sizing (off unless SH_LONG_CONV_LO<1). The VECTOR as a continuous dial, # not a 0/1 gate: when a LONG's BTC regime is in the topping/decelerating zone (btc_dev 7d-slope # in the (LO,HI] dead-band that signal-tested as the loser) SIZE DOWN to SH_LONG_CONV_LO instead # of skipping. A smaller position does NOT free a slot for an equally-topping replacement, so the # de-risk actually reduces topping-regime exposure rather than just reshuffling which pair holds it. conv_lo = _envf("SH_LONG_CONV_LO", 1.0) if conv_lo < 1.0 and side == "long" and self.dp is not None: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df is not None and len(df) and "btc_dev_sl" in df.columns: r = df.loc[df["date"] <= current_time, "btc_dev_sl"] if len(r): devsl = r.iloc[-1] lo = _envf("SH_LONG_CONV_TLO", 0.0) hi = _envf("SH_LONG_CONV_THI", 0.03) if lo < devsl <= hi: stake = stake * conv_lo # SHORT-side analog (off unless SH_SHORT_CONV_LO<1): size DOWN shorts when the bear is reverting # up fast (btc_dev 7d-slope >= squeeze threshold) instead of gating them — keeps the slot occupied # so the SH_MAX_SHORT-capped book isn't reshuffled. Tests whether the long-side win transfers. sconv_lo = _envf("SH_SHORT_CONV_LO", 1.0) if sconv_lo < 1.0 and side == "short" and self.dp is not None: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if df is not None and len(df) and "btc_dev_sl" in df.columns: r = df.loc[df["date"] <= current_time, "btc_dev_sl"] if len(r) and r.iloc[-1] >= _envf("SH_SHORT_CONV_THI", 0.03): stake = stake * sconv_lo # DRAWDOWN-CONVICTION sizing (off unless SH_DD_CONV<1): the GRADED alternative to the # hard MaxDrawdown global LOCK. Instead of going 100% to cash when the trailing closed-trade # drawdown trips, SIZE DOWN all new entries to SH_DD_CONV (e.g. 0.25). Keeps money in the # market through the cluster so a real recovery is captured at reduced size, rather than # head-in-sand cash for 48h on stale P&L. Same "ratios" DD calc as the protection. To test # as a REPLACEMENT for the hard brake, run with TR_DD_MAX=99 (protection off) + SH_DD_CONV<1. dd_conv = _envf("SH_DD_CONV", 1.0) if dd_conv < 1.0: try: from datetime import timedelta from freqtrade.data.metrics import calculate_max_drawdown lb = int(_envf("SH_DD_CONV_LB", 168)) back_to = current_time - timedelta(minutes=lb * 60) tw = Trade.get_trades_proxy(is_open=False, close_date=back_to) if len(tw) >= int(_envf("SH_DD_CONV_LIM", 10)): tdf = pd.DataFrame([{"close_date": t.close_date_utc, "close_profit": t.close_profit} for t in tw]) dobj = calculate_max_drawdown(tdf, value_col="close_profit") if dobj.drawdown_abs > _envf("SH_DD_CONV_TRIP", 0.12): stake = stake * dd_conv except (ValueError, Exception): pass return max(min_stake or 0.0, stake) def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs) -> bool: # Correlation/concurrency guard. The worst losses are CLUSTERS: many alt LONGS # entered the same candle (all correlated to BTC) then all stop out together on # one BTC dump (e.g. 2025-08-14: 7 longs -> -10% each in 7h). Capping concurrent # longs turns "one correlated bet x7" into a bounded book. Off unless SH_MAX_LONG # /SH_MAX_SHORT set (0 = unlimited), so it never touches the proven short engine. # CASH / stand-aside mode — the explicit "do nothing" 4th state. Vol acts OPPOSITELY on # the two sides (empirically): high vol = big up-trends (good for longs) but squeeze risk # (bad for shorts); dead-calm = no trend to follow (bad for both = chop). So the vetoes # are side-aware, off unless set: # SH_VOL_MIN — below this BTC vol, market is dead → stand aside (both sides) # SH_VOL_MAX_SHORT — above this BTC vol, shorts get squeezed → stand aside (shorts only) # SH_VOL_MAX — symmetric ceiling (both sides); legacy/blunt, usually leave off # NET-AWARE drawdown brake (off unless SH_NETDD_TRIP>0): the built-in MaxDrawdown locks on the # worst peak-to-trough POCKET in the window and IGNORES the window's NET — so a 3-loser pocket # inside a net-WINNING week locks us out mid-rally (live 2026-06-25: -16.5% pocket but the 19-trade # week netted ~+60% / +$64). This variant blocks new entries only if BOTH the pocket drawdown # > SH_NETDD_TRIP AND the window net return < SH_NETDD_NET. Dynamic (re-checked each candle, no # fixed 48h lock; unblocks the instant net recovers). Run as REPLACEMENT with TR_DD_MAX=99. netdd_trip = _envf("SH_NETDD_TRIP", 0.0) if netdd_trip > 0: try: from datetime import timedelta from freqtrade.data.metrics import calculate_max_drawdown lb = int(_envf("SH_NETDD_LB", 168)) back_to = current_time - timedelta(minutes=lb * 60) tw = Trade.get_trades_proxy(is_open=False, close_date=back_to) if len(tw) >= int(_envf("SH_NETDD_LIM", 10)): net = sum(t.close_profit for t in tw) tdf = pd.DataFrame([{"close_date": t.close_date_utc, "close_profit": t.close_profit} for t in tw]) dobj = calculate_max_drawdown(tdf, value_col="close_profit") if dobj.drawdown_abs > netdd_trip and net < _envf("SH_NETDD_NET", 0.0): return False except (ValueError, Exception): pass vol_min = _envf("SH_VOL_MIN", 0.0) vol_max_short = _envf("SH_VOL_MAX_SHORT", 0.0) vol_max = _envf("SH_VOL_MAX", 0.0) if vol_min > 0 or vol_max_short > 0 or vol_max > 0: bv = self._btc_vol(current_time) if bv is not None: if vol_min > 0 and bv < vol_min: return False if vol_max > 0 and bv > vol_max: return False if vol_max_short > 0 and side == "short" and bv > vol_max_short: return False max_long = int(_envf("SH_MAX_LONG", 0)) max_short = int(_envf("SH_MAX_SHORT", 0)) if max_long <= 0 and max_short <= 0: return True open_trades = Trade.get_open_trades() if side == "long" and max_long > 0: if sum(1 for t in open_trades if not t.is_short) >= max_long: return False if side == "short" and max_short > 0: if sum(1 for t in open_trades if t.is_short) >= max_short: return False return True def adjust_trade_position(self, trade, current_time, current_rate, current_profit, min_stake, max_stake, current_entry_rate, current_exit_rate, current_entry_profit, current_exit_profit, **kwargs): # GRID engine — STATELESS per-lot ladder (custom-data does NOT persist across candles in # backtest, so we reconstruct state from freqtrade's own order counts, which DO). Fixed # levels k=0..L-1 at anchor*(1-g*k); the initial entry is level 0. We ladder strictly deeper # (next buy = level n_entries) and bank shallowest-first (next sell target = level n_exits). # Selling a lot as price rises g above ITS level — not the average — is what avoids the # martingale trap. Position is hard-bounded by SH_GRID_LEVELS * one-lot, so it can't run away. # (freqtrade realises partial exits on average cost, so harvest is under the per-lot sim, and # within one trade a sold level isn't re-bought — re-harvest happens across re-entries.) if not getattr(self, "grid_on", False) or trade.enter_tag != "grid": return None g = _envf("SH_GRID_G", 0.02) levels = int(_envf("SH_GRID_LEVELS", 8)) entries = trade.select_filled_orders(trade.entry_side) n_ent = len(entries) if n_ent == 0: return None anchor = entries[0].safe_price unit = entries[0].cost # one lot's notional (1x leverage) n_exit = trade.nr_of_successful_exits # partial sells so far (shallowest-first) n_open = n_ent - n_exit # SCALE OUT: bank the shallowest still-open lot (level n_exit) once price is g above its level if n_open > 0 and current_rate >= anchor * (1.0 - g * n_exit) * (1.0 + g): if min_stake is None or unit >= min_stake: return -unit return None # SCALE IN: drop to the next deeper level (index n_ent) and add a lot, while levels remain. # KNIFE GUARD (SH_GRID_TREND_GUARD, on by default): don't ladder deeper into a confirmed # per-pair downtrend (alt Supertrend == down) — that catching-knife inventory is what hits # the -25% stop while BTC sits inside its band. We only add levels in genuine chop. if n_ent < levels and current_rate <= anchor * (1.0 - g * n_ent): if _envf("SH_GRID_TREND_GUARD", 1.0) >= 1.0: df, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) if df is not None and len(df) and df["st_dir"].iloc[-1] == -1: return None return min(unit, max_stake) if max_stake else unit return None def confirm_trade_exit(self, pair, trade, order_type, amount, rate, time_in_force, exit_reason, current_time, **kwargs) -> bool: # Grids ignore Supertrend-flip / ROI / trailing signal exits — they are managed rung-by-rung # by adjust_trade_position and flattened wholesale on band-break (custom_exit -> grid_flat). # Stoploss and grid_flat still pass through as backstops. if getattr(self, "grid_on", False) and trade.enter_tag == "grid": if exit_reason in ("long_end", "exit_signal", "trailing_stop_loss", "roi"): return False return True def informative_pairs(self): return [("BTC/USD:USD", "1h"), ("BTC/USD:USD", "1d")] def _supertrend_dir(self, dataframe, length, mult): st = pta.supertrend(dataframe["high"], dataframe["low"], dataframe["close"], length=length, multiplier=mult) dcol = [c for c in st.columns if c.startswith("SUPERTd")][0] return st[dcol] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: rs_lb = int(_envf("TR_RS_LB", self.rs_lb)) st_len = int(_envf("SH_ST_LEN", self.st_len)) st_mult = _envf("SH_ST_MULT", self.st_mult) dataframe["st_dir"] = self._supertrend_dir(dataframe, st_len, st_mult).values dataframe["ema_trend"] = ta.EMA(dataframe, timeperiod=self.ema_len) dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) dataframe["ext"] = dataframe["close"] / dataframe["ema_trend"] # 1.0 = at EMA dataframe["atrp"] = ta.ATR(dataframe, timeperiod=14) / dataframe["close"] # IDEA2: vol for ATR stop # RS vs BTC perp dataframe["pair_mom"] = dataframe["close"] / dataframe["close"].shift(rs_lb) - 1.0 dataframe["btc_mom"] = 0.0 if self.dp is not None: btc = self.dp.get_pair_dataframe("BTC/USD:USD", self.timeframe) if btc is not None and len(btc) > rs_lb: btc = btc.copy() btc["bmom"] = btc["close"] / btc["close"].shift(rs_lb) - 1.0 dataframe = pd.merge(dataframe, btc[["date", "bmom"]], on="date", how="left") dataframe["btc_mom"] = dataframe["bmom"].ffill().fillna(0.0) dataframe.drop(columns=["bmom"], inplace=True) dataframe["rs"] = dataframe["pair_mom"] - dataframe["btc_mom"] # BTC daily regime — GRADED, not a single global flip. bull = clearly above SMA (by # SH_REGIME_BAND), bear = clearly below; the band in between is a NEUTRAL no-entry zone # (sit out chop instead of forcing the whole bot to one side). SH_MKT_SLOPE optionally # also requires the SMA itself to be rising (bull) / falling (bear). Defaults (sma100, # band0, slope0) reproduce the old binary close>SMA gate exactly. mkt_sma = int(_envf("SH_MKT_SMA", self.mkt_sma)) band = _envf("SH_REGIME_BAND", 0.0) slope_lb = int(_envf("SH_MKT_SLOPE", 0)) dataframe["btc_bull"] = True # default when BTC data missing: longs on (matches old btc_up=True) dataframe["btc_bear"] = False # default: shorts OFF (matches old ~btc_up=False) — no shorting on uncertainty dataframe["btc_dev"] = 0.0 # BTC close vs SMA, as a fraction (neg = below SMA); 0 when data missing if self.dp is not None: btc_d = self.dp.get_pair_dataframe("BTC/USD:USD", "1d") if btc_d is not None and len(btc_d) > mkt_sma + slope_lb: btc_d = btc_d.copy() sma = ta.SMA(btc_d, timeperiod=mkt_sma) bull = btc_d["close"] > sma * (1.0 + band) bear = btc_d["close"] < sma * (1.0 - band) if slope_lb > 0: rising = sma > sma.shift(slope_lb) bull = bull & rising bear = bear & (~rising) # Regime hysteresis (off unless SH_REGIME_HYST>1): the raw band9 regime flickers # every ~4 days as BTC oscillates across the band; the chop-straddle loss months are # brief false breakouts that immediately reverse. Require the regime to hold for N # consecutive days before committing => skip the head-fakes. Causal rolling = no lookahead. hyst = int(_envf("SH_REGIME_HYST", 1)) if hyst > 1: bull = bull.astype(int).rolling(hyst, min_periods=hyst).min().fillna(0) == 1 bear = bear.astype(int).rolling(hyst, min_periods=hyst).min().fillna(0) == 1 btc_d["btc_bull"] = bull.astype(float) btc_d["btc_bear"] = bear.astype(float) btc_d["btc_dev"] = (btc_d["close"] / sma - 1.0) dataframe = merge_informative_pair( dataframe, btc_d[["date", "btc_bull", "btc_bear", "btc_dev"]], self.timeframe, "1d", ffill=True) dataframe["btc_bull"] = dataframe["btc_bull_1d"].fillna(1.0) > 0.5 # warmup → bull (longs on) dataframe["btc_bear"] = dataframe["btc_bear_1d"].fillna(0.0) > 0.5 # warmup → not bear (shorts off) dataframe["btc_dev"] = dataframe["btc_dev_1d"].fillna(0.0) # warmup → 0 (no depth signal) dataframe["btc_up"] = dataframe["btc_bull"] # back-compat alias # btc_dev SLOPE (regime VELOCITY) — for conviction sizing, not a gate. 7d default. dataframe["btc_dev_sl"] = (dataframe["btc_dev"] - dataframe["btc_dev"].shift(int(_envf("SH_DEVSL_LB", 168)))).fillna(0.0) # Per-pair funding (crowding/squeeze signal). Negative funding = shorts pay longs # = crowded shorts = squeeze fuel. Only loaded if the filter is enabled (SH_FUND_MIN). dataframe["fund"] = 0.0 if _envf("SH_FUND_MIN", -1.0) > -1.0 or _envf("SH_FUND_MAX_LONG", 9.0) < 9.0: base = metadata["pair"].split("/")[0] dd = str(self.config.get("datadir") or "user_data/data/krakenfutures") fn = f"{base}_USD_USD-1h-funding_rate.feather" cands = [os.path.join(dd, "futures", fn), # datadir already = .../krakenfutures os.path.join(dd, "krakenfutures", "futures", fn)] # datadir = .../data fp = next((c for c in cands if os.path.exists(c)), None) if fp: fr = pd.read_feather(fp)[["date", "open"]].rename(columns={"open": "fr"}) fr = fr[~fr["date"].duplicated()] fr["date"] = fr["date"].astype(dataframe["date"].dtype) # ms->ns so merge matches dataframe = pd.merge(dataframe, fr, on="date", how="left") dataframe["fund"] = dataframe["fr"].rolling(24, min_periods=1).mean().ffill().fillna(0.0) dataframe.drop(columns=["fr"], inplace=True) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: long_on = _envf("SH_LONG", 1.0) >= 1.0 short_on = _envf("SH_SHORT", 1.0) >= 1.0 rsi_min = _envf("SH_RSI_MIN", 35.0) # don't short below this RSI (oversold => squeeze) max_ext = _envf("SH_MAX_EXT", 0.15) # don't short if >15% below EMA (stretched) btc_conv = _envf("SH_BTC_CONV", 0.0) # require BTC 7d-mom <= this to short (0=off); dodges flat/squeeze bears rs_max = _envf("SH_RS_MAX", 0.0) # short only names this much WEAKER than BTC (0=any weaker; -0.05=much weaker) fund_min = _envf("SH_FUND_MIN", -1.0) # skip shorts where 24h funding < this (crowded shorts; -1=off) # ── short-leg BTC-depth band-pass (off by default). Shorts empirically only pay in a # confirmed-but-not-exhausted downtrend: BTC ~9-20% below its SMA. Marginal (<9%, downtrend # failing) and extreme (>20%, capitulation/squeeze) both lose. Units = whole percent. ── short_dev_min = _envf("SH_SHORT_DEV_MIN", 0.0) # require BTC at least this % below SMA (0=off) short_dev_max = _envf("SH_SHORT_DEV_MAX", 0.0) # don't short if BTC more than this % below SMA (0=off; squeeze guard) # ── long-leg refinement knobs (off by default) ── long_btc_min = _envf("SH_LONG_BTC_MIN", -9.0) # IDEA1: require BTC 7d-mom >= this to go long (harder regime gate) long_rsi_max = _envf("SH_LONG_RSI_MAX", 100.0) # IDEA3 anti-knife: skip longs with RSI above this (overbought top) long_max_ext = _envf("SH_LONG_MAX_EXT", 9.0) # IDEA3 anti-knife: skip longs >X above EMA (stretched; ext=close/ema-1) long_fund_max = _envf("SH_FUND_MAX_LONG", 9.0) # FUNDING contrarian: skip longs where 24h funding > this (crowded longs => predicts reversal down; 9=off) long_gate = _envf("SH_LONG_GATE", 1.0) >= 1.0 # apply BTC regime gate to longs (1=on) short_gate = _envf("SH_SHORT_GATE", 1.0) >= 1.0 # apply BTC regime gate to shorts (1=on) if long_on: long_cond = ( (dataframe["st_dir"] == 1) & (dataframe["close"] > dataframe["ema_trend"]) & (dataframe["rs"] >= 0) & (dataframe["btc_bull"] if long_gate else True) & (dataframe["volume"] > 0) ) if long_btc_min > -9.0: # IDEA1: only long a confirmed BTC uptrend long_cond = long_cond & (dataframe["btc_mom"] >= long_btc_min) if long_rsi_max < 100.0: # IDEA3: not buying an overbought top long_cond = long_cond & (dataframe["rsi"] <= long_rsi_max) if long_max_ext < 9.0: # IDEA3: not buying a stretched move long_cond = long_cond & ((dataframe["ext"] - 1.0) <= long_max_ext) if long_fund_max < 9.0: # FUNDING contrarian: not buying a crowded-long top long_cond = long_cond & (dataframe["fund"] <= long_fund_max) dataframe.loc[long_cond, ["enter_long", "enter_tag"]] = (1, "trend_long") if short_on: short_cond = ( (dataframe["st_dir"] == -1) & (dataframe["close"] < dataframe["ema_trend"]) & (dataframe["rs"] <= rs_max) # selection: only names weaker than BTC (by rs_max) & (dataframe["btc_bear"] if short_gate else True) & (dataframe["rsi"] >= rsi_min) # squeeze guard: not oversold & (dataframe["ext"] >= (1.0 - max_ext)) # squeeze guard: not stretched & (dataframe["volume"] > 0) ) if btc_conv < 0: # conviction gate: only short a CONFIRMED BTC downtrend short_cond = short_cond & (dataframe["btc_mom"] <= btc_conv) if fund_min > -1.0: # crowding guard: skip pairs with deeply negative (crowded-short) funding short_cond = short_cond & (dataframe["fund"] >= fund_min) if short_dev_min > 0: # band-pass floor: BTC deep enough below SMA (downtrend confirmed) short_cond = short_cond & (dataframe["btc_dev"] <= -short_dev_min / 100.0) if short_dev_max > 0: # band-pass ceiling: BTC not TOO deep (capitulation/squeeze guard) short_cond = short_cond & (dataframe["btc_dev"] >= -short_dev_max / 100.0) dataframe.loc[short_cond, ["enter_short", "enter_tag"]] = (1, "trend_short") # GRID entries (off unless SH_GRID_ON=1): seed a grid in the NEUTRAL band only # (~btc_bull & ~btc_bear) on a mild pullback (rsi < SH_GRID_RSI). Mutually exclusive with # trend entries (those require bull/bear), so there is no overlap; the ladder is then run # by adjust_trade_position and flattened on band-break by custom_exit. if _envf("SH_GRID_ON", 0.0) >= 1.0: neutral = (~dataframe["btc_bull"].astype(bool)) & (~dataframe["btc_bear"].astype(bool)) grid_cond = neutral & (dataframe["rsi"] < _envf("SH_GRID_RSI", 55.0)) & (dataframe["volume"] > 0) dataframe.loc[grid_cond, ["enter_long", "enter_tag"]] = (1, "grid") return dataframe def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs): # Side-aware take-profit: bank profit on SHORTS (relief-rally round-trip risk), # but leave LONGS alone to ride sustained bull trends to the Supertrend flip. # Evaluated on candle close (no intra-candle peak assumption) => realistic fills. if getattr(self, "grid_on", False) and trade.enter_tag == "grid": # Band-break kill-switch: when BTC leaves the neutral band, flatten the whole grid and # hand the regime back to the trend book (the move that dodged the 2026-06 breakdown). # Otherwise return None and let adjust_trade_position manage the rungs. band = _envf("SH_REGIME_BAND", 0.0) dev = self._btc_dev(current_time) if dev is not None and abs(dev) > band: return "grid_flat" return None if trade.is_short: tp = _envf("SH_TP", 0.0) if tp <= 0: return None # optional decay: after SH_TP_MIN hours of no follow-through, accept smaller SH_TP_T tp_t = _envf("SH_TP_T", 0.0) if tp_t > 0: held_h = (current_time - trade.open_date_utc).total_seconds() / 3600.0 if held_h >= _envf("SH_TP_MIN", 24) and current_profit >= tp_t: return "short_tp_decay" if current_profit >= tp: return "short_tp" else: # Regime-conditional LONG take-profit (off unless SH_LONG_TP>0). Longs in mild/moderate # bull peak ~8% then round-trip (13-20%-above-SMA bucket = the long loss zone); bank them. # But ONLY when BTC is NOT in strong bull (dev <= SH_LONG_TP_MAXDEV) — strong-bull longs # ride to +500% and ARE the edge, so never cap those. Exit-side => edge knowable at decision. ltp = _envf("SH_LONG_TP", 0.0) if ltp > 0 and current_profit >= ltp: maxdev = _envf("SH_LONG_TP_MAXDEV", 0.0) # 0 = apply at any regime depth if maxdev <= 0: return "long_tp" dev = self._btc_dev(current_time) if dev is not None and dev <= maxdev: return "long_tp" return None def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[dataframe["st_dir"] == -1, ["exit_long", "exit_tag"]] = (1, "long_end") dataframe.loc[dataframe["st_dir"] == 1, ["exit_short", "exit_tag"]] = (1, "short_end") return dataframe