Finder

import numpy as np
import pandas as pd
import talib.abstract as ta
import pandas_ta as pta
from itertools import product
from functools import reduce
from pandas import DataFrame, Series
from datetime import datetime
from typing import Any, Dict, List
from freqtrade.constants import Config
from freqtrade.strategy import BooleanParameter, CategoricalParameter, IStrategy
from freqtrade.optimize.space import Categorical, SKDecimal, Dimension


class Finder(IStrategy):
    INTERFACE_VERSION = 3
    process_only_new_candles = True
    can_short = False
    stoploss = -0.20
    minimal_roi = {"0": 0.30}
    max_open_trades = 10
    trailing_stop = False
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True
    use_custom_stoploss = False
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False
    ignore_buying_expired_candle_after = 300
    timeframe = "4h"
    NUM_RULES = 2
    SIDES = ["buy", "sell"]

    class HyperOpt:
        def stoploss_space() -> List[Dimension]:
            return [SKDecimal(-0.30, -0.01, decimals=2, name="stoploss")]

        def roi_space() -> List[Dimension]:
            return [SKDecimal(0.01, 0.30, decimals=2, name="roi")]

        def generate_roi_table(p: Dict) -> Dict[int, float]:
            return {0: p["roi"]}

        def max_open_trades_space() -> List[Dimension]:
            return [Categorical(range(5, 10, 1), name="max_open_trades")]

    def __init__(self, config: Config):
        super().__init__(config)
        self.config = config
        self._initialize_parameters()

    def _initialize_parameters(self):
        INDICATORS = """
        upperband lowerband macd macdsignal slowk slowd fastk fastd
        dema ema kama midpoint midprice sar sma t3 tema trima wma
        adx aroonosc cci cmo mfi mom roc rsi trix willr
        atr natr
        supertrend
        """.split()

        OPERATORS = [
            "DISABLED",
            "GOING_UP",
            "GOING_DOWN",
            "RISING",
            "FALLING",
            "ABOVE_AVERAGED",
            "BELOW_AVERAGED",
        ]
        PERIOD_RANGE = list(range(3, 15)) + list(range(15, 55, 5))
        SHIFT_RANGE = range(0, 12, 2)
        DOW_RANGE = range(0, 7)

        for side in self.SIDES:
            for idx in range(self.NUM_RULES):
                setattr(self, f"{side}_{idx}_indicator",
                    CategoricalParameter(INDICATORS, default="sma", space=side),
                )
                setattr(self, f"{side}_{idx}_period",
                    CategoricalParameter(PERIOD_RANGE, default=10, space=side),
                )
                setattr(self, f"{side}_{idx}_shift",
                    CategoricalParameter(SHIFT_RANGE, default=0, space=side),
                )
                setattr(self, f"{side}_{idx}_operator",
                    CategoricalParameter(OPERATORS, default="DISABLED", space=side),
                )
            setattr(self, f"{side}_min_dow", 
                CategoricalParameter(DOW_RANGE, default=0, space=side))
            setattr(self, f"{side}_max_dow", 
                CategoricalParameter(DOW_RANGE, default=0, space=side))

    def _is_rule_disabled(self, side: str, idx: int) -> bool:
        return getattr(self, f"{side}_{idx}_operator").value == "DISABLED"

    def _get_rule_parameters(self, side: str, idx: int):
        ind = getattr(self, f"{side}_{idx}_indicator").value
        prd = int(getattr(self, f"{side}_{idx}_period").value)
        sft = int(getattr(self, f"{side}_{idx}_shift").value)
        return ind, prd, sft

    def _calculate_indicator(self, dataframe, indicator, period):
        indicator_mapping = {
            "upperband": "bbands",
            "lowerband": "bbands",
            "macd": "macd",
            "macdsignal": "macd",
            "slowk": "stoch",
            "slowd": "stoch",
            "fastk": "stochf",
            "fastd": "stochf",
        }

        if indicator in indicator_mapping:
            ta_func = getattr(ta, indicator_mapping[indicator].upper())
            result = ta_func(dataframe, timeperiod=period)
            result = result[indicator]
        elif indicator == "supertrend":
            result = pta.supertrend(
                dataframe["high"], dataframe["low"], dataframe["close"], length=period
            )[f"SUPERT_{period}_3.0"]
        else:
            ta_func = getattr(ta, indicator.upper())
            result = ta_func(dataframe, timeperiod=period)

        return pd.Series(result)

    def _apply_operator(self, series, side, idx, shift):
        operator = getattr(self, f"{side}_{idx}_operator").value
        averaged = pd.Series(ta.EMA(series, timeperiod=10))
        shifted = series.shift(shift)

        if operator in ["ABOVE_AVERAGED", "BELOW_AVERAGED"]:
            comparison = (
                series > averaged.shift(shift)
                if operator == "ABOVE_AVERAGED"
                else series < averaged.shift(shift)
            )
            return comparison

        value0, value1, value2 = shifted, shifted.shift(1), shifted.shift(2)

        comparison_operators = {
            "GOING_UP": (value0 > value1) & (value1 < value2),
            "GOING_DOWN": (value0 < value1) & (value1 > value2),
            "RISING": (value0 > value1) & (value1 > value2),
            "FALLING": (value0 < value1) & (value1 < value2),
        }

        return comparison_operators.get(operator, pd.Series(False, index=series.index))

    def _apply_dow_conditions(self, dataframe, side):
        min_dow = getattr(self, f"{side}_min_dow").value
        max_dow = getattr(self, f"{side}_max_dow").value
        if min_dow < max_dow:
            return [(dataframe["dow"] >= min_dow) & (dataframe["dow"] <= max_dow)]
        else:
            return [(dataframe["dow"] <= min_dow) & (dataframe["dow"] >= max_dow)]

    def _populate_trend(self, dataframe: pd.DataFrame, side: str) -> pd.DataFrame:
        conditions = []

        for idx in range(self.NUM_RULES):
            if self._is_rule_disabled(side, idx):
                continue

            indicator, period, shift = self._get_rule_parameters(side, idx)
            result = self._calculate_indicator(dataframe, indicator, period)
            conditions.append(self._apply_operator(result, side, idx, shift))

        conditions.append(dataframe["volume"] > 0)
        conditions.extend(self._apply_dow_conditions(dataframe, side))

        return conditions

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["dow"] = dataframe["date"].dt.dayofweek
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        conditions = self._populate_trend(dataframe, "buy")
        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), "enter_long"] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        conditions = self._populate_trend(dataframe, "sell")
        if conditions:
            dataframe.loc[reduce(lambda x, y: x & y, conditions), "exit_long"] = 1
        return dataframe