Merge pull request #365 from pikers/ppu_history

Ppu history

piker/brokers/ib/broker.py

@@ -383,7 +383,7 @@ async def update_and_audit_msgs(
                 symbol=ibppmsg.symbol,
                 currency=ibppmsg.currency,
                 size=p.size,
-                avg_price=p.be_price,
+                avg_price=p.ppu,
             )
             msgs.append(msg)
 
@@ -430,7 +430,7 @@ async def update_and_audit_msgs(
                 symbol=p.symbol.front_fqsn(),
                 # currency=ibppmsg.currency,
                 size=p.size,
-                avg_price=p.be_price,
+                avg_price=p.ppu,
             )
             if validate and p.size:
                 raise ValueError(

piker/clearing/_allocate.py

@@ -126,7 +126,7 @@ def next_order_info(
             l_sub_pp = self.units_limit - abs_live_size
 
         elif size_unit == 'currency':
-            live_cost_basis = abs_live_size * live_pp.be_price
+            live_cost_basis = abs_live_size * live_pp.ppu
             slot_size = currency_per_slot / price
             l_sub_pp = (self.currency_limit - live_cost_basis) / price
 
@@ -158,7 +158,7 @@ def next_order_info(
             if size_unit == 'currency':
                 # compute the "projected" limit's worth of units at the
                 # current pp (weighted) price:
-                slot_size = currency_per_slot / live_pp.be_price
+                slot_size = currency_per_slot / live_pp.ppu
 
             else:
                 slot_size = u_per_slot
@@ -200,7 +200,7 @@ def next_order_info(
                 Position(
                     symbol=sym,
                     size=order_size,
-                    be_price=price,
+                    ppu=price,
                     bsuid=sym,
                 )
             )
@@ -229,8 +229,8 @@ def slots_used(
         abs_pp_size = abs(pp.size)
 
         if self.size_unit == 'currency':
-            # live_currency_size = size or (abs_pp_size * pp.be_price)
-            live_currency_size = abs_pp_size * pp.be_price
+            # live_currency_size = size or (abs_pp_size * pp.ppu)
+            live_currency_size = abs_pp_size * pp.ppu
             prop = live_currency_size / self.currency_limit
 
         else:
@@ -303,7 +303,7 @@ def mk_allocator(
     # if the current position is already greater then the limit
     # settings, increase the limit to the current position
     if alloc.size_unit == 'currency':
-        startup_size = startup_pp.size * startup_pp.be_price
+        startup_size = startup_pp.size * startup_pp.ppu
 
         if startup_size > alloc.currency_limit:
             alloc.currency_limit = round(startup_size, ndigits=2)

piker/clearing/_paper_engine.py

@@ -22,17 +22,25 @@
 from datetime import datetime
 from operator import itemgetter
 import time
-from typing import Tuple, Optional, Callable
+from typing import (
+    Any,
+    Optional,
+    Callable,
+)
 import uuid
 
 from bidict import bidict
+import pendulum
 import trio
 import tractor
 from dataclasses import dataclass
 
 from .. import data
 from ..data._source import Symbol
-from ..pp import Position
+from ..pp import (
+    Position,
+    Transaction,
+)
 from ..data._normalize import iterticks
 from ..data._source import unpack_fqsn
 from ..log import get_logger
@@ -63,11 +71,12 @@ class PaperBoi:
     _buys: bidict
     _sells: bidict
     _reqids: bidict
-    _positions: dict[str, BrokerdPosition]
+    _positions: dict[str, Position]
+    _trade_ledger: dict[str, Any]
 
     # init edge case L1 spread
-    last_ask: Tuple[float, float] = (float('inf'), 0)  # price, size
-    last_bid: Tuple[float, float] = (0, 0)
+    last_ask: tuple[float, float] = (float('inf'), 0)  # price, size
+    last_bid: tuple[float, float] = (0, 0)
 
     async def submit_limit(
         self,
@@ -77,22 +86,23 @@ async def submit_limit(
         action: str,
         size: float,
         reqid: Optional[str],
+
     ) -> int:
-        """Place an order and return integer request id provided by client.
+        '''
+        Place an order and return integer request id provided by client.
 
-        """
+        '''
         is_modify: bool = False
-        if reqid is None:
-            reqid = str(uuid.uuid4())
 
-        else:
+        entry = self._reqids.get(reqid)
+        if entry:
             # order is already existing, this is a modify
-            (oid, symbol, action, old_price) = self._reqids[reqid]
+            (oid, symbol, action, old_price) = entry
             assert old_price != price
             is_modify = True
-
-        # register order internally
-        self._reqids[reqid] = (oid, symbol, action, price)
+        else:
+            # register order internally
+            self._reqids[reqid] = (oid, symbol, action, price)
 
         if action == 'alert':
             # bypass all fill simulation
@@ -197,16 +207,15 @@ async def fake_fill(
         """
         # TODO: net latency model
         await trio.sleep(0.05)
+        fill_time_ns = time.time_ns()
+        fill_time_s = time.time()
 
-        msg = BrokerdFill(
-
+        fill_msg = BrokerdFill(
             reqid=reqid,
-            time_ns=time.time_ns(),
-
+            time_ns=fill_time_ns,
             action=action,
             size=size,
             price=price,
-
             broker_time=datetime.now().timestamp(),
             broker_details={
                 'paper_info': {
@@ -216,7 +225,9 @@ async def fake_fill(
                 'name': self.broker + '_paper',
             },
         )
-        await self.ems_trades_stream.send(msg)
+        await self.ems_trades_stream.send(fill_msg)
+
+        self._trade_ledger.update(fill_msg.to_dict())
 
         if order_complete:
 
@@ -243,36 +254,45 @@ async def fake_fill(
 
         # lookup any existing position
         token = f'{symbol}.{self.broker}'
-        pp_msg = self._positions.setdefault(
+        pp = self._positions.setdefault(
             token,
-            BrokerdPosition(
-                broker=self.broker,
-                account='paper',
-                symbol=symbol,
-                # TODO: we need to look up the asset currency from
-                # broker info. i guess for crypto this can be
-                # inferred from the pair?
-                currency='',
-                size=0.0,
-                avg_price=0,
+            Position(
+                Symbol(key=symbol),
+                size=size,
+                ppu=price,
+                bsuid=symbol,
             )
         )
-
-        # delegate update to `.pp.Position.lifo_update()`
-        pp = Position(
-            Symbol(key=symbol),
-            size=pp_msg.size,
-            be_price=pp_msg.avg_price,
+        t = Transaction(
+            fqsn=symbol,
+            tid=oid,
+            size=size,
+            price=price,
+            cost=1.,  # todo cost model
+            dt=pendulum.from_timestamp(fill_time_s),
             bsuid=symbol,
         )
-        pp_msg.size, pp_msg.avg_price = pp.lifo_update(size, price)
+        pp.add_clear(t)
+
+        pp_msg = BrokerdPosition(
+            broker=self.broker,
+            account='paper',
+            symbol=symbol,
+            # TODO: we need to look up the asset currency from
+            # broker info. i guess for crypto this can be
+            # inferred from the pair?
+            currency='',
+            size=pp.size,
+            avg_price=pp.ppu,
+        )
 
         await self.ems_trades_stream.send(pp_msg)
 
 
 async def simulate_fills(
     quote_stream: 'tractor.ReceiveStream',  # noqa
     client: PaperBoi,
+
 ) -> None:
 
     # TODO: more machinery to better simulate real-world market things:
@@ -389,6 +409,24 @@ async def handle_order_requests(
             # validate
             order = BrokerdOrder(**request_msg)
 
+            if order.reqid is None:
+                reqid = str(uuid.uuid4())
+            else:
+                reqid = order.reqid
+
+            # deliver ack that order has been submitted to broker routing
+            await ems_order_stream.send(
+                BrokerdOrderAck(
+
+                    # ems order request id
+                    oid=order.oid,
+
+                    # broker specific request id
+                    reqid=reqid,
+
+                )
+            )
+
             # call our client api to submit the order
             reqid = await client.submit_limit(
 
@@ -402,20 +440,7 @@ async def handle_order_requests(
                 # there is no existing order so ask the client to create
                 # a new one (which it seems to do by allocating an int
                 # counter - collision prone..)
-                reqid=order.reqid,
-            )
-
-            # deliver ack that order has been submitted to broker routing
-            await ems_order_stream.send(
-                BrokerdOrderAck(
-
-                    # ems order request id
-                    oid=order.oid,
-
-                    # broker specific request id
-                    reqid=reqid,
-
-                )
+                reqid=reqid,
             )
 
         elif action == 'cancel':
@@ -468,6 +493,9 @@ async def trades_dialogue(
 
                 # TODO: load paper positions from ``positions.toml``
                 _positions={},
+
+                # TODO: load postions from ledger file
+                _trade_ledger={},
             )
 
             n.start_soon(handle_order_requests, client, ems_stream)
@@ -510,5 +538,4 @@ async def open_paperboi(
                 loglevel=loglevel,
 
         ) as (ctx, first):
-
             yield ctx, first