# source: https://raw.githubusercontent.com/DerSalvador/freqtrade-helm-chart/f80d4d8b77c53435e9c0a9045636f1bfb2b8c539/chart/deployed_strategies/bot-ssc-03-k8s-namespace/AutoArimaTripleV1.py%20GodStraHo.py
kubectl --context=gke_vaulted-gift-406223_europe-west1-b_private-cluster-3 -n bot-ssc-03 exec -it pod/freqtrade-bot-ssc-03-7479445cbd-xdnxj -c freqtrade -- cat /freqtrade/user_data/strategies/Github_DerSalvador_freqtrade_helm_chart__AutoArimaTripleV1_pyGodStraHo__20260416_224245.py GodStraHo.py
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401

# --- Do not remove these libs ---
import numpy as np  # noqa
import pandas as pd  # noqa
from pandas import DataFrame
import os

from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter,
                                IStrategy, IntParameter, informative)
from freqtrade.data.dataprovider import DataProvider

from statsmodels.tsa.statespace.sarimax import SARIMAX, SARIMAXResults
from pmdarima.arima import auto_arima

from datetime import datetime

pd.set_option('display.max_columns', None)


class Github_DerSalvador_freqtrade_helm_chart__AutoArimaTripleV1_pyGodStraHo__20260416_224245(IStrategy):
    """
    This is a strategy template to get you started.
    More information in https://www.freqtrade.io/en/latest/strategy-customization/

    You can:
        :return: a Dataframe with all mandatory indicators for the strategies
    - Rename the class name (Do not forget to update class_name)
    - Add any methods you want to build your strategy
    - Add any lib you need to build your strategy

    You must keep:
    - the lib in the section "Do not remove these libs"
    - the methods: populate_indicators, populate_entry_trend, populate_exit_trend
    You should keep:
    - timeframe, minimal_roi, stoploss, trailing_*
    """
    # Strategy interface version - allow new iterations of the strategy interface.
    # Check the documentation or the Sample strategy to get the latest version.
    INTERFACE_VERSION = 2

    log = open(os.path.join(os.path.dirname(os.path.realpath(__file__)), '../logs/strategy-log.log'), 'w')

    # Minimal ROI designed for the strategy.
    # This attribute will be overridden if the config file contains "minimal_roi".
    minimal_roi = {
        "60": 0.01,
        "30": 0.02,
        "0": 0.04
    }

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

    # Trailing stoploss
    trailing_stop = False
    # trailing_only_offset_is_reached = False
    # trailing_stop_positive = 0.01
    # trailing_stop_positive_offset = 0.0  # Disabled / not configured

    # Optimal timeframe for the strategy.
    timeframe = '5m'

    # Run "populate_indicators()" only for new candle.
    process_only_new_candles = True

    # These values can be overridden in the "ask_strategy" section in the config.
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # Number of candles the strategy requires before producing valid signals
    startup_candle_count: int = 30

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

    # Optional order time in force.
    order_time_in_force = {
        'entry': 'gtc',
        'exit': 'gtc'
    }

    plot_config = {
        # Main plot indicators (Moving averages, ...)
        'main_plot': {
            'tema': {},
            'sar': {'color': 'white'},
        },
        'subplots': {
            # Subplots - each dict defines one additional plot
            "MACD": {
                'macd': {'color': 'blue'},
                'macdsignal': {'color': 'orange'},
            },
            "RSI": {
                'rsi': {'color': 'red'},
            }
        }
    }

    predictor_30m = None
    predictor_1h = None
    predictor_5m = None
    last_predicted_frame_30m = None, None
    last_predicted_frame_1h = None, None
    last_predicted_frame_5m = None, None

    @informative('30m')
    def populate_indicators_30m(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        current_date = dataframe.loc[len(dataframe) - 1].date
        last_date, last_prediction = self.last_predicted_frame_30m
        if last_date == current_date:
            self.__print('Use stored predicted value as 30m indicator.')
            dataframe['prediction'] = dataframe.apply(lambda row: last_prediction if row.date == current_date else None, axis=1)
            return dataframe

        if not self.predictor_30m:
            self.predictor_30m = ArimaPredictor(self.dp, metadata['pair'], '30m', '30min')

        self.__print('===30m===')
        self.__print(dataframe.tail(2))
        prediction = self.predictor_30m.predict(dataframe, current_date)
        dataframe['prediction'] = dataframe.apply(lambda row: prediction if row.date == current_date else None, axis=1)
        self.last_predicted_frame_30m = current_date, prediction
        self.__print(f'Prediction 30m {current_date}: {prediction:0.2f}')

        # dry run: Remember last candle
        # backtest: process everything at the same time from head to tail
        # TODO Model generation in background
        # TODO Save and restore
        return dataframe

    def __print(self, message):
        self.log.write(f'{message}\n')

    @informative('1h')
    def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        current_date = dataframe.loc[len(dataframe) - 1].date
        last_date, last_prediction = self.last_predicted_frame_1h
        if last_date == current_date:
            self.__print('Use stored predicted value as 1h indicator.')
            dataframe['prediction'] = dataframe.apply(lambda row: last_prediction if row.date == current_date else None, axis=1)
            return dataframe

        if not self.predictor_1h:
            self.predictor_1h = ArimaPredictor(self.dp, metadata['pair'], '1h', 'H')

        self.__print('===1h===')
        self.__print(dataframe.tail(2))
        prediction = self.predictor_1h.predict(dataframe, current_date)
        dataframe['prediction'] = dataframe.apply(lambda row: prediction if row.date == current_date else None, axis=1)
        self.last_predicted_frame_1h = current_date, prediction
        self.__print(f'Prediction 1h {current_date}: {prediction:0.2f}')

        return dataframe

    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.
        :param dataframe: Dataframe with data from the exchange
        :param metadata: Additional information, like the currently traded pair
        :return: a Dataframe with all mandatory indicators for the strategies
        """
        current_date = dataframe.loc[len(dataframe) - 1].date
        last_date, last_prediction = self.last_predicted_frame_5m
        if last_date == current_date:
            self.__print('Use stored predicted value as 5m indicator.')
            dataframe['prediction'] = dataframe.apply(lambda row: last_prediction if row.date == current_date else None, axis=1)
            return dataframe

        if not self.predictor_5m:
            self.predictor_5m = ArimaPredictor(self.dp, metadata['pair'], '5m', '5min')

        self.__print('===5m===')
        self.__print(dataframe.tail(2))
        prediction = self.predictor_5m.predict(dataframe.loc[:, ['date', 'open', 'high', 'low', 'close', 'volume']], current_date)
        dataframe['prediction'] = dataframe.apply(lambda row: prediction if row.date == current_date else None, axis=1)
        self.last_predicted_frame_5m = current_date, prediction
        self.__print(f'Prediction 5m {current_date}: {prediction:0.2f}')

        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the entry signal for the given dataframe
        :param dataframe: DataFrame populated with indicators
        :param metadata: Additional information, like the currently traded pair
        :return: DataFrame with entry column
        """

        self.__print('===entry===')
        self.__print(dataframe.tail(1))
        dataframe.loc[
            (
            ),
            'entry'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Based on TA indicators, populates the exit signal for the given dataframe
        :param dataframe: DataFrame populated with indicators
        :param metadata: Additional information, like the currently traded pair
        :return: DataFrame with entry column
        """

        self.__print('===exit===')
        self.__print(dataframe.tail(1))
        dataframe.loc[
            (
            ),
            'exit'] = 1
        return dataframe


class ArimaPredictor(object):
    train_size = 10000
    steps = -1
    scale = None
    model: SARIMAXResults = None
    last_x = None

    def __init__(self, dp: DataProvider, pair: str, t_frame: str, frequency: str):
        self.dp = dp
        self.pair = pair
        self.t_frame = t_frame
        self.frequency = frequency
        self.data = self.__download_data()

    def predict(self, c_data: DataFrame, current_prediction_date):
        self.__combine_data(c_data)
        must_train = self.__must_train()
        # must be executed before get_current_data because the scale function is needed
        if must_train:
            self.__train(current_prediction_date)
        x, y = self.__get_current_data(current_prediction_date)
        if not must_train:
            self.model.append(y, exog=self.last_x, refit=False)
        prediction = self.model.forecast(steps=self.steps*-1, exog=x)
        self.last_x = x

        # TODO Confidence interval used in trend (only if lower and upper is in interval)
        prediction = prediction.tail(1).mean()
        current = x['mean'].mean()
        return (prediction-current)/current*100

    def __get_current_data(self, current_prediction_date):
        index = self.data.index.get_loc(current_prediction_date)
        row = self.data.iloc[index:index+1, :].copy().asfreq(freq=self.frequency)
        row['mean'] = (row['low'] + row['high']) / 2
        self.scale(row)
        return row, row.rename(columns={'mean': 'step_ahead'})\
                       .drop('open', axis=1)\
                       .drop('high', axis=1)\
                       .drop('low', axis=1)\
                       .drop('volume', axis=1)\
                       .drop('close', axis=1)

    def __must_train(self):
        return self.model is None

    def __train(self, current_prediction_date):
        start = datetime.now()
        print(f'Started {self.t_frame} training at {start}.')
        end_index = self.data.index.get_loc(current_prediction_date)
        x, y = self.__prepare_data(self.data, self.frequency)
        train_x, train_y, self.scale = self.__get_train_data(x, y, end_index)
        step_wise = auto_arima(train_y,
                               exogenous=train_x,
                               start_p=1, start_q=1,
                               max_p=7, max_q=7,
                               d=1, max_d=7,
                               trace=False,
                               error_action='ignore',
                               suppress_warnings=True,
                               stepwise=True)
        print(f'Optimal parameter for {self.t_frame} are {step_wise.order}.')
        self.model = SARIMAX(train_y,
                             exog=train_x,
                             order=step_wise.order,
                             enforce_invertibility=False,
                             enforce_stationarity=False).fit()
        print(f'Finished training of {self.t_frame} in {datetime.now()-start}.')

    def __download_data(self):
        h_data = self.dp.historic_ohlcv(self.pair, self.t_frame).set_index('date')
        return h_data

    def __combine_data(self, c_data: DataFrame):
        c_data = c_data.copy().set_index('date')
        self.data = self.data.append(c_data[c_data.index.isin(self.data.index) == False], verify_integrity=True)

    def __prepare_data(self, df: DataFrame, frequency: str):
        df = df.copy()
        df['mean'] = (df['low'] + df['high']) / 2
        df['step_ahead'] = df['mean'].shift(self.steps)
        df = df.dropna()

        x = df.copy().drop('step_ahead', axis=1).asfreq(freq=frequency).interpolate()

        y = df.copy().drop('open', axis=1).drop('high', axis=1).drop('low', axis=1).drop('volume', axis=1).drop('mean',
                                                                                                                axis=1).drop(
            'close', axis=1).asfreq(freq=frequency).interpolate()
        return x, y

    def __get_train_data(self, x, y, current_prediction_index):
        if current_prediction_index - self.train_size < 0:
            raise Exception('Not enough data in history. Please first download enough data')

        train_x = x.iloc[current_prediction_index - self.train_size:current_prediction_index, :].copy()
        train_y = y.iloc[current_prediction_index - self.train_size:current_prediction_index, :].copy()

        fix_scaler = FixedScaler(train_x['mean'].min(), train_x['mean'].max(), train_x['volume'].min(),
                                 train_x['volume'].max())

        def scale(df):
            nonlocal fix_scaler
            df['volume'] = df['volume'].apply(lambda v: fix_scaler.scale(v))

        scale(train_x)
        return train_x, train_y, scale


class FixedScaler(object):

    def __init__(self, min_in, max_in, min_out, max_out):
        self.min_in = min_in
        self.max_in = max_in
        self.min_out = min_out
        self.max_out = max_out

    def scale(self, value):
        value_std = (value - self.min_out) / (self.max_out - self.min_out)
        return value_std * (self.max_in - self.min_in) + self.min_incat: GodStraHo.py: No such file or directory