DCAv2/utils/fetch_pair_volatility.py

'''
Downloads data from the exchange, calculates the volatilities and writes the results to a database.
'''


import ccxt
import time
import sqlite3
import math
import statistics
from credentials import get_credentials
from threading import Thread


def yang_zhang(candles):
    '''
    Calculates Yang-Zhang volatility
    
    candles: list of candles in the form of [time, open, high, low, close, volume]
    
    Yang, D., & Zhang, Q. (2000). Drift‐Independent Volatility Estimation Based on High, Low, Open, and Close Prices. The Journal of Business, 73(3), 477–492. https://doi.org/10.1086/209650
    '''
    normalized_open = []
    normalized_high = []
    normalized_down = []
    normalized_close = []
    rogers_satchell = []

    #Calculating normalized values
    for x in range(1,len(candles)):
        normalized_open.append(math.log(candles[x][1]/candles[x-1][4]))
        normalized_high.append(math.log(candles[x][2]/candles[x][1]))
        normalized_down.append(math.log(candles[x][3]/candles[x][1]))
        normalized_close.append(math.log(candles[x][4]/candles[x][1]))
        rogers_satchell.append(normalized_high[x-1]*(normalized_high[x-1]-normalized_close[x-1])+normalized_down[x-1]*(normalized_down[x-1]-normalized_close[x-1]))

    #Calculating volatilities
    rs_volatility = math.sqrt(sum(rogers_satchell)/(len(rogers_satchell)))
    no_volatility = statistics.stdev(normalized_open)
    nc_volatility = statistics.stdev(normalized_close)

    #Calculate constant k
    k = 0.34/(1.34+((len(candles)+1)/(len(candles)-1)))

    return math.sqrt(no_volatility**2+k*nc_volatility**2+(1-k)*rs_volatility**2)


def parkinson(candles):
    '''
    Calculates Parkison volatility.
    
    candles: list of lists with the format [timestamp,open,high,low,close,volume]
    
    Parkinson, M. (1980) The Extreme Value Method for Estimating the Variance of the Rate of Return. Journal of Business, 53, 61-65. https://doi.org/10.1086/296071 
    '''    
    
    logsquared = [math.log(float(x[2])/float(x[3]))**2 for x in candles]
    avg = sum(logsquared)/len(logsquared)
    return math.sqrt(avg/(4*math.log(2)))


def write_to_db(broker,data):
    """
    Write data to database
    """

    database_connection = sqlite3.connect(f"{get_credentials('VOL_DB_PATH')['path']}{broker}.db")
    database_cursor = database_connection.cursor()
    for item in data:
        database_cursor.execute('INSERT INTO volatilities_table VALUES(?, ?, ?, ?, ?)', [item, int(time.time()), data[item][0], data[item][1], data[item][2]])
    database_connection.commit()
    database_connection.close()    
    return 0


def get_pair_list(broker, inclusions = ["/USDT"], exclusions = []):
    """
    Get the list of pairs from exchange
    """
    
    exchange = getattr(ccxt, broker)
    exchange = exchange({   
        "apiKey": "",
        "secret": "",
        "password": "",                                 
        "timeout": 30000,
        "enableRateLimit": True
    })
    
    pair_list = [pair for pair in exchange.fetch_tickers().keys()]
    
    for inclusion in inclusions:
        for pair in pair_list.copy():
            if inclusion not in pair:
                pair_list.remove(pair)
    
    for exclusion in exclusions:
        for pair in pair_list.copy():
            if exclusion in pair:
                pair_list.remove(pair)
    
    return pair_list
    

def fetch_data(broker: str, pair_list: list, timeframe: str, samples: int):
    """
    Fetch data from exchange
    """

    global volatilities

    wait_time = .5       #Sleep time between requests
    index = 0

    exchange = getattr(ccxt, broker)
    exchange = exchange({   
        "apiKey": "",
        "secret": "",
        "password": "",                                 
        "timeout": 30000,
        "enableRateLimit": True
    })
    
    for pair in pair_list:
        trading_volume = 0
        index += 1
        try:
            data = exchange.fetch_ohlcv(pair,timeframe=timeframe,limit=samples)
        except Exception as e:
            print(e)
            continue
        try:
            parkinson_volatility = parkinson(data)
            yangzhang_volatility = yang_zhang(data)
        except Exception as e:
            print(e)
            continue
        for item in data:
            trading_volume += item[4]*item[5]    
        volatilities[broker][pair] = [round(yangzhang_volatility*100,2),round(parkinson_volatility*100,2),int(trading_volume)]
        print(f"{pair} on {broker} ready, {len(pair_list)-index} pairs remaining.")
        time.sleep(wait_time)
    
    return 0


if __name__=="__main__":

    threads = []
    exchanges = ["binance","gateio","kucoin","okx"]
    pair_list = []
    volatilities = {item:{} for item in exchanges}
    exchange_list = [item for item in volatilities]
    samples = 288
    timeframe = "5m"
    minimum_volume = 0

    #Create databases for each exchange
    for item in exchange_list:
        database_connection = sqlite3.connect(f"{get_credentials('VOL_DB_PATH')['path']}{item}.db")
        database_cursor = database_connection.cursor()           
        database_cursor.execute('''
            CREATE TABLE IF NOT EXISTS volatilities_table (
            pair TEXT,
            timestamp INTEGER,
            yang_zhang REAL,
            parkinson REAL,
            volume REAL)''')
        database_connection.commit()
        database_connection.close()


    for broker in exchange_list:
        threads.append(Thread(target=fetch_data,args=(broker, get_pair_list(broker), timeframe, samples,)))
        
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    
    for item in exchange_list:
        write_to_db(item,volatilities[item])