# source: https://raw.githubusercontent.com/remiotore/freqtrade/44beaeb6a420cd8e9f2e4ea93e11d6cfa192ee03/strategies/kac_index_v2_10x.py # ============================================================= # ➤ Stratégie KAC-Index : ATR-CCI + Indexation sur Total3 # # Fichier généré par Moutonneux 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 from typing import Optional, Union, Any, Dict import talib.abstract as ta import ta as clean_ta from freqtrade.strategy import timeframe_to_minutes, BooleanParameter, IntParameter, CategoricalParameter, DecimalParameter, IStrategy, merge_informative_pair import logging from freqtrade.optimize.space import Categorical, Integer, SKDecimal from freqtrade.optimize.hyperopt import IHyperOptLoss # ❌ 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 = {} # 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: tv = TvDatafeed() 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_v2 --config futures_binance_fix.json --timerange 20210101- --timeframe 1h --max-open-trades -1 --cache none --dry-run-wallet 1000 class Github_remiotore_freqtrade__kac_index_v2_10x__20260111_210550(IStrategy): trailing_stop = False # minimal_roi = { # "0": 4.05, # "6000": 3.42, # "12000": 2.58, # "16560": 0.53, # "19440": -0.02 # } minimal_roi = {'0': 4.05, '15': 1, '30': 0.8, '60': 0.5, '90': 0.37, '2880': -0.3} 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 stoploss = -0.4 trailing_stop = False order_types = {'entry': 'limit', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': False} order_time_in_force = {'entry': 'GTC', 'exit': 'GTC'} def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: """ Customize leverage for each new trade. This method is only called in futures mode. :param pair: Pair that's currently analyzed :param current_time: datetime object, containing the current datetime :param current_rate: Rate, calculated based on pricing settings in exit_pricing. :param proposed_leverage: A leverage proposed by the bot. :param max_leverage: Max leverage allowed on this pair :param entry_tag: Optional entry_tag (buy_tag) if provided with the buy signal. :param side: "long" or "short" - indicating the direction of the proposed trade :return: A leverage amount, which is between 1.0 and max_leverage. """ return 10.0 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: in_backtest = True if self.dp.runmode.value in ('backtest', 'hyperopt') else False dataframe = add_tv_graph(dataframe, backtest=in_backtest) dataframe['TOTAL3'] = dataframe['1d_TOTAL3'].rolling(28).mean() for column in ['close', 'high', 'low', 'open']: column_old = column + '_old' dataframe[column_old] = dataframe[column] dataframe[column] = dataframe[column] / dataframe['TOTAL3'] dataframe['ATR'] = ta.ATR(dataframe, timeperiod=8) / dataframe['close'] * 100 dataframe['CCI'] = clean_ta.trend.CCIIndicator(high=dataframe['high'], low=dataframe['low'], close=dataframe['close'], window=6, constant=0.01625).cci() for column in ['close', 'high', 'low', 'open']: column_old = column + '_old' dataframe[column] = dataframe[column_old] dataframe['I-ATR'] = np.where(dataframe['ATR'] > 2.75, dataframe['ATR'], np.nan) dataframe['C-ATR'] = np.where(dataframe['ATR'] < dataframe['ATR'].shift(1), dataframe['ATR'], np.nan) dataframe['C-CCI'] = np.where(dataframe['CCI'] > dataframe['CCI'].shift(1), dataframe['CCI'], np.nan) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[reduce(lambda x, y: x & y, [dataframe['ATR'] > 2.75]), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: conditions = [] conditions.append(dataframe['ATR'] < dataframe['ATR'].shift(1)) conditions.append(dataframe['CCI'] > dataframe['CCI'].shift(1)) dataframe.loc[reduce(lambda x, y: x & y, conditions), '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'}}}}