pikers
/
piker
3
0
Fork 0
Code Issues 2 Pull Requests 5 Packages Projects Releases Wiki Activity

ib.feed: start drafting out `get_mkt_info()` endpoint

master
Tyler Goodlet 2023-05-15 15:35:57 -04:00
parent 1263835034
commit f20e2d6ee2
1 changed files with 55 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

97
piker/brokers/ib/feed.py
View File

@ -19,7 +19,10 @@ Data feed endpoints pre-wrapped and ready for use with ``tractor``/``trio``.
""" """
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from contextlib import asynccontextmanager as acm from contextlib import (
asynccontextmanager as acm,
nullcontext,
)
from decimal import Decimal from decimal import Decimal
from dataclasses import asdict from dataclasses import asdict
from datetime import datetime from datetime import datetime
@ -59,6 +62,9 @@ from .api import (
Contract, Contract,
) )
from ._util import data_reset_hack from ._util import data_reset_hack
from piker._cacheables import (
async_lifo_cache,
)
# https://interactivebrokers.github.io/tws-api/tick_types.html # https://interactivebrokers.github.io/tws-api/tick_types.html
@ -733,41 +739,55 @@ def normalize(
return data return data
# TODO! @async_lifo_cache()
# async def get_mkt_info( async def get_mkt_info(
# fqme: str, fqme: str,
# _cache: dict[str, MktPair] = {} proxy: MethodProxy | None = None,
# ) -> tuple[MktPair, Pair]: ) -> tuple[MktPair, Pair]:
# both = _cache.get(fqme) # we don't need to split off any fqme broker part?
# if both: # bs_fqme, _, broker = fqme.partition('.')
# return both
# proxy: MethodProxy proxy: MethodProxy
# async with open_data_client() as proxy: if proxy is not None:
client_ctx = nullcontext(proxy)
else:
client_ctx = open_data_client
# pair: Pair = await client.exch_info(fqme.upper()) async with client_ctx as proxy:
# mkt = MktPair(
# dst=Asset( try:
# name=pair.baseAsset, (
# atype='crypto', con, # Contract
# tx_tick=digits_to_dec(pair.baseAssetPrecision), details, # ContractDetails
# ), ) = await proxy.get_sym_details(symbol=fqme)
# src=Asset( except ConnectionError:
# name=pair.quoteAsset, log.exception(f'Proxy is ded {proxy._aio_ns}')
# atype='crypto', raise
# tx_tick=digits_to_dec(pair.quoteAssetPrecision),
# ), # pair: Pair = await client.exch_info(fqme.upper())
# price_tick=pair.price_tick,
# size_tick=pair.size_tick, # mkt = MktPair(
# bs_mktid=pair.symbol, # dst=Asset(
# broker='binance', # name=pair.baseAsset,
# ) # atype='crypto',
# both = mkt, pair # tx_tick=digits_to_dec(pair.baseAssetPrecision),
# _cache[fqme] = both # ),
# return both # src=Asset(
# name=pair.quoteAsset,
# atype='crypto',
# tx_tick=digits_to_dec(pair.quoteAssetPrecision),
# ),
# price_tick=pair.price_tick,
# size_tick=pair.size_tick,
# bs_mktid=pair.symbol,
# broker='binance',
# )
# return both
return con, details
async def stream_quotes( async def stream_quotes(
@ -794,18 +814,11 @@ async def stream_quotes(
proxy: MethodProxy proxy: MethodProxy
async with open_data_client() as proxy: async with open_data_client() as proxy:
try: con, details = await get_mkt_info(sym, proxy=proxy)
(
con, # Contract
first_ticker, # Ticker
details, # ContractDetails
) = await proxy.get_sym_details(symbol=sym)
except ConnectionError:
log.exception(f'Proxy is ded {proxy._aio_ns}')
raise
first_ticker = await proxy.get_quote(contract=con)
first_quote = normalize(first_ticker) first_quote = normalize(first_ticker)
# print(f'first quote: {first_quote}') log.runtime(f'FIRST QUOTE: {first_quote}')
def mk_init_msgs() -> dict[str, dict]: def mk_init_msgs() -> dict[str, dict]:
''' '''
@ -879,7 +892,7 @@ async def stream_quotes(
# TODO: we should instead spawn a task that waits on a feed to start # TODO: we should instead spawn a task that waits on a feed to start
# and let it wait indefinitely..instead of this hard coded stuff. # and let it wait indefinitely..instead of this hard coded stuff.
with trio.move_on_after(1): with trio.move_on_after(1):
contract, first_ticker, details = await proxy.get_quote(symbol=sym) first_ticker = await proxy.get_quote(contract=con)
# it might be outside regular trading hours so see if we can at # it might be outside regular trading hours so see if we can at
# least grab history. # least grab history.