predict_arb/tradovate.py

import base64
import aiohttp
import websockets
import os
import hmac
import asyncio
import operator
from decimal import Decimal as D
import dill as pickle
import hashlib
import json
import requests
import datetime
import time
from utils import *
from piecewise import *

TRADOVATE_AUTH = read_auth("tradovate")
TRADOVATE_SESSION_FILE = ".tradovate_session"
TRADOVATE_DEVICEID = "b21da153-4e25-4679-b958-053cc5dc8eeb"
# My Own: "76fdc8d9-e156-46a4-b2b2-187087fcd35e"
TRADOVATE_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36"
# My Own: "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36",
TRADOVATE_ROOT = "https://demo.tradovateapi.com/v1" if DEMO else "https://live.tradovateapi.com/v1"
TRADOVATE_LIVE_ROOT = "https://live.tradovateapi.com/v1"
TRADOVATE_WS = "wss://demo.tradovateapi.com/v1/websocket" if DEMO else "wss://live.tradovateapi.com/v1/websocket"
TRADOVATE_MD_WS = "wss://md-demo.tradovateapi.com/v1/websocket" if DEMO else "wss://md.tradovateapi.com/v1/websocket"
TRADOVATE_APPID = "tradovate_trader(web)"
TRADOVATE_ORIGIN = "https://trader.tradovate.com"

COMMON_HEADERS = {
    "accept": "*/*",
    "accept-encoding": "gzip, deflate, br",
    "accept-language": "en-US,en;q=0.9",
    "cache-control": "no-cache",
    "origin": TRADOVATE_ORIGIN,
    "pragma": "no-cache",
    "referer": TRADOVATE_ORIGIN+"/",
    "user-agent": TRADOVATE_USER_AGENT,
}

def conv_ask_to_pc(lims, p, s):
    # TODO: make sure this is correct
    fee = 11 * 5
    return (conv_lim_pc(*lims) + (-p - fee), s, p)

def convert_contract_hundred_lower(data):
    rem = 0
    ret = []
    for p, s in data:
        s += rem
        if s//5: ret.append((p*5, s//5))
        rem = s%5
    return ret

class TradovateSocket:
    # TODO: figure out where the weird r query param comes from
    def __init__(self, url, cookie_jar):
        self.context = websockets.connect(
            url,
            extra_headers = {
                "Accept-Encoding": "gzip, deflate, br",
                "Accept-Language": "en-US,en;q=0.9",
                "Cache-Control": "no-cache",
                "Connection": "Upgrade",
                "Pragma": "no-cache",
                "Upgrade": "websocket",
                "User-Agent": TRADOVATE_USER_AGENT,
                "Cookie": cookie_jar.filter_cookies(url).output(header=''),
            },
            origin = TRADOVATE_ORIGIN,
        )
        self.id = 1
        self.cur_time = time.time()

    async def __aenter__(self):
        self.ws = await self.context.__aenter__()
        return self

    async def __aexit__(self, *args, **kwargs):
        await self.context.__aexit__(*args, **kwargs)

    async def send(self, op, query="", body=""):
        msg = "%s\n%d\n%s\n%s" % (op, self.id, query, body)
        self.id += 1
        await self.ws.send(msg)

    async def stream(self):
        async for msg in self.ws:
            print(msg)
            now = time.time()

            if now - self.cur_time > 2.5:
                await self.ws.send('[]')
                self.cur_time = now

            if msg[0]!='a': continue
            yield msg[1:]

class TradovateSession:
    def __init__(self, s):
        self.s = s
        self.ticker_lookup = {}

    def __repr__(self):
        return "tradovate"

    @staticmethod
    def make_login_payload(username, password):
        ts = str(int(time.time()*1000 - 1581e9))

        sec = hmac.new(
            "035a1259-11e7-485a-aeae-9b6016579351".encode(),
            (ts + TRADOVATE_DEVICEID + username + password + TRADOVATE_APPID).encode(),
            digestmod = hashlib.sha256
        ).hexdigest()

        return {
            "name": username,
            "password": password,
            "appId": TRADOVATE_APPID,
            "appVersion": "1.221209.1",
            "deviceId": TRADOVATE_DEVICEID,
            # "userAgent": "Dart/2.17.6; ec/1.221209.0", # This depends on version too
            "environment": "demo",
            "cid": 1, # This is client id
            "sec": sec,
            "chl": ts,
        }

    # Making descision not to allow not to cache logins
    # a) They have limited logins allowed, so until we can detect auth fail, this would be bad
    # b) EC seems to have short login window anyways
    # Still should allow it for development work when I'm slamming them
    # TODO: renew these tokens, expiry is every 80 min
    async def login(self, use_cached=False):
        self.s.headers.clear()
        self.s.headers.update(COMMON_HEADERS)
        self.s.cookie_jar.clear()

        if use_cached and os.path.exists(TRADOVATE_SESSION_FILE):
            with open(TRADOVATE_SESSION_FILE, 'rb') as f:
                self.user_id, self.access_token, self.mdaccess_token, cookies = pickle.load(f)
            self.s.headers['authorization'] = "Bearer " + self.access_token
            self.s.cookie_jar._cookies = cookies
        else:
            print("doing new login")
            async with self.s.post(
                TRADOVATE_LIVE_ROOT + "/auth/accesstokenrequest",
                data = json.dumps(TradovateSession.make_login_payload(**TRADOVATE_AUTH)),
                headers = {
                    "content-type": "text/plain; charset=utf-8",
                },
            ) as resp:
                data = await resp.json()
                self.access_token = data["accessToken"]
                self.mdaccess_token = data["mdAccessToken"]
                self.user_id = data["userId"]
                self.s.headers['authorization'] = "Bearer " + self.access_token

            with open(TRADOVATE_SESSION_FILE, 'wb') as f:
                pickle.dump((self.user_id, self.access_token, self.mdaccess_token, self.s.cookie_jar._cookies), f)

    async def get_ldep(self):
        async with self.s.get(
            TRADOVATE_ROOT + "/account/ldeps",
            params = {'masterids': self.user_id},
        ) as resp:
            data = await resp.json()
            self.ldep_id = data[0]['id']

    async def setup_ordering(self):
        await self.get_ldep()

        self.order_sock = TradovateSocket(TRADOVATE_WS, self.s.cookie_jar)

        await self.order_sock.__aenter__()
        await self.order_sock.send("authorize", body=self.access_token)

        async def order_ws_log():
            async for msg in self.order_sock.stream():
                print("ORDER LOG: " + msg)
                print()

        asyncio.create_task(order_ws_log())

## Only myster here is tickers and what accountId is in this case
#  user/registeraudituseraction
#  209
#  
#  {"accountId":1368262,"actionType":"OrderTicketBuy","details":"ORDERTICKET OECES27Z2 C38000: Order Ticket Buy, Buy 3  Limit 15.00, TIF Day"}
#  
#  
#  order/placeorder
#  210
#  
#  {"accountId":1368262,"action":"Buy","symbol":"OECES27Z2 C38000","orderQty":3,"orderType":"Limit","price":15,"timeInForce":"Day","text":"Ticket"}

    async def execute_order(self, mid, p, s, buy):
        p = D(p/5)/100
        s *= 5
        assert buy

        name = self.ticker_lookup[mid]

        audit_msg = {
            "accountId": self.ldep_id,
            "actionType": "OrderTicketBuy",
            "details": "ORDERTICKET %s: Order Ticket Buy, Buy %d  Limit %0.2f, TIF Day"
                % (name, s, p), 
        }

        order_msg = {
            "accountId": self.ldep_id,
            "action": "Buy",
            "symbol": name,
            "orderQty": s,
            "orderType": "Limit",
            "price": float(p),
            "timeInForce": "Day",
            "text": "Ticket",
        }

        print(json.dumps(audit_msg))
        print(json.dumps(order_msg))
        await self.order_sock.send("user/registeraudituseraction", body=json.dumps(audit_msg))
        await self.order_sock.send("order/placeorder", body=json.dumps(order_msg))


    async def markets_from_maturity(self, maturityid):
        async with self.s.get(
            TRADOVATE_ROOT + "/contract/deps",
            params = {'masterid': maturityid},
        ) as resp:
            data = await resp.json()

            lims = {}
            for contract in data:
                mid = contract['id']
                if mid%10!=5: continue
                strike = cents(contract['strikePrice'])
                put = contract['isPut']

                self.ticker_lookup[mid] = contract['name']

                # TODO: figure out boundary
                if put:
                    lims[mid] = (None, strike)
                else:
                    lims[mid] = (strike, None)
            print(lims)
            return lims

    async def orderbook_stream(self, maturityid):
        lims = await self.markets_from_maturity(maturityid)

        book = {}

        async with TradovateSocket(TRADOVATE_MD_WS, self.s.cookie_jar) as ws:
            await ws.send("authorize", body=self.mdaccess_token)
            for symbol in lims:
                await ws.send("md/subscribedom", body='{"symbol":"%d"}' %symbol)

            async for msg in ws.stream():
                data = json.loads(msg)

                doms = []
                for ev in data:
                    if ev.get('e')!='md': continue
                    for dom_ev in ev.get('d', {}).get('doms', []):
                        ts = datetime.datetime.fromisoformat(dom_ev['timestamp'][:-1])
                        doms.append((ts, dom_ev))
                if not doms: continue
                doms.sort(key=operator.itemgetter(0))

                for ts, upd in doms:
                    mid = upd['contractId']

                    prices = [(cents(o['price']), o['size']) for o in upd['offers']]
                    prices.sort()

                    prices = convert_contract_hundred_lower(prices)

                    lim = lims[mid]
                    book[mid] = Digital(
                        *lim,
                        bids = [],
                        asks = [conv_ask_to_pc(lim, p, s) for p, s in prices],
                        exchange = self,
                        market_id = mid,
                    )

                    print(book[mid])
                    if prices:
                        await self.execute_order(mid, book[mid].asks[0][-1], 4, True)
                        import sys
                        sys.exit(0)

                yield [*book.values()]

async def main():
    # print(TradovateSession.make_login_payload(**TRADOVATE_AUTH))
    # return
    async with aiohttp.ClientSession() as ts:
        s = TradovateSession(ts)
        await s.login(use_cached=True)
        await s.setup_ordering()
        async for book in s.orderbook_stream(maturityid=51047):
            print(book)
        # await asyncio.sleep(10000)
        # print(await s.test_req())


if __name__=="__main__":
    asyncio.run(main())