My CCXT connector to BackTesting.py (my gift)

[pliciwebstephane]

Hello,

Your work is awsome and realy cool.
I am a PHP developper and i started Python yesteday (really :) ).
I am french, sorry for my poor english.

I think many people want to use CCXT with Backtesting.py, that's why, i want to share my code with you (it is my gift for you).
This load as many candle you want from now based on 1000 candles by 1000 candles.

In the Backtest file run with :

exchange = 'binance'  # ccxt exchange
myccxt = MyCCXT(exchange)
df_ohlcv = myccxt.getOHLC(nbcandles=bt_nbcandles, ut=bt_ut, symbol=bt_coin)              # récupération des candles
bt = Backtest(df_ohlcv, myStrategy, cash = 100000, commission=bt_fees, exclusive_orders=True)   # Init backtesting

The MyCCXT class

import sys
import ccxt
import pandas
import datetime

class MyCCXT:

    exchange = None
    maxcandles_one_time = 1000

    def __init__(self, exchange = 'bitmex'):
        print("CCXT : Run CCXT on ", exchange)
        exchange_class = getattr(ccxt, exchange)
        self.exchange = exchange_class(
            # {
            # 'apiKey': 'YOUR_API_KEY',
            # 'secret': 'YOUR_SECRET',
            # }
        )
        # self.exchange.verbose = True
        self.exchange.options['fetchOHLCVOpenTimestamp'] = False


    def getOHLC(self, nbcandles = 500, ut = '1h', symbol = 'BTC/USD'):
        

        orig_length = nbcandles
        candle_in_seconds = ccxt.Exchange.parse_timeframe(ut)* 1000
        # première demande de l'ohlc, elle peut suffir 
        a_ohlc = self.__subGetOHLC(nbcandles=nbcandles, ut=ut, symbol=symbol)

        print("Recevied {} candles".format(len(a_ohlc)))

        need_more_candle = False
        # On a déjà le bon nombre de candles ?
        if (len(a_ohlc) == orig_length):
            need_more_candle = False # oui, alors on est bon
        else: 
            need_more_candle = True
        

        global_ohlc = a_ohlc.copy()

        


        if need_more_candle :
            max_candles = len(a_ohlc)
            while True:
                length = orig_length - len(global_ohlc) # on a encore besoin de  X candles
                # print("Besoin de {} candles".format(length))
                if (length == 0):
                    break
                old_time = global_ohlc['Time'].iloc[0] # // le plus vieux timestamp reçu
                # print("OldTime {} ".format(old_time))
                step = min(length, max_candles)*(candle_in_seconds) #; // on remonte de length ou de max possible
                # print("On remonte de step : {} ".format(step))
                more_candle = min(length, max_candles+1)
                since=(old_time-step)

                # print("===> Ask for time END {}".format(self.exchange.ymdhms(old_time-step)))
                # print("===> Ask for time START {}".format(self.exchange.ymdhms(old_time)))

                # print("Since {} ".format(since))
                a_ohlc = self.__subGetOHLC(since = since, nbcandles = more_candle, ut = ut, symbol = symbol)
                # print("Recevied {} ".format(len(a_ohlc)))

                global_ohlc = pandas.concat([global_ohlc, a_ohlc])
                # print("TOTAL Recevied {} ".format(len(global_ohlc)))

                # // sort by timestamp
                # print("Sorting")
                global_ohlc.sort_values(by=['Time'], inplace=True)
                global_ohlc.drop_duplicates(inplace=True)

                print("We Are at {} candles, ask for more {} candles".format(len(global_ohlc), more_candle))

                if (length < 0):
                    print("break {} ".format(length))
                    break
        

        # // test de cohérence des timestamp, même écart tout le temps
        if (len(global_ohlc) != orig_length):
            print("Sorry not all candles recevied ... recevied {} wanted {} ".format(len(global_ohlc), orig_length))
            sys.exit()
        else:
            print("Size is OK")

        # print(global_ohlc)

        # vérification de la cohérence des données reçues dont le timing
        start_time = None
        print("Checking Time is well incremntal")
        for x in range(0, len(global_ohlc)-1):
            # print("Checking X => {}".format(x))
            if start_time is not None:
                if (start_time+candle_in_seconds) != global_ohlc['Time'].iloc[x]:
                    print("Probleme il manque des bougies au milieu on dirait...")
                    # print("on a traité {}".format(start_time))
                    # print(self.exchange.ymdhms(start_time))
                    # print("on attend {}".format((start_time+candle_in_seconds)))
                    # print(self.exchange.ymdhms((start_time+candle_in_seconds)))
                    # print("on trouve {}".format(global_ohlc['Time'].iloc[x]))
                    # print(self.exchange.ymdhms(global_ohlc['Time'].iloc[x]))
                    sys.exit()
            start_time=global_ohlc['Time'].iloc[x]
        print("Check DONE Time coherence is => OK")


        return global_ohlc

    def __subGetOHLC(self, since = None, nbcandles = 500, ut = '1h', symbol = 'BTC/USD'):

        nbcandles = min([self.maxcandles_one_time, nbcandles])
        # print('CCXT : __subGetOHLC : {}'.format(since))

        ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe=ut, since=since, limit=nbcandles)
        # print('CCXT : Nombre de candles received : {}'.format(len(ohlcv)))

        df_ohlcv = pandas.DataFrame(ohlcv, columns = ['Time', 'Open', 'High', 'Low', 'Close', 'Volume'])
        # conversion des datetime unix en date humaine
        df_ohlcv['HTime'] = [datetime.datetime.fromtimestamp(float(time)/1000) for time in df_ohlcv['Time']]
        # df_ohlcv.drop('Volume', axis=1, inplace=True) # pour laisser les volumes
        # on dit que Time c'est la clé finalement
        df_ohlcv.set_index('HTime', inplace=True)
        df_ohlcv.tail() # pour applatir je pense après le Time en index

        # print(df_ohlcv)
        return df_ohlcv

[pliciwebstephane]

If you use this script, please make a comment :) I will like it :)

[eervin123]

Love this script. Much appreciated. I've shared it with several friends who aren't really savvy with python and they all picked it up very quickly.

Just like backtesting.py, this makes life easy for people to roll up their sleeves and learn. So, thank you to you and thank you to @kernc

[pliciwebstephane]

Nice, thanks !