""" ShortKeltnerV2HL — Hyperliquid (USDC-margined) port of ShortKeltnerV2. IDENTICAL logic to ShortKeltnerV2 (inverse-Keltner short-side mean-reversion + BTC daily-200d-MA macro bear guard). ONLY difference: Hyperliquid perps are USDC-margined (BTC/USDC:USDC), so the BTC macro-gate informatives reference BTC/USDC:USDC and the generated columns are btc_usdc_* (not btc_usdt_*). Purpose: forward-measured Hyperliquid strategy. It began as a dry-run instrument because Hyperliquid did not provide the historical download needed by the original validation workflow. The operator authorized a bounded micro-live epoch on 2026-08-23; the retired dry curve remains frozen as chart lineage and the live config uses a dedicated account plus an exchange-resident limit stop. See docs/hyperliquid_short_validation_2026-05-29.md and docs/audits/2026-08-24-live-fleet-review-remediation.md. Generated 2026-05-29. """ import logging import pandas as pd import talib.abstract as ta from freqtrade.strategy import IStrategy, informative from pandas import DataFrame logger = logging.getLogger(__name__) class ShortKeltnerV2HL(IStrategy): INTERFACE_VERSION = 3 can_short = True trading_mode = "futures" margin_mode = "isolated" timeframe = "1h" minimal_roi = {"0": 0.06, "360": 0.04, "720": 0.025, "1440": 0.015, "2160": 0.0} stoploss = -0.05 trailing_stop = False process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False startup_candle_count = 250 kelt_period = 25 kelt_atr_mult = 2.5 vol_multiplier = 1.75 vol_sma_period = 20 rsi_overbought = 60 btc_slope_lookback = 24 # Single source of truth for the macro bear gate's informative inputs. # populate_indicators builds btc_bear from a frame reindexed to exactly # these columns and the fail-closed watchdog checks exactly these columns, # so the two cannot drift apart. They did drift once: the gate read # btc_usdc_sma50_slope_1h while the watchdog did not check it, so a NaN # slope forced btc_bear to 0 and blocked every entry with no warning — # the precise silent failure the watchdog exists to make visible. # populate_exit_trend's inputs (close_1h, sma50_1h) are a subset, so # covering this list also covers the exit path's macro inputs. MACRO_GATE_COLUMNS = ( "btc_usdc_close_1h", "btc_usdc_sma50_1h", "btc_usdc_sma200_1h", "btc_usdc_sma50_slope_1h", "btc_usdc_sma200_1d", ) @property def protections(self): return [ {"method": "CooldownPeriod", "stop_duration_candles": 4}, {"method": "StoplossGuard", "lookback_period_candles": 48, "trade_limit": 3, "stop_duration_candles": 24, "only_per_pair": False}, {"method": "MaxDrawdown", "lookback_period_candles": 72, "max_allowed_drawdown": 0.12, "stop_duration_candles": 48, "trade_limit": 2}, ] def leverage(self, pair, current_time, current_rate, proposed_leverage, max_leverage, entry_tag, side, **kwargs) -> float: return min(2.0, max_leverage) @informative("1d", "BTC/USDC:USDC") def populate_indicators_btc_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["sma200"] = ta.SMA(dataframe, timeperiod=200) return dataframe @informative("1h", "BTC/USDC:USDC") def populate_indicators_btc_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe["sma50"] = ta.SMA(dataframe, timeperiod=50) dataframe["sma200"] = ta.SMA(dataframe, timeperiod=200) dataframe["sma50_slope"] = dataframe["sma50"] - dataframe["sma50"].shift(self.btc_slope_lookback) return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: sma = dataframe["close"].rolling(self.kelt_period).mean() atr = _atr(dataframe, self.kelt_period) dataframe["kelt_upper"] = sma + self.kelt_atr_mult * atr dataframe["vol_sma"] = dataframe["volume"].rolling(self.vol_sma_period).mean() dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14) # Reindex to the declared gate columns: a column the informative never # produced becomes all-NaN instead of raising, so every comparison # below is False and the gate fails CLOSED. The watchdog then reports # it as a missing/NaN input rather than the bot dying each candle. gate = dataframe.reindex(columns=list(self.MACRO_GATE_COLUMNS)) dataframe["btc_bear"] = ( (gate["btc_usdc_close_1h"] < gate["btc_usdc_sma50_1h"]) & (gate["btc_usdc_close_1h"] < gate["btc_usdc_sma200_1h"]) & (gate["btc_usdc_sma50_slope_1h"] < 0) & (gate["btc_usdc_close_1h"] < gate["btc_usdc_sma200_1d"]) ).astype(int) # Fail-closed observability: NaN gate inputs force btc_bear to 0, # indistinguishable from a legitimately closed gate. HL serves limited # history, so a short informative fetch would silently zero entries # forever. Warn while the condition persists. # # sma50_slope needs btc_sma50_period + btc_slope_lookback = 74 BTC 1h # candles and sma200_1d needs 200 BTC daily candles — both deeper than # the pair's own 250-candle startup, so this is the realistic fault. last = gate.iloc[-1] nan_cols = [c for c in self.MACRO_GATE_COLUMNS if pd.isna(last[c])] if nan_cols: logger.warning( "ShortKeltnerV2HL %s: macro-gate inputs NaN at last bar (%s) — " "btc_bear forced 0, entries blocked. Check BTC/USDC informative depth.", metadata["pair"], ", ".join(nan_cols), ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: rsi_was_overbought = ( (dataframe["rsi"].shift(1) > self.rsi_overbought) | (dataframe["rsi"].shift(2) > self.rsi_overbought) ) dataframe.loc[ ( (dataframe["close"] < dataframe["kelt_upper"]) & (dataframe["close"].shift(1) >= dataframe["kelt_upper"].shift(1)) & (dataframe["volume"] > self.vol_multiplier * dataframe["vol_sma"]) & rsi_was_overbought & (dataframe["btc_bear"] == 1) & (dataframe["volume"] > 0) ), ["enter_short", "enter_tag"], ] = (1, "kelt_upper_reject") return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( ( (dataframe["btc_usdc_close_1h"] > dataframe["btc_usdc_sma50_1h"]) | (dataframe["rsi"] < 30) ) & (dataframe["volume"] > 0) ), ["exit_short", "exit_tag"], ] = (1, "regime_flip_or_oversold") return dataframe def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs): if not trade.is_short: return None hours = (current_time - trade.open_date_utc).total_seconds() / 3600 if hours >= 36: return "time_exit_36h" return None def _atr(df: DataFrame, period: int) -> pd.Series: high = df["high"]; low = df["low"]; close = df["close"] tr = pd.concat([high - low, (high - close.shift(1)).abs(), (low - close.shift(1)).abs()], axis=1).max(axis=1) return tr.rolling(period).mean()