mds-tools/qhp-download-futures.py

#!/usr/bin/env python3

import sys
import argparse
import zmq
import io
import json
import csv
import datetime
import struct
import dateutil.tz

def timeframe_to_seconds(tf):
    if tf == 'M1':
        return 60
    elif tf == 'M5':
        return 5 * 60
    elif tf == 'M15':
        return 15 * 60
    elif tf == 'H1':
        return 3600
    elif tf == 'D':
        return 86400
    elif tf == 'W':
        return 7 * 86400
    else:
        raise ValueError('Invalid value')

class BarAggregator:
    def __init__(self, timeframe, tz):
        self.open_ = 0
        self.high = 0
        self.low = 0
        self.close = 0
        self.volume = 0
        self.timestamp = None
        self.current_bar_number = None
        self.timeframe = timeframe
        self.tz = tz

    def clear(self):
        self.open_ = 0
        self.high = 0
        self.low = 0
        self.close = 0
        self.volume = 0
        self.timestamp = None
        self.current_bar_number = None
    
    def push_bar(self, timestamp, open_, high, low, close, volume):
        bar_number = timestamp.timestamp() // self.timeframe
        if bar_number != self.current_bar_number:
            b_open = self.open_
            b_high = self.high
            b_low = self.low
            b_close = self.close
            b_volume = self.volume
            if self.current_bar_number is not None:
                b_timestamp = datetime.datetime.fromtimestamp(self.current_bar_number * self.timeframe, self.tz)

            self.open_ = open_
            self.high = high
            self.low = low
            self.close = close
            self.volume = volume
            self.timestamp = timestamp
            prev_bar_number = self.current_bar_number
            self.current_bar_number = bar_number

            if prev_bar_number is not None:
                return (b_timestamp, b_open, b_high, b_low, b_close, b_volume)
        else:
            self.high = max(high, self.high)
            self.low = min(low, self.low)
            self.close = close
            self.volume += volume
            return None

    def get_bar(self):
        if self.timestamp is None:
            return None
        b_open = self.open_
        b_high = self.high
        b_low = self.low
        b_close = self.close
        b_volume = self.volume
        b_timestamp = datetime.datetime.fromtimestamp(self.timeframe * ( self.timestamp.timestamp() // self.timeframe), self.tz)

        return (b_timestamp, b_open, b_high, b_low, b_close, b_volume)

def get_data(qhp, ticker, start_time, end_time, period, tz, timedelta):
    utctz = dateutil.tz.gettz('UTC')
    rq = {
        "ticker" : ticker,
        "from" : start_time.strftime("%Y-%m-%dT%H:%M:%S"),
        "to" : end_time.strftime("%Y-%m-%dT%H:%M:%S"),
        "timeframe" : period
    }

    qhp.send_multipart([bytes(json.dumps(rq), "utf-8")])
    resp = qhp.recv()

    if resp != b'OK':
        errmsg = qhp.recv_string()
        return None

    bar_count = 0
    result = []
    while True:
        if qhp.getsockopt(zmq.RCVMORE) == 0:
            break
        rawdata = qhp.recv()
        for line in struct.iter_unpack("<qddddQ", rawdata):

            timestamp = int(line[0])
            open_ = float(line[1])
            high = float(line[2])
            low = float(line[3])
            close = float(line[4])
            volume = int(line[5])
            dt = datetime.datetime.fromtimestamp(timestamp, utctz) + timedelta
            dt = dt.astimezone(utctz)

            bar_count += 1
            result.append((dt, open_, high, low, close, volume))

    return result

def write_to_file(writer, bars, ticker, period, agg):
    for bar in bars:
        if agg is not None:
            mbar = agg.push_bar(bar[0], bar[1], bar[2], bar[3], bar[4], bar[5])
            if mbar is not None:
                writer.writerow([ticker, agg.timeframe, mbar[0].strftime("%Y%m%d"), mbar[0].strftime("%H%M%S"), mbar[1], mbar[2], mbar[3], mbar[4], mbar[5]])
        else:
            writer.writerow([ticker, period, bar[0].strftime("%Y%m%d"), bar[0].strftime("%H%M%S"), bar[1], bar[2], bar[3], bar[4], bar[5]])
    if agg is not None:
        mbar = agg.get_bar()
        if mbar is not None:
            writer.writerow([ticker, agg.timeframe, mbar[0].strftime("%Y%m%d"), mbar[0].strftime("%H%M%S"), mbar[1], mbar[2], mbar[3], mbar[4], mbar[5]])
            agg.clear()
        else:
            print('none bar' ,ticker )

def make_tickers_list(base, start_time, end_time, futures_interval):
    result = []
    month = start_time.date().month
    year = start_time.date().year

    while True:
        if month % futures_interval == 0:
            result.append(base + '-' + str(month) + '.' + str(year)[-2:])
            if month > end_time.date().month and year >= end_time.date().year:
                break

        month += 1
        if month > 12:
            month = 1
            year += 1

    return result
            

def main():
    parser = argparse.ArgumentParser(description='QHP client')
    parser.add_argument('-o', '--output-file', action='store', dest='output_file', help='Output filename', required=True)
    parser.add_argument('-p', '--timeframe', action='store', dest='timeframe', help='Data timeframe', required=True)
    parser.add_argument('-q', '--qhp', action='store', dest='qhp', help='QHP endpoint', required=True)
    parser.add_argument('-y', '--symbol', action='store', dest='symbol', help='Base symbol', required=True)
    parser.add_argument('-f', '--from', action='store', dest='from_', help='Starting date', required=True)
    parser.add_argument('-t', '--to', action='store', dest='to', help='Ending date', required=True)
    parser.add_argument('-r', '--rescale', action='store', dest='rescale', help='Rescale to timeframe')
    parser.add_argument('-d', '--time-delta', action='store', dest='time_delta', help='Add given time delta (in seconds)', required=False)
    parser.add_argument('-i', '--futures-interval', action='store', dest='futures_interval', help='Futures interval between exprations in month', required=True)
    parser.add_argument('-s', '--stitch-delta', action='store', dest='stitch_delta', help='Futures interval between exprations in month', required=True)
    parser.add_argument('-e', '--replace-ticker', action='store', dest='replace_ticker', help='Replace ticker id in file', required=False)
    parser.add_argument('-z', '--timezone', action='store', dest='timezone', help='Convert bar timestamps to given timezone', required=False)

    args = parser.parse_args()

    period = args.timeframe
    symbol = args.symbol
    filename = args.output_file

    ctx = zmq.Context.instance()
    s = ctx.socket(zmq.REQ)
    s.connect(args.qhp)

    start_time = datetime.datetime.strptime(args.from_, "%Y%m%d")
    end_time = datetime.datetime.strptime(args.to, "%Y%m%d")

    if filename == "!":
        if args.rescale is not None:
            filename = "{}_{}_{}_{}.csv".format(symbol, args.from_, args.to, args.rescale)
        else:
            filename = "{}_{}_{}_{}.csv".format(symbol, args.from_, args.to, args.period)

        print("Assuming filename: {}".format(filename))

    timedelta = datetime.timedelta()
    if args.time_delta:
        timedelta = datetime.timedelta(seconds=int(args.time_delta))

    tz = dateutil.tz.gettz('UTC')
    if args.timezone is not None:
        tz = dateutil.tz.gettz(args.timezone)

    delta = int(args.stitch_delta)

    agg = None
    if args.rescale:
        agg = BarAggregator(int(args.rescale), tz)

    data = {}
    tickers = make_tickers_list(symbol, start_time, end_time, int(args.futures_interval))
    print("Tickers: {}".format(tickers))
    empty_tickers = []
    for ticker in tickers:
        print("Requesting data: {}".format(ticker))
        bars = get_data(s, ticker, start_time, end_time, period, tz, timedelta)

        if len(bars) > 0:
            data[ticker] = { 'bars' : bars }
        else:
            empty_tickers.append(ticker)
            
    tickers = [t for t in tickers if not (t in empty_tickers)]

    for ticker in tickers:
        print("Cutting off trailing data: {}".format(ticker))
        end_date = data[ticker]['bars'][-1][0].date()
        if ticker != tickers[-1]:
            cutoff_date = datetime.date.fromordinal(end_date.toordinal() - delta)
        else:
            cutoff_date = datetime.date.fromordinal(end_date.toordinal())

        data[ticker]['bars'] = [s for s in data[ticker]['bars'] if s[0].date() <= cutoff_date]
        data[ticker]['end_date'] = cutoff_date

    prev_ticker = None
    for k, v in sorted(data.items(), key=lambda x: x[1]['end_date']):
        print("Cutting off starting data: {}".format(k))
        if prev_ticker is not None:
            start_date = data[prev_ticker]['bars'][-1][0]
            v['bars'] = [s for s in data[k]['bars'] if s[0] > start_date]
            print("{} : {}".format(data[prev_ticker]['bars'][-1][0], data[k]['bars'][0][0]), v['bars'][0][0])
        prev_ticker = k
        
    with open(filename, 'w+') as f:
        writer = csv.writer(f)
        writer.writerow(['<TICKER>', '<PER>', '<DATE>', '<TIME>', '<OPEN>', '<HIGH>', '<LOW>', '<CLOSE>', '<VOLUME>'])
        for k, v in sorted(data.items(), key=lambda x: x[1]['end_date']):
            ticker = args.replace_ticker
            if ticker is None:
                ticker = k
            write_to_file(writer, v['bars'], ticker, period, agg)

if __name__ == '__main__':
    main()