# source: https://raw.githubusercontent.com/jatinpruthvi/forex-crypto/1411e0e4b4ccec7546b7bb186d02f7ca67ec6a09/services/freqtrade/user_data/strategies/OptionB_MultiAsset_Portfolio.py
# directory_url: https://github.com/jatinpruthvi/forex-crypto/blob/main/services/freqtrade/user_data/strategies/
# User: jatinpruthvi
# Repository: forex-crypto
# --------------------"""
Option B: All-Weather Multi-Strategy Concurrent Portfolio (BTC, SOL, BNB, LINK)
- Capital Allocation Weights:
    • 40% BTC/USDT: 12-Bar Compressed Donchian Breakout
    • 30% SOL/USDT: Pin Bar & Bollinger Band Reversal Scalp
    • 15% BNB/USDT: Institutional Reversal Mean-Reversion
    • 15% LINK/USDT: Dynamic Trendline & Pivot Breakout
- Multi-Timeframe Alignment: 4H 50-EMA & VWAP macro filtration on each asset
- Execution Timeframe: 15-Minute (M15)
- Concurrent Open Positions: Up to 4 simultaneous non-correlated positions
- Target: +17.06% Compounded Monthly ROI, 100+ trades/month, ~8.40% Max Drawdown
"""
from typing import Optional, TYPE_CHECKING, Dict, Any
from pandas import DataFrame
import numpy as np
import pandas as pd
from datetime import datetime, timezone

try:
    from freqtrade.strategy import IStrategy, informative
except ImportError:
    class IStrategy:
        def __init__(self, config: Optional[Dict[str, Any]] = None):
            self.config = config or {}
    def informative(timeframe):
        def decorator(f):
            return f
        return decorator

if TYPE_CHECKING:
    from freqtrade.persistence import Trade


class Github_jatinpruthvi_forex_crypto__OptionB_MultiAsset_Portfolio__20261105_061721(IStrategy):
    INTERFACE_VERSION = 3
    timeframe = "15m"
    can_short: bool = True

    def __init__(self, config: Optional[Dict[str, Any]] = None):
        cfg = dict(config or {})
        if "candle_type_def" not in cfg:
            cfg["candle_type_def"] = "spot"
        super().__init__(cfg)
        # Custom stake weighting per pair
        self.custom_stake_weights = {
            "BTC/USDT:USDT": 0.40,
            "SOL/USDT:USDT": 0.30,
            "BNB/USDT:USDT": 0.15,
            "LINK/USDT:USDT": 0.15,
            "BTC/USDT": 0.40,
            "SOL/USDT": 0.30,
            "BNB/USDT": 0.15,
            "LINK/USDT": 0.15,
        }

    # Global emergency stop
    stoploss = -0.05
    use_custom_stoploss = True
    minimal_roi = {}

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

    # Parameters
    atr_period = 14
    adx_period = 14
    adx_min = 18.0
    trailing_atr_mult = 2.5

    @informative("4h")
    def populate_indicators_4h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema50"] = dataframe["close"].ewm(span=50).mean()
        typical_p = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3.0
        dataframe["vwap"] = (typical_p * dataframe["volume"]).rolling(30).sum() / dataframe["volume"].rolling(30).sum().replace(0, 1)
        dataframe["macro_trend"] = np.where(dataframe["close"] > dataframe["ema50"], 1, -1)
        return dataframe

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

        tr = np.maximum(high - low, np.maximum(abs(high - close.shift(1)), abs(low - close.shift(1))))
        # no .bfill(): warmup rows must not borrow future ATR (lookahead)
        dataframe["atr"] = tr.rolling(14).mean()
        dataframe["ema20"] = close.ewm(span=20).mean()
        dataframe["ema50"] = close.ewm(span=50).mean()

        typical_p = (high + low + close) / 3.0
        dataframe["vwap"] = (typical_p * vol).rolling(30).sum() / vol.rolling(30).sum().replace(0, 1)

        # ADX Regime Filter
        plus_dm = high.diff()
        minus_dm = low.diff()
        plus_dm[plus_dm < 0] = 0
        minus_dm[minus_dm > 0] = 0
        minus_dm = minus_dm.abs()
        plus_di = 100 * (plus_dm.ewm(alpha=1/self.adx_period).mean() / dataframe["atr"])
        minus_di = 100 * (minus_dm.ewm(alpha=1/self.adx_period).mean() / dataframe["atr"])
        dx = 100 * (abs(plus_di - minus_di) / (plus_di + minus_di).replace(0, 1))
        dataframe["adx"] = dx.ewm(alpha=1/self.adx_period).mean()

        # BTC: 12-bar Donchian
        dataframe["donchian_high"] = high.rolling(12).max().shift(1)
        dataframe["donchian_low"] = low.rolling(12).min().shift(1)

        # SOL & BNB: Bollinger Bands for Reversal
        bb_mid = close.rolling(20).mean()
        bb_std = close.rolling(20).std()
        dataframe["bb_lower"] = bb_mid - 2.0 * bb_std
        dataframe["bb_upper"] = bb_mid + 2.0 * bb_std

        # LINK: Trendline Pivot Breakout (8-bar pivot + ATR expansion)
        dataframe["pivot_high"] = high.rolling(8).max().shift(1)
        dataframe["pivot_low"] = low.rolling(8).min().shift(1)
        dataframe["atr_expanding"] = tr > (dataframe["atr"] * 1.15)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        pair = metadata.get("pair", "")
        macro = dataframe.get("macro_trend_4h", pd.Series(1, index=dataframe.index))

        # 1. BTC: Donchian 12 Breakout
        if "BTC" in pair:
            long_cond = (
                (macro == 1) &
                (dataframe["close"] > dataframe["open"]) &
                (dataframe["close"] > dataframe["vwap"]) &
                (dataframe["ema20"] > dataframe["ema50"]) &
                (dataframe["high"] >= dataframe["donchian_high"]) &
                (dataframe["adx"] >= self.adx_min)
            )
            short_cond = (
                (macro == -1) &
                (dataframe["close"] < dataframe["open"]) &
                (dataframe["close"] < dataframe["vwap"]) &
                (dataframe["ema20"] < dataframe["ema50"]) &
                (dataframe["low"] <= dataframe["donchian_low"]) &
                (dataframe["adx"] >= self.adx_min)
            )
            dataframe.loc[long_cond, "enter_tag"] = "btc_donchian_long"
            dataframe.loc[short_cond, "enter_tag"] = "btc_donchian_short"

        # 2. SOL & BNB: Pin Bar & Bollinger Reversal Scalp
        elif "SOL" in pair or "BNB" in pair:
            long_cond = (macro == 1) & (dataframe["low"] <= dataframe["bb_lower"]) & (dataframe["close"] > dataframe["open"])
            short_cond = (macro == -1) & (dataframe["high"] >= dataframe["bb_upper"]) & (dataframe["close"] < dataframe["open"])
            tag_prefix = "sol" if "SOL" in pair else "bnb"
            dataframe.loc[long_cond, "enter_tag"] = f"{tag_prefix}_reversal_long"
            dataframe.loc[short_cond, "enter_tag"] = f"{tag_prefix}_reversal_short"

        # 3. LINK: Dynamic Trendline Pivot Breakout
        elif "LINK" in pair:
            long_cond = (macro == 1) & (dataframe["high"] > dataframe["pivot_high"]) & dataframe["atr_expanding"] & (dataframe["close"] > dataframe["vwap"])
            short_cond = (macro == -1) & (dataframe["low"] < dataframe["pivot_low"]) & dataframe["atr_expanding"] & (dataframe["close"] < dataframe["vwap"])
            dataframe.loc[long_cond, "enter_tag"] = "link_pivot_long"
            dataframe.loc[short_cond, "enter_tag"] = "link_pivot_short"

        else:
            long_cond = pd.Series(False, index=dataframe.index)
            short_cond = pd.Series(False, index=dataframe.index)

        dataframe.loc[long_cond, "enter_long"] = 1
        dataframe.loc[short_cond, "enter_short"] = 1
        return dataframe

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

    def custom_stoploss(
        self,
        pair: str,
        trade: "Trade",
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs,
    ) -> Optional[float]:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return None

        last_candle = dataframe.iloc[-1]
        atr = float(last_candle["atr"])
        entry_price = float(trade.open_rate)
        is_short = getattr(trade, "is_short", False)
        atr_pct = atr / entry_price

        # Asset-tailored ATR parameters
        if "BTC" in pair or "LINK" in pair:
            sl_mult = 1.3
            be_active_mult = 1.0
        else: # SOL, BNB
            sl_mult = 1.2
            be_active_mult = 1.0

        sl_pct = sl_mult * atr_pct
        be_active_pct = be_active_mult * atr_pct
        be_buffer_pct = 0.25 * atr_pct

        # Dynamic Trailing Ratchet: Ratchet SL from current price once profit >= 1.5x ATR
        trailing_sl_mult = getattr(self, "trailing_atr_mult", 2.5)
        if current_profit >= (1.5 * atr_pct):
            trailing_stop_distance = -(trailing_sl_mult * atr) / current_rate
            if is_short:
                target_be_sl = entry_price - (0.25 * atr)
                be_stop_distance = (current_rate - target_be_sl) / current_rate
            else:
                target_be_sl = entry_price + (0.25 * atr)
                be_stop_distance = (target_be_sl - current_rate) / current_rate
            be_stop_distance = min(-0.0001, be_stop_distance)
            if trailing_stop_distance > be_stop_distance:
                return trailing_stop_distance
            return be_stop_distance

        if current_profit >= be_active_pct:
            if is_short:
                target_be_sl = entry_price - (0.25 * atr)
                be_stop_distance = (current_rate - target_be_sl) / current_rate
            else:
                target_be_sl = entry_price + (0.25 * atr)
                be_stop_distance = (target_be_sl - current_rate) / current_rate
            return min(-0.0001, be_stop_distance)

        return -abs(sl_pct)

    def custom_exit(
        self,
        pair: str,
        trade: "Trade",
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[str]:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return None

        last_candle = dataframe.iloc[-1]
        atr = last_candle["atr"]
        tp_mult = 2.0 if ("BTC" in pair or "LINK" in pair) else 1.8
        tp_pct = (atr * tp_mult) / trade.open_rate

        if current_profit >= tp_pct:
            return "tp_atr_hit"

        return None