Tractor typed msg hackin

piker/_daemon.py

@@ -22,13 +22,13 @@
 from contextlib import asynccontextmanager as acm
 from collections import defaultdict
 
-from pydantic import BaseModel
+import tractor
 import trio
 from trio_typing import TaskStatus
-import tractor
 
 from .log import get_logger, get_console_log
 from .brokers import get_brokermod
+from .data.types import Struct
 
 
 log = get_logger(__name__)
@@ -47,16 +47,13 @@
 ]
 
 
-class Services(BaseModel):
+class Services(Struct):
 
     actor_n: tractor._supervise.ActorNursery
     service_n: trio.Nursery
     debug_mode: bool  # tractor sub-actor debug mode flag
     service_tasks: dict[str, tuple[trio.CancelScope, tractor.Portal]] = {}
 
-    class Config:
-        arbitrary_types_allowed = True
-
     async def start_service_task(
         self,
         name: str,
@@ -207,23 +204,26 @@ async def open_piker_runtime(
     assert _services is None
 
     # XXX: this may open a root actor as well
-    async with (
-        tractor.open_root_actor(
+    with tractor.msg.configure_native_msgs(
+        [Struct],
+    ):
+        async with (
+            tractor.open_root_actor(
 
-            # passed through to ``open_root_actor``
-            arbiter_addr=_registry_addr,
-            name=name,
-            loglevel=loglevel,
-            debug_mode=debug_mode,
-            start_method=start_method,
+                # passed through to ``open_root_actor``
+                arbiter_addr=_registry_addr,
+                name=name,
+                loglevel=loglevel,
+                debug_mode=debug_mode,
+                start_method=start_method,
 
-            # TODO: eventually we should be able to avoid
-            # having the root have more then permissions to
-            # spawn other specialized daemons I think?
-            enable_modules=_root_modules,
-        ) as _,
-    ):
-        yield tractor.current_actor()
+                # TODO: eventually we should be able to avoid
+                # having the root have more then permissions to
+                # spawn other specialized daemons I think?
+                enable_modules=_root_modules,
+            ) as _,
+        ):
+            yield tractor.current_actor()
 
 
 @acm
@@ -263,27 +263,31 @@ async def maybe_open_pikerd(
     if loglevel:
         get_console_log(loglevel)
 
-    # subtle, we must have the runtime up here or portal lookup will fail
-    async with maybe_open_runtime(loglevel, **kwargs):
+    # XXX: this may open a root actor as well
+    with tractor.msg.configure_native_msgs(
+        [Struct],
+    ):
+        # subtle, we must have the runtime up here or portal lookup will fail
+        async with maybe_open_runtime(loglevel, **kwargs):
 
-        async with tractor.find_actor(_root_dname) as portal:
-            # assert portal is not None
-            if portal is not None:
-                yield portal
-                return
+            async with tractor.find_actor(_root_dname) as portal:
+                # assert portal is not None
+                if portal is not None:
+                    yield portal
+                    return
 
-    # presume pikerd role since no daemon could be found at
-    # configured address
-    async with open_pikerd(
+        # presume pikerd role since no daemon could be found at
+        # configured address
+        async with open_pikerd(
 
-        loglevel=loglevel,
-        debug_mode=kwargs.get('debug_mode', False),
+            loglevel=loglevel,
+            debug_mode=kwargs.get('debug_mode', False),
 
-    ) as _:
-        # in the case where we're starting up the
-        # tractor-piker runtime stack in **this** process
-        # we return no portal to self.
-        yield None
+        ) as _:
+            # in the case where we're starting up the
+            # tractor-piker runtime stack in **this** process
+            # we return no portal to self.
+            yield None
 
 
 # brokerd enabled modules
@@ -445,7 +449,7 @@ async def spawn_brokerd(
     )
 
     # non-blocking setup of brokerd service nursery
-    from .data import _setup_persistent_brokerd
+    from .data.feed import _setup_persistent_brokerd
 
     await _services.start_service_task(
         dname,

piker/brokers/binance.py

@@ -34,13 +34,13 @@
 import numpy as np
 import tractor
 from pydantic.dataclasses import dataclass
-from pydantic import BaseModel
 import wsproto
 
 from .._cacheables import open_cached_client
 from ._util import resproc, SymbolNotFound
 from ..log import get_logger, get_console_log
 from ..data import ShmArray
+from ..data.types import Struct
 from ..data._web_bs import open_autorecon_ws, NoBsWs
 
 log = get_logger(__name__)
@@ -79,12 +79,14 @@
 
 
 # https://binance-docs.github.io/apidocs/spot/en/#exchange-information
-class Pair(BaseModel):
+class Pair(Struct, frozen=True):
     symbol: str
     status: str
 
     baseAsset: str
     baseAssetPrecision: int
+    cancelReplaceAllowed: bool
+    allowTrailingStop: bool
     quoteAsset: str
     quotePrecision: int
     quoteAssetPrecision: int
@@ -287,7 +289,7 @@ async def get_client() -> Client:
 
 
 # validation type
-class AggTrade(BaseModel):
+class AggTrade(Struct):
     e: str  # Event type
     E: int  # Event time
     s: str  # Symbol
@@ -341,7 +343,9 @@ async def stream_messages(ws: NoBsWs) -> AsyncGenerator[NoBsWs, dict]:
 
         elif msg.get('e') == 'aggTrade':
 
-            # validate
+            # NOTE: this is purely for a definition, ``msgspec.Struct``
+            # does not runtime-validate until you decode/encode.
+            # see: https://jcristharif.com/msgspec/structs.html#type-validation
             msg = AggTrade(**msg)
 
             # TODO: type out and require this quote format
@@ -352,8 +356,8 @@ async def stream_messages(ws: NoBsWs) -> AsyncGenerator[NoBsWs, dict]:
                 'brokerd_ts': time.time(),
                 'ticks': [{
                     'type': 'trade',
-                    'price': msg.p,
-                    'size': msg.q,
+                    'price': float(msg.p),
+                    'size': float(msg.q),
                     'broker_ts': msg.T,
                 }],
             }
@@ -448,7 +452,7 @@ async def stream_quotes(
             d = cache[sym.upper()]
             syminfo = Pair(**d)  # validation
 
-            si = sym_infos[sym] = syminfo.dict()
+            si = sym_infos[sym] = syminfo.to_dict()
 
             # XXX: after manually inspecting the response format we
             # just directly pick out the info we need

piker/brokers/ib/broker.py

@@ -148,7 +148,7 @@ async def handle_order_requests(
                 oid=request_msg['oid'],
                 symbol=request_msg['symbol'],
                 reason=f'No account found: `{account}` ?',
-            ).dict())
+            ))
             continue
 
         client = _accounts2clients.get(account)
@@ -161,7 +161,7 @@ async def handle_order_requests(
                 oid=request_msg['oid'],
                 symbol=request_msg['symbol'],
                 reason=f'No api client loaded for account: `{account}` ?',
-            ).dict())
+            ))
             continue
 
         if action in {'buy', 'sell'}:
@@ -188,7 +188,7 @@ async def handle_order_requests(
                     oid=request_msg['oid'],
                     symbol=request_msg['symbol'],
                     reason='Order already active?',
-                ).dict())
+                ))
 
             # deliver ack that order has been submitted to broker routing
             await ems_order_stream.send(
@@ -197,9 +197,8 @@ async def handle_order_requests(
                     oid=order.oid,
                     # broker specific request id
                     reqid=reqid,
-                    time_ns=time.time_ns(),
                     account=account,
-                ).dict()
+                )
             )
 
         elif action == 'cancel':
@@ -559,7 +558,7 @@ async def open_stream(
                     cids2pps,
                     validate=True,
                 )
-                all_positions.extend(msg.dict() for msg in msgs)
+                all_positions.extend(msg for msg in msgs)
 
         if not all_positions and cids2pps:
             raise RuntimeError(
@@ -665,7 +664,7 @@ async def emit_pp_update(
             msg = msgs[0]
             break
 
-    await ems_stream.send(msg.dict())
+    await ems_stream.send(msg)
 
 
 async def deliver_trade_events(
@@ -743,7 +742,7 @@ async def deliver_trade_events(
 
                     broker_details={'name': 'ib'},
                 )
-                await ems_stream.send(msg.dict())
+                await ems_stream.send(msg)
 
             case 'fill':
 
@@ -803,7 +802,7 @@ async def deliver_trade_events(
                     broker_time=trade_entry['broker_time'],
 
                 )
-                await ems_stream.send(msg.dict())
+                await ems_stream.send(msg)
 
                 # 2 cases:
                 # - fill comes first or
@@ -879,7 +878,7 @@ async def deliver_trade_events(
                 cid, msg = pack_position(item)
                 # acctid = msg.account = accounts_def.inverse[msg.account]
                 # cuck ib and it's shitty fifo sys for pps!
-                # await ems_stream.send(msg.dict())
+                # await ems_stream.send(msg)
 
             case 'event':
 
@@ -891,7 +890,7 @@ async def deliver_trade_events(
                 # level...
                 # reqid = item.get('reqid', 0)
                 # if getattr(msg, 'reqid', 0) < -1:
-                # log.info(f"TWS triggered trade\n{pformat(msg.dict())}")
+                # log.info(f"TWS triggered trade\n{pformat(msg)}")
 
                 # msg.reqid = 'tws-' + str(-1 * reqid)
 

piker/brokers/kraken/README.rst

@@ -0,0 +1,64 @@
+``kraken`` backend
+------------------
+though they don't have the most liquidity of all the cexes they sure are
+accommodating to those of us who appreciate a little ``xmr``.
+
+status
+******
+current support is *production grade* and both real-time data and order
+management should be correct and fast. this backend is used by core devs
+for live trading.
+
+
+config
+******
+In order to get order mode support your ``brokers.toml``
+needs to have something like the following:
+
+.. code:: toml
+
+   [kraken]
+   accounts.spot = 'spot'
+   key_descr = "spot"
+   api_key = "69696969696969696696969696969696969696969696969696969696"
+   secret = "BOOBSBOOBSBOOBSBOOBSBOOBSSMBZ69696969696969669969696969696"
+
+
+If everything works correctly you should see any current positions
+loaded in the pps pane on chart load and you should also be able to
+check your trade records in the file::
+
+    <pikerk_conf_dir>/ledgers/trades_kraken_spot.toml
+
+
+An example ledger file will have entries written verbatim from the
+trade events schema:
+
+.. code:: toml
+
+    [TFJBKK-SMBZS-VJ4UWS]
+    ordertxid = "SMBZSA-7CNQU-3HWLNJ"
+    postxid = "SMBZSE-M7IF5-CFI7LT"
+    pair = "XXMRZEUR"
+    time = 1655691993.4133966
+    type = "buy"
+    ordertype = "limit"
+    price = "103.97000000"
+    cost = "499.99999977"
+    fee = "0.80000000"
+    vol = "4.80907954"
+    margin = "0.00000000"
+    misc = ""
+
+
+your ``pps.toml`` file will have position entries like,
+
+.. code:: toml
+
+   [kraken.spot."xmreur.kraken"]
+   size = 4.80907954
+   be_price = 103.97000000
+   bsuid = "XXMRZEUR"
+   clears = [
+    { tid = "TFJBKK-SMBZS-VJ4UWS", cost = 0.8, price = 103.97, size = 4.80907954, dt = "2022-05-20T02:26:33.413397+00:00" },
+   ]

piker/brokers/kraken/api.py

@@ -282,6 +282,7 @@ async def submit_limit(
                 "volume": str(size),
             }
             return await self.endpoint('AddOrder', data)
+
         else:
             # Edit order data for kraken api
             data["txid"] = reqid