Merge pull request #442 from pikers/misc_brokerd_backend_repairs

Misc brokerd backend repairs

Author: goodboy

.github/workflows/ci.yml

@@ -3,9 +3,8 @@ name: CI
 
 on:
   # Triggers the workflow on push or pull request events but only for the master branch
-  push:
-    branches: [ master ]
   pull_request:
+  push:
     branches: [ master ]
 
   # Allows you to run this workflow manually from the Actions tab

dockering/ib/docker-compose.yml

@@ -8,7 +8,7 @@ services:
     # https://github.com/waytrade/ib-gateway-docker#supported-tags
     # image: waytrade/ib-gateway:981.3j
     image: waytrade/ib-gateway:1012.2i
-    restart: always  # restart whenev there's a crash or user clicsk
+    restart: 'no'  # restart on boot whenev there's a crash or user clicsk
     network_mode: 'host'
 
     volumes:
@@ -64,7 +64,7 @@ services:
 
   # ib_gw_live:
   #   image: waytrade/ib-gateway:1012.2i
-  #   restart: always
+  #   restart: no
   #   network_mode: 'host'
 
   #   volumes:

piker/brokers/binance.py

@@ -519,14 +519,15 @@ async def subscribe(ws: wsproto.WSConnection):
                 subs.append("{sym}@bookTicker")
 
             # unsub from all pairs on teardown
-            await ws.send_msg({
-                "method": "UNSUBSCRIBE",
-                "params": subs,
-                "id": uid,
-            })
-
-            # XXX: do we need to ack the unsub?
-            # await ws.recv_msg()
+            if ws.connected():
+                await ws.send_msg({
+                    "method": "UNSUBSCRIBE",
+                    "params": subs,
+                    "id": uid,
+                })
+
+                # XXX: do we need to ack the unsub?
+                # await ws.recv_msg()
 
         async with open_autorecon_ws(
             'wss://stream.binance.com/ws',

piker/brokers/deribit/feed.py

@@ -94,21 +94,6 @@ async def get_ohlc(
         yield get_ohlc, {'erlangs': 3, 'rate': 3}
 
 
-async def backfill_bars(
-    symbol: str,
-    shm: ShmArray,  # type: ignore # noqa
-    task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED,
-) -> None:
-    """Fill historical bars into shared mem / storage afap.
-    """
-    instrument = symbol
-    with trio.CancelScope() as cs:
-        async with open_cached_client('deribit') as client:
-            bars = await client.bars(instrument)
-            shm.push(bars)
-            task_status.started(cs)
-
-
 async def stream_quotes(
 
     send_chan: trio.abc.SendChannel,

piker/brokers/ib/api.py

@@ -162,6 +162,7 @@ def __init__(self):
     'CMECRYPTO',
     'COMEX',
     'CMDTY',  # special name case..
+    'CBOT',  # (treasury) yield futures
 )
 
 _adhoc_futes_set = {
@@ -197,6 +198,21 @@ def __init__(self):
     'xagusd.cmdty',  # silver spot
     'ni.comex',  # silver futes
     'qi.comex',  # mini-silver futes
+
+    # treasury yields
+    # etfs by duration:
+    # SHY -> IEI -> IEF -> TLT
+    'zt.cbot',  # 2y
+    'z3n.cbot',  # 3y
+    'zf.cbot',  # 5y
+    'zn.cbot',  # 10y
+    'zb.cbot',  # 30y
+
+    # (micros of above)
+    '2yy.cbot',
+    '5yy.cbot',
+    '10y.cbot',
+    '30y.cbot',
 }
 
 

piker/brokers/ib/broker.py

@@ -611,7 +611,7 @@ async def trades_dialogue(
                         pp = table.pps[bsuid]
                         if msg.size != pp.size:
                             log.error(
-                                'Position mismatch {pp.symbol.front_fqsn()}:\n'
+                                f'Position mismatch {pp.symbol.front_fqsn()}:\n'
                                 f'ib: {msg.size}\n'
                                 f'piker: {pp.size}\n'
                             )

piker/brokers/ib/feed.py

@@ -135,7 +135,10 @@ async def open_history_client(
             # fx cons seem to not provide this endpoint?
             'idealpro' not in fqsn
         ):
-            head_dt = await proxy.get_head_time(fqsn=fqsn)
+            try:
+                head_dt = await proxy.get_head_time(fqsn=fqsn)
+            except RequestError:
+                head_dt = None
 
         async def get_hist(
             timeframe: float,

piker/brokers/kraken/api.py

@@ -510,10 +510,6 @@ def normalize_symbol(
 
         '''
         ticker = cls._ntable[ticker]
-        symlen = len(ticker)
-        if symlen != 6:
-            raise ValueError(f'Unhandled symbol: {ticker}')
-
         return ticker.lower()
 
 

piker/brokers/kraken/broker.py

@@ -413,19 +413,26 @@ async def trades_dialogue(
 
 ) -> AsyncIterator[dict[str, Any]]:
 
-    # XXX: required to propagate ``tractor`` loglevel to piker logging
+    # XXX: required to propagate ``tractor`` loglevel to ``piker`` logging
     get_console_log(loglevel or tractor.current_actor().loglevel)
 
     async with get_client() as client:
 
+        if not client._api_key:
+            raise RuntimeError(
+                'Missing Kraken API key in `brokers.toml`!?!?')
+
         # TODO: make ems flip to paper mode via
         # some returned signal if the user only wants to use
         # the data feed or we return this?
-        # await ctx.started(({}, ['paper']))
+        # else:
+        #     await ctx.started(({}, ['paper']))
 
-        if not client._api_key:
-            raise RuntimeError(
-                'Missing Kraken API key in `brokers.toml`!?!?')
+        # NOTE: currently we expect the user to define a "source fiat"
+        # (much like the web UI let's you set an "account currency")
+        # such that all positions (nested or flat) will be translated to
+        # this source currency's terms.
+        src_fiat = client.conf['src_fiat']
 
         # auth required block
         acctid = client._name
@@ -444,10 +451,9 @@ async def trades_dialogue(
         # NOTE: testing code for making sure the rt incremental update
         # of positions, via newly generated msgs works. In order to test
         # this,
-        # - delete the *ABSOLUTE LAST* entry from accont's corresponding
+        # - delete the *ABSOLUTE LAST* entry from account's corresponding
         #   trade ledgers file (NOTE this MUST be the last record
-        #   delivered from the
-        #   api ledger),
+        #   delivered from the api ledger),
         # - open you ``pps.toml`` and find that same tid and delete it
         #   from the pp's clears table,
         # - set this flag to `True`
@@ -486,40 +492,84 @@ async def trades_dialogue(
             # and do diff with ledger to determine
             # what amount of trades-transactions need
             # to be reloaded.
-            sizes = await client.get_balances()
-            for dst, size in sizes.items():
+            balances = await client.get_balances()
+            for dst, size in balances.items():
                 # we don't care about tracking positions
                 # in the user's source fiat currency.
-                if dst == client.conf['src_fiat']:
+                if dst == src_fiat:
                     continue
 
-                def has_pp(dst: str) -> Position | bool:
-                    pps_dst_assets = {bsuid[:3]: bsuid for bsuid in table.pps}
-                    pair = pps_dst_assets.get(dst)
-                    pp = table.pps.get(pair)
+                def has_pp(
+                    dst: str,
+                    size: float,
 
-                    if (
-                        not pair or not pp
-                        or not math.isclose(pp.size, size)
-                    ):
-                        return False
+                ) -> Position | bool:
 
-                    return pp
+                    src2dst: dict[str, str] = {}
+                    for bsuid in table.pps:
+                        try:
+                            dst_name_start = bsuid.rindex(src_fiat)
+                        except (
+                            ValueError,   # substr not found
+                        ):
+                            # TODO: handle nested positions..(i.e.
+                            # positions where the src fiat was used to
+                            # buy some other dst which was furhter used
+                            # to buy another dst..)
+                            log.warning(
+                                f'No src fiat {src_fiat} found in {bsuid}?'
+                            )
+                            continue
 
-                pos = has_pp(dst)
+                        _dst = bsuid[:dst_name_start]
+                        if _dst != dst:
+                            continue
+
+                        src2dst[src_fiat] = dst
+
+                    for src, dst in src2dst.items():
+                        pair = f'{dst}{src_fiat}'
+                        pp = table.pps.get(pair)
+                        if (
+                            pp
+                            and math.isclose(pp.size, size)
+                        ):
+                            return pp
+
+                        elif (
+                            size == 0
+                            and pp.size
+                        ):
+                            log.warning(
+                                f'`kraken` account says you have  a ZERO '
+                                f'balance for {bsuid}:{pair}\n'
+                                f'but piker seems to think `{pp.size}`\n'
+                                'This is likely a discrepancy in piker '
+                                'accounting if the above number is'
+                                "large,' though it's likely to due lack"
+                                "f tracking xfers fees.."
+                            )
+                            return pp
+
+                    return False
+
+                pos = has_pp(dst, size)
                 if not pos:
 
                     # we have a balance for which there is no pp
                     # entry? so we have to likely update from the
                     # ledger.
                     updated = table.update_from_trans(ledger_trans)
                     log.info(f'Updated pps from ledger:\n{pformat(updated)}')
-                    pos = has_pp(dst)
+                    pos = has_pp(dst, size)
 
-                    if not pos and not simulate_pp_update:
+                    if (
+                        not pos
+                        and not simulate_pp_update
+                    ):
                         # try reloading from API
                         table.update_from_trans(api_trans)
-                        pos = has_pp(dst)
+                        pos = has_pp(dst, size)
                         if not pos:
 
                             # get transfers to make sense of abs balances.
@@ -557,7 +607,7 @@ def has_pp(dst: str) -> Position | bool:
                                         f'{pformat(updated)}'
                                     )
 
-                        if not has_pp(dst):
+                        if has_pp(dst, size):
                             raise ValueError(
                                 'Could not reproduce balance:\n'
                                 f'dst: {dst}, {size}\n'

piker/brokers/kraken/feed.py

@@ -303,24 +303,6 @@ async def get_ohlc(
         yield get_ohlc, {'erlangs': 1, 'rate': 1}
 
 
-async def backfill_bars(
-
-    sym: str,
-    shm: ShmArray,  # type: ignore # noqa
-    count: int = 10,  # NOTE: any more and we'll overrun the underlying buffer
-    task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED,
-
-) -> None:
-    '''
-    Fill historical bars into shared mem / storage afap.
-    '''
-    with trio.CancelScope() as cs:
-        async with open_cached_client('kraken') as client:
-            bars = await client.bars(symbol=sym)
-            shm.push(bars)
-            task_status.started(cs)
-
-
 async def stream_quotes(
 
     send_chan: trio.abc.SendChannel,
@@ -419,14 +401,15 @@ async def subscribe(ws: NoBsWs):
             yield
 
             # unsub from all pairs on teardown
-            await ws.send_msg({
-                'pair': list(ws_pairs.values()),
-                'event': 'unsubscribe',
-                'subscription': ['ohlc', 'spread'],
-            })
-
-            # XXX: do we need to ack the unsub?
-            # await ws.recv_msg()
+            if ws.connected():
+                await ws.send_msg({
+                    'pair': list(ws_pairs.values()),
+                    'event': 'unsubscribe',
+                    'subscription': ['ohlc', 'spread'],
+                })
+
+                # XXX: do we need to ack the unsub?
+                # await ws.recv_msg()
 
         # see the tips on reconnection logic:
         # https://support.kraken.com/hc/en-us/articles/360044504011-WebSocket-API-unexpected-disconnections-from-market-data-feeds