kac_index_v1

# =============================================================
#  ➤ Stratégie KAC-Index : ATR-CCI + Indexation sur Total3
#
#  Fichier rédigé par Titouannwtt pour Freqtrade France 🇫🇷
#  ➤ Soutenez-nous sur : https://coff.ee/freqtrade_france et obtenez davantage de stratégies
#
#  📌 Ce code est fourni à des fins personnelles ou éducatives.
#  ❌ Il ne doit pas être partagé, distribué, ni publié dans un autre cadre.
# =============================================================


import numpy as np
import pandas as pd
from pandas import DataFrame
from functools import reduce
import time
from datetime import datetime, timedelta
from pandas import DataFrame
import talib.abstract as ta
import ta as clean_ta
from freqtrade.strategy import IStrategy
import logging

# ❌ Cette stratégie nécessite l'utilisation de tvdatafeed, une librairie externe !
# Pour l'installer, utilise : 
        # Installations commands : 
        # pip install requests websockets
        # git clone https://github.com/rongardF/tvdatafeed
        # cd tvdatafeed
        # pip install .

from tvDatafeed import TvDatafeed, Interval
from pytz import UTC

logger = logging.getLogger(__name__)
cache = {}
tv = TvDatafeed()


# Cette fonction permet de récupérer les valeurs du TOTAL3 depuis Trading View (https://fr.tradingview.com/symbols/TOTAL3/) pour les mettre dans un dataframe
# Libre à vous de tester cette fonction avec d'autres symboles disponibles sur Trading View
def add_tv_graph(dataframe: pd.DataFrame, backtest=True):
    symbol = 'TOTAL3'
    exchange = 'CRYPTOCAP'
    interval = Interval.in_daily
    column_name = f'1d_{symbol}'
    
    # Liste des colonnes que l'on veut récupéré sur Trading View
    ohlc_types = ['close', 'open', 'high', 'low', 'volume']
    
    # Lorsque le bot sera en live, il sera important de ne pas retélécharger constamment les données
    # donc on utilisera 'current_hour' pour regarder si on a déjà les données liées à l'heure actuel dans notre cache ou non
    current_hour = datetime.now().replace(minute=0, second=0, microsecond=0)

    # Si les données sont déjà présentes et valides, on sort directement
    if f'{column_name}_close' in dataframe.columns and dataframe[f'{column_name}_close'].max() != -1:
        dataframe[f'{column_name}'] = dataframe[f'{column_name}_close'].ffill()
        return dataframe

    # Clé de cache pour éviter de recharger les mêmes données si on les a déjà chargé
    cache_key = (symbol, exchange, interval.value, 4000) if backtest else (current_hour, symbol, exchange, interval.value, 4000)

    # Chargement des données depuis TradingView, avec fallback si erreur
    if cache_key not in cache:
        try:
            cache[cache_key] = tv.get_hist(symbol, exchange, interval, n_bars=4000)
        except:
            time.sleep(10)
            try:
                cache[cache_key] = tv.get_hist(symbol, exchange, interval, n_bars=2000)
            except:
                time.sleep(20)
                cache[cache_key] = tv.get_hist(symbol, exchange, interval, n_bars=2000)

    # Nettoyage de cache obsolète si le bot est en live (sinon au bout de plusieurs jours, le cache peut devenir assez conséquent)
    if not backtest:
        cache.pop((current_hour - timedelta(hours=1), symbol, exchange, interval.value, 4000), None)

    # Mise en forme du dataframe téléchargé
    tv_df = pd.DataFrame(cache[cache_key])
    if tv_df.empty:
        print(f"Aucune donnée de tradingview {symbol} récupérée.")
        for ohlc in [''] + ohlc_types:
            dataframe[f'{column_name}_{ohlc}'.rstrip('_')] = dataframe['close'] * 0 - 1
        return dataframe

    tv_df.drop(columns=["symbol"], errors="ignore", inplace=True)
    tv_df.index = pd.to_datetime(tv_df.index).tz_localize(UTC)

    # Décalage des valeurs pour ne pas utiliser la bougie en cours
    for ohlc in ohlc_types:
        if ohlc in tv_df.columns:
            tv_df[ohlc] = tv_df[ohlc].shift(1)
            tv_df.rename(columns={ohlc: f'{column_name}_{ohlc}'}, inplace=True)

    # Préparation du dataframe principal pour le merge
    dataframe = dataframe.copy()
    dataframe['date'] = pd.to_datetime(dataframe['date'])
    if dataframe['date'].dt.tz is None:
        dataframe['date'] = dataframe['date'].dt.tz_localize(UTC)
    dataframe.set_index('date', inplace=True)

    # Merge des données, à partir de la première date disponible
    start_date = dataframe.index.min().normalize()
    tv_df = tv_df[tv_df.index >= start_date]
    dataframe = dataframe.join(tv_df, how='left')

    # Remplissage des trous par propagation
    for ohlc in ohlc_types:
        col = f'{column_name}_{ohlc}'
        if col in dataframe.columns:
            dataframe[col] = dataframe[col].ffill()

    # Ajout d'une colonne "valeur principale" = close
    dataframe[f'{column_name}'] = dataframe.get(f'{column_name}_close', dataframe['close'] * 0 - 1).ffill()

    # Retour au format d'origine
    dataframe.reset_index(inplace=True)
    return dataframe





###############################################
# Stratégie KAC-Index : ATR-CCI + Indexation sur Total3

# La stratégie a été trouvée avec les paramètres suivants :
#   max positions = -1
#   timeframe = 1h
#   config = futures_binance_fix.json
#   timerange = 20210101-

#command = freqtrade backtesting --strategy kac_index_v1 --config futures_binance_fix.json --timerange 20210101- --timeframe 1h --max-open-trades -1 --cache none --dry-run-wallet 1000

class kac_index_v1(IStrategy):

    use_exit_signal = True
    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False
    startup_candle_count: int = 10
    position_adjustment_enable = True

    # Détermine la taille personnalisée du premier ordre d'achat.
    # Dans le cas où on utiliserait Hyperliquid, on doit impérativement avoir une position minimum de 15 usdc
    # donc si la stratégie veut ouvrir en dessous, on fixe à 15 usdc minimum
    # Normalement c'est une condition qui ne nous servira que au début de la stratégie, si elle devient gagnante, on dépassera un certain capital qui rendra cette condition inutile.
    def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float, max_stake: float, **kwargs) -> float:
        # Forcer un minimum de 15 USDC pour hyperliquid
        if proposed_stake < 15.0:
            if self.dp.runmode.value != "hyperopt" and self.dp.runmode.value != "backtest":
                logger.info(f"Stake amount for {pair} is too small ({proposed_stake} < 15.0), adjusting to 15.0.")
            proposed_stake = 15.0
        return proposed_stake
    
    
    INTERFACE_VERSION = 3
    timeframe = '1h'
    can_short: bool = False
    process_only_new_candles = True

    # Takeprofit
    minimal_roi = {
        # Si à n'importe quel moment dès l'ouverture de la position : la position atteint 143% de profit, alors on ferme la position
        '0': 1.45,
        
        # Si 4 jours (5760 minutes) après l'ouverture de la position : la position atteint 133% de profit ou est déjà au dessus, alors on ferme la position
        '5760': 1.331,
        
        # Si 5 jours et demi (131 heures soit 7860 minutes) après l'ouverture de la position : la position atteint 31,5% de profit ou est déjà au dessus, alors on ferme la position
        '7860': 0.314,
        
        # Si 7 jours (169 heures soit 10140) après l'ouverture de la position : la position est au moins supérieur à -28%, alors on ferme la position.
        # Si la position est à -30%, alors elle restera ouverte jusqu'à atteindre -28% ou recevoir un signal de fermeture.
        '10140': -0.282
    }

    stoploss = -0.9

    trailing_stop = False

    order_types = {'entry': 'limit','exit': 'limit','stoploss': 'market','stoploss_on_exchange': False}
    order_time_in_force = {'entry': 'GTC', 'exit': 'GTC'}


    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        in_backtest=True if self.dp.runmode.value in ("backtest", "hyperopt") else False
        
        #Indexation sur le TOTAL3 : On récupère les valeurs du TOTAL3 sur tradingview
        # (comme l'historique Trading view ne permet pas de récupérer la donnée en timeframe 1h depuis 2021, on récupère la donnée en timeframe 1d)
        dataframe = add_tv_graph(dataframe, backtest=in_backtest)
        
        # Ensuite, on lisse la donnée récupérée au format 1d, pour la transformer sur 24 heures grâce à une moyenne sur un rolling
        # (on convertit ainsi notre donnée en tf 1d en une donnée lisée en tf 1h)
        dataframe['TOTAL3'] = dataframe['1d_TOTAL3'].rolling(24).mean()
        
        #Indexation sur le TOTAL3 : On divise toutes les colonnes ohlc (open, high, low, close) par notre TOTAL3 lissée
        # Nos valeurs ohlc de la paire seront donc indexées par rapport au TOTAL3 et plus par rapport à l'USDC ou l'USDT
        for column in ['close', 'high', 'low', 'open']:
            column_old=column+'_old'
            dataframe[column_old]=dataframe[column]
            dataframe[column]=dataframe[column]/dataframe['TOTAL3']
            
        # Indicateur ATR (mais sur une indexation au TOTAL3 puisque ohlc avec une indexation sur le TOTAL3 sont utilisés)
        dataframe["ATR"] = (ta.ATR(dataframe, timeperiod=16) / dataframe["close"]) * 100
        
        # Indicateur CCI (avec indexation sur le TOTAL3 également)
        dataframe['CCI'] = clean_ta.trend.CCIIndicator(high=dataframe['high'], low=dataframe['low'], close=dataframe['close'], window=10, constant=0.0175).cci()
        
        
        #Indexation sur le TOTAL3 : On remet les valeurs ohlc à leurs valeurs d'origines (sinon FreqUI, backtest, etc. ne comprennent pas)
        for column in ['close', 'high', 'low', 'open']:
            column_old=column+'_old'
            dataframe[column]=dataframe[column_old]
        
        
        # Plot configs - Permet de réprésenter les zones où les indicateurs respectent les conditions
        # Points d'entrée
        dataframe['I-ATR'] = np.where((dataframe['ATR'] > 1.5) & (dataframe['ATR'] > dataframe['ATR'].shift(1)), dataframe['ATR'], np.nan)
        dataframe['I-CCI'] = np.where((dataframe['CCI'] < 0), dataframe['CCI'], np.nan)

        # Points de sortie
        dataframe['C-ATR'] = np.where((dataframe['ATR'] < -2) & (dataframe['ATR'] < dataframe['ATR'].shift(1)), dataframe['ATR'], np.nan)
        dataframe['C-CCI'] = np.where((dataframe['CCI'] > 200) & (dataframe['CCI'] > dataframe['CCI'].shift(1)), dataframe['CCI'], np.nan)

        return dataframe


    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # On applique les conditions d'entrée de notre stratégie
        dataframe.loc[reduce(lambda x, y: x & y, [
            # ATR croissant et supérieur à 1.5
            (dataframe['ATR'] > 1.5),
            (dataframe['ATR'] > dataframe['ATR'].shift(1)),
            
            # CCI en dessous de 0
            (dataframe['CCI'] < 0),
        ]), 'enter_long'] = 1
        return dataframe


    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # On applique les conditions de sortie de notre stratégie
        dataframe.loc[reduce(lambda x, y: x & y, [
            # ATR décroissant et en dessous de -2
            (dataframe['ATR'] < -2),
            (dataframe['ATR'] < dataframe['ATR'].shift(1)),
            
            # CCI croissant et au dessus de 200
            (dataframe['CCI'] > 200),
            (dataframe['CCI'] > dataframe['CCI'].shift(1))
        ]), 'exit_long'] = 1
        return dataframe


    plot_config = {
        "main_plot": {},
        "subplots": {
            "CCI": {
                "CCI": {
                    "color": "#53387a",
                    "type": "line"
                },
                "C-CCI": {
                    "color": "#ff0000",
                    "type": "scatter",
                    "scatterSymbolSize": 6
                },
                "I-CCI": {
                    "color": "#00ff37",
                    "type": "scatter"
                }
            },
            "ATR": {
                "ATR": {
                    "color": "#eba1b1"
                },
                "I-ATR": {
                    "color": "#15ff00",
                    "type": "line"
                },
                "C-ATR": {
                    "color": "#ff0000",
                    "type": "line"
                }
            }
        }
    }