# ============================================================= # ➤ 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 kac_index_v2(IStrategy): trailing_stop = False minimal_roi = { "0": 4.05, "6000": 3.42, "12000": 2.58, "16560": 0.53, "19440": -0.02 } 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.40 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 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" } } } }