A script to fetch OHLCV data for FTX MOVE daily contracts (e.g. BTC-MOVE-0815)

fetch_BTCMOVE.py

import os
import sys 
argvs = sys.argv  
from datetime import datetime, timedelta, timezone
import json
import pytz
import pandas as pd 
import ciso8601

# For FtxClient ----------------------------------------
import time
import urllib.parse
from typing import Optional, Dict, Any, List

from requests import Request, Session, Response
import hmac
# ------------------------------------------------------


##########################################################################################################
#
#  Global settings
#
##########################################################################################################

# get yesterday's date in UTC 
#  ****Assuming this script will be executed at midnight (0:00)!!****
dt = datetime.utcnow() - timedelta(days=1)
# timezone setting 
jst = pytz.timezone('Asia/Tokyo')

# Market and interval 
prefix = 'BTC-MOVE-' 
market = prefix + dt.strftime('%m%d')
print(f'market is {market}')
#interval = float(sys.argv[1]) if len(sys.argv) >= 2 else 1 # unit: min 
interval = 1 # unit: min 
# Make sure interval is either 0.4/1/5/15/60/240/1440. See get_klines() comment.
if interval not in [0.4, 1, 5, 15, 60, 240, 1440]:
    print(f'Kline interval {interval} has to be either 0.4/1/5/15/60/240/1440.')    
    sys.exit(0)
elif interval >= 1: 
    interval = int(interval)
    print(f'Kline interval is {interval} min.')
else:     
    print(f'Kline interval is {int(interval*60)} sec.')

##########################################################################################################
#
#  FtxClient class 
#    from https://github.com/ftexchange/ftx/blob/master/rest/client.py
#
##########################################################################################################

class FtxClient:
    _ENDPOINT = 'https://ftx.com/api/'

    def __init__(self, api_key=None, api_secret=None, subaccount_name=None) -> None:
        self._session = Session()
        self._api_key = api_key
        self._api_secret = api_secret
        self._subaccount_name = subaccount_name

    def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any:
        return self._request('GET', path, params=params)

    def _post(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any:
        return self._request('POST', path, json=params)

    def _delete(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any:
        return self._request('DELETE', path, json=params)

    def _request(self, method: str, path: str, **kwargs) -> Any:
        request = Request(method, self._ENDPOINT + path, **kwargs)
        self._sign_request(request)
        response = self._session.send(request.prepare())
        return self._process_response(response)

    def _sign_request(self, request: Request) -> None:
        ts = int(time.time() * 1000)
        prepared = request.prepare()
        signature_payload = f'{ts}{prepared.method}{prepared.path_url}'.encode()
        if prepared.body:
            signature_payload += prepared.body
        signature = hmac.new(self._api_secret.encode(), signature_payload, 'sha256').hexdigest()
        request.headers['FTX-KEY'] = self._api_key
        request.headers['FTX-SIGN'] = signature
        request.headers['FTX-TS'] = str(ts)
        if self._subaccount_name:
            request.headers['FTX-SUBACCOUNT'] = urllib.parse.quote(self._subaccount_name)

    def _process_response(self, response: Response) -> Any:
        try:
            data = response.json()
        except ValueError:
            response.raise_for_status()
            raise
        else:
            if not data['success']:
                raise Exception(data['error'])
            return data['result']

    def get_klines(self, market_name: str, resolution: int = 3600, limit: int = 1440, 
                    start_time: int = None, end_time: int = None):
        # resolution: window length in seconds: 15, 60, 300, 900, 3600, 14400, 86400
        # limit: max number to fetch, optional, max 5000
        return self._get(f'markets/{market_name}/candles', {'resolution': resolution, 
                                                            'limit': limit,
                                                            'start_time': start_time,
                                                            'end_time': end_time})


    def get_trades(self, market_name: str, limit: int = 1440, start_time: int = None, end_time: int = None):
        # resolution: window length in seconds: 15, 60, 300, 900, 3600, 14400, 86400
        # limit: max number to fetch, optional, max 5000
        return self._get(f'markets/{market_name}/trades', {'limit': limit,
                                                            'start_time': start_time,
                                                            'end_time': end_time})



# --------------------------------------------------------------------------------------------------------

# connect to FTX 
dir_path = os.getcwd()
APIfile_path = dir_path + '/ftx_config_mainnet.json'
with open(APIfile_path,'r') as f:
    ftx_key = json.load(f)
client = FtxClient(api_key=ftx_key['apiKey'], 
                    api_secret=ftx_key['secret']) 


##########################################################################################################
#
#  Fetch OHLCV function
#
##########################################################################################################

# Get candle stick data from FTX
def get_historical_klines(market, interval, limit=2880, start_time=None, end_time=None):

    # make sure limtit is below 5000 
    if limit > 5000:
        smart_message(f'Max klines is 5000 per request. Getting 5000 klines instead of {limit}.')
        limit = 5000 

    for _ in range(10): 
        try:
            temp_dict = client.get_klines(market_name=market, resolution=int(interval*60), 
                                            limit=limit, start_time=start_time, end_time=end_time)
            #print(temp_dict)
        except Exception as e: 
            print(e)
            print("Failed to get historical kline. Retrying....")
            time.sleep(2)
        else: 
            if len(temp_dict) > 0:  
                break  
            else: 
                time.sleep(1)
    else: # when all the retries failed
        print("(get_historical_klines_simple) Failed 10 times to get historical kline data.")
        # If you like, terminate the program (or report this to Discord/LINE/whatever)
        #sys.exit(0)

    # convert to data frame 
    df = pd.DataFrame.from_dict(temp_dict)
    df.columns = ['Date UTC', 'TimeStamp', 'Open', 'High', 'Low', 'Close', 'Volume']
    # change OHLCV data types from 'object' to 'float'
    df[['Open', 'High', 'Low', 'Close', 'Volume']] = df[['Open', 'High', 'Low', 'Close', 'Volume']].astype('float64')
    df['Date JST'] = [datetime.fromtimestamp(i/1000, jst).strftime('%Y-%m-%d %H:%M:%S.%d')[:-3] for i in df['TimeStamp']] # use JST to convert time
    # change df column order to OHLCV
    df = df[['Open', 'High', 'Low', 'Close', 'Volume', 'TimeStamp', 'Date UTC', 'Date JST']]

    return df

##########################################################################################################
#
#  Main routine
#
##########################################################################################################

def main(): 

    first_date = datetime(2019, 9, 23, 0, 0, 0) # the first day of FTX MOVE contract
    #first_date = datetime(2020, 12, 23, 0, 0, 0) 
    last_date = datetime.utcnow()
    delta = timedelta(days=1)
    
    curr_date = first_date
    while curr_date < last_date:
        sdate = curr_date - delta    
        edate = curr_date + delta
        sts = ciso8601.parse_datetime(sdate.strftime('%Y-%m-%dT00:00:00')).replace(tzinfo=timezone.utc).timestamp()
        ets = ciso8601.parse_datetime(edate.strftime('%Y-%m-%dT00:00:00')).replace(tzinfo=timezone.utc).timestamp()
        market = curr_date.strftime('BTC-MOVE-%Y%m%d') 
        #if market == 'BTC-MOVE-20201231':
        #    market = 'BTC-MOVE-1231'
        if curr_date.year == 2021:
            market = curr_date.strftime('BTC-MOVE-%m%d') 
        print(f'market is {market}')
        df = get_historical_klines(market, interval=interval, 
                                    limit=2880, start_time=sts, end_time=ets)  
        df.to_csv(f'{dir_path}/data/FTX_OHLCV_{curr_date.year}_BTC-MOVE-{curr_date.strftime("%m%d")}_{interval}min.csv')

        curr_date += delta
        

if __name__ == "__main__":
    main()