# source: https://raw.githubusercontent.com/geovanybramanthya/AzuraAzureQuant/6ca3d7649eda9ce16f0c6fc60b8a5ca180bee6be/user_data/strategies/ApexDualAlpha_Omni_V14_HighVelocity.py
# directory_url: https://github.com/geovanybramanthya/AzuraAzureQuant/blob/main/user_data/strategies/
# User: geovanybramanthya
# Repository: AzuraAzureQuant
# --------------------﻿"""
Github_geovanybramanthya_AzuraAzureQuant__ApexDualAlpha_Omni_V14_HighVelocity__20260918_180015 - Tri-Engine Confluence & Maker Velocity Strategy
Optimized for Bybit Linear Futures 7-Pair Universe: BTC, ETH, SOL, ADA, DOGE, LINK, PAXG.
"""

import numpy as np
import pandas as pd
from pandas import DataFrame
from datetime import datetime
from typing import Optional
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_V14_HighVelocity__20260918_180015(IStrategy):
    INTERFACE_VERSION = 3

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

    # Fast-Harvest Capital Velocity Profit Ladder
    minimal_roi = {
        "0": 0.380,
        "180": 0.140,
        "360": 0.080,
        "720": 0.040,
        "1440": 0.010
    }

    stoploss = -0.297

    trailing_stop = True
    trailing_stop_positive = 0.160
    trailing_stop_positive_offset = 0.200
    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_entry_price(self, pair: str, current_time: datetime, proposed_rate: float,
                           entry_tag: Optional[str], side: str, **kwargs) -> float:
        # Passive resting limit order for maker rebate and support fill
        if side == "long":
            return round(proposed_rate * 0.9990, 6)
        else:
            return round(proposed_rate * 1.0010, 6)

    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 explosive drops (> +4.0%)
        if trade.is_short and current_profit > 0.040:
            return "short_explosive_dump_tp"

        # 2. Surgical Volatility Scalp for PAXG
        if "PAXG" in pair and trade_duration >= 4.0 and current_profit >= 0.0065:
            return "paxg_volatility_tp"

        # 3. Structural Profit Banking for ETH
        if "ETH" in pair and trade_duration >= 14.0 and current_profit >= 0.0080:
            return "eth_profit_bank_tp"

        # 4. Rapid Stagnation Cutoff at 18 hours (unfreezes execution slots)
        if trade_duration > 18.0 and current_profit < -0.015:
            return "stagnation_18h_cutoff"

        # 5. Flat Time-Decay Cutoff at 32 hours
        if trade_duration > 32.0 and current_profit < 0.004:
            return "time_decay_32h_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)
        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:
            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}"] = 22.0

        dataframe["ema_9"] = ta.EMA(dataframe, timeperiod=9)
        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_width"] = (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"]

        # 2-Candle Stochastic Inflection Window
        stoch_cross_up_0 = qtpylib.crossed_above(dataframe["slowk"], dataframe["slowd"])
        stoch_cross_up_1 = qtpylib.crossed_above(dataframe["slowk"].shift(1), dataframe["slowd"].shift(1))
        dataframe["k_cross_up_window"] = stoch_cross_up_0 | (stoch_cross_up_1 & (dataframe["slowk"] > dataframe["slowd"]))

        stoch_cross_down_0 = qtpylib.crossed_below(dataframe["slowk"], dataframe["slowd"])
        stoch_cross_down_1 = qtpylib.crossed_below(dataframe["slowk"].shift(1), dataframe["slowd"].shift(1))
        dataframe["k_cross_down_window"] = stoch_cross_down_0 | (stoch_cross_down_1 & (dataframe["slowk"] < dataframe["slowd"]))

        # Squeeze detection: BB Width bottom 25th percentile compression
        dataframe["bb_width_q25"] = dataframe["bb_width"].rolling(50).quantile(0.25)
        dataframe["in_squeeze"] = dataframe["bb_width"] < dataframe["bb_width_q25"]
        dataframe["squeeze_released"] = (dataframe["in_squeeze"].shift(1) == True) & (dataframe["in_squeeze"] == False)

        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", "")

        adx_long_gate = 19.0 if ("BTC" in pair or "SOL" in pair) else 18.0
        rsi_min_long = 42.0 if "SOL" in pair else 40.0

        # Archetype 1: Trend Pullback Alpha
        e1_long = (
            (dataframe[f"macro_bull_4h_{self.informative_timeframe}"] == True) &
            (dataframe[f"adx_{self.informative_timeframe}"] > adx_long_gate) &
            (
                (dataframe["close"] <= dataframe["ema_21"] * 1.008) |
                (dataframe["low"] <= dataframe["bb_middleband"])
            ) &
            (dataframe["close"] >= dataframe["ema_50"] * 0.988) &
            (dataframe["rsi"] >= rsi_min_long) &
            (dataframe["rsi"] <= 64.0) &
            (dataframe["k_cross_up_window"] == True) &
            (dataframe["slowk"] < 56.0) &
            (
                (dataframe["is_green"] == True) |
                (dataframe["lower_wick"] > dataframe["body_size"] * 0.65)
            ) &
            (dataframe["volume"] > dataframe["volume_mean_20"] * 0.65)
        )

        # Archetype 2: Volatility Squeeze Breakout Alpha
        e2_squeeze = (
            (dataframe["squeeze_released"] == True) &
            (dataframe[f"macro_bull_4h_{self.informative_timeframe}"] == True) &
            (dataframe["close"] > dataframe["bb_middleband"]) &
            (dataframe["close"] < dataframe["bb_upperband"] * 0.998) &
            (dataframe["close"] > dataframe["ema_9"]) &
            (dataframe["rsi"] >= 52.0) &
            (dataframe["rsi"] <= 68.0) &
            (dataframe["slowk"] > dataframe["slowd"]) &
            (dataframe["is_green"] == True) &
            (dataframe["volume"] > dataframe["volume_mean_20"] * 1.05)
        )

        long_conditions = e1_long | e2_squeeze
        dataframe.loc[e1_long, ["enter_long", "enter_tag"]] = (1, "long_pullback_alpha")
        dataframe.loc[e2_squeeze & ~e1_long, ["enter_long", "enter_tag"]] = (1, "long_squeeze_expansion")

        # Archetype 3: Active Resistance Exhaustion Short
        if "BTC" not in pair:
            adx_short_gate = 20.0 if "SOL" in pair else 18.5
            rsi_min_short = 54.0 if "SOL" in pair else 52.0

            short_conditions = (
                (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.32 * dataframe["atr"]) &
                (dataframe["rsi"] >= rsi_min_short) &
                (dataframe["rsi"] <= 68.0) &
                (dataframe["k_cross_down_window"] == True) &
                (dataframe["slowk"] > 52.0) &
                (
                    (dataframe["is_red"] == True) |
                    (dataframe["upper_wick"] > dataframe["body_size"] * 0.65)
                ) &
                (dataframe["volume"] > dataframe["volume_mean_20"] * 0.68)
            )

            if "SOL" in pair:
                short_conditions = short_conditions & (dataframe["close"] < dataframe["ema_200"])

            dataframe.loc[short_conditions & ~long_conditions, ["enter_short", "enter_tag"]] = (1, "short_atr_fade_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"] >= 72) &
            (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
