Epoch time-align (indexing)

piker/_daemon.py

@@ -22,9 +22,9 @@
 import os
 from typing import (
     Optional,
-    Union,
     Callable,
     Any,
+    ClassVar,
 )
 from contextlib import (
     asynccontextmanager as acm,
@@ -257,7 +257,7 @@ async def open_piker_runtime(
     # and spawn the service tree distributed per that.
     start_method: str = 'trio',
 
-    tractor_kwargs: dict = {},
+    **tractor_kwargs,
 
 ) -> tuple[
     tractor.Actor,
@@ -393,7 +393,7 @@ async def maybe_open_pikerd(
 
     **kwargs,
 
-) -> Union[tractor._portal.Portal, Services]:
+) -> tractor._portal.Portal | ClassVar[Services]:
     '''
     If no ``pikerd`` daemon-root-actor can be found start it and
     yield up (we should probably figure out returning a portal to self
@@ -445,7 +445,7 @@ async def maybe_open_pikerd(
         # tractor-piker runtime stack in **this** process
         # we return no portal to self.
         assert service_manager
-        yield None
+        yield service_manager
 
 
 # `brokerd` enabled modules

piker/_profile.py

@@ -152,9 +152,14 @@ def __new__(
             # don't do anything
             return cls._disabledProfiler
 
-        # create an actual profiling object
         cls._depth += 1
         obj = super(Profiler, cls).__new__(cls)
+        obj._msgs = []
+
+        # create an actual profiling object
+        if cls._depth < 1:
+            cls._msgs = []
+
         obj._name = msg or func_qualname
         obj._delayed = delayed
         obj._markCount = 0
@@ -174,8 +179,12 @@ def __call__(self, msg=None):
 
         self._markCount += 1
         newTime = perf_counter()
+        tot_ms = (newTime - self._firstTime) * 1000
         ms = (newTime - self._lastTime) * 1000
-        self._newMsg("  %s: %0.4f ms", msg, ms)
+        self._newMsg(
+            f"  {msg}: {ms:0.4f}, tot:{tot_ms:0.4f}"
+        )
+
         self._lastTime = newTime
 
     def mark(self, msg=None):

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/kraken/broker.py

@@ -509,7 +509,9 @@ def has_pp(
                     for bsuid in table.pps:
                         try:
                             dst_name_start = bsuid.rindex(src_fiat)
-                        except IndexError:
+                        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

piker/clearing/__init__.py

@@ -18,3 +18,9 @@
 Market machinery for order executions, book, management.
 
 """
+from ._client import open_ems
+
+
+__all__ = [
+    'open_ems',
+]

piker/clearing/_client.py

@@ -18,20 +18,27 @@
 Orders and execution client API.
 
 """
+from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from pprint import pformat
+from typing import TYPE_CHECKING
 
 import trio
 import tractor
 from tractor.trionics import broadcast_receiver
 
 from ..log import get_logger
 from ..data.types import Struct
-from ._ems import _emsd_main
 from .._daemon import maybe_open_emsd
 from ._messages import Order, Cancel
 from ..brokers import get_brokermod
 
+if TYPE_CHECKING:
+    from ._messages import (
+        BrokerdPosition,
+        Status,
+    )
+
 
 log = get_logger(__name__)
 
@@ -167,12 +174,19 @@ async def relay_order_cmds_from_sync_code(
 @acm
 async def open_ems(
     fqsn: str,
+    mode: str = 'live',
 
-) -> (
+) -> tuple[
     OrderBook,
     tractor.MsgStream,
-    dict,
-):
+    dict[
+        # brokername, acctid
+        tuple[str, str],
+        list[BrokerdPosition],
+    ],
+    list[str],
+    dict[str, Status],
+]:
     '''
     Spawn an EMS daemon and begin sending orders and receiving
     alerts.
@@ -213,14 +227,16 @@ async def open_ems(
     from ..data._source import unpack_fqsn
     broker, symbol, suffix = unpack_fqsn(fqsn)
 
-    mode: str = 'live'
-
     async with maybe_open_emsd(broker) as portal:
 
         mod = get_brokermod(broker)
-        if not getattr(mod, 'trades_dialogue', None):
+        if (
+            not getattr(mod, 'trades_dialogue', None)
+            or mode == 'paper'
+        ):
             mode = 'paper'
 
+        from ._ems import _emsd_main
         async with (
             # connect to emsd
             portal.open_context(

piker/clearing/_ems.py

@@ -46,6 +46,7 @@
 from ..data._source import (
     unpack_fqsn,
     mk_fqsn,
+    float_digits,
 )
 from ..data.feed import (
     Feed,
@@ -418,7 +419,7 @@ async def maybe_open_brokerd_dialog(
 
             # load the paper trading engine
             exec_mode = 'paper'
-            log.warning(f'Entering paper trading mode for {broker}')
+            log.info(f'{broker}: Entering `paper` trading mode')
 
             # load the paper trading engine as a subactor of this emsd
             # actor to simulate the real IPC load it'll have when also
@@ -1250,6 +1251,7 @@ async def process_client_order_cmds(
 
                 spread_slap: float = 5
                 min_tick = flume.symbol.tick_size
+                min_tick_digits = float_digits(min_tick)
 
                 if action == 'buy':
                     tickfilter = ('ask', 'last', 'trade')
@@ -1258,12 +1260,18 @@ async def process_client_order_cmds(
                     # TODO: we probably need to scale this based
                     # on some near term historical spread
                     # measure?
-                    abs_diff_away = spread_slap * min_tick
+                    abs_diff_away = round(
+                        spread_slap * min_tick,
+                        ndigits=min_tick_digits,
+                    )
 
                 elif action == 'sell':
                     tickfilter = ('bid', 'last', 'trade')
                     percent_away = -0.005
-                    abs_diff_away = -spread_slap * min_tick
+                    abs_diff_away = round(
+                        -spread_slap * min_tick,
+                        ndigits=min_tick_digits,
+                    )
 
                 else:  # alert
                     tickfilter = ('trade', 'utrade', 'last')
@@ -1367,7 +1375,15 @@ async def _emsd_main(
     exec_mode: str,  # ('paper', 'live')
     loglevel: str = 'info',
 
-) -> None:
+) -> tuple[
+    dict[
+        # brokername, acctid
+        tuple[str, str],
+        list[BrokerdPosition],
+    ],
+    list[str],
+    dict[str, Status],
+]:
     '''
     EMS (sub)actor entrypoint providing the execution management
     (micro)service which conducts broker order clearing control on