# source: https://raw.githubusercontent.com/geovanybramanthya/AzuraAzureQuant/6ca3d7649eda9ce16f0c6fc60b8a5ca180bee6be/user_data/strategies/ApexDualAlpha_Omni_V12_Active.py
# directory_url: https://github.com/geovanybramanthya/AzuraAzureQuant/blob/main/user_data/strategies/
# User: geovanybramanthya
# Repository: AzuraAzureQuant
# --------------------"""
Github_geovanybramanthya_AzuraAzureQuant__ApexDualAlpha_Omni_V12_Active__20260918_180015 - Pullback Paradox Resolution & Chop-Resilient Calibration
Curated for Bybit Linear Futures 7-Pair Universe: BTC, ETH, SOL, ADA, DOGE, LINK, PAXG.

Key Structural Advancements:
1. Resolves the Pullback Paradox:
   - Macro trend orientation is defined structurally: EMA_50_4h > EMA_200_4h (macro bull)
   - Pullback tolerance: Price is permitted to test dynamic 4H EMA50 support zone (close >= EMA50_4h * 0.985)
2. 2-Candle Crossover Window:
   - Captures entries even if confirmation occurs 1 candle after instantaneous cross.
3. Adaptive ADX Regime:
   - Calibrated to > 14.0 (BTC > 15.0) to operate through consolidation compressions.
4. Preserves High-Yield ROI & Asymmetric Payoff:
   - Same Take Profit ladder, trailing stop, and dynamic leverage (7x BTC, 3x Altcoins).
"""

import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from freqtrade.strategy import (
    IStrategy,
    merge_informative_pair,
)
from freqtrade.persistence import Trade
import talib.abstract as ta
from technical import qtpylib


class Github_geovanybramanthya_AzuraAzureQuant__ApexDualAlpha_Omni_V12_Active__20260918_180015(IStrategy):
    INTERFACE_VERSION = 3

    can_short: bool = True
    timeframe = "1h"
    informative_timeframe = "4h"

    # High-Yield Precision Profit Ladder
    minimal_roi = {
        "0": 0.526,
        "467": 0.161,
        "688": 0.090,
        "1173": 0.011
    }

    stoploss = -0.297

    trailing_stop = True
    trailing_stop_positive = 0.217
    trailing_stop_positive_offset = 0.249
    trailing_only_offset_is_reached = False

    order_types = {
        "entry": "limit",
        "exit": "limit",
        "stoploss": "market",
        "stoploss_on_exchange": False
    }

    order_time_in_force = {
        "entry": "GTC",
        "exit": "GTC"
    }

    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False
    startup_candle_count: int = 200

    def leverage(self, pair: str, current_time: datetime, current_rate: float,
                 proposed_leverage: float, max_leverage: float, entry_tag: str,
                 side: str, **kwargs) -> float:
        if "BTC" in pair:
            return 7.0
        return 3.0

    def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
                    current_profit: float, **kwargs):
        trade_duration = (current_time - trade.open_date_utc).total_seconds() / 3600
        
        # 1. Fast profit taking for Shorts on sudden explosive dumps (> +4.5% profit)
        if trade.is_short and current_profit > 0.045:
            return "short_explosive_dump_tp"

        # 2. Surgical Volatility Scalp for PAXG (Gold bullion low-volatility cycle)
        if "PAXG" in pair and trade_duration >= 4.0 and current_profit >= 0.0065:
            return "paxg_volatility_tp"

        # 3. Structural Profit Banking for ETH (protecting against macro ETH/BTC bleed)
        if "ETH" in pair and trade_duration >= 16.0 and current_profit >= 0.0080:
            return "eth_profit_bank_tp"

        # 4. Stale decay cutoff at 36 hours
        if trade_duration > 36.0 and current_profit < -0.010:
            return "time_decay_stale_loss"
        
        # 5. Micro-profit closure at 48 hours
        if trade_duration > 48.0 and current_profit < 0.005:
            return "time_decay_48h_cutoff"
        
        return None

    def informative_pairs(self):
        pairs = self.dp.current_whitelist() if hasattr(self, "dp") and self.dp else []
        return [(pair, self.informative_timeframe) for pair in pairs]

    def informative_4h_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        
        # Structural macro trend orientation: decoupled from immediate close position
        dataframe["macro_bull_4h"] = dataframe["ema_50"] > dataframe["ema_200"]
        dataframe["macro_bear_4h"] = dataframe["ema_50"] < dataframe["ema_200"]
        return dataframe

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        if hasattr(self, "dp") and self.dp:
            inf_4h = self.dp.get_pair_dataframe(pair=metadata["pair"], timeframe=self.informative_timeframe)
            if not inf_4h.empty:
                inf_4h = self.informative_4h_indicators(inf_4h, metadata)
                dataframe = merge_informative_pair(dataframe, inf_4h, self.timeframe, self.informative_timeframe, ffill=True)

        for col in [f"macro_bull_4h_{self.informative_timeframe}", f"macro_bear_4h_{self.informative_timeframe}"]:
            if col not in dataframe.columns:
                dataframe[col] = True if "bull" in col else False
        if f"adx_{self.informative_timeframe}" not in dataframe.columns:
            dataframe[f"adx_{self.informative_timeframe}"] = 25.0
        if f"ema_50_{self.informative_timeframe}" not in dataframe.columns:
            dataframe[f"ema_50_{self.informative_timeframe}"] = dataframe["close"]

        # Streamlined 1H Indicators
        dataframe["ema_21"] = ta.EMA(dataframe, timeperiod=21)
        dataframe["ema_50"] = ta.EMA(dataframe, timeperiod=50)
        dataframe["ema_200"] = ta.EMA(dataframe, timeperiod=200)

        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)

        bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2.0)
        dataframe["bb_lowerband"] = bollinger["lower"]
        dataframe["bb_middleband"] = bollinger["mid"]
        dataframe["bb_upperband"] = bollinger["upper"]

        stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3)
        dataframe["slowk"] = stoch["slowk"]
        dataframe["slowd"] = stoch["slowd"]

        dataframe["volume_mean_20"] = dataframe["volume"].rolling(window=20).mean()
        dataframe["is_green"] = dataframe["close"] > dataframe["open"]
        dataframe["is_red"] = dataframe["close"] < dataframe["open"]
        dataframe["body_size"] = (dataframe["close"] - dataframe["open"]).abs()
        dataframe["lower_wick"] = dataframe[["open", "close"]].min(axis=1) - dataframe["low"]
        dataframe["upper_wick"] = dataframe["high"] - dataframe[["open", "close"]].max(axis=1)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[:, "enter_long"] = 0
        dataframe.loc[:, "enter_short"] = 0
        dataframe.loc[:, "enter_tag"] = ""

        pair = metadata.get("pair", "")

        # Long Configuration - Dynamic Chop Tolerance
        adx_long_gate = 15.0 if "BTC" in pair else 14.0
        rsi_min_long = 42.0 if "SOL" in pair else 40.0
        pullback_threshold = 1.008

        # 2-candle crossover window
        stoch_cross_up = qtpylib.crossed_above(dataframe["slowk"], dataframe["slowd"])
        cross_window_up = stoch_cross_up | ((dataframe["slowk"].shift(1) > dataframe["slowd"].shift(1)) & (dataframe["slowk"].shift(2) <= dataframe["slowd"].shift(2)))

        # 1. LONG: Paradox-Resolved Bullish Pullback
        long_conditions = (
            (dataframe[f"macro_bull_4h_{self.informative_timeframe}"] == True) &
            (dataframe[f"adx_{self.informative_timeframe}"] > adx_long_gate) &
            (dataframe["close"] >= dataframe[f"ema_50_{self.informative_timeframe}"] * 0.985) &
            (
                (dataframe["close"] <= dataframe["ema_21"] * pullback_threshold) |
                (dataframe["low"] <= dataframe["bb_middleband"])
            ) &
            (dataframe["close"] >= dataframe["ema_50"] * 0.988) &
            (dataframe["rsi"] >= rsi_min_long) &
            (dataframe["rsi"] <= 64.0) &
            cross_window_up &
            (dataframe["slowk"] < 58.0) &
            (
                (dataframe["is_green"] == True) |
                (dataframe["lower_wick"] > dataframe["body_size"] * 0.65)
            ) &
            (dataframe["volume"] > dataframe["volume_mean_20"] * 0.65)
        )

        # 2. SHORT: Resistance Exhaustion Fade
        if "BTC" not in pair:
            adx_short_gate = 16.0 if "SOL" in pair else 15.0
            rsi_min_short = 54.0 if "SOL" in pair else 52.0

            stoch_cross_down = qtpylib.crossed_below(dataframe["slowk"], dataframe["slowd"])
            cross_window_down = stoch_cross_down | ((dataframe["slowk"].shift(1) < dataframe["slowd"].shift(1)) & (dataframe["slowk"].shift(2) >= dataframe["slowd"].shift(2)))

            short_base = (
                (dataframe[f"macro_bear_4h_{self.informative_timeframe}"] == True) &
                (dataframe[f"adx_{self.informative_timeframe}"] > adx_short_gate) &
                (dataframe["close"] <= dataframe[f"ema_50_{self.informative_timeframe}"] * 1.015) &
                (
                    (dataframe["high"] >= dataframe["ema_21"] * 0.996) |
                    (dataframe["high"] >= dataframe["bb_middleband"] * 0.998)
                ) &
                (dataframe["close"] <= dataframe["ema_50"] + 0.35 * dataframe["atr"]) &
                (dataframe["rsi"] >= rsi_min_short) &
                (dataframe["rsi"] <= 68.0) &
                cross_window_down &
                (dataframe["slowk"] > 52.0) &
                (
                    (dataframe["is_red"] == True) |
                    (dataframe["upper_wick"] > dataframe["body_size"] * 0.65)
                ) &
                (dataframe["volume"] > dataframe["volume_mean_20"] * 0.65)
            )

            dataframe.loc[short_base & ~long_conditions, ["enter_short", "enter_tag"]] = (1, "short_atr_fade_alpha")

        dataframe.loc[long_conditions, ["enter_long", "enter_tag"]] = (1, "long_pullback_alpha")

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[:, "exit_long"] = 0
        dataframe.loc[:, "exit_short"] = 0
        dataframe.loc[:, "exit_tag"] = ""

        long_exit = (
            (dataframe["rsi"] >= 71) &
            (dataframe["close"] >= dataframe["bb_upperband"] * 0.998)
        )

        short_exit = (
            (dataframe["rsi"] <= 28) &
            (dataframe["close"] <= dataframe["bb_lowerband"] * 1.002)
        )

        dataframe.loc[long_exit, ["exit_long", "exit_tag"]] = (1, "exit_upper_bb_tp")
        dataframe.loc[short_exit, ["exit_short", "exit_tag"]] = (1, "exit_lower_bb_tp")

        return dataframe
