# source: https://raw.githubusercontent.com/IvPalmer/Master-Trader/c7955a15d4bfe82e567bc2bd315a372000a442fd/ft_userdata/user_data/strategies/ShortKeltnerV2HL.py
# directory_url: https://github.com/IvPalmer/Master-Trader/blob/main/ft_userdata/user_data/strategies/
# User: IvPalmer
# Repository: Master-Trader
# --------------------"""
Github_IvPalmer_Master_Trader__ShortKeltnerV2HL__20260923_012354 — 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 Github_IvPalmer_Master_Trader__ShortKeltnerV2HL__20260923_012354(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(
                "Github_IvPalmer_Master_Trader__ShortKeltnerV2HL__20260923_012354 %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()
