`binance`: pre-process `Pair` filters at init

Allows us to keep the struct frozen as well avoid complexity in the pure
data type. Also changes `.price/size_tick` to plain ol' properties.
pre_overruns_ctxcancelled
Tyler Goodlet 2023-03-23 17:34:55 -04:00
parent a50452dbfd
commit 62a40c57a0
1 changed files with 28 additions and 32 deletions

View File

@ -95,7 +95,7 @@ _show_wap_in_history = False
# TODO: make this frozen again by pre-processing the # TODO: make this frozen again by pre-processing the
# filters list to a dict at init time? # filters list to a dict at init time?
class Pair(Struct): # , frozen=True): class Pair(Struct, frozen=True):
symbol: str symbol: str
status: str status: str
@ -121,40 +121,21 @@ class Pair(Struct): # , frozen=True):
defaultSelfTradePreventionMode: str defaultSelfTradePreventionMode: str
allowedSelfTradePreventionModes: list[str] allowedSelfTradePreventionModes: list[str]
filters: list[ filters: dict[
dict[
str, str,
Union[str, int, float] Union[str, int, float]
] ]
]
permissions: list[str] permissions: list[str]
_filtersbykey: dict | None = None @property
def get_filter(self) -> dict[str, dict]:
filters = self._filtersbykey
if self._filtersbykey:
return filters
filters = self._filtersbykey = {}
for entry in self.filters:
ftype = entry['filterType']
filters[ftype] = entry
return filters
def size_tick(self) -> Decimal: def size_tick(self) -> Decimal:
# XXX: lul, after manually inspecting the response format we # XXX: lul, after manually inspecting the response format we
# just directly pick out the info we need # just directly pick out the info we need
return Decimal( return Decimal(self.filters['PRICE_FILTER']['tickSize'].rstrip('0'))
self.get_filter()['PRICE_FILTER']['tickSize'].rstrip('0')
)
@property
def price_tick(self) -> Decimal: def price_tick(self) -> Decimal:
return Decimal( return Decimal(self.filters['LOT_SIZE']['stepSize'].rstrip('0'))
self.get_filter()['LOT_SIZE']['stepSize'].rstrip('0')
)
class OHLC(Struct): class OHLC(Struct):
@ -238,9 +219,24 @@ class Client:
if not entries: if not entries:
raise SymbolNotFound(f'{sym} not found:\n{resp}') raise SymbolNotFound(f'{sym} not found:\n{resp}')
pairs = { # pre-process .filters field into a table
item['symbol']: Pair(**item) for item in entries pairs = {}
} for item in entries:
symbol = item['symbol']
filters = {}
filters_ls: list = item.pop('filters')
for entry in filters_ls:
ftype = entry['filterType']
filters[ftype] = entry
pairs[symbol] = Pair(
filters=filters,
**item,
)
# pairs = {
# item['symbol']: Pair(**item) for item in entries
# }
self._pairs.update(pairs) self._pairs.update(pairs)
if sym is not None: if sym is not None:
@ -503,8 +499,8 @@ async def stream_quotes(
for sym in symbols: for sym in symbols:
pair: Pair = pairs[sym.upper()] pair: Pair = pairs[sym.upper()]
price_tick = pair.price_tick() price_tick = pair.price_tick
size_tick = pair.size_tick() size_tick = pair.size_tick
mkt_infos[sym] = MktPair( mkt_infos[sym] = MktPair(
dst=Asset( dst=Asset(