minor changes, prepare for rebase of overlays branch

piker/clearing/_messages.py

@@ -100,7 +100,7 @@ class Order(Struct):
     price: float
     size: float  # -ve is "sell", +ve is "buy"
 
-    brokers: Optional[list[str]] = []
+    brokers: list[str] = []
 
 
 class Cancel(Struct):

piker/clearing/_paper_engine.py

@@ -280,7 +280,7 @@ class PaperBoi(Struct):
         # Store txn in state for PP update
         self._txn_dict[oid] = t
         self._trade_ledger.update(ledger_entry)
-        
+
         # Write to ledger toml  
         with open_trade_ledger(self.broker, 'paper') as ledger:
             ledger.update(self._trade_ledger)   
@@ -330,6 +330,7 @@ async def simulate_fills(
     # https://github.com/quantopian/zipline/blob/master/zipline/finance/ledger.py
 
     # this stream may eventually contain multiple symbols
+
     async for quotes in quote_stream:
         for sym, quote in quotes.items():
             for tick in iterticks(
@@ -423,9 +424,9 @@ async def simulate_fills(
                     # simulated live orders prematurely.
                     case _:
                         continue
-
+                
-                # iterate all potentially clearable book prices
+                # iterate alcl potentially clearable book prices
-                # in FIFO order per side.
+                # in FIFO order per side.c
                 for order_info, pred in iter_entries:
                     (our_price, size, reqid, action) = order_info
 
@@ -438,6 +439,8 @@ async def simulate_fills(
                             'sell': sells
                         }[action].inverse.pop(order_info)
 
+                        log.warning(f'order_info: {order_info}')
+                           
                         # clearing price would have filled entirely
                         await client.fake_fill(
                             fqsn=sym,
@@ -553,10 +556,11 @@ async def trades_dialogue(
         ) as feed,
 
     ):
+
         with open_pps(broker, 'paper-id') as table: 
             # save pps in local state
             _positions.update(table.pps)
-
+        
         pp_msgs: list[BrokerdPosition] = []
         pos: Position
         token: str  # f'{symbol}.{self.broker}'
@@ -589,7 +593,6 @@ async def trades_dialogue(
 
                 _reqids=_reqids,
 
-                # TODO: load paper positions from ``positions.toml``
                 _positions=_positions,
 
                 # TODO: load postions from ledger file
@@ -628,7 +631,6 @@ async def open_paperboi(
         # (we likely don't need more then one proc for basic
         # simulated order clearing)
         if portal is None:
-            log.info('Starting new paper-engine actor')
             portal = await tn.start_actor(
                 service_name,
                 enable_modules=[__name__]

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+trio_mode=True
+log_cli=1

tests/test_paper.py

@@ -0,0 +1,143 @@
+from datetime import datetime
+import time
+import trio
+import pytest
+import tractor
+import math
+from piker.log import get_logger
+from piker.clearing._messages import ( 
+    Order                                    
+)
+from typing import AsyncContextManager
+
+from piker.pp import (
+    Position,
+    Transaction,
+    open_trade_ledger,
+    open_pps
+)
+
+from piker._daemon import (
+    find_service,
+    check_for_service,
+    Services,
+)
+from piker.data import (
+    open_feed,
+)
+from piker.clearing import (
+    open_ems,
+)
+from piker.clearing._messages import (
+    BrokerdPosition,
+    Status,
+)
+from piker.clearing._client import (
+    OrderBook,
+)
+
+log = get_logger(__name__)
+
+
+def test_paper_trade(
+    open_test_pikerd: AsyncContextManager
+):
+
+    async def main():
+        # type declares
+        book: OrderBook
+        trades_stream: tractor.MsgStream
+        pps: dict[str, list[BrokerdPosition]]
+        accounts: list[str]
+        dialogs: dict[str, Status]
+
+        async with (
+            open_test_pikerd() as (_, _, _, services),
+
+            open_ems(
+                'xbtusdt.kraken',
+                mode='paper',
+            ) as (
+                book,
+                trades_stream,
+                pps,
+                accounts,
+                dialogs,
+            ),
+        ):
+            
+            test_exec_mode='live'
+            test_action = 'buy'
+            test_oid = '560beac8-b1b1-4dee-bd1e-6604a704c9ea'
+            test_account = 'paper'
+            test_size = 1 
+            test_price = 30000
+            test_broker = 'kraken'
+            test_brokers = [test_broker]
+            test_symbol = 'xbtusdt' 
+            test_fqsn = f'{test_symbol}.{test_broker}'
+            test_pp_account = 'piker-paper'
+
+            order = Order(
+                exec_mode=test_exec_mode,
+                action=test_action,
+                oid=test_oid,
+                account=test_account,
+                size=test_size,
+                symbol=test_fqsn,
+                price=test_price,
+                brokers=test_brokers
+            )
+
+            book.send(order)
+            
+            await trio.sleep(1)
+
+            cleared_ledger_entry = {}
+            # check if trades have been updated in in ledge and pp 
+            with open_trade_ledger(test_broker, test_account) as ledger:
+                log.warning(f'ledger: {ledger}')
+                cleared_ledger_entry = ledger[test_oid]
+                assert list(ledger.keys())[0] == test_oid
+                assert cleared_ledger_entry['size'] == test_size
+                assert cleared_ledger_entry['fqsn'] == test_fqsn
+                 
+            with open_pps(test_broker, test_pp_account) as table: 
+                # save pps in local state
+                assert table.brokername == test_broker 
+                assert table.acctid == test_pp_account 
+#                assert cleared_ledger_entry['price'] == table.conf.clears[0].price
+                pp_price = table.conf[test_broker][test_pp_account][test_fqsn]["ppu"] 
+                assert math.isclose(pp_price, cleared_ledger_entry['size'], rel_tol=1)
+
+            raise KeyboardInterrupt
+        
+
+    with pytest.raises(
+        trio.MultiError
+    ) as exc_info:
+        trio.run(main)
+
+
+def test_trades_persist(
+    open_test_pikerd: AsyncContextManager
+):
+
+    async def main():
+        async with (
+            open_test_pikerd() as (_, _, _, services),
+
+            open_ems(
+                'xbtusdt.kraken',
+                mode='paper',
+            ) as (
+                book,
+                trades_stream,
+                pps,
+                accounts,
+                dialogs,
+            ),
+        ):
+            print(f'pps: {pps}')
+
+    trio.run(main)          

tests/test_services.py

@@ -9,6 +9,13 @@ import pytest
 import trio
 import tractor
 
+from datetime import datetime
+import time
+from piker.log import get_logger
+from piker.clearing._messages import ( 
+    BrokerdFill
+)
+
 from piker._daemon import (
     find_service,
     Services,
@@ -27,6 +34,8 @@ from piker.clearing._client import (
     OrderBook,
 )
 
+log = get_logger(__name__)
+
 
 def test_runtime_boot(
     open_test_pikerd: AsyncContextManager
@@ -174,5 +183,8 @@ def test_ensure_ems_in_paper_actors(
     ) as exc_info:
         trio.run(main)
 
-    cancel_msg: str = '`_emsd_main()` was remotely cancelled by its caller'
+    cancel_msg: str = '_emsd_main was remotely cancelled by its caller'
     assert cancel_msg in exc_info.value.args[0]
+
+
+