TopGunStrategy

# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401

"""
╔══════════════════════════════════════════════════════════════════════╗
║              TopGun Trading Strategy  —  Freqtrade                  ║
║                                                                      ║
║  Indicators (exact Pine Script ports):                               ║
║   1. Pi Cycle High/Low       (cycle top & bottom detection)          ║
║   2. [RS] JMA Modified RSI   (3-length: 10 / 60 / 420)              ║
║   3. TTM Squeeze             (BB vs KC momentum)                     ║
║   4. RSI + Pivot Divergence  (bull/bear labels)                      ║
║   5. Kalit MA Stack          (MA5/10/20/50/100/200 + RSI2 rules)     ║
║                                                                      ║
║  Timeframe : 4h (momentum) with 1d informative (trend)              ║
║  Pairs     : BTC/USDT  ETH/USDT  SOL/USDT                          ║
╚══════════════════════════════════════════════════════════════════════╝
"""

from datetime import datetime
from typing import Optional
import numpy as np
import pandas as pd
from pandas import DataFrame
from freqtrade.strategy import IStrategy, informative
import talib.abstract as ta


class TopGunStrategy(IStrategy):
    """
    TopGun Trading Strategy
    All indicators ported exactly from Pine Script.
    """

    # ── Strategy metadata ────────────────────────────────────────────
    INTERFACE_VERSION = 3
    timeframe = "4h"
    can_short = False

    # ── ROI & Stoploss ───────────────────────────────────────────────
    minimal_roi = {
        "0":   0.06,   # 6% take profit immediately
        "60":  0.04,   # 4% after 60 min
        "120": 0.02,   # 2% after 120 min
        "240": 0.01,   # 1% after 240 min
    }
    stoploss = -0.03   # 3% stop loss

    # ── Trailing stop ────────────────────────────────────────────────
    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.04
    trailing_only_offset_is_reached = True

    # ── Misc ─────────────────────────────────────────────────────────
    startup_candle_count: int = 500   # need lots of history for MA300/Pi Cycle
    process_only_new_candles = True

    # ══════════════════════════════════════════════════════════════════
    #  HELPER FUNCTIONS
    # ══════════════════════════════════════════════════════════════════

    def _rma(self, series: pd.Series, length: int) -> pd.Series:
        """Wilder's RMA — same as Pine ta.rma"""
        return series.ewm(alpha=1 / length, adjust=False).mean()

    def _jmax(self, src: np.ndarray, length: int, exp: float) -> np.ndarray:
        """
        Exact port of capissimo's jmax() Pine Script function.
        Jurik Moving Average core calculation.
        """
        beta  = 0.45 * (length - 1) / (0.45 * (length - 1) + 2)
        alpha = beta ** exp
        n     = len(src)
        L0 = L1 = L3 = L4 = 0.0
        out = np.full(n, np.nan)
        for i in range(n):
            v = src[i]
            if np.isnan(v):
                continue
            L0 = (1 - alpha) * v + alpha * L0
            L1 = (v - L0) * (1 - beta) + beta * L1
            L2 = L0 + L1
            L3 = (L2 - L4) * ((1 - alpha) ** 2) + (alpha ** 2) * L3
            L4 = L4 + L3
            out[i] = L4
        return out

    def _jma_mod_rsi(self, src: np.ndarray, length: int, exp: float = 2.0) -> np.ndarray:
        """
        [RS] JMA Modified RSI — exact Pine Script port.
        RSI where the average is replaced by JMA instead of a simple mean.
        """
        avg = self._jmax(src, length, exp)
        u   = np.maximum(src - avg, 0)
        d   = np.maximum(avg - src, 0)
        ru  = self._rma(pd.Series(u), length).values
        rd  = self._rma(pd.Series(d), length).values
        with np.errstate(divide="ignore", invalid="ignore"):
            rs  = np.where(rd == 0, np.inf, ru / rd)
            rsi = np.where(rd == 0, 100.0,
                  np.where(ru == 0, 0.0,
                  100 - np.round(100 / (1 + rs))))
        return rsi

    # ══════════════════════════════════════════════════════════════════
    #  INFORMATIVE — 1D timeframe for trend bias
    # ══════════════════════════════════════════════════════════════════

    @informative("1d")
    def populate_indicators_1d(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Daily timeframe indicators:
        - Pi Cycle High/Low MAs
        - JMA RSI slow (length=420) for macro momentum
        - EMA alignment for trend
        """
        c = dataframe["close"]

        # ── Pi Cycle MAs ─────────────────────────────────────────────
        dataframe["pi_low_long"]  = c.rolling(471).mean() * (745 / 1000)
        dataframe["pi_low_short"] = c.ewm(span=150, adjust=False).mean()
        dataframe["pi_hi_long"]   = c.rolling(350).mean() * 2.0
        dataframe["pi_hi_short"]  = c.rolling(111).mean()

        # Pi Cycle crossunder signals
        dataframe["pi_cycle_low"] = (
            (dataframe["pi_low_short"].shift(1) >= dataframe["pi_low_long"].shift(1)) &
            (dataframe["pi_low_short"] < dataframe["pi_low_long"])
        ).astype(int)

        dataframe["pi_cycle_hi"] = (
            (dataframe["pi_hi_long"].shift(1) >= dataframe["pi_hi_short"].shift(1)) &
            (dataframe["pi_hi_long"] < dataframe["pi_hi_short"])
        ).astype(int)

        # ── JMA RSI slow (420) — macro momentum ──────────────────────
        dataframe["jrsi2"] = self._jma_mod_rsi(c.values, 420, 2.0)

        # ── EMA stack for trend alignment ────────────────────────────
        dataframe["ema10"]  = c.ewm(span=10,  adjust=False).mean()
        dataframe["ema20"]  = c.ewm(span=20,  adjust=False).mean()
        dataframe["ema50"]  = c.ewm(span=50,  adjust=False).mean()
        dataframe["ema100"] = c.ewm(span=100, adjust=False).mean()
        dataframe["ma200"]  = c.rolling(200).mean()

        dataframe["ema_bullish"] = (
            (dataframe["ema10"]  > dataframe["ema20"])  &
            (dataframe["ema20"]  > dataframe["ema50"])  &
            (dataframe["ema50"]  > dataframe["ema100"]) &
            (dataframe["ema100"] > dataframe["ma200"])
        ).astype(int)

        dataframe["ema_bearish"] = (
            (dataframe["ema10"]  < dataframe["ema20"])  &
            (dataframe["ema20"]  < dataframe["ema50"])  &
            (dataframe["ema50"]  < dataframe["ema100"]) &
            (dataframe["ema100"] < dataframe["ma200"])
        ).astype(int)

        return dataframe

    # ══════════════════════════════════════════════════════════════════
    #  MAIN INDICATORS — 4H timeframe
    # ══════════════════════════════════════════════════════════════════

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        c = dataframe["close"]
        h = dataframe["high"]
        lo = dataframe["low"]

        # ── 1. JMA Modified RSI (all 3 lengths) ──────────────────────
        dataframe["jrsi0"] = self._jma_mod_rsi(c.values, 10,  2.0)  # fast
        dataframe["jrsi1"] = self._jma_mod_rsi(c.values, 60,  2.0)  # mid
        # jrsi2 comes from 1d informative

        # ── 2. TTM Squeeze ───────────────────────────────────────────
        length = 20
        # Bollinger Bands
        bb_mid   = c.rolling(length).mean()
        bb_upper = bb_mid + 2.0 * c.rolling(length).std()

        # Keltner Channels
        tr = pd.concat([
            h - lo,
            (h - c.shift()).abs(),
            (lo - c.shift()).abs()
        ], axis=1).max(axis=1)
        kc_mid   = c.ewm(span=length, adjust=False).mean()
        kc_upper = kc_mid + 1.0 * tr.ewm(span=length, adjust=False).mean()

        # Squeeze ON = BB inside KC
        dataframe["squeeze_on"]  = (bb_upper < kc_upper).astype(int)
        dataframe["squeeze_off"] = (bb_upper >= kc_upper).astype(int)

        # Momentum histogram
        hh   = h.rolling(length).max()
        ll   = lo.rolling(length).min()
        mid  = (hh + ll) / 2
        delta = c - (mid + kc_mid) / 2
        dataframe["ttm_mom"] = delta.rolling(length).apply(
            lambda x: np.polyval(np.polyfit(np.arange(len(x)), x, 1), len(x) - 1),
            raw=True
        )
        dataframe["ttm_rising"] = (
            dataframe["ttm_mom"] > dataframe["ttm_mom"].shift(1)
        ).astype(int)
        dataframe["ttm_pos"] = (dataframe["ttm_mom"] >= 0).astype(int)

        # ── 3. Standard RSI + Divergence ─────────────────────────────
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        # Pivot-based divergence (simplified for Freqtrade compatibility)
        rsi   = dataframe["rsi"]
        lbL   = 5
        lbR   = 5
        rLo   = 5
        rHi   = 60

        bull_div = pd.Series(0, index=dataframe.index)
        bear_div = pd.Series(0, index=dataframe.index)

        rsi_arr  = rsi.values
        low_arr  = lo.values
        high_arr = h.values
        n        = len(dataframe)

        for i in range(lbL + lbR, n):
            # Pivot low check
            window = rsi_arr[i - lbL - lbR: i + 1]
            if len(window) > 0 and rsi_arr[i - lbR] == np.nanmin(window):
                for j in range(i - lbR - rLo, max(i - lbR - rHi, lbL) - 1, -1):
                    if j < 0:
                        break
                    w2 = rsi_arr[j - lbL: j + lbR + 1] if j >= lbL and j + lbR < n else None
                    if w2 is not None and rsi_arr[j] == np.nanmin(w2):
                        if rsi_arr[i - lbR] > rsi_arr[j] and low_arr[i - lbR] < low_arr[j]:
                            bull_div.iloc[i] = 1
                        break
            # Pivot high check
            if rsi_arr[i - lbR] == np.nanmax(rsi_arr[i - lbL - lbR: i + 1]):
                for j in range(i - lbR - rLo, max(i - lbR - rHi, lbL) - 1, -1):
                    if j < 0:
                        break
                    w2 = rsi_arr[j - lbL: j + lbR + 1] if j >= lbL and j + lbR < n else None
                    if w2 is not None and rsi_arr[j] == np.nanmax(w2):
                        if rsi_arr[i - lbR] < rsi_arr[j] and high_arr[i - lbR] > high_arr[j]:
                            bear_div.iloc[i] = 1
                        break

        dataframe["bull_div"] = bull_div
        dataframe["bear_div"] = bear_div

        # ── 4. Kalit MA Stack + RSI(2) ───────────────────────────────
        dataframe["ma5"]   = c.rolling(5).mean()
        dataframe["ma200_4h"] = c.rolling(200).mean()
        dataframe["rsi2"]  = ta.RSI(dataframe, timeperiod=2)

        dataframe["kalit_long"]  = (
            (c > dataframe["ma200_4h"]) &
            (c < dataframe["ma5"])      &
            (dataframe["rsi2"] < 10)
        ).astype(int)

        dataframe["kalit_short"] = (
            (c < dataframe["ma200_4h"]) &
            (c > dataframe["ma5"])      &
            (dataframe["rsi2"] > 90)
        ).astype(int)

        # ── 5. Pi Cycle position score ───────────────────────────────
        # (pulled from 1d informative, used in signal scoring)
        # Available as dataframe["1d_pi_cycle_low"] etc after merge

        return dataframe

    # ══════════════════════════════════════════════════════════════════
    #  BUY SIGNAL
    # ══════════════════════════════════════════════════════════════════

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Multi-indicator buy signal.
        Requires confluence of indicators across timeframes.
        """
        conditions = []

        # ── MUST HAVE: Not in a squeeze (or squeeze just fired) ───────
        squeeze_ok = (
            (dataframe["squeeze_off"] == 1) |
            (dataframe["squeeze_on"].shift(1) == 1) & (dataframe["squeeze_off"] == 1)
        )
        conditions.append(squeeze_ok)

        # ── MUST HAVE: TTM momentum bullish ──────────────────────────
        conditions.append(dataframe["ttm_rising"] == 1)
        conditions.append(dataframe["ttm_pos"] == 1)

        # ── MUST HAVE: JMA RSI mid not overbought ────────────────────
        conditions.append(dataframe["jrsi1"] < 75)
        conditions.append(dataframe["jrsi1"] > 20)

        # ── SHOULD HAVE: At least one of these ───────────────────────
        extra = (
            (dataframe["bull_div"] == 1) |          # RSI bull divergence
            (dataframe["kalit_long"] == 1) |         # Kalit RSI(2) setup
            (dataframe["jrsi0"] < 30) |              # JMA RSI fast oversold
            (dataframe.get("1d_ema_bullish", 0) == 1) # Daily EMAs bullish
        )
        conditions.append(extra)

        # ── VOLUME: Not dead market ───────────────────────────────────
        conditions.append(dataframe["volume"] > 0)

        # Combine all conditions
        dataframe.loc[
            pd.concat([c.to_frame() if isinstance(c, pd.Series) else pd.Series(c, index=dataframe.index).to_frame()
                      for c in conditions], axis=1).all(axis=1),
            "enter_long"
        ] = 1

        return dataframe

    # ══════════════════════════════════════════════════════════════════
    #  SELL SIGNAL
    # ══════════════════════════════════════════════════════════════════

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Exit when momentum reverses or overbought.
        ROI and stoploss handle most exits — this catches trend reversals.
        """
        conditions = []

        # TTM momentum turning negative
        conditions.append(
            (dataframe["ttm_rising"] == 0) &
            (dataframe["ttm_pos"] == 0)
        )

        # OR: JMA RSI fast overbought
        conditions.append(dataframe["jrsi0"] > 80)

        # OR: RSI bear divergence
        conditions.append(dataframe["bear_div"] == 1)

        # OR: Pi Cycle HIGH signal (major top warning)
        pi_hi_col = "1d_pi_cycle_hi"
        if pi_hi_col in dataframe.columns:
            conditions.append(dataframe[pi_hi_col] == 1)

        dataframe.loc[
            pd.concat([c.to_frame() if isinstance(c, pd.Series)
                      else pd.Series(c, index=dataframe.index).to_frame()
                      for c in conditions], axis=1).any(axis=1),
            "exit_long"
        ] = 1

        return dataframe