beta_factors_model

# --- Do not remove these libs ---
from datetime import timedelta
from pathlib import Path

import joblib
import numpy as np
import pandas as pd

# --------------------------------
from freqtrade.strategy import DecimalParameter, IStrategy
from pandas import DataFrame


class beta_factors_model(IStrategy):

    INTERFACE_VERSION: int = 3
    # Minimal ROI designed for the strategy.
    # This attribute will be overridden if the config file contains "minimal_roi"
    minimal_roi = {}

    can_short = False

    # Optimal stoploss designed for the strategy
    # This attribute will be overridden if the config file contains "stoploss"
    stoploss = -0.15

    # Optimal timeframe for the strategy
    timeframe = '1w'

    # Minimum Candle count for indicator to populate
    startup_candle_count = 10

    # trailing stoploss
    trailing_stop = False
    trailing_stop_positive = 0.01
    trailing_stop_positive_offset = 0.02

    # Optional order type mapping
    order_types = {
        'entry': 'limit',
        'exit': 'limit',
        'stoploss': 'market',
        'stoploss_on_exchange': False
    }

    # Hyperoptable parameters
    buy_threshold = DecimalParameter(0.01, 0.09, decimals=3, default=0.07, space="buy")
    sell_threshold = DecimalParameter(0.01, 0.09, decimals=3, default=0.02, space="sell")

    # market cap dataframe
    btc_cap = None

    # load LSTM model
    MODEL_PATH = Path(__file__).resolve().parent / "beta_factors_model.joblib"
    _model = None

    def get_model(self):
        if self._model is None:
            self._model = joblib.load(self.MODEL_PATH)
        return self._model


    # load market cap data
    def load_market_cap_data(self):
        if self.btc_cap is not None:
            return self.btc_cap
        BASE_DIR = Path(__file__).resolve().parent
        PATH = BASE_DIR / "Bitcoin_marketcap.csv"
        data = pd.read_csv(PATH, sep=";")

        # modify timeClose to utc datetime
        data["timeClose"] = pd.to_datetime(data["timeClose"], utc=True)
        data["timeClose"] = data["timeClose"].dt.floor('D')

        # shift market cap to align with next day's prices
        data["marketCap_shifted"] = data["marketCap"].shift(1)

        self.btc_cap = data
        return data
    
    def informative_pairs(self):
        """
        Define additional, informative pair/interval combinations to be cached from the exchange.
        These pair/interval combinations are non-tradeable, unless they are part
        of the whitelist as well.
        For more information, please consult the documentation
        :return: List of tuples in the format (pair, interval)
            Sample: return [("ETH/USDT", "5m"),
                            ("BTC/USDT", "15m"),
                            ]
        """
        return []
    
    HOLD_DAYS = 7

    def custom_exit(self, pair, trade, current_time, current_rate, current_profit, **kwargs):
        # Exit after HOLD_DAYS regardless (time stop)
        if current_time - trade.open_date_utc >= timedelta(days=self.HOLD_DAYS):
            return "time_exit"
        return None


    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Adds several different TA indicators to the given DataFrame

        Performance Note: For the best performance be frugal on the number of indicators
        you are using. Let uncomment only the indicator you are using in your strategies
        or your hyperopt configuration, otherwise you will waste your memory and CPU usage.

        lookback timeframe: 15H
        Number of 5 min intervals = 15 x 60 / 5 = 180
        """

        # Load market cap data
        btc_cap = self.load_market_cap_data()

        # Merge market cap data into the dataframe
        dataframe["date"] = pd.to_datetime(dataframe["date"], utc=True)
        dataframe["week_end"] = (dataframe["date"].dt.floor('D') + pd.Timedelta(days=6)).dt.floor('D')
        sliced_btc_cap = btc_cap[["timeClose", "marketCap_shifted"]]
        dataframe = dataframe.merge(sliced_btc_cap, left_on="week_end", right_on="timeClose", how="left")

        # Design columns for ML model
        # define weekly returns based on closing prices
        dataframe['ret'] = dataframe['close'].pct_change()

        # Define CMKT Proxy as the weekly return of BTC-USD
        dataframe['cmkt'] = dataframe['ret']

        # Calculate CMOM(returns momentum) over a 2-week period
        dataframe['cmom'] = dataframe['ret'].rolling(window=2).sum()

        # Define mcap-cmkt interaction term
        dataframe['csize'] = np.log(dataframe['marketCap_shifted'])
        dataframe['csize_cmkt'] = dataframe['csize'] * dataframe['cmkt']

        # Higher Order terms
        dataframe['cmkt_2'] = dataframe['cmkt'] ** 2
        dataframe['cmom_3'] = dataframe['cmom'] ** 3

        # show dataframe with NaN values
        # print("Dataframe with NaN values:")
        # print(dataframe.head(20))

        # Replace NaN values with zero
        dataframe.fillna(0, inplace=True)

        # check the dataframe
        # print(dataframe.head())

        # Prepare data for prediction
        feature_cols = [
            'cmkt', 'cmom', 'csize', 'csize_cmkt', 'cmkt_2', 'cmom_3'
        ]
        X = dataframe[feature_cols].values
        # print(X.shape)

        # Predict
        pred_norm = self.get_model().predict(X)
        dataframe["pred_ret"] = np.nan

        # Align predictions to dataframe rows
        dataframe["pred_ret"] = pred_norm.flatten()
        # print("predicted returns:", pred_norm)

        dataframe["ml_signal"] = 0
        dataframe.loc[dataframe["pred_ret"] > 0.07, "ml_signal"] = 1
        dataframe.loc[dataframe["pred_ret"] < -0.07, "ml_signal"] = -1

        # ml signal = 0 if any features = 0 (to avoid trading on no info)
        dataframe.loc[
            (dataframe['cmkt'] == 0) | (dataframe['cmom'] == 0) | (dataframe['csize'] == 0) 
            | (dataframe['csize_cmkt'] == 0) | (dataframe['cmkt_2'] == 0) | (dataframe['cmom_3'] == 0), 
            'ml_signal'] = 0

        # print("first rows of dataframe with indicators:")
        # print(dataframe.head(50))

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the buy signal for the given dataframe
        :param dataframe: DataFrame
        :return: DataFrame with buy column
        """

        # Buy when predicted price increase is above threshold
        dataframe.loc[
            (
                (dataframe['ml_signal'] == 1) & 
                (dataframe['volume'] > 0)
            ),
            'enter_long'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the sell signal for the given dataframe
        :param dataframe: DataFrame
        :return: DataFrame with buy column
        """

        # Sell when predicted price decrease is above threshold
        dataframe.loc[
            (
                (dataframe['ml_signal'] == -1) & 
                (dataframe['volume'] > 0) 
            ),
            'exit_long'] = 1
    
        
        return dataframe