FreqSignalsAiDataProvider.py

import logging
import os
from functools import reduce
from typing import Dict
import pandas as pd
import numpy as np
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib

from freqtrade.strategy import CategoricalParameter, IStrategy, merge_informative_pair
from freqsignals import FreqSignalsStrategy, FreqSignalsMixin
DATA_SET_ID = os.environ.get("FREQSIGNALS_AI_DATA_SET_ID")


logger = logging.getLogger(__name__)


class FreqSignalsAiDataProvider(IStrategy, FreqSignalsMixin):
# class FreqSignalsAiDataProvider(IStrategy):
    """
    Example strategy showing how the user connects their own
    IFreqaiModel to the strategy. Namely, the user uses:
    self.freqai.start(dataframe, metadata)
    to make predictions on their data. populate_any_indicators() automatically
    generates the variety of features indicated by the user in the
    canonical freqtrade configuration file under config['freqai'].
    """

    minimal_roi = {"0": 0.1, "240": -1}

    plot_config = {
        "main_plot": {},
        "subplots": {
            "prediction": {"prediction": {"color": "blue"}},
            "do_predict": {
                "do_predict": {"color": "brown"},
            },
        },
    }

    process_only_new_candles = True
    stoploss = -0.05
    use_exit_signal = True
    # this is the maximum period fed to talib (timeframe independent)
    startup_candle_count: int = 40
    can_short = False

    std_dev_multiplier_buy = CategoricalParameter(
        [0.75, 1, 1.25, 1.5, 1.75], default=1.25, space="buy", optimize=True)
    std_dev_multiplier_sell = CategoricalParameter(
        [0.75, 1, 1.25, 1.5, 1.75], space="sell", default=1.25, optimize=True)

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.freqsignals_init()
        self.signal_update_time_by_pair: Dict[str, str] = {}

    def populate_any_indicators(
        self, pair, df, tf, informative=None, set_generalized_indicators=False
    ):
        """
        Function designed to automatically generate, name and merge features
        from user indicated timeframes in the configuration file. User controls the indicators
        passed to the training/prediction by prepending indicators with `f'%-{pair}`
        (see convention below). I.e. user should not prepend any supporting metrics
        (e.g. bb_lowerband below) with % unless they explicitly want to pass that metric to the
        model.
        :param pair: pair to be used as informative
        :param df: strategy dataframe which will receive merges from informatives
        :param tf: timeframe of the dataframe which will modify the feature names
        :param informative: the dataframe associated with the informative pair
        """

        if informative is None:
            informative = self.dp.get_pair_dataframe(pair, tf)

        # first loop is automatically duplicating indicators for time periods
        for t in self.freqai_info["feature_parameters"]["indicator_periods_candles"]:

            t = int(t)
            informative[f"%-{pair}rsi-period_{t}"] = ta.RSI(informative, timeperiod=t)
            # informative[f"%-{pair}mfi-period_{t}"] = ta.MFI(informative, timeperiod=t)
            informative[f"%-{pair}adx-period_{t}"] = ta.ADX(informative, timeperiod=t)
            informative[f"%-{pair}sma-period_{t}"] = ta.SMA(informative, timeperiod=t)
            informative[f"%-{pair}ema-period_{t}"] = ta.EMA(informative, timeperiod=t)

            # bollinger = qtpylib.bollinger_bands(
            #     qtpylib.typical_price(informative), window=t, stds=2.2
            # )
            # informative[f"{pair}bb_lowerband-period_{t}"] = bollinger["lower"]
            # informative[f"{pair}bb_middleband-period_{t}"] = bollinger["mid"]
            # informative[f"{pair}bb_upperband-period_{t}"] = bollinger["upper"]

            # informative[f"%-{pair}bb_width-period_{t}"] = (
            #     informative[f"{pair}bb_upperband-period_{t}"]
            #     - informative[f"{pair}bb_lowerband-period_{t}"]
            # ) / informative[f"{pair}bb_middleband-period_{t}"]
            # informative[f"%-{pair}close-bb_lower-period_{t}"] = (
            #     informative["close"] / informative[f"{pair}bb_lowerband-period_{t}"]
            # )

            # informative[f"%-{pair}roc-period_{t}"] = ta.ROC(informative, timeperiod=t)

            informative[f"%-{pair}relative_volume-period_{t}"] = (
                informative["volume"] / informative["volume"].rolling(t).mean()
            )

        informative[f"%-{pair}pct-change"] = informative["close"].pct_change()
        informative[f"%-{pair}raw_volume"] = informative["volume"]
        informative[f"%-{pair}raw_price"] = informative["close"]

        indicators = [col for col in informative if col.startswith("%")]
        # This loop duplicates and shifts all indicators to add a sense of recency to data
        for n in range(self.freqai_info["feature_parameters"]["include_shifted_candles"] + 1):
            if n == 0:
                continue
            informative_shift = informative[indicators].shift(n)
            informative_shift = informative_shift.add_suffix("_shift-" + str(n))
            informative = pd.concat((informative, informative_shift), axis=1)

        df = merge_informative_pair(df, informative, self.config["timeframe"], tf, ffill=True)
        skip_columns = [
            (s + "_" + tf) for s in ["date", "open", "high", "low", "close", "volume"]
        ]
        df = df.drop(columns=skip_columns)

        # Add generalized indicators here (because in live, it will call this
        # function to populate indicators during training). Notice how we ensure not to
        # add them multiple times
        if set_generalized_indicators:
            df["%-day_of_week"] = (df["date"].dt.dayofweek + 1) / 7
            df["%-hour_of_day"] = (df["date"].dt.hour + 1) / 25

            # user adds targets here by prepending them with &- (see convention below)
            df["&-s_close"] = (
                df["close"]
                .shift(-self.freqai_info["feature_parameters"]["label_period_candles"])
                .rolling(self.freqai_info["feature_parameters"]["label_period_candles"])
                .mean()
                / df["close"]
                - 1
            )

            # User "looks into the future" here to figure out if the future
            # will be "up" or "down". This same column name is available to
            # the user
            # df['&s-up_or_down'] = np.where(df["close"].shift(-50) >
            #                                df["close"], 'up', 'down')


            # Classifiers are typically set up with strings as targets:
            # df['&s-up_or_down'] = np.where( df["close"].shift(-100) >
            #                                 df["close"], 'up', 'down')

            # If user wishes to use multiple targets, they can add more by
            # appending more columns with '&'. User should keep in mind that multi targets
            # requires a multioutput prediction model such as
            # templates/CatboostPredictionMultiModel.py,

            # df["&-s_range"] = (
            #     df["close"]
            #     .shift(-self.freqai_info["feature_parameters"]["label_period_candles"])
            #     .rolling(self.freqai_info["feature_parameters"]["label_period_candles"])
            #     .max()
            #     -
            #     df["close"]
            #     .shift(-self.freqai_info["feature_parameters"]["label_period_candles"])
            #     .rolling(self.freqai_info["feature_parameters"]["label_period_candles"])
            #     .min()
            # )

        return df

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:

        # All indicators must be populated by populate_any_indicators() for live functionality
        # to work correctly.

        # the model will return all labels created by user in `populate_any_indicators`
        # (& appended targets), an indication of whether or not the prediction should be accepted,
        # the target mean/std values for each of the labels created by user in
        # `populate_any_indicators()` for each training period.

        dataframe = self.freqai.start(dataframe, metadata, self)
        for val in self.std_dev_multiplier_buy.range:
            dataframe[f'target_roi_{val}'] = (
                dataframe["&-s_close_mean"] + dataframe["&-s_close_std"] * val
                )
        for val in self.std_dev_multiplier_sell.range:
            dataframe[f'sell_roi_{val}'] = (
                dataframe["&-s_close_mean"] - dataframe["&-s_close_std"] * val
                )

        # Begin FreqSignals. Above here is from the FreqaiExampleStrategy
        # https://github.com/freqtrade/freqtrade/blob/develop/freqtrade/templates/FreqaiExampleStrategy.py
        last_update_time = self.signal_update_time_by_pair.get(metadata['pair'])
        current_candle_time = dataframe.iloc[-1].date

        if (last_update_time is None or last_update_time != current_candle_time):
            # If there's a move, upload the signal with time TTL (minutes)
            self.signal_update_time_by_pair[metadata['pair']] = current_candle_time
            # up_down_value = dataframe.iloc[-1]["&s-up_or_down"]
            up_down_value = dataframe.iloc[-1]["&-s_close"]
            signal_data = {
                # required fields
                "symbol": metadata['pair'],
                "value": 1 if up_down_value == 'up' else -1 if up_down_value == 'down' else 0,
                "ttl_minutes": 60,
                "data_set_id": DATA_SET_ID,
                # any additional context
                # "up_or_down": 1 if up_down_value == 'up' else -1 if up_down_value == 'down' else 0,
                "s_close": round(dataframe.iloc[-1]["&-s_close"], 4),
                "s_close_mean": round(dataframe.iloc[-1]["&-s_close_mean"], 4),
                "do_predict": round(dataframe.iloc[-1]["do_predict"], 4),
                "DI_values": round(dataframe.iloc[-1]["DI_values"], 4),
                "DI_values": round(dataframe.iloc[-1]["do_predict"], 4),
                "price": round(dataframe.iloc[-1]["close"], 4),
                "last_move": round(dataframe.iloc[-1]["close"] - dataframe.iloc[-2]["close"], 4),
            }
            logger.info(f"setting signal for {metadata['pair']} at {current_candle_time}")
            logger.info(signal_data)
            self.freqsignals_client.post_signal(signal_data)
        # End FreqSignals. Above here is from the FreqaiExampleStrategy
    
        return dataframe

    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:

        enter_long_conditions = [
            df["do_predict"] == 1,
            df["&-s_close"] > df[f"target_roi_{self.std_dev_multiplier_buy.value}"],
            ]

        if enter_long_conditions:
            df.loc[
                reduce(lambda x, y: x & y, enter_long_conditions), ["enter_long", "enter_tag"]
            ] = (1, "long")

        enter_short_conditions = [
            df["do_predict"] == 1,
            df["&-s_close"] < df[f"sell_roi_{self.std_dev_multiplier_sell.value}"],
            ]

        if enter_short_conditions:
            df.loc[
                reduce(lambda x, y: x & y, enter_short_conditions), ["enter_short", "enter_tag"]
            ] = (1, "short")

        return df

    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        exit_long_conditions = [
            df["do_predict"] == 1,
            df["&-s_close"] < df[f"sell_roi_{self.std_dev_multiplier_sell.value}"] * 0.25,
            ]
        if exit_long_conditions:
            df.loc[reduce(lambda x, y: x & y, exit_long_conditions), "exit_long"] = 1

        exit_short_conditions = [
            df["do_predict"] == 1,
            df["&-s_close"] > df[f"target_roi_{self.std_dev_multiplier_buy.value}"] * 0.25,
            ]
        if exit_short_conditions:
            df.loc[reduce(lambda x, y: x & y, exit_short_conditions), "exit_short"] = 1

        return df

    def get_ticker_indicator(self):
        return int(self.config["timeframe"][:-1])

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time,
        entry_tag,
        side: str,
        **kwargs,
    ) -> bool:

        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = df.iloc[-1].squeeze()

        if side == "long":
            if rate > (last_candle["close"] * (1 + 0.0025)):
                return False
        else:
            if rate < (last_candle["close"] * (1 - 0.0025)):
                return False

        return True