# source: https://raw.githubusercontent.com/geovanybramanthya-bot/AzuraAzureQuant/5395c3a9395f9137ef13617814a8b2443dad0665/user_data/strategies/ApexDualAlpha_Omni_V7_Evolution.py
# directory_url: https://github.com/geovanybramanthya-bot/AzuraAzureQuant/blob/main/user_data/strategies/
# User: geovanybramanthya-bot
# Repository: AzuraAzureQuant
# --------------------"""
Github_geovanybramanthya_bot_AzuraAzureQuant__ApexDualAlpha_Omni_V7_Evolution__20260913_152406 - Cognitive Multi-Pattern & Structural Invalidation Engine
Curated for Bybit Linear Futures 7-Pair Universe: BTC, ETH, SOL, ADA, DOGE, LINK, PAXG.

Overcoming the 4 Logical Fallacies of V6:
1. Fallacy 1 (Sunk Cost / Stale Loss):
   - Replaces the passive 36h stale hold with an active Structural Invalidation Exit.
   - If trade duration > 20h, current_profit < -0.015 (-4.5% leveraged), AND price < 1H EMA 50:
     exit immediately via "structural_invalidation_cut". Unfreezes capital slots for new alpha!
2. Fallacy 2 (Correlated Altcoin Sinkholes):
   - Micro-Pattern BTC Flash-Dump Shield: Altcoins do not open Longs if BTC's current 1H candle
     is an aggressive red sell bar (close < open and close < btc_ema_21). Prevents multi-coin wipes.
3. Fallacy 3 (Distribution Volume Illusion):
   - Requires confirmed absorption on pullbacks: green candle OR strong hammer lower wick (> body_size * 0.85).
4. Fallacy 4 (Fearful Under-Trading Solved via Pattern 2 Breakout):
   - Adds Volatility Squeeze Expansion Engine (Bollinger Bandwidth contraction breakout)
     to capture rapid momentum legs that never pull back to EMA21, expanding trade frequency!
"""

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_bot_AzuraAzureQuant__ApexDualAlpha_Omni_V7_Evolution__20260913_152406(IStrategy):
    INTERFACE_VERSION = 3

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

    # Fee-Proof Wide Profit Ladder
    minimal_roi = {
        "0": 0.526,
        "467": 0.161,
        "688": 0.090,
        "1173": 0
    }

    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
        
        # 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"

        # Structural Invalidation Cut (Cuts failed bounces early to free capital)
        # If trade is > 20h and losing > -1.5% raw (-4.5% on 3x stake), unfreeze slot!
        if trade_duration > 20.0 and current_profit < -0.015:
            return "structural_invalidation_cut"

        # Standard stale cutoff for mild drifts at 36h
        if trade_duration > 36.0 and current_profit < -0.005:
            return "time_decay_stale_loss"
        
        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 []
        informative = [(pair, self.informative_timeframe) for pair in pairs]
        # Ensure BTC 1H and 4H are available for micro-regime protection
        if ("BTC/USDT:USDT", self.timeframe) not in informative:
            informative.append(("BTC/USDT:USDT", self.timeframe))
        if ("BTC/USDT:USDT", self.informative_timeframe) not in informative:
            informative.append(("BTC/USDT:USDT", self.informative_timeframe))
        return informative

    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)
        dataframe["macro_bull_4h"] = (dataframe["close"] > dataframe["ema_50"]) & (dataframe["ema_50"] > dataframe["ema_200"])
        dataframe["macro_bear_4h"] = (dataframe["close"] < dataframe["ema_50"]) & (dataframe["ema_50"] < dataframe["ema_200"])
        return dataframe

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        if hasattr(self, "dp") and self.dp:
            # Pair 4H informative
            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)

            # BTC 1H Micro-trend Informative for Altcoins
            if metadata["pair"] != "BTC/USDT:USDT":
                btc_1h = self.dp.get_pair_dataframe(pair="BTC/USDT:USDT", timeframe=self.timeframe)
                if not btc_1h.empty:
                    btc_1h["btc_ema_21"] = ta.EMA(btc_1h, timeperiod=21)
                    btc_1h["btc_dumping"] = (btc_1h["close"] < btc_1h["open"]) & (btc_1h["close"] < btc_1h["btc_ema_21"])
                    dataframe = merge_informative_pair(dataframe, btc_1h, self.timeframe, self.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"btc_dumping_{self.timeframe}" not in dataframe.columns:
            dataframe[f"btc_dumping_{self.timeframe}"] = False

        # 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"]
        dataframe["bb_bandwidth"] = (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe["bb_middleband"]

        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", "")
        is_btc = "BTC" in pair

        # Long Thresholds
        if is_btc:
            adx_long_gate = 24.0
        elif "ETH" in pair or "SOL" in pair:
            adx_long_gate = 22.5
        else:
            adx_long_gate = 21.0

        rsi_min_long = 44.0 if "SOL" in pair else 43.0
        pullback_threshold = 1.006

        # Microstructure Altcoin Protection: do not long if BTC is flash dumping
        btc_safe = True if is_btc else (dataframe[f"btc_dumping_{self.timeframe}"] == False)

        # 1. LONG PATTERN A: Proven High-Conviction Pullback Bounce
        long_pullback = (
            btc_safe &
            (dataframe[f"macro_bull_4h_{self.informative_timeframe}"] == True) &
            (dataframe[f"adx_{self.informative_timeframe}"] > adx_long_gate) &
            (
                (dataframe["close"] <= dataframe["ema_21"] * pullback_threshold) |
                (dataframe["low"] <= dataframe["bb_middleband"])
            ) &
            (dataframe["close"] >= dataframe["ema_50"] * 0.990) &
            (dataframe["rsi"] >= rsi_min_long) &
            (dataframe["rsi"] <= 63.0) &
            (qtpylib.crossed_above(dataframe["slowk"], dataframe["slowd"])) &
            (dataframe["slowk"] < 54.0) &
            (
                (dataframe["is_green"] == True) |
                (dataframe["lower_wick"] > dataframe["body_size"] * 0.80)
            ) &
            (dataframe["volume"] > dataframe["volume_mean_20"] * 0.70)
        )

        # 2. LONG PATTERN B: Volatility Expansion Breakout Alpha (New Trade Frequency Driver)
        # Captures fast runaway trends that never pull back to EMA21
        long_breakout = (
            btc_safe &
            (dataframe[f"macro_bull_4h_{self.informative_timeframe}"] == True) &
            (dataframe[f"adx_{self.informative_timeframe}"] > (adx_long_gate + 2.0)) &
            (dataframe["close"] > dataframe["bb_upperband"]) &
            (dataframe["close"] > dataframe["ema_21"] * 1.010) &
            (dataframe["rsi"] >= 58.0) &
            (dataframe["rsi"] <= 68.0) &
            (dataframe["volume"] > dataframe["volume_mean_20"] * 1.40) &
            (dataframe["is_green"] == True) &
            (dataframe["close"] > dataframe["close"].shift(1))
        )

        # 3. SHORT PATTERN: Precision Resistance Exhaustion Fade
        # BTC is kept Long-Only to eliminate 7.0x leverage counter-trend risk
        if not is_btc:
            if "SOL" in pair:
                adx_short_gate = 25.0
                rsi_min_short = 56.0
                require_below_ema200 = True
            else:
                adx_short_gate = 22.0
                rsi_min_short = 53.0
                require_below_ema200 = False

            short_base = (
                (dataframe[f"macro_bear_4h_{self.informative_timeframe}"] == True) &
                (dataframe[f"adx_{self.informative_timeframe}"] > adx_short_gate) &
                (
                    (dataframe["high"] >= dataframe["ema_21"] * 0.996) |
                    (dataframe["high"] >= dataframe["bb_middleband"] * 0.998)
                ) &
                (dataframe["close"] <= dataframe["ema_50"] + 0.30 * dataframe["atr"]) &
                (dataframe["rsi"] >= rsi_min_short) &
                (dataframe["rsi"] <= 67.0) &
                (qtpylib.crossed_below(dataframe["slowk"], dataframe["slowd"])) &
                (dataframe["slowk"] > 55.0) &
                (
                    (dataframe["is_red"] == True) |
                    (dataframe["upper_wick"] > dataframe["body_size"] * 0.70)
                ) &
                (dataframe["volume"] > dataframe["volume_mean_20"] * 0.72)
            )

            if require_below_ema200:
                short_conditions = short_base & (dataframe["close"] < dataframe["ema_200"])
            else:
                short_conditions = short_base

            dataframe.loc[short_conditions & ~long_pullback & ~long_breakout, ["enter_short", "enter_tag"]] = (1, "short_atr_fade_alpha")

        dataframe.loc[long_pullback, ["enter_long", "enter_tag"]] = (1, "long_pullback_alpha")
        dataframe.loc[long_breakout & ~long_pullback, ["enter_long", "enter_tag"]] = (1, "long_breakout_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"] = ""

        # Dynamic Bollinger Runner Exit (Yield-Boosting)
        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