# source: https://raw.githubusercontent.com/4tie/fortiesr/d5cc40760e3cbaff381058a0ea8733d2be469578/backend/services/auto_quant/generator/advanced_generators.py
# directory_url: https://github.com/4tie/fortiesr/blob/main/backend/services/auto_quant/generator/
# User: 4tie
# Repository: fortiesr
# --------------------"""Advanced strategy template generators for the Auto-Quant pipeline.

Produces complex freqtrade strategies with sophisticated features including
omni-strategies with Boolean switches, ensemble voting with weighted consensus,
and advanced position sizing.
"""

from __future__ import annotations


def generate_strategy_source_omni(class_name: str, timeframe: str = "5m", winning_pairs: list | None = None) -> str:
    """Return a single massive Omni-Strategy with Boolean switches for every indicator.

    Hyperopt searches CategoricalParameter (True/False) switches to decide which
    indicators participate in the entry signal, plus numeric thresholds for each one.
    This guarantees exhaustive search over all indicator combinations:
      RSI · MACD · Bollinger Bands · EMA 50/200 cross · ADX · ATR trend filter

    ROI / stoploss defaults are calibrated to the chosen timeframe so the genetic
    algorithm starts from a sensible prior for the target trading style.

    Args:
        class_name: Name of the strategy class
        timeframe: Candle timeframe for ROI/stoploss calibration
        winning_pairs: List of winning pairs with ATR values for position sizing
    """
    # Timeframe-aware ROI / stoploss / trailing defaults
    scalping_tfs = {"1m", "3m", "5m", "15m"}
    intraday_tfs = {"30m"}
    # swing / position = everything else

    if timeframe in scalping_tfs:
        roi_block = '{"0": 0.03, "10": 0.02, "30": 0.01, "60": 0}'
        stoploss_val = "-0.025"
        trailing_val = "True"
        trailing_stop_pos = "0.008"
        trailing_only_offset = "False"
    elif timeframe in intraday_tfs:
        roi_block = '{"0": 0.06, "30": 0.04, "60": 0.02, "120": 0}'
        stoploss_val = "-0.04"
        trailing_val = "True"
        trailing_stop_pos = "0.012"
        trailing_only_offset = "False"
    elif timeframe in ("4h", "1d"):
        roi_block = '{"0": 0.20, "120": 0.10, "300": 0.05, "600": 0}'
        stoploss_val = "-0.10"
        trailing_val = "True"
        trailing_stop_pos = "0.025"
        trailing_only_offset = "True"
    else:  # 1h default swing
        roi_block = '{"0": 0.12, "60": 0.06, "180": 0.03, "360": 0}'
        stoploss_val = "-0.07"
        trailing_val = "True"
        trailing_stop_pos = "0.018"
        trailing_only_offset = "False"

    return f'''\
# Auto-generated by Strategy Lab — Auto-Quant Factory (Omni-Strategy Edition)
# One massive template with Boolean switches for every indicator.
# Hyperopt mutates True/False switches + numeric thresholds to find the best combo.
# Optimise with: freqtrade hyperopt --strategy Github_4tie_fortiesr__advanced_generators__20260709_213639 --spaces buy roi stoploss

from __future__ import annotations
from datetime import datetime
from functools import reduce
import operator

from freqtrade.strategy import CategoricalParameter, IntParameter, DecimalParameter, IStrategy, Trade, stoploss_from_open
from pandas import DataFrame
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
import numpy as np


class Github_4tie_fortiesr__advanced_generators__20260709_213639(IStrategy):
    INTERFACE_VERSION: int = 3

    minimal_roi = {roi_block}

    stoploss = {stoploss_val}
    timeframe = "{timeframe}"
    trailing_stop = False
    use_custom_stoploss = True
    process_only_new_candles = True

    # ── Boolean indicator switches (True = active, False = ignored) ───────────
    use_rsi       = CategoricalParameter([True, False], default=True,  space="buy", optimize=True)
    use_macd      = CategoricalParameter([True, False], default=True,  space="buy", optimize=True)
    use_bb        = CategoricalParameter([True, False], default=True,  space="buy", optimize=True)
    use_ema_cross = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_adx       = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_atr       = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_stoch     = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_cci       = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_mfi       = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_willr     = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_obv       = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_sar       = CategoricalParameter([True, False], default=False, space="buy", optimize=True)
    use_natr      = CategoricalParameter([True, False], default=False, space="buy", optimize=True)

    # ── Per-indicator numeric thresholds ─────────────────────────────────────
    rsi_threshold  = IntParameter(20, 50, default=30, space="buy", optimize=True)
    adx_threshold  = IntParameter(15, 40, default=25, space="buy", optimize=True)
    macd_hist_min  = DecimalParameter(0.0, 0.005, default=0.0,   decimals=4, space="buy", optimize=True)
    bb_factor      = DecimalParameter(0.95, 1.0,  default=1.0,   decimals=3, space="buy", optimize=True)
    ema_fast_p     = IntParameter(5, 30,   default=50,  space="buy", optimize=False)
    ema_slow_p     = IntParameter(50, 250, default=200, space="buy", optimize=False)
    stoch_threshold = IntParameter(20, 40, default=30, space="buy", optimize=True)
    cci_threshold  = IntParameter(-100, -50, default=-100, space="buy", optimize=True)
    mfi_threshold  = IntParameter(20, 40, default=30, space="buy", optimize=True)
    willr_threshold = IntParameter(-80, -60, default=-80, space="buy", optimize=True)

    # ── ATR regime window ─────────────────────────────────────────────────────
    atr_window     = IntParameter(10, 60, default=20, space="buy", optimize=True)

    # ── Stepped trailing profit lock-in tiers ─────────────────────────────────
    ts_tier1_trigger = DecimalParameter(0.020, 0.040, default=0.030, decimals=3, space="buy", optimize=True)
    ts_tier1_lock    = DecimalParameter(0.001, 0.010, default=0.003, decimals=3, space="buy", optimize=True)
    ts_tier2_trigger = DecimalParameter(0.050, 0.080, default=0.060, decimals=3, space="buy", optimize=True)
    ts_tier2_lock    = DecimalParameter(0.030, 0.050, default=0.035, decimals=3, space="buy", optimize=True)
    ts_tier3_trigger = DecimalParameter(0.100, 0.180, default=0.120, decimals=3, space="buy", optimize=True)
    ts_tier3_lock    = DecimalParameter(0.060, 0.100, default=0.080, decimals=3, space="buy", optimize=True)

    # ── ATR-based position sizing for winning pairs ───────────────────────────
    # ATR dictionary for each winning pair (pair -> ATR value)
    atr_dict = {{}}
    target_risk_pct = 0.02  # 2% risk per trade

    # ── Trading window filter (Session Optimizer) ───────────────────────────
    # Blocked hours and days based on historical performance analysis
    blocked_hours = []  # e.g., [0, 1, 2, 3, 4, 5] for early morning UTC
    blocked_days = []   # e.g., [5, 6] for weekends (0=Monday, 6=Sunday)

    def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime,
                        current_rate: float, current_profit: float,
                        after_fill: bool, **kwargs) -> float | None:
        if current_profit >= self.ts_tier3_trigger.value:
            return stoploss_from_open(
                self.ts_tier3_lock.value,
                current_profit,
                is_short=trade.is_short,
                leverage=trade.leverage,
            ) or 1
        if current_profit >= self.ts_tier2_trigger.value:
            return stoploss_from_open(
                self.ts_tier2_lock.value,
                current_profit,
                is_short=trade.is_short,
                leverage=trade.leverage,
            ) or 1
        if current_profit >= self.ts_tier1_trigger.value:
            return stoploss_from_open(
                self.ts_tier1_lock.value,
                current_profit,
                is_short=trade.is_short,
                leverage=trade.leverage,
            ) or 1
        return None

    def custom_stake_amount(self, pair: str, current_time, current_rate: float,
                            proposed_stake: float, min_stake: float | None, max_stake: float,
                            leverage: float, entry_tag: str | None, side: str, **kwargs) -> float:
        """Calculate position size based on ATR to normalize risk across pairs.

        Formula: position_size = base_stake * (target_risk_pct / (ATR / current_price))
        High-volatility pairs get smaller position sizes to keep risk normalized.
        """
        # Division by zero protection
        if current_rate <= 0:
            return proposed_stake
        
        atr = self.atr_dict.get(pair, current_rate * 0.02)  # fallback to 2% of price
        atr_pct = atr / current_rate
        
        # Division by zero protection for ATR
        if atr_pct <= 0:
            return proposed_stake
        
        position_size = proposed_stake * (self.target_risk_pct / atr_pct)
        # Clamp to reasonable bounds (0.5x to 2.0x of base stake)
        position_size = min(max_stake, max(min_stake if min_stake else proposed_stake * 0.5, position_size))
        position_size = min(position_size, proposed_stake * 2.0)
        return position_size

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # RSI
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)

        # MACD
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe["macd"]       = macd["macd"]
        dataframe["macdsignal"] = macd["macdsignal"]
        dataframe["macdhist"]   = macd["macdhist"]

        # Bollinger Bands
        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=20, stds=2
        )
        dataframe["bb_lowerband"] = bollinger["lower"]
        dataframe["bb_upperband"] = bollinger["upper"]

        # EMA 50 / 200
        dataframe["ema_fast"] = ta.EMA(dataframe, timeperiod=self.ema_fast_p.value)
        dataframe["ema_slow"] = ta.EMA(dataframe, timeperiod=self.ema_slow_p.value)

        # ADX
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)

        # ATR + rolling median (trend filter)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)
        dataframe["atr_median"] = (
            dataframe["atr"]
            .rolling(window=self.atr_window.value, min_periods=1)
            .median()
        )

        # STOCH (Stochastic Oscillator)
        stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3)
        dataframe["stoch_slowk"] = stoch["slowk"]
        dataframe["stoch_slowd"] = stoch["slowd"]

        # CCI (Commodity Channel Index)
        dataframe["cci"] = ta.CCI(dataframe, timeperiod=20)

        # MFI (Money Flow Index)
        dataframe["mfi"] = ta.MFI(dataframe, timeperiod=14)

        # WILLR (Williams %R)
        dataframe["willr"] = ta.WILLR(dataframe, timeperiod=14)

        # OBV (On Balance Volume)
        dataframe["obv"] = ta.OBV(dataframe)

        # SAR (Parabolic SAR)
        dataframe["sar"] = ta.SAR(dataframe, acceleration=0.02, maximum=0.2)

        # NATR (Normalized ATR)
        dataframe["natr"] = ta.NATR(dataframe, timeperiod=14)

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        conditions = [dataframe["volume"] > 0]

        # Trading window filter: prevent entries during blocked hours/days
        if self.blocked_hours or self.blocked_days:
            # Get current hour and day from dataframe index (assumes datetime index)
            dataframe["hour"] = dataframe.index.hour
            dataframe["day"] = dataframe.index.dayofweek

            # Block entries during excluded hours
            if self.blocked_hours:
                hour_filter = ~dataframe["hour"].isin(self.blocked_hours)
                conditions.append(hour_filter)

            # Block entries during excluded days
            if self.blocked_days:
                day_filter = ~dataframe["day"].isin(self.blocked_days)
                conditions.append(day_filter)

        if self.use_rsi.value:
            conditions.append(dataframe["rsi"] < self.rsi_threshold.value)

        if self.use_macd.value:
            conditions.append(
                qtpylib.crossed_above(dataframe["macd"], dataframe["macdsignal"])
                & (dataframe["macdhist"] > self.macd_hist_min.value)
            )

        if self.use_bb.value:
            conditions.append(
                dataframe["close"] <= dataframe["bb_lowerband"] * self.bb_factor.value
            )

        if self.use_ema_cross.value:
            conditions.append(dataframe["ema_fast"] > dataframe["ema_slow"])

        if self.use_adx.value:
            conditions.append(dataframe["adx"] > self.adx_threshold.value)

        if self.use_atr.value:
            conditions.append(dataframe["atr"] > dataframe["atr_median"])

        if self.use_stoch.value:
            conditions.append(dataframe["stoch_slowk"] < self.stoch_threshold.value)

        if self.use_cci.value:
            conditions.append(dataframe["cci"] < self.cci_threshold.value)

        if self.use_mfi.value:
            conditions.append(dataframe["mfi"] < self.mfi_threshold.value)

        if self.use_willr.value:
            conditions.append(dataframe["willr"] < self.willr_threshold.value)

        if self.use_obv.value:
            conditions.append(dataframe["obv"] > dataframe["obv"].rolling(20).mean())

        if self.use_sar.value:
            conditions.append(dataframe["close"] > dataframe["sar"])

        if self.use_natr.value:
            conditions.append(dataframe["natr"] > dataframe["natr"].rolling(20).mean())

        # Need at least one signal beyond the volume guard
        if len(conditions) > 1:
            combined = reduce(operator.and_, conditions)
            dataframe.loc[combined, "enter_long"] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe
'''


def generate_strategy_source_ensemble(class_name: str) -> str:
    """Return a complete freqtrade strategy using weighted Alpha Consensus Voting.

    Three independent signals (RSI oversold, MACD cross, BB breakout) each cast a
    binary vote.  The votes are combined into a *normalized* weighted score:

        score = (rsi_w * rsi_vote + macd_w * macd_vote + bb_w * bb_vote)
                / (rsi_w + macd_w + bb_w)

    The strategy enters long only when score >= consensus_threshold.

    Key properties:
    - All three weights and the threshold are DecimalParameters in the ``buy``
      space, so freqtrade's hyperopt can tune them alongside ROI/stoploss.
    - Setting a weight to 0 effectively switches that signal off — hyperopt
      discovers this automatically.
    - The normalized score reaches exactly 1.0 when every non-zero signal votes
      buy (unanimous consensus).
    - Weights can be overridden at runtime (without a recompile) by writing a
      JSON file at ``<user_data_dir>/ensemble_weights.json``:

          {{
            "rsi_weight": 0.5,
            "macd_weight": 0.3,
            "bb_weight": 0.2,
            "consensus_threshold": 0.4
          }}

      The strategy reads this file on every ``populate_entry_trend`` call,
      falling back to the hyperopt-optimised DecimalParameter values when the
      file is absent or unreadable.
    """
    return f'''\
# Auto-generated by Strategy Lab — Auto-Quant Factory (Alpha Ensemble Edition)
# Entry logic uses weighted consensus voting across RSI, MACD, and BB signals.
# Optimise with: freqtrade hyperopt --strategy Github_4tie_fortiesr__advanced_generators__20260709_213639 --spaces buy roi stoploss
#
# Live weight updates: write user_data/ensemble_weights.json to change weights
# without restarting freqtrade (changes take effect on the next candle).

import json
from pathlib import Path

from freqtrade.strategy import DecimalParameter, IntParameter, IStrategy
from pandas import DataFrame
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib


class Github_4tie_fortiesr__advanced_generators__20260709_213639(IStrategy):
    INTERFACE_VERSION: int = 3

    minimal_roi = {{
        "0": 0.10,
        "30": 0.05,
        "60": 0.02,
        "120": 0,
    }}

    stoploss = -0.05
    timeframe = "5m"
    trailing_stop = False
    process_only_new_candles = True

    # ── Alpha Weights — optimised by hyperopt (buy space) ────────────────────
    # Setting a weight to 0.0 effectively switches that signal off.
    rsi_weight  = DecimalParameter(0.0, 1.0, default=0.4, decimals=2, space="buy", optimize=True)
    macd_weight = DecimalParameter(0.0, 1.0, default=0.3, decimals=2, space="buy", optimize=True)
    bb_weight   = DecimalParameter(0.0, 1.0, default=0.3, decimals=2, space="buy", optimize=True)

    # ── Consensus threshold — how high the weighted score must be to enter ───
    consensus_threshold = DecimalParameter(0.1, 0.9, default=0.5, decimals=2,
                                           space="buy", optimize=True)

    # ── Individual signal parameters ─────────────────────────────────────────
    rsi_oversold  = IntParameter(20, 40, default=30, space="buy", optimize=True)
    rsi_period    = IntParameter(10, 20, default=14, space="buy", optimize=False)

    # ── Live-weight helper ────────────────────────────────────────────────────

    def _load_live_weights(self):
        """Attempt to read override weights from ensemble_weights.json.

        Falls back to the hyperopt-optimised DecimalParameter values when the
        file is absent, malformed, or any key is missing.
        """
        try:
            p = Path(self.config.get("user_data_dir", ".")) / "ensemble_weights.json"
            if p.exists():
                cfg = json.loads(p.read_text(encoding="utf-8"))
                return (
                    float(cfg.get("rsi_weight",          self.rsi_weight.value)),
                    float(cfg.get("macd_weight",         self.macd_weight.value)),
                    float(cfg.get("bb_weight",           self.bb_weight.value)),
                    float(cfg.get("consensus_threshold", self.consensus_threshold.value)),
                )
        except Exception as exc:
            logger.warning("Generator | failed to load ensemble_weights.json: %s", exc)
        return (
            self.rsi_weight.value,
            self.macd_weight.value,
            self.bb_weight.value,
            self.consensus_threshold.value,
        )

    # ── Indicator computation ─────────────────────────────────────────────────

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # RSI
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)

        # MACD (fast=12, slow=26, signal=9)
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe["macd"]       = macd["macd"]
        dataframe["macdsignal"] = macd["macdsignal"]
        dataframe["macdhist"]   = macd["macdhist"]

        # Bollinger Bands (period=20, stddev=2)
        bollinger = qtpylib.bollinger_bands(
            qtpylib.typical_price(dataframe), window=20, stds=2
        )
        dataframe["bb_lowerband"] = bollinger["lower"]
        dataframe["bb_upperband"] = bollinger["upper"]

        return dataframe

    # ── Weighted consensus entry logic ────────────────────────────────────────

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        rsi_w, macd_w, bb_w, threshold = self._load_live_weights()

        # ── Binary votes (1 = bullish signal present, 0 = absent) ─────────
        rsi_vote  = (dataframe["rsi"] < self.rsi_oversold.value).astype(float)
        macd_vote = qtpylib.crossed_above(
            dataframe["macd"], dataframe["macdsignal"]
        ).astype(float)
        bb_vote   = (dataframe["close"] < dataframe["bb_lowerband"]).astype(float)

        # ── Weighted, normalised consensus score ───────────────────────────
        total_weight = rsi_w + macd_w + bb_w
        if total_weight > 0:
            score = (rsi_vote * rsi_w + macd_vote * macd_w + bb_vote * bb_w) / total_weight
        else:
            score = dataframe["close"] * 0  # all weights zero → never enter

        # ── Enter long when consensus clears the threshold ─────────────────
        dataframe.loc[
            (score >= threshold) & (dataframe["volume"] > 0),
            "enter_long",
        ] = 1

        return dataframe

    # ── Exit stub (relies on ROI / stoploss) ──────────────────────────────────

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        return dataframe
'''