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
import sys
from credentials import get_credentials
from threading import Thread

def write_to_log(message):
    '''
    Writes a message to the log file
    '''
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")


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
    })
    
    markets = exchange.load_markets()
    pair_list = [pair for pair in markets if ":" not in pair and markets[pair]["active"]]
    
    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: str, timeframe: str, samples: int):
    """
    Fetch data from exchange
    """
 
    index = 0

    trading_volume = 0
    index += 1
    try:
        data = exchange.fetch_ohlcv(pair,timeframe=timeframe,limit=samples)
    except Exception as e:
        write_to_log(str(e))
        return None
    try:
        parkinson_volatility = parkinson(data)
        yangzhang_volatility = yang_zhang(data)
    except Exception as e:
        write_to_log(str(e))
        return None
    for item in data:
        trading_volume += item[4]*item[5]    
    
    print(f"{pair} on {broker} ready, {len(pair_list)-index} pairs remaining.")
    return [round(yangzhang_volatility*100,2),round(parkinson_volatility*100,2),int(trading_volume)]


if __name__=="__main__":

    threads = []
    volatilities = {}
    samples = 288
    timeframe = "5m"
    minimum_volume = 0
    wait_time = .5

    #Create database if it does not exist
    database_connection = sqlite3.connect(f"{get_credentials('VOL_DB_PATH')['path']}{sys.argv[1]}.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()

    exchange = getattr(ccxt, sys.argv[1])
    exchange = exchange({   
        "apiKey": "",
        "secret": "",
        "password": "",                                 
        "timeout": 30000,
        "enableRateLimit": True
    })
    
    pair_list = get_pair_list(sys.argv[1],inclusions = ["/USDT"], exclusions = [])
    
    for pair in pair_list:
        data = fetch_data(exchange, pair, timeframe, samples)
        if data is not None:
            volatilities[pair] = data
        time.sleep(wait_time)
        
    write_to_db(sys.argv[1],volatilities)