Finally solve the last-price-is-`nan` issue..

Not sure why I put this off for so long but the check is in now such
that if the market isn't open or no rt quote comes in from the first
query, we just pull from the last shm history 'close' value.
Includes another fix to avoid raising when a double remove on the client
side stream from the registry sometimes happens.

piker/brokers/ib/__init__.py

@@ -20,15 +20,10 @@ Interactive Brokers API backend.
 Sub-modules within break into the core functionalities:
 
 - ``broker.py`` part for orders / trading endpoints
-- ``data.py`` for real-time data feed endpoints
+- ``feed.py`` for real-time data feed endpoints
-
+- ``api.py`` for the core API machinery which is ``trio``-ized
-- ``client.py`` for the core API machinery which is ``trio``-ized
   wrapping around ``ib_insync``.
 
-- ``report.py`` for the hackery to build manual pp calcs
-  to avoid ib's absolute bullshit FIFO style position
-  tracking..
-
 """
 from .api import (
     get_client,
@@ -38,7 +33,10 @@ from .feed import (
     open_symbol_search,
     stream_quotes,
 )
-from .broker import trades_dialogue
+from .broker import (
+    trades_dialogue,
+    norm_trade_records,
+)
 
 __all__ = [
     'get_client',

piker/brokers/ib/api.py

@@ -38,15 +38,21 @@ import time
 from types import SimpleNamespace
 
 
+from bidict import bidict
 import trio
 import tractor
 from tractor import to_asyncio
+import ib_insync as ibis
 from ib_insync.wrapper import RequestError
 from ib_insync.contract import Contract, ContractDetails
 from ib_insync.order import Order
 from ib_insync.ticker import Ticker
-from ib_insync.objects import Position
+from ib_insync.objects import (
-import ib_insync as ibis
+    Position,
+    Fill,
+    Execution,
+    CommissionReport,
+)
 from ib_insync.wrapper import Wrapper
 from ib_insync.client import Client as ib_Client
 import numpy as np
@@ -155,30 +161,23 @@ class NonShittyIB(ibis.IB):
         self.client.apiEnd += self.disconnectedEvent
 
 
-# map of symbols to contract ids
-_adhoc_cmdty_data_map = {
-    # https://misc.interactivebrokers.com/cstools/contract_info/v3.10/index.php?action=Conid%20Info&wlId=IB&conid=69067924
-
-    # NOTE: some cmdtys/metals don't have trade data like gold/usd:
-    # https://groups.io/g/twsapi/message/44174
-    'XAUUSD': ({'conId': 69067924}, {'whatToShow': 'MIDPOINT'}),
-}
-
 _futes_venues = (
     'GLOBEX',
     'NYMEX',
     'CME',
     'CMECRYPTO',
+    'COMEX',
+    'CMDTY',  # special name case..
 )
 
 _adhoc_futes_set = {
 
     # equities
     'nq.globex',
-    'mnq.globex',
+    'mnq.globex',  # micro
 
     'es.globex',
-    'mes.globex',
+    'mes.globex',  # micro
 
     # cypto$
     'brr.cmecrypto',
@@ -195,20 +194,46 @@ _adhoc_futes_set = {
     # metals
     'xauusd.cmdty',  # gold spot
     'gc.nymex',
-    'mgc.nymex',
+    'mgc.nymex',  # micro
+
+    # oil & gas
+    'cl.nymex',
 
     'xagusd.cmdty',  # silver spot
     'ni.nymex',  # silver futes
     'qi.comex',  # mini-silver futes
 }
 
+
+# map of symbols to contract ids
+_adhoc_symbol_map = {
+    # https://misc.interactivebrokers.com/cstools/contract_info/v3.10/index.php?action=Conid%20Info&wlId=IB&conid=69067924
+
+    # NOTE: some cmdtys/metals don't have trade data like gold/usd:
+    # https://groups.io/g/twsapi/message/44174
+    'XAUUSD': ({'conId': 69067924}, {'whatToShow': 'MIDPOINT'}),
+}
+for qsn in _adhoc_futes_set:
+    sym, venue = qsn.split('.')
+    assert venue.upper() in _futes_venues, f'{venue}'
+    _adhoc_symbol_map[sym.upper()] = (
+        {'exchange': venue},
+        {},
+    )
+
+
 # exchanges we don't support at the moment due to not knowing
 # how to do symbol-contract lookup correctly likely due
 # to not having the data feeds subscribed.
 _exch_skip_list = {
+
     'ASX',  # aussie stocks
     'MEXI',  # mexican stocks
-    'VALUE',  # no idea
+
+    # no idea
+    'VALUE',
+    'FUNDSERV',
+    'SWB2',
 }
 
 # https://misc.interactivebrokers.com/cstools/contract_info/v3.10/index.php?action=Conid%20Info&wlId=IB&conid=69067924
@@ -261,27 +286,29 @@ class Client:
 
         # NOTE: the ib.client here is "throttled" to 45 rps by default
 
-    async def trades(
+    async def trades(self) -> dict[str, Any]:
-        self,
+        '''
-        # api_only: bool = False,
+        Return list of trade-fills from current session in ``dict``.
 
-    ) -> dict[str, Any]:
+        '''
-
+        fills: list[Fill] = self.ib.fills()
-        # orders = await self.ib.reqCompletedOrdersAsync(
+        norm_fills: list[dict] = []
-        #     apiOnly=api_only
-        # )
-        fills = await self.ib.reqExecutionsAsync()
-        norm_fills = []
         for fill in fills:
             fill = fill._asdict()  # namedtuple
-            for key, val in fill.copy().items():
+            for key, val in fill.items():
-                if isinstance(val, Contract):
+                match val:
+                    case Contract() | Execution() | CommissionReport():
                         fill[key] = asdict(val)
 
             norm_fills.append(fill)
 
         return norm_fills
 
+    async def orders(self) -> list[Order]:
+        return await self.ib.reqAllOpenOrdersAsync(
+            apiOnly=False,
+        )
+
     async def bars(
         self,
         fqsn: str,
@@ -483,6 +510,14 @@ class Client:
 
         return con
 
+    async def get_con(
+        self,
+        conid: int,
+    ) -> Contract:
+        return await self.ib.qualifyContractsAsync(
+            ibis.Contract(conId=conid)
+        )
+
     async def find_contract(
         self,
         pattern: str,
@@ -553,7 +588,7 @@ class Client:
 
         # commodities
         elif exch == 'CMDTY':  # eg. XAUUSD.CMDTY
-            con_kwargs, bars_kwargs = _adhoc_cmdty_data_map[sym]
+            con_kwargs, bars_kwargs = _adhoc_symbol_map[sym]
             con = ibis.Commodity(**con_kwargs)
             con.bars_kwargs = bars_kwargs
 
@@ -811,10 +846,23 @@ _scan_ignore: set[tuple[str, int]] = set()
 
 def get_config() -> dict[str, Any]:
 
-    conf, path = config.load()
+    conf, path = config.load('brokers')
-
     section = conf.get('ib')
 
+    accounts = section.get('accounts')
+    if not accounts:
+        raise ValueError(
+            'brokers.toml -> `ib.accounts` must be defined\n'
+            f'location: {path}'
+        )
+
+    names = list(accounts.keys())
+    accts = section['accounts'] = bidict(accounts)
+    log.info(
+        f'brokers.toml defines {len(accts)} accounts: '
+        f'{pformat(names)}'
+    )
+
     if section is None:
         log.warning(f'No config section found for ib in {path}')
         return {}
@@ -990,7 +1038,7 @@ async def load_aio_clients(
         for acct, client in _accounts2clients.items():
             log.info(f'Disconnecting {acct}@{client}')
             client.ib.disconnect()
-            _client_cache.pop((host, port))
+            _client_cache.pop((host, port), None)
 
 
 async def load_clients_for_trio(
@@ -1019,9 +1067,6 @@ async def load_clients_for_trio(
             await asyncio.sleep(float('inf'))
 
 
-_proxies: dict[str, MethodProxy] = {}
-
-
 @acm
 async def open_client_proxies() -> tuple[
     dict[str, MethodProxy],
@@ -1044,13 +1089,14 @@ async def open_client_proxies() -> tuple[
         if cache_hit:
             log.info(f'Re-using cached clients: {clients}')
 
+        proxies = {}
         for acct_name, client in clients.items():
             proxy = await stack.enter_async_context(
                 open_client_proxy(client),
             )
-            _proxies[acct_name] = proxy
+            proxies[acct_name] = proxy
 
-        yield _proxies, clients
+        yield proxies, clients
 
 
 def get_preferred_data_client(
@@ -1199,11 +1245,13 @@ async def open_client_proxy(
     event_table = {}
 
     async with (
+
         to_asyncio.open_channel_from(
             open_aio_client_method_relay,
             client=client,
             event_consumers=event_table,
         ) as (first, chan),
+
         trio.open_nursery() as relay_n,
     ):
 

piker/brokers/ib/broker.py

@@ -26,8 +26,10 @@ from typing import (
     Any,
     Optional,
     AsyncIterator,
+    Union,
 )
 
+from bidict import bidict
 import trio
 from trio_typing import TaskStatus
 import tractor
@@ -42,10 +44,13 @@ from ib_insync.order import (
 from ib_insync.objects import (
     Fill,
     Execution,
+    CommissionReport,
 )
 from ib_insync.objects import Position
+import pendulum
 
 from piker import config
+from piker import pp
 from piker.log import get_console_log
 from piker.clearing._messages import (
     BrokerdOrder,
@@ -56,13 +61,16 @@ from piker.clearing._messages import (
     BrokerdFill,
     BrokerdError,
 )
+from piker.data._source import Symbol
 from .api import (
     _accounts2clients,
-    _adhoc_futes_set,
+    # _adhoc_futes_set,
+    _adhoc_symbol_map,
     log,
     get_config,
     open_client_proxies,
     Client,
+    MethodProxy,
 )
 
 
@@ -80,29 +88,39 @@ def pack_position(
         # TODO: lookup fqsn even for derivs.
         symbol = con.symbol.lower()
 
+    # try our best to figure out the exchange / venue
     exch = (con.primaryExchange or con.exchange).lower()
-    symkey = '.'.join((symbol, exch))
     if not exch:
-        # attempt to lookup the symbol from our
+        # for wtv cucked reason some futes don't show their
-        # hacked set..
+        # exchange (like CL.NYMEX) ...
-        for sym in _adhoc_futes_set:
+        entry = _adhoc_symbol_map.get(
-            if symbol in sym:
+            con.symbol or con.localSymbol
-                symkey = sym
+        )
-                break
+        if entry:
+            meta, kwargs = entry
+            cid = meta.get('conId')
+            if cid:
+                assert con.conId == meta['conId']
+            exch = meta['exchange']
+
+    assert exch, f'No clue:\n {con}'
+    fqsn = '.'.join((symbol, exch))
 
     expiry = con.lastTradeDateOrContractMonth
     if expiry:
-        symkey += f'.{expiry}'
+        fqsn += f'.{expiry}'
 
     # TODO: options contracts into a sane format..
-
+    return (
-    return BrokerdPosition(
+        con.conId,
+        BrokerdPosition(
             broker='ib',
             account=pos.account,
-        symbol=symkey,
+            symbol=fqsn,
             currency=con.currency,
             size=float(pos.position),
             avg_price=float(pos.avgCost) / float(con.multiplier or 1.0),
+        ),
     )
 
 
@@ -205,19 +223,35 @@ async def recv_trade_updates(
     # sync with trio task
     to_trio.send_nowait(None)
 
-    def push_tradesies(eventkit_obj, obj, fill=None):
+    def push_tradesies(
-        """Push events to trio task.
+        eventkit_obj,
+        obj,
+        fill: Optional[Fill] = None,
+        report: Optional[CommissionReport] = None,
+    ):
+        '''
+        Push events to trio task.
 
-        """
+        '''
-        if fill is not None:
+        match eventkit_obj.name():
+
+            case 'orderStatusEvent':
+                item = ('status', obj)
+
+            case 'commissionReportEvent':
+                assert report
+                item = ('cost', report)
+
+            case 'execDetailsEvent':
                 # execution details event
                 item = ('fill', (obj, fill))
 
-        elif eventkit_obj.name() == 'positionEvent':
+            case 'positionEvent':
                 item = ('position', obj)
 
-        else:
+            case _:
-            item = ('status', obj)
+                log.error(f'Error unknown event {obj}')
+                return
 
         log.info(f'eventkit event ->\n{pformat(item)}')
 
@@ -233,15 +267,15 @@ async def recv_trade_updates(
         'execDetailsEvent',  # all "fill" updates
         'positionEvent',  # avg price updates per symbol per account
 
-        # 'commissionReportEvent',
         # XXX: ugh, it is a separate event from IB and it's
         # emitted as follows:
         # self.ib.commissionReportEvent.emit(trade, fill, report)
+        'commissionReportEvent',
 
         # XXX: not sure yet if we need these
         # 'updatePortfolioEvent',
 
-        # XXX: these all seem to be weird ib_insync intrernal
+        # XXX: these all seem to be weird ib_insync internal
         # events that we probably don't care that much about
         # given the internal design is wonky af..
         # 'newOrderEvent',
@@ -257,6 +291,149 @@ async def recv_trade_updates(
     await client.ib.disconnectedEvent
 
 
+async def update_ledger_from_api_trades(
+    trade_entries: list[dict[str, Any]],
+    client: Union[Client, MethodProxy],
+
+) -> tuple[
+    dict[str, pp.Transaction],
+    dict[str, dict],
+]:
+
+    conf = get_config()
+
+    # XXX; ERRGGG..
+    # pack in the "primary/listing exchange" value from a
+    # contract lookup since it seems this isn't available by
+    # default from the `.fills()` method endpoint...
+    for entry in trade_entries:
+        condict = entry['contract']
+        conid = condict['conId']
+        pexch = condict['primaryExchange']
+
+        if not pexch:
+            cons = await client.get_con(conid=conid)
+            if cons:
+                con = cons[0]
+                pexch = con.primaryExchange or con.exchange
+            else:
+                # for futes it seems like the primary is always empty?
+                pexch = condict['exchange']
+
+        entry['listingExchange'] = pexch
+
+    entries = trades_to_ledger_entries(
+        conf['accounts'].inverse,
+        trade_entries,
+    )
+
+    # write recent session's trades to the user's (local) ledger file.
+    records: dict[str, pp.Transactions] = {}
+
+    for acctid, trades_by_id in entries.items():
+        # normalize to transaction form
+        records[acctid] = norm_trade_records(trades_by_id)
+
+    return records, entries
+
+
+async def update_and_audit_msgs(
+    acctid: str,  # no `ib.` prefix is required!
+    pps: list[pp.Position],
+    cids2pps: dict[tuple[str, int], BrokerdPosition],
+    validate: bool = False,
+
+) -> list[BrokerdPosition]:
+
+    msgs: list[BrokerdPosition] = []
+    # pps: dict[int, pp.Position] = {}
+
+    for p in pps:
+        bsuid = p.bsuid
+
+        # build trade-session-actor local table
+        # of pps from unique symbol ids.
+        # pps[bsuid] = p
+
+        # retreive equivalent ib reported position message
+        # for comparison/audit versus the piker equivalent
+        # breakeven pp calcs.
+        ibppmsg = cids2pps.get((acctid, bsuid))
+
+        if ibppmsg:
+            msg = BrokerdPosition(
+                broker='ib',
+
+                # XXX: ok so this is annoying, we're relaying
+                # an account name with the backend suffix prefixed
+                # but when reading accounts from ledgers we don't
+                # need it and/or it's prefixed in the section
+                # table..
+                account=ibppmsg.account,
+                # XXX: the `.ib` is stripped..?
+                symbol=ibppmsg.symbol,
+                currency=ibppmsg.currency,
+                size=p.size,
+                avg_price=p.be_price,
+            )
+            msgs.append(msg)
+
+            if validate:
+                ibsize = ibppmsg.size
+                pikersize = msg.size
+                diff = pikersize - ibsize
+
+                # if ib reports a lesser pp it's not as bad since we can
+                # presume we're at least not more in the shit then we
+                # thought.
+                if diff:
+                    raise ValueError(
+                        f'POSITION MISMATCH ib <-> piker ledger:\n'
+                        f'ib: {ibppmsg}\n'
+                        f'piker: {msg}\n'
+                        'YOU SHOULD FIGURE OUT WHY TF YOUR LEDGER IS OFF!?!?'
+                    )
+                    msg.size = ibsize
+
+            if ibppmsg.avg_price != msg.avg_price:
+
+                # TODO: make this a "propoganda" log level?
+                log.warning(
+                    'The mega-cucks at IB want you to believe with their '
+                    f'"FIFO" positioning for {msg.symbol}:\n'
+                    f'"ib" mega-cucker avg price: {ibppmsg.avg_price}\n'
+                    f'piker, LIFO breakeven PnL price: {msg.avg_price}'
+                )
+
+        else:
+            # make brand new message
+            msg = BrokerdPosition(
+                broker='ib',
+
+                # XXX: ok so this is annoying, we're relaying
+                # an account name with the backend suffix prefixed
+                # but when reading accounts from ledgers we don't
+                # need it and/or it's prefixed in the section
+                # table.. we should just strip this from the message
+                # right since `.broker` is already included?
+                account=f'ib.{acctid}',
+                # XXX: the `.ib` is stripped..?
+                symbol=p.symbol.front_fqsn(),
+                # currency=ibppmsg.currency,
+                size=p.size,
+                avg_price=p.be_price,
+            )
+            if validate and p.size:
+                raise ValueError(
+                    f'UNEXPECTED POSITION ib <-> piker ledger:\n'
+                    f'piker: {msg}\n'
+                    'YOU SHOULD FIGURE OUT WHY TF YOUR LEDGER IS OFF!?!?'
+                )
+            msgs.append(msg)
+
+    return msgs
+
+
 @tractor.context
 async def trades_dialogue(
 
@@ -277,6 +454,14 @@ async def trades_dialogue(
     accounts = set()
     clients: list[tuple[Client, trio.MemoryReceiveChannel]] = []
 
+    # TODO: this causes a massive tractor bug when you run marketstored
+    # with ``--tsdb``... you should get:
+    # - first error the assertion
+    # - chart should get that error and die
+    # - pikerd goes to debugger again from trio nursery multi-error
+    # - hitting final control-c to kill daemon will lead to hang
+    # assert 0
+
     async with (
         trio.open_nursery() as nurse,
         open_client_proxies() as (proxies, aioclients),
@@ -306,22 +491,83 @@ async def trades_dialogue(
             assert account in accounts_def
             accounts.add(account)
 
+        cids2pps: dict[str, BrokerdPosition] = {}
+        update_records: dict[str, bidict] = {}
+
+        # process pp value reported from ib's system. we only use these
+        # to cross-check sizing since average pricing on their end uses
+        # the so called (bs) "FIFO" style which more or less results in
+        # a price that's not useful for traders who want to not lose
+        # money.. xb
         for client in aioclients.values():
             for pos in client.positions():
 
-                msg = pack_position(pos)
+                cid, msg = pack_position(pos)
-                msg.account = accounts_def.inverse[msg.account]
+                acctid = msg.account = accounts_def.inverse[msg.account]
-
+                acctid = acctid.strip('ib.')
+                cids2pps[(acctid, cid)] = msg
                 assert msg.account in accounts, (
                     f'Position for unknown account: {msg.account}')
 
-                all_positions.append(msg.dict())
+                # collect all ib-pp reported positions so that we can be
+                # sure know which positions to update from the ledger if
+                # any are missing from the ``pps.toml``
+                update_records.setdefault(acctid, bidict())[cid] = msg.symbol
 
-        trades: list[dict] = []
+        # update trades ledgers for all accounts from
-        for proxy in proxies.values():
+        # connected api clients which report trades for **this session**.
-            trades.append(await proxy.trades())
+        new_trades = {}
+        for account, proxy in proxies.items():
+            trades = await proxy.trades()
+            (
+                records_by_acct,
+                ledger_entries,
+            ) = await update_ledger_from_api_trades(
+                trades,
+                proxy,
+            )
+            new_trades.update(records_by_acct)
 
-        log.info(f'Loaded {len(trades)} from this session')
+        for acctid, trans in new_trades.items():
+            for t in trans:
+                bsuid = t.bsuid
+                if bsuid in update_records:
+                    assert update_records[bsuid] == t.fqsn
+                else:
+                    update_records.setdefault(acctid, bidict())[bsuid] = t.fqsn
+
+        # load all positions from `pps.toml`, cross check with ib's
+        # positions data, and relay re-formatted pps as msgs to the ems.
+        # __2 cases__:
+        # - new trades have taken place this session that we want to
+        #   always reprocess indempotently,
+        # - no new trades yet but we want to reload and audit any
+        #   positions reported by ib's sys that may not yet be in
+        #   piker's ``pps.toml`` state-file.
+        for acctid, to_update in update_records.items():
+            trans = new_trades.get(acctid)
+            active, closed = pp.update_pps_conf(
+                'ib',
+                acctid,
+                trade_records=trans,
+                ledger_reload=to_update,
+            )
+            for pps in [active, closed]:
+                msgs = await update_and_audit_msgs(
+                    acctid,
+                    pps.values(),
+                    cids2pps,
+                    validate=True,
+                )
+                all_positions.extend(msg.dict() for msg in msgs)
+
+        if not all_positions and cids2pps:
+            raise RuntimeError(
+                'Positions reported by ib but not found in `pps.toml`!?\n'
+                f'{pformat(cids2pps)}'
+            )
+
+        # log.info(f'Loaded {len(trades)} from this session')
         # TODO: write trades to local ``trades.toml``
         # - use above per-session trades data and write to local file
         # - get the "flex reports" working and pull historical data and
@@ -332,6 +578,16 @@ async def trades_dialogue(
             tuple(name for name in accounts_def if name in accounts),
         ))
 
+        # TODO: maybe just write on teardown?
+        # we might also want to delegate a specific actor for
+        # ledger writing / reading for speed?
+
+        # write ledger with all new trades **AFTER** we've updated the
+        # `pps.toml` from the original ledger state!
+        for acctid, trades_by_id in ledger_entries.items():
+            with pp.open_trade_ledger('ib', acctid) as ledger:
+                ledger.update(trades_by_id)
+
         async with (
             ctx.open_stream() as ems_stream,
             trio.open_nursery() as n,
@@ -345,32 +601,96 @@ async def trades_dialogue(
                     deliver_trade_events,
                     stream,
                     ems_stream,
-                    accounts_def
+                    accounts_def,
+                    cids2pps,
+                    proxies,
                 )
 
             # block until cancelled
             await trio.sleep_forever()
 
 
+async def emit_pp_update(
+    ems_stream: tractor.MsgStream,
+    trade_entry: dict,
+    accounts_def: bidict,
+    proxies: dict,
+    cids2pps: dict,
+
+) -> None:
+
+    # compute and relay incrementally updated piker pp
+    acctid = accounts_def.inverse[trade_entry['execution']['acctNumber']]
+    proxy = proxies[acctid]
+
+    acctname = acctid.strip('ib.')
+    records_by_acct, ledger_entries = await update_ledger_from_api_trades(
+        [trade_entry],
+        proxy,
+    )
+    records = records_by_acct[acctname]
+    r = records[0]
+
+    # update and load all positions from `pps.toml`, cross check with
+    # ib's positions data, and relay re-formatted pps as msgs to the
+    # ems. we report both the open and closed updates in one map since
+    # for incremental update we may have just fully closed a pp and need
+    # to relay that msg as well!
+    active, closed = pp.update_pps_conf(
+        'ib',
+        acctname,
+        trade_records=records,
+        ledger_reload={r.bsuid: r.fqsn},
+    )
+
+    # NOTE: write ledger with all new trades **AFTER** we've updated the
+    # `pps.toml` from the original ledger state!
+    for acctid, trades_by_id in ledger_entries.items():
+        with pp.open_trade_ledger('ib', acctid) as ledger:
+            ledger.update(trades_by_id)
+
+    for pos in filter(
+        bool,
+        [active.get(r.bsuid), closed.get(r.bsuid)]
+    ):
+        msgs = await update_and_audit_msgs(
+            acctname,
+            [pos],
+            cids2pps,
+
+            # ib pp event might not have arrived yet
+            validate=False,
+        )
+        if msgs:
+            msg = msgs[0]
+            break
+
+    await ems_stream.send(msg.dict())
+
+
 async def deliver_trade_events(
 
     trade_event_stream: trio.MemoryReceiveChannel,
     ems_stream: tractor.MsgStream,
-    accounts_def: dict[str, str],
+    accounts_def: dict[str, str],  # eg. `'ib.main'` -> `'DU999999'`
+    cids2pps: dict[tuple[str, str], BrokerdPosition],
+    proxies: dict[str, MethodProxy],
 
 ) -> None:
-    '''Format and relay all trade events for a given client to the EMS.
+    '''
+    Format and relay all trade events for a given client to emsd.
 
     '''
     action_map = {'BOT': 'buy', 'SLD': 'sell'}
+    ids2fills: dict[str, dict] = {}
 
     # TODO: for some reason we can receive a ``None`` here when the
     # ib-gw goes down? Not sure exactly how that's happening looking
     # at the eventkit code above but we should probably handle it...
     async for event_name, item in trade_event_stream:
-
         log.info(f'ib sending {event_name}:\n{pformat(item)}')
 
+        match event_name:
             # TODO: templating the ib statuses in comparison with other
             # brokers is likely the way to go:
             # https://interactivebrokers.github.io/tws-api/interfaceIBApi_1_1EWrapper.html#a17f2a02d6449710b6394d0266a353313
@@ -394,7 +714,7 @@ async def deliver_trade_events(
             #   reqId 1550: Order held while securities are located.'),
             # status='PreSubmitted', message='')],
 
-        if event_name == 'status':
+            case 'status':
 
                 # XXX: begin normalization of nonsense ib_insync internal
                 # object-state tracking representations...
@@ -423,8 +743,9 @@ async def deliver_trade_events(
 
                     broker_details={'name': 'ib'},
                 )
+                await ems_stream.send(msg.dict())
 
-        elif event_name == 'fill':
+            case 'fill':
 
                 # for wtv reason this is a separate event type
                 # from IB, not sure why it's needed other then for extra
@@ -438,17 +759,35 @@ async def deliver_trade_events(
                 # https://www.python.org/dev/peps/pep-0526/#global-and-local-variable-annotations
                 trade: Trade
                 fill: Fill
+
+                # TODO: maybe we can use matching to better handle these cases.
                 trade, fill = item
                 execu: Execution = fill.execution
+                execid = execu.execId
 
-            # TODO: normalize out commissions details?
+                # TODO:
-            details = {
+                # - normalize out commissions details?
+                # - this is the same as the unpacking loop above in
+                # ``trades_to_ledger_entries()`` no?
+                trade_entry = ids2fills.setdefault(execid, {})
+                cost_already_rx = bool(trade_entry)
+
+                # if the costs report was already received this
+                # should be not empty right?
+                comms = fill.commissionReport.commission
+                if cost_already_rx:
+                    assert comms
+
+                trade_entry.update(
+                    {
                         'contract': asdict(fill.contract),
                         'execution': asdict(fill.execution),
-                'commissions': asdict(fill.commissionReport),
+                        # 'commissionReport': asdict(fill.commissionReport),
-                'broker_time': execu.time,   # supposedly server fill time
+                        # supposedly server fill time?
+                        'broker_time': execu.time,
                         'name': 'ib',
                     }
+                )
 
                 msg = BrokerdFill(
                     # should match the value returned from `.submit_limit()`
@@ -459,14 +798,68 @@ async def deliver_trade_events(
                     size=execu.shares,
                     price=execu.price,
 
-                broker_details=details,
+                    broker_details=trade_entry,
                     # XXX: required by order mode currently
-                broker_time=details['broker_time'],
+                    broker_time=trade_entry['broker_time'],
 
                 )
+                await ems_stream.send(msg.dict())
 
-        elif event_name == 'error':
+                # 2 cases:
+                # - fill comes first or
+                # - comms report comes first
+                comms = fill.commissionReport.commission
+                if comms:
+                    # UGHHH since the commision report object might be
+                    # filled in **after** we already serialized to dict..
+                    # def need something better for all this.
+                    trade_entry.update(
+                        {'commissionReport': asdict(fill.commissionReport)}
+                    )
 
+                if comms or cost_already_rx:
+                    # only send a pp update once we have a cost report
+                    await emit_pp_update(
+                        ems_stream,
+                        trade_entry,
+                        accounts_def,
+                        proxies,
+                        cids2pps,
+                    )
+
+            case 'cost':
+
+                cr: CommissionReport = item
+                execid = cr.execId
+
+                trade_entry = ids2fills.setdefault(execid, {})
+                fill_already_rx = bool(trade_entry)
+
+                # only fire a pp msg update if,
+                # - we haven't already
+                # - the fill event has already arrived
+                #   but it didn't yet have a commision report
+                #   which we fill in now.
+                if (
+                    fill_already_rx
+                    and 'commissionReport' not in trade_entry
+                ):
+                    # no fill msg has arrived yet so just fill out the
+                    # cost report for now and when the fill arrives a pp
+                    # msg can be emitted.
+                    trade_entry.update(
+                        {'commissionReport': asdict(cr)}
+                    )
+
+                    await emit_pp_update(
+                        ems_stream,
+                        trade_entry,
+                        accounts_def,
+                        proxies,
+                        cids2pps,
+                    )
+
+            case 'error':
                 err: dict = item
 
                 # f$#$% gawd dammit insync..
@@ -480,13 +873,15 @@ async def deliver_trade_events(
                 # TODO: what schema for this msg if we're going to make it
                 # portable across all backends?
                 # msg = BrokerdError(**err)
-            continue
 
-        elif event_name == 'position':
+            case 'position':
-            msg = pack_position(item)
-            msg.account = accounts_def.inverse[msg.account]
 
-        elif event_name == 'event':
+                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())
+
+            case 'event':
 
                 # it's either a general system status event or an external
                 # trade event?
@@ -498,8 +893,6 @@ async def deliver_trade_events(
                 # if getattr(msg, 'reqid', 0) < -1:
                 # log.info(f"TWS triggered trade\n{pformat(msg.dict())}")
 
-            continue
-
                 # msg.reqid = 'tws-' + str(-1 * reqid)
 
                 # mark msg as from "external system"
@@ -507,19 +900,200 @@ async def deliver_trade_events(
                 # considering multiplayer/group trades tracking
                 # msg.broker_details['external_src'] = 'tws'
 
-        # XXX: we always serialize to a dict for msgpack
+            case _:
-        # translations, ideally we can move to an msgspec (or other)
+                log.error(f'WTF: {event_name}: {item}')
-        # encoder # that can be enabled in ``tractor`` ahead of
+
-        # time so we can pass through the message types directly.
+
-        await ems_stream.send(msg.dict())
+def norm_trade_records(
+    ledger: dict[str, Any],
+
+) -> list[pp.Transaction]:
+    '''
+    Normalize a flex report or API retrieved executions
+    ledger into our standard record format.
+
+    '''
+    records: list[pp.Transaction] = []
+
+    for tid, record in ledger.items():
+
+        conid = record.get('conId') or record['conid']
+        comms = record.get('commission') or -1*record['ibCommission']
+        price = record.get('price') or record['tradePrice']
+
+        # the api doesn't do the -/+ on the quantity for you but flex
+        # records do.. are you fucking serious ib...!?
+        size = record.get('quantity') or record['shares'] * {
+            'BOT': 1,
+            'SLD': -1,
+        }[record['side']]
+
+        exch = record['exchange']
+        lexch = record.get('listingExchange')
+
+        suffix = lexch or exch
+        symbol = record['symbol']
+
+        # likely an opts contract record from a flex report..
+        # TODO: no idea how to parse ^ the strike part from flex..
+        # (00010000 any, or 00007500 tsla, ..)
+        # we probably must do the contract lookup for this?
+        if '   ' in symbol or '--' in exch:
+            underlying, _, tail = symbol.partition('   ')
+            suffix = exch = 'opt'
+            expiry = tail[:6]
+            # otype = tail[6]
+            # strike = tail[7:]
+
+            print(f'skipping opts contract {symbol}')
+            continue
+
+        # timestamping is way different in API records
+        date = record.get('date')
+        if not date:
+            # probably a flex record with a wonky non-std timestamp..
+            date, ts = record['dateTime'].split(';')
+            dt = pendulum.parse(date)
+            ts = f'{ts[:2]}:{ts[2:4]}:{ts[4:]}'
+            tsdt = pendulum.parse(ts)
+            dt.set(hour=tsdt.hour, minute=tsdt.minute, second=tsdt.second)
+
+        else:
+            # epoch_dt = pendulum.from_timestamp(record.get('time'))
+            dt = pendulum.parse(date)
+
+        # special handling of symbol extraction from
+        # flex records using some ad-hoc schema parsing.
+        instr = record.get('assetCategory')
+        if instr == 'FUT':
+            symbol = record['description'][:3]
+
+        # try to build out piker fqsn from record.
+        expiry = record.get(
+            'lastTradeDateOrContractMonth') or record.get('expiry')
+        if expiry:
+            expiry = str(expiry).strip(' ')
+            suffix = f'{exch}.{expiry}'
+            expiry = pendulum.parse(expiry)
+
+        fqsn = Symbol.from_fqsn(
+            fqsn=f'{symbol}.{suffix}.ib',
+            info={},
+        ).front_fqsn().rstrip('.ib')
+
+        # NOTE: for flex records the normal fields for defining an fqsn
+        # sometimes won't be available so we rely on two approaches for
+        # the "reverse lookup" of piker style fqsn keys:
+        # - when dealing with API trade records received from
+        #   `IB.trades()` we do a contract lookup at he time of processing
+        # - when dealing with flex records, it is assumed the record
+        #   is at least a day old and thus the TWS position reporting system
+        #   should already have entries if the pps are still open, in
+        #   which case, we can pull the fqsn from that table (see
+        #   `trades_dialogue()` above).
+
+        records.append(pp.Transaction(
+            fqsn=fqsn,
+            tid=tid,
+            size=size,
+            price=price,
+            cost=comms,
+            dt=dt,
+            expiry=expiry,
+            bsuid=conid,
+        ))
+
+    return records
+
+
+def trades_to_ledger_entries(
+    accounts: bidict,
+    trade_entries: list[object],
+    source_type: str = 'api',
+
+) -> dict:
+    '''
+    Convert either of API execution objects or flex report
+    entry objects into ``dict`` form, pretty much straight up
+    without modification.
+
+    '''
+    trades_by_account = {}
+
+    for t in trade_entries:
+        if source_type == 'flex':
+            entry = t.__dict__
+
+            # XXX: LOL apparently ``toml`` has a bug
+            # where a section key error will show up in the write
+            # if you leave a table key as an `int`? So i guess
+            # cast to strs for all keys..
+
+            # oddly for some so-called "BookTrade" entries
+            # this field seems to be blank, no cuckin clue.
+            # trade['ibExecID']
+            tid = str(entry.get('ibExecID') or entry['tradeID'])
+            # date = str(entry['tradeDate'])
+
+            # XXX: is it going to cause problems if a account name
+            # get's lost? The user should be able to find it based
+            # on the actual exec history right?
+            acctid = accounts[str(entry['accountId'])]
+
+        elif source_type == 'api':
+            # NOTE: example of schema we pull from the API client.
+            # {
+            #     'commissionReport': CommissionReport(...
+            #     'contract': {...
+            #     'execution': Execution(...
+            #     'time': 1654801166.0
+            # }
+
+            # flatten all sub-dicts and values into one top level entry.
+            entry = {}
+            for section, val in t.items():
+                match section:
+                    case 'contract' | 'execution' | 'commissionReport':
+                        # sub-dict cases
+                        entry.update(val)
+
+                    case 'time':
+                        # ib has wack ns timestamps, or is that us?
+                        continue
+
+                    case _:
+                        entry[section] = val
+
+            tid = str(entry['execId'])
+            dt = pendulum.from_timestamp(entry['time'])
+            # TODO: why isn't this showing seconds in the str?
+            entry['date'] = str(dt)
+            acctid = accounts[entry['acctNumber']]
+
+        if not tid:
+            # this is likely some kind of internal adjustment
+            # transaction, likely one of the following:
+            # - an expiry event that will show a "book trade" indicating
+            #   some adjustment to cash balances: zeroing or itm settle.
+            # - a manual cash balance position adjustment likely done by
+            #   the user from the accounts window in TWS where they can
+            #   manually set the avg price and size:
+            #   https://api.ibkr.com/lib/cstools/faq/web1/index.html#/tag/DTWS_ADJ_AVG_COST
+            log.warning(f'Skipping ID-less ledger entry:\n{pformat(entry)}')
+            continue
+
+        trades_by_account.setdefault(
+            acctid, {}
+        )[tid] = entry
+
+    return trades_by_account
 
 
 def load_flex_trades(
     path: Optional[str] = None,
 
-) -> dict[str, str]:
+) -> dict[str, Any]:
 
-    from pprint import pprint
     from ib_insync import flexreport, util
 
     conf = get_config()
@@ -555,36 +1129,38 @@ def load_flex_trades(
         report = flexreport.FlexReport(path=path)
 
     trade_entries = report.extract('Trade')
-    trades = {
+    ln = len(trade_entries)
-        # XXX: LOL apparently ``toml`` has a bug
+    # log.info(f'Loaded {ln} trades from flex query')
-        # where a section key error will show up in the write
+    print(f'Loaded {ln} trades from flex query')
-        # if you leave this as an ``int``?
-        str(t.__dict__['tradeID']): t.__dict__
-        for t in trade_entries
-    }
 
-    ln = len(trades)
+    trades_by_account = trades_to_ledger_entries(
-    log.info(f'Loaded {ln} trades from flex query')
+        # get reverse map to user account names
+        conf['accounts'].inverse,
+        trade_entries,
+        source_type='flex',
+    )
 
-    trades_by_account = {}
+    ledgers = {}
-    for tid, trade in trades.items():
+    for acctid, trades_by_id in trades_by_account.items():
-        trades_by_account.setdefault(
+        with pp.open_trade_ledger('ib', acctid) as ledger:
-            # oddly for some so-called "BookTrade" entries
+            ledger.update(trades_by_id)
-            # this field seems to be blank, no cuckin clue.
-            # trade['ibExecID']
-            str(trade['accountId']), {}
-        )[tid] = trade
 
-    section = {'ib': trades_by_account}
+        ledgers[acctid] = ledger
-    pprint(section)
 
-    # TODO: load the config first and append in
+    return ledgers
-    # the new trades loaded here..
-    try:
-        config.write(section, 'trades')
-    except KeyError:
-        import pdbpp; pdbpp.set_trace()  # noqa
 
 
 if __name__ == '__main__':
+    import sys
+    import os
+
+    args = sys.argv
+    if len(args) > 1:
+        args = args[1:]
+        for arg in args:
+            path = os.path.abspath(arg)
+            load_flex_trades(path=path)
+    else:
+        # expect brokers.toml to have an entry and
+        # pull from the web service.
         load_flex_trades()

piker/brokers/ib/feed.py

@@ -217,8 +217,8 @@ async def get_bars(
                 )
 
             elif (
-                err.code == 162
+                err.code == 162 and
-                and 'HMDS query returned no data' in err.message
+                'HMDS query returned no data' in err.message
             ):
                 # XXX: this is now done in the storage mgmt layer
                 # and we shouldn't implicitly decrement the frame dt
@@ -237,6 +237,13 @@ async def get_bars(
                     frame_size=2000,
                 )
 
+            # elif (
+            #     err.code == 162 and
+            #     'Trading TWS session is connected from a different IP address' in err.message
+            # ):
+            #     log.warning("ignoring ip address warning")
+            #     continue
+
             elif _pacing in msg:
 
                 log.warning(
@@ -909,17 +916,17 @@ async def open_symbol_search(
                     # trigger async request
                     await trio.sleep(0)
 
-                    # match against our ad-hoc set immediately
+                    # # match against our ad-hoc set immediately
-                    adhoc_matches = fuzzy.extractBests(
+                    # adhoc_matches = fuzzy.extractBests(
-                        pattern,
+                    #     pattern,
-                        list(_adhoc_futes_set),
+                    #     list(_adhoc_futes_set),
-                        score_cutoff=90,
+                    #     score_cutoff=90,
-                    )
+                    # )
-                    log.info(f'fuzzy matched adhocs: {adhoc_matches}')
+                    # log.info(f'fuzzy matched adhocs: {adhoc_matches}')
-                    adhoc_match_results = {}
+                    # adhoc_match_results = {}
-                    if adhoc_matches:
+                    # if adhoc_matches:
-                        # TODO: do we need to pull contract details?
+                    #     # TODO: do we need to pull contract details?
-                        adhoc_match_results = {i[0]: {} for i in adhoc_matches}
+                    #     adhoc_match_results = {i[0]: {} for i in adhoc_matches}
 
                 log.debug(f'fuzzy matching stocks {stock_results}')
                 stock_matches = fuzzy.extractBests(
@@ -928,7 +935,8 @@ async def open_symbol_search(
                     score_cutoff=50,
                 )
 
-                matches = adhoc_match_results | {
+                # matches = adhoc_match_results | {
+                matches = {
                     item[0]: {} for item in stock_matches
                 }
                 # TODO: we used to deliver contract details

piker/brokers/kraken/__init__.py

@@ -0,0 +1,61 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Kraken backend.
+
+Sub-modules within break into the core functionalities:
+
+- ``broker.py`` part for orders / trading endpoints
+- ``feed.py`` for real-time data feed endpoints
+- ``api.py`` for the core API machinery which is ``trio``-ized
+  wrapping around ``ib_insync``.
+
+'''
+
+from piker.log import get_logger
+
+log = get_logger(__name__)
+
+from .api import (
+    get_client,
+)
+from .feed import (
+    open_history_client,
+    open_symbol_search,
+    stream_quotes,
+)
+from .broker import (
+    trades_dialogue,
+    norm_trade_records,
+)
+
+__all__ = [
+    'get_client',
+    'trades_dialogue',
+    'open_history_client',
+    'open_symbol_search',
+    'stream_quotes',
+    'norm_trade_records',
+]
+
+
+# tractor RPC enable arg
+__enable_modules__: list[str] = [
+    'api',
+    'feed',
+    'broker',
+]

piker/brokers/kraken/api.py

@@ -0,0 +1,468 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Kraken web API wrapping.
+
+'''
+from contextlib import asynccontextmanager as acm
+from dataclasses import field
+from datetime import datetime
+import itertools
+from typing import (
+    Any,
+    Optional,
+    Union,
+)
+import time
+
+# import trio
+# import tractor
+import pendulum
+import asks
+from fuzzywuzzy import process as fuzzy
+import numpy as np
+from pydantic.dataclasses import dataclass
+import urllib.parse
+import hashlib
+import hmac
+import base64
+
+from piker import config
+from piker.brokers._util import (
+    resproc,
+    SymbolNotFound,
+    BrokerError,
+    DataThrottle,
+)
+from . import log
+
+# <uri>/<version>/
+_url = 'https://api.kraken.com/0'
+
+
+# Broker specific ohlc schema which includes a vwap field
+_ohlc_dtype = [
+    ('index', int),
+    ('time', int),
+    ('open', float),
+    ('high', float),
+    ('low', float),
+    ('close', float),
+    ('volume', float),
+    ('count', int),
+    ('bar_wap', float),
+]
+
+# UI components allow this to be declared such that additional
+# (historical) fields can be exposed.
+ohlc_dtype = np.dtype(_ohlc_dtype)
+
+_show_wap_in_history = True
+_symbol_info_translation: dict[str, str] = {
+    'tick_decimals': 'pair_decimals',
+}
+
+
+@dataclass
+class OHLC:
+    '''
+    Description of the flattened OHLC quote format.
+
+    For schema details see:
+        https://docs.kraken.com/websockets/#message-ohlc
+
+    '''
+    chan_id: int  # internal kraken id
+    chan_name: str  # eg. ohlc-1  (name-interval)
+    pair: str  # fx pair
+    time: float  # Begin time of interval, in seconds since epoch
+    etime: float  # End time of interval, in seconds since epoch
+    open: float  # Open price of interval
+    high: float  # High price within interval
+    low: float  # Low price within interval
+    close: float  # Close price of interval
+    vwap: float  # Volume weighted average price within interval
+    volume: float  # Accumulated volume **within interval**
+    count: int  # Number of trades within interval
+    # (sampled) generated tick data
+    ticks: list[Any] = field(default_factory=list)
+
+
+def get_config() -> dict[str, Any]:
+
+    conf, path = config.load()
+    section = conf.get('kraken')
+
+    if section is None:
+        log.warning(f'No config section found for kraken in {path}')
+        return {}
+
+    return section
+
+
+def get_kraken_signature(
+    urlpath: str,
+    data: dict[str, Any],
+    secret: str
+) -> str:
+    postdata = urllib.parse.urlencode(data)
+    encoded = (str(data['nonce']) + postdata).encode()
+    message = urlpath.encode() + hashlib.sha256(encoded).digest()
+
+    mac = hmac.new(base64.b64decode(secret), message, hashlib.sha512)
+    sigdigest = base64.b64encode(mac.digest())
+    return sigdigest.decode()
+
+
+class InvalidKey(ValueError):
+    '''
+    EAPI:Invalid key
+    This error is returned when the API key used for the call is
+    either expired or disabled, please review the API key in your
+    Settings -> API tab of account management or generate a new one
+    and update your application.
+
+    '''
+
+
+class Client:
+
+    def __init__(
+        self,
+        name: str = '',
+        api_key: str = '',
+        secret: str = ''
+    ) -> None:
+        self._sesh = asks.Session(connections=4)
+        self._sesh.base_location = _url
+        self._sesh.headers.update({
+            'User-Agent':
+                'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
+        })
+        self._pairs: list[str] = []
+        self._name = name
+        self._api_key = api_key
+        self._secret = secret
+
+    @property
+    def pairs(self) -> dict[str, Any]:
+        if self._pairs is None:
+            raise RuntimeError(
+                "Make sure to run `cache_symbols()` on startup!"
+            )
+            # retreive and cache all symbols
+
+        return self._pairs
+
+    async def _public(
+        self,
+        method: str,
+        data: dict,
+    ) -> dict[str, Any]:
+        resp = await self._sesh.post(
+            path=f'/public/{method}',
+            json=data,
+            timeout=float('inf')
+        )
+        return resproc(resp, log)
+
+    async def _private(
+        self,
+        method: str,
+        data: dict,
+        uri_path: str
+    ) -> dict[str, Any]:
+        headers = {
+            'Content-Type':
+                'application/x-www-form-urlencoded',
+            'API-Key':
+                self._api_key,
+            'API-Sign':
+                get_kraken_signature(uri_path, data, self._secret)
+        }
+        resp = await self._sesh.post(
+            path=f'/private/{method}',
+            data=data,
+            headers=headers,
+            timeout=float('inf')
+        )
+        return resproc(resp, log)
+
+    async def endpoint(
+        self,
+        method: str,
+        data: dict[str, Any]
+
+    ) -> dict[str, Any]:
+        uri_path = f'/0/private/{method}'
+        data['nonce'] = str(int(1000*time.time()))
+        return await self._private(method, data, uri_path)
+
+    async def get_trades(
+        self,
+
+    ) -> dict[str, Any]:
+        '''
+        Get the trades (aka cleared orders) history from the rest endpoint:
+        https://docs.kraken.com/rest/#operation/getTradeHistory
+
+        '''
+        ofs = 0
+        trades_by_id: dict[str, Any] = {}
+
+        for i in itertools.count():
+
+            # increment 'ofs' pagination offset
+            ofs = i*50
+
+            resp = await self.endpoint(
+                'TradesHistory',
+                {'ofs': ofs},
+            )
+            by_id = resp['result']['trades']
+            trades_by_id.update(by_id)
+
+            # we can get up to 50 results per query
+            if (
+                len(by_id) < 50
+            ):
+                err = resp.get('error')
+                if err:
+                    raise BrokerError(err)
+
+                # we know we received the max amount of
+                # trade results so there may be more history.
+                # catch the end of the trades
+                count = resp['result']['count']
+                break
+
+        # santity check on update
+        assert count == len(trades_by_id.values())
+        return trades_by_id
+
+    async def submit_limit(
+        self,
+        symbol: str,
+        price: float,
+        action: str,
+        size: float,
+        reqid: str = None,
+        validate: bool = False  # set True test call without a real submission
+
+    ) -> dict:
+        '''
+        Place an order and return integer request id provided by client.
+
+        '''
+        # Build common data dict for common keys from both endpoints
+        data = {
+            "pair": symbol,
+            "price": str(price),
+            "validate": validate
+        }
+        if reqid is None:
+            # Build order data for kraken api
+            data |= {
+                "ordertype": "limit",
+                "type": action,
+                "volume": str(size),
+            }
+            return await self.endpoint('AddOrder', data)
+        else:
+            # Edit order data for kraken api
+            data["txid"] = reqid
+            return await self.endpoint('EditOrder', data)
+
+    async def submit_cancel(
+        self,
+        reqid: str,
+    ) -> dict:
+        '''
+        Send cancel request for order id ``reqid``.
+
+        '''
+        # txid is a transaction id given by kraken
+        return await self.endpoint('CancelOrder', {"txid": reqid})
+
+    async def symbol_info(
+        self,
+        pair: Optional[str] = None,
+    ):
+        if pair is not None:
+            pairs = {'pair': pair}
+        else:
+            pairs = None  # get all pairs
+
+        resp = await self._public('AssetPairs', pairs)
+        err = resp['error']
+        if err:
+            symbolname = pairs['pair'] if pair else None
+            raise SymbolNotFound(f'{symbolname}.kraken')
+
+        pairs = resp['result']
+
+        if pair is not None:
+            _, data = next(iter(pairs.items()))
+            return data
+        else:
+            return pairs
+
+    async def cache_symbols(
+        self,
+    ) -> dict:
+        if not self._pairs:
+            self._pairs = await self.symbol_info()
+
+        return self._pairs
+
+    async def search_symbols(
+        self,
+        pattern: str,
+        limit: int = None,
+    ) -> dict[str, Any]:
+        if self._pairs is not None:
+            data = self._pairs
+        else:
+            data = await self.symbol_info()
+
+        matches = fuzzy.extractBests(
+            pattern,
+            data,
+            score_cutoff=50,
+        )
+        # repack in dict form
+        return {item[0]['altname']: item[0] for item in matches}
+
+    async def bars(
+        self,
+        symbol: str = 'XBTUSD',
+
+        # UTC 2017-07-02 12:53:20
+        since: Optional[Union[int, datetime]] = None,
+        count: int = 720,  # <- max allowed per query
+        as_np: bool = True,
+
+    ) -> dict:
+
+        if since is None:
+            since = pendulum.now('UTC').start_of('minute').subtract(
+                minutes=count).timestamp()
+
+        elif isinstance(since, int):
+            since = pendulum.from_timestamp(since).timestamp()
+
+        else:  # presumably a pendulum datetime
+            since = since.timestamp()
+
+        # UTC 2017-07-02 12:53:20 is oldest seconds value
+        since = str(max(1499000000, int(since)))
+        json = await self._public(
+            'OHLC',
+            data={
+                'pair': symbol,
+                'since': since,
+            },
+        )
+        try:
+            res = json['result']
+            res.pop('last')
+            bars = next(iter(res.values()))
+
+            new_bars = []
+
+            first = bars[0]
+            last_nz_vwap = first[-3]
+            if last_nz_vwap == 0:
+                # use close if vwap is zero
+                last_nz_vwap = first[-4]
+
+            # convert all fields to native types
+            for i, bar in enumerate(bars):
+                # normalize weird zero-ed vwap values..cmon kraken..
+                # indicates vwap didn't change since last bar
+                vwap = float(bar.pop(-3))
+                if vwap != 0:
+                    last_nz_vwap = vwap
+                if vwap == 0:
+                    vwap = last_nz_vwap
+
+                # re-insert vwap as the last of the fields
+                bar.append(vwap)
+
+                new_bars.append(
+                    (i,) + tuple(
+                        ftype(bar[j]) for j, (name, ftype) in enumerate(
+                            _ohlc_dtype[1:]
+                        )
+                    )
+                )
+            array = np.array(new_bars, dtype=_ohlc_dtype) if as_np else bars
+            return array
+        except KeyError:
+            errmsg = json['error'][0]
+
+            if 'not found' in errmsg:
+                raise SymbolNotFound(errmsg + f': {symbol}')
+
+            elif 'Too many requests' in errmsg:
+                raise DataThrottle(f'{symbol}')
+
+            else:
+                raise BrokerError(errmsg)
+
+
+@acm
+async def get_client() -> Client:
+
+    section = get_config()
+    if section:
+        client = Client(
+            name=section['key_descr'],
+            api_key=section['api_key'],
+            secret=section['secret']
+        )
+    else:
+        client = Client()
+
+    # at startup, load all symbols locally for fast search
+    await client.cache_symbols()
+
+    yield client
+
+
+def normalize_symbol(
+    ticker: str
+) -> str:
+    '''
+    Normalize symbol names to to a 3x3 pair.
+
+    '''
+    remap = {
+        'XXBTZEUR': 'XBTEUR',
+        'XXMRZEUR': 'XMREUR',
+
+        # ws versions? pretty weird..
+        'XBT/EUR': 'XBTEUR',
+        'XMR/EUR': 'XMREUR',
+    }
+    symlen = len(ticker)
+    if symlen != 6:
+        ticker = remap[ticker]
+    else:
+        raise ValueError(f'Unhandled symbol: {ticker}')
+
+    return ticker.lower()

piker/brokers/kraken/broker.py

@@ -0,0 +1,540 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Order api and machinery
+
+'''
+from contextlib import asynccontextmanager as acm
+from functools import partial
+from itertools import chain
+from pprint import pformat
+import time
+from typing import (
+    Any,
+    AsyncIterator,
+    # Callable,
+    # Optional,
+    # Union,
+)
+
+import pendulum
+from pydantic import BaseModel
+import trio
+import tractor
+import wsproto
+
+from piker import pp
+from piker.clearing._messages import (
+    BrokerdCancel,
+    BrokerdError,
+    BrokerdFill,
+    BrokerdOrder,
+    BrokerdOrderAck,
+    BrokerdPosition,
+    BrokerdStatus,
+)
+from . import log
+from .api import (
+    Client,
+    BrokerError,
+    get_client,
+    normalize_symbol,
+)
+from .feed import (
+    get_console_log,
+    open_autorecon_ws,
+    NoBsWs,
+    stream_messages,
+)
+
+
+class Trade(BaseModel):
+    '''
+    Trade class that helps parse and validate ownTrades stream
+
+    '''
+    reqid: str  # kraken order transaction id
+    action: str  # buy or sell
+    price: float  # price of asset
+    size: float  # vol of asset
+    broker_time: str  # e.g GTC, GTD
+
+
+async def handle_order_requests(
+
+    client: Client,
+    ems_order_stream: tractor.MsgStream,
+
+) -> None:
+
+    request_msg: dict
+    order: BrokerdOrder
+
+    async for request_msg in ems_order_stream:
+        log.info(
+            'Received order request:\n'
+            f'{pformat(request_msg)}'
+        )
+
+        action = request_msg['action']
+
+        if action in {'buy', 'sell'}:
+
+            account = request_msg['account']
+            if account != 'kraken.spot':
+                log.error(
+                    'This is a kraken account, \
+                    only a `kraken.spot` selection is valid'
+                )
+                await ems_order_stream.send(BrokerdError(
+                    oid=request_msg['oid'],
+                    symbol=request_msg['symbol'],
+
+                    # reason=f'Kraken only, No account found: `{account}` ?',
+                    reason=(
+                        'Kraken only, order mode disabled due to '
+                        'https://github.com/pikers/piker/issues/299'
+                    ),
+
+                ).dict())
+                continue
+
+            # validate
+            order = BrokerdOrder(**request_msg)
+            # call our client api to submit the order
+            resp = await client.submit_limit(
+                symbol=order.symbol,
+                price=order.price,
+                action=order.action,
+                size=order.size,
+                reqid=order.reqid,
+            )
+
+            err = resp['error']
+            if err:
+                oid = order.oid
+                log.error(f'Failed to submit order: {oid}')
+
+                await ems_order_stream.send(
+                    BrokerdError(
+                        oid=order.oid,
+                        reqid=order.reqid,
+                        symbol=order.symbol,
+                        reason="Failed order submission",
+                        broker_details=resp
+                    ).dict()
+                )
+            else:
+                # TODO: handle multiple orders (cancels?)
+                #       txid is an array of strings
+                if order.reqid is None:
+                    reqid = resp['result']['txid'][0]
+                else:
+                    # update the internal pairing of oid to krakens
+                    # txid with the new txid that is returned on edit
+                    reqid = resp['result']['txid']
+
+                # 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,
+
+                        # account the made the order
+                        account=order.account
+
+                    ).dict()
+                )
+
+        elif action == 'cancel':
+            msg = BrokerdCancel(**request_msg)
+
+            # Send order cancellation to kraken
+            resp = await client.submit_cancel(
+                reqid=msg.reqid
+            )
+
+            # Check to make sure there was no error returned by
+            # the kraken endpoint. Assert one order was cancelled.
+            try:
+                result = resp['result']
+                count = result['count']
+
+            # check for 'error' key if we received no 'result'
+            except KeyError:
+                error = resp.get('error')
+
+                await ems_order_stream.send(
+                    BrokerdError(
+                        oid=msg.oid,
+                        reqid=msg.reqid,
+                        symbol=msg.symbol,
+                        reason="Failed order cancel",
+                        broker_details=resp
+                    ).dict()
+                )
+
+                if not error:
+                    raise BrokerError(f'Unknown order cancel response: {resp}')
+
+            else:
+                if not count:  # no orders were cancelled?
+
+                    # XXX: what exactly is this from and why would we care?
+                    # there doesn't seem to be any docs here?
+                    # https://docs.kraken.com/rest/#operation/cancelOrder
+
+                    # Check to make sure the cancellation is NOT pending,
+                    # then send the confirmation to the ems order stream
+                    pending = result.get('pending')
+                    if pending:
+                        log.error(f'Order {oid} cancel was not yet successful')
+
+                        await ems_order_stream.send(
+                            BrokerdError(
+                                oid=msg.oid,
+                                reqid=msg.reqid,
+                                symbol=msg.symbol,
+                                # TODO: maybe figure out if pending
+                                # cancels will eventually get cancelled
+                                reason="Order cancel is still pending?",
+                                broker_details=resp
+                            ).dict()
+                        )
+
+                else:  # order cancel success case.
+
+                    await ems_order_stream.send(
+                        BrokerdStatus(
+                            reqid=msg.reqid,
+                            account=msg.account,
+                            time_ns=time.time_ns(),
+                            status='cancelled',
+                            reason='Order cancelled',
+                            broker_details={'name': 'kraken'}
+                        ).dict()
+                    )
+    else:
+        log.error(f'Unknown order command: {request_msg}')
+
+
+@acm
+async def subscribe(
+    ws: wsproto.WSConnection,
+    token: str,
+    subs: list[str] = ['ownTrades', 'openOrders'],
+):
+    '''
+    Setup ws api subscriptions:
+    https://docs.kraken.com/websockets/#message-subscribe
+
+    By default we sign up for trade and order update events.
+
+    '''
+    # more specific logic for this in kraken's sync client:
+    # https://github.com/krakenfx/kraken-wsclient-py/blob/master/kraken_wsclient_py/kraken_wsclient_py.py#L188
+
+    assert token
+    for sub in subs:
+        msg = {
+            'event': 'subscribe',
+            'subscription': {
+                'name': sub,
+                'token': token,
+            }
+        }
+
+        # TODO: we want to eventually allow unsubs which should
+        # be completely fine to request from a separate task
+        # since internally the ws methods appear to be FIFO
+        # locked.
+        await ws.send_msg(msg)
+
+    yield
+
+    for sub in subs:
+        # unsub from all pairs on teardown
+        await ws.send_msg({
+            'event': 'unsubscribe',
+            'subscription': [sub],
+        })
+
+    # XXX: do we need to ack the unsub?
+    # await ws.recv_msg()
+
+
+@tractor.context
+async def trades_dialogue(
+    ctx: tractor.Context,
+    loglevel: str = None,
+) -> AsyncIterator[dict[str, Any]]:
+
+    # XXX: required to propagate ``tractor`` loglevel to piker logging
+    get_console_log(loglevel or tractor.current_actor().loglevel)
+
+    async with get_client() as client:
+
+        # 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']))
+
+        if not client._api_key:
+            raise RuntimeError(
+                'Missing Kraken API key in `brokers.toml`!?!?')
+
+        # auth required block
+        acctid = client._name
+        acc_name = 'kraken.' + acctid
+
+        # pull and deliver trades ledger
+        trades = await client.get_trades()
+        log.info(
+            f'Loaded {len(trades)} trades from account `{acc_name}`'
+        )
+        trans = await update_ledger(acctid, trades)
+        active, closed = pp.update_pps_conf(
+            'kraken',
+            acctid,
+            trade_records=trans,
+            ledger_reload={}.fromkeys(t.bsuid for t in trans),
+        )
+
+        position_msgs: list[dict] = []
+        pps: dict[int, pp.Position]
+        for pps in [active, closed]:
+            for tid, p in pps.items():
+                msg = BrokerdPosition(
+                    broker='kraken',
+                    account=acc_name,
+                    symbol=p.symbol.front_fqsn(),
+                    size=p.size,
+                    avg_price=p.be_price,
+                    currency='',
+                )
+                position_msgs.append(msg.dict())
+
+        await ctx.started(
+            (position_msgs, [acc_name])
+        )
+
+        # Get websocket token for authenticated data stream
+        # Assert that a token was actually received.
+        resp = await client.endpoint('GetWebSocketsToken', {})
+
+        err = resp.get('error')
+        if err:
+            raise BrokerError(err)
+
+        token = resp['result']['token']
+
+        ws: NoBsWs
+        async with (
+            ctx.open_stream() as ems_stream,
+            open_autorecon_ws(
+                'wss://ws-auth.kraken.com/',
+                fixture=partial(
+                    subscribe,
+                    token=token,
+                ),
+            ) as ws,
+            trio.open_nursery() as n,
+        ):
+            # task for processing inbound requests from ems
+            n.start_soon(handle_order_requests, client, ems_stream)
+
+            count: int = 0
+
+            # process and relay trades events to ems
+            # https://docs.kraken.com/websockets/#message-ownTrades
+            async for msg in stream_messages(ws):
+                match msg:
+                    case [
+                        trades_msgs,
+                        'ownTrades',
+                        {'sequence': seq},
+                    ]:
+                        # XXX: do we actually need this orrr?
+                        # ensure that we are only processing new trades?
+                        assert seq > count
+                        count += 1
+
+                        # flatten msgs for processing
+                        trades = {
+                            tid: trade
+                            for entry in trades_msgs
+                            for (tid, trade) in entry.items()
+
+                            # only emit entries which are already not-in-ledger
+                            if tid not in {r.tid for r in trans}
+                        }
+                        for tid, trade in trades.items():
+
+                            # parse-cast
+                            reqid = trade['ordertxid']
+                            action = trade['type']
+                            price = float(trade['price'])
+                            size = float(trade['vol'])
+                            broker_time = float(trade['time'])
+
+                            # send a fill msg for gui update
+                            fill_msg = BrokerdFill(
+                                reqid=reqid,
+                                time_ns=time.time_ns(),
+
+                                action=action,
+                                size=size,
+                                price=price,
+                                # TODO: maybe capture more msg data
+                                # i.e fees?
+                                broker_details={'name': 'kraken'},
+                                broker_time=broker_time
+                            )
+                            await ems_stream.send(fill_msg.dict())
+
+                            filled_msg = BrokerdStatus(
+                                reqid=reqid,
+                                time_ns=time.time_ns(),
+
+                                account=acc_name,
+                                status='filled',
+                                filled=size,
+                                reason='Order filled by kraken',
+                                broker_details={
+                                    'name': 'kraken',
+                                    'broker_time': broker_time
+                                },
+
+                                # TODO: figure out if kraken gives a count
+                                # of how many units of underlying were
+                                # filled. Alternatively we can decrement
+                                # this value ourselves by associating and
+                                # calcing from the diff with the original
+                                # client-side request, see:
+                                # https://github.com/pikers/piker/issues/296
+                                remaining=0,
+                            )
+                            await ems_stream.send(filled_msg.dict())
+
+                        # update ledger and position tracking
+                        trans = await update_ledger(acctid, trades)
+                        active, closed = pp.update_pps_conf(
+                            'kraken',
+                            acctid,
+                            trade_records=trans,
+                            ledger_reload={}.fromkeys(
+                                t.bsuid for t in trans),
+                        )
+
+                        # emit pp msgs
+                        for pos in filter(
+                            bool,
+                            chain(active.values(), closed.values()),
+                        ):
+                            pp_msg = BrokerdPosition(
+                                broker='kraken',
+
+                                # XXX: ok so this is annoying, we're
+                                # relaying an account name with the
+                                # backend suffix prefixed but when
+                                # reading accounts from ledgers we
+                                # don't need it and/or it's prefixed
+                                # in the section table.. we should
+                                # just strip this from the message
+                                # right since `.broker` is already
+                                # included?
+                                account=f'kraken.{acctid}',
+                                symbol=pos.symbol.front_fqsn(),
+                                size=pos.size,
+                                avg_price=pos.be_price,
+
+                                # TODO
+                                # currency=''
+                            )
+                            await ems_stream.send(pp_msg.dict())
+
+                    case [
+                        trades_msgs,
+                        'openOrders',
+                        {'sequence': seq},
+                    ]:
+                        # TODO: async order update handling which we
+                        # should remove from `handle_order_requests()`
+                        # above:
+                        # https://github.com/pikers/piker/issues/293
+                        # https://github.com/pikers/piker/issues/310
+                        log.info(f'Order update {seq}:{trades_msgs}')
+
+                    case _:
+                        log.warning(f'Unhandled trades msg: {msg}')
+                        await tractor.breakpoint()
+
+
+def norm_trade_records(
+    ledger: dict[str, Any],
+
+) -> list[pp.Transaction]:
+
+    records: list[pp.Transaction] = []
+
+    for tid, record in ledger.items():
+
+        size = record.get('vol') * {
+            'buy': 1,
+            'sell': -1,
+        }[record['type']]
+        bsuid = record['pair']
+        norm_sym = normalize_symbol(bsuid)
+
+        records.append(
+            pp.Transaction(
+                fqsn=f'{norm_sym}.kraken',
+                tid=tid,
+                size=float(size),
+                price=float(record['price']),
+                cost=float(record['fee']),
+                dt=pendulum.from_timestamp(float(record['time'])),
+                bsuid=bsuid,
+
+                # XXX: there are no derivs on kraken right?
+                # expiry=expiry,
+            )
+        )
+
+    return records
+
+
+async def update_ledger(
+    acctid: str,
+    trade_entries: list[dict[str, Any]],
+
+) -> list[pp.Transaction]:
+
+    # write recent session's trades to the user's (local) ledger file.
+    with pp.open_trade_ledger(
+        'kraken',
+        acctid,
+    ) as ledger:
+        ledger.update(trade_entries)
+
+    # normalize to transaction form
+    records = norm_trade_records(trade_entries)
+    return records

piker/brokers/kraken/feed.py

@@ -0,0 +1,464 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Real-time and historical data feed endpoints.
+
+'''
+from contextlib import asynccontextmanager as acm
+from dataclasses import asdict
+from datetime import datetime
+from typing import (
+    Any,
+    Optional,
+    Callable,
+)
+import time
+
+from fuzzywuzzy import process as fuzzy
+import numpy as np
+import pendulum
+from pydantic import BaseModel
+from trio_typing import TaskStatus
+import tractor
+import trio
+import wsproto
+
+from piker._cacheables import open_cached_client
+from piker.brokers._util import (
+    BrokerError,
+    DataThrottle,
+    DataUnavailable,
+)
+from piker.log import get_console_log
+from piker.data import ShmArray
+from piker.data._web_bs import open_autorecon_ws, NoBsWs
+from . import log
+from .api import (
+    Client,
+    OHLC,
+)
+
+
+# https://www.kraken.com/features/api#get-tradable-pairs
+class Pair(BaseModel):
+    altname: str  # alternate pair name
+    wsname: str  # WebSocket pair name (if available)
+    aclass_base: str  # asset class of base component
+    base: str  # asset id of base component
+    aclass_quote: str  # asset class of quote component
+    quote: str  # asset id of quote component
+    lot: str  # volume lot size
+
+    pair_decimals: int  # scaling decimal places for pair
+    lot_decimals: int  # scaling decimal places for volume
+
+    # amount to multiply lot volume by to get currency volume
+    lot_multiplier: float
+
+    # array of leverage amounts available when buying
+    leverage_buy: list[int]
+    # array of leverage amounts available when selling
+    leverage_sell: list[int]
+
+    # fee schedule array in [volume, percent fee] tuples
+    fees: list[tuple[int, float]]
+
+    # maker fee schedule array in [volume, percent fee] tuples (if on
+    # maker/taker)
+    fees_maker: list[tuple[int, float]]
+
+    fee_volume_currency: str  # volume discount currency
+    margin_call: str  # margin call level
+    margin_stop: str  # stop-out/liquidation margin level
+    ordermin: float  # minimum order volume for pair
+
+
+async def stream_messages(
+    ws: NoBsWs,
+):
+    '''
+    Message stream parser and heartbeat handler.
+
+    Deliver ws subscription messages as well as handle heartbeat logic
+    though a single async generator.
+
+    '''
+    too_slow_count = last_hb = 0
+
+    while True:
+
+        with trio.move_on_after(5) as cs:
+            msg = await ws.recv_msg()
+
+        # trigger reconnection if heartbeat is laggy
+        if cs.cancelled_caught:
+
+            too_slow_count += 1
+
+            if too_slow_count > 20:
+                log.warning(
+                    "Heartbeat is too slow, resetting ws connection")
+
+                await ws._connect()
+                too_slow_count = 0
+                continue
+
+        if isinstance(msg, dict):
+            if msg.get('event') == 'heartbeat':
+
+                now = time.time()
+                delay = now - last_hb
+                last_hb = now
+
+                # XXX: why tf is this not printing without --tl flag?
+                log.debug(f"Heartbeat after {delay}")
+                # print(f"Heartbeat after {delay}")
+
+                continue
+
+            err = msg.get('errorMessage')
+            if err:
+                raise BrokerError(err)
+        else:
+            yield msg
+
+
+async def process_data_feed_msgs(
+    ws: NoBsWs,
+):
+    '''
+    Parse and pack data feed messages.
+
+    '''
+    async for msg in stream_messages(ws):
+
+        chan_id, *payload_array, chan_name, pair = msg
+
+        if 'ohlc' in chan_name:
+
+            yield 'ohlc', OHLC(chan_id, chan_name, pair, *payload_array[0])
+
+        elif 'spread' in chan_name:
+
+            bid, ask, ts, bsize, asize = map(float, payload_array[0])
+
+            # TODO: really makes you think IB has a horrible API...
+            quote = {
+                'symbol': pair.replace('/', ''),
+                'ticks': [
+                    {'type': 'bid', 'price': bid, 'size': bsize},
+                    {'type': 'bsize', 'price': bid, 'size': bsize},
+
+                    {'type': 'ask', 'price': ask, 'size': asize},
+                    {'type': 'asize', 'price': ask, 'size': asize},
+                ],
+            }
+            yield 'l1', quote
+
+        # elif 'book' in msg[-2]:
+        #     chan_id, *payload_array, chan_name, pair = msg
+        #     print(msg)
+
+        else:
+            print(f'UNHANDLED MSG: {msg}')
+            yield msg
+
+
+def normalize(
+    ohlc: OHLC,
+
+) -> dict:
+    quote = asdict(ohlc)
+    quote['broker_ts'] = quote['time']
+    quote['brokerd_ts'] = time.time()
+    quote['symbol'] = quote['pair'] = quote['pair'].replace('/', '')
+    quote['last'] = quote['close']
+    quote['bar_wap'] = ohlc.vwap
+
+    # seriously eh? what's with this non-symmetry everywhere
+    # in subscription systems...
+    # XXX: piker style is always lowercases symbols.
+    topic = quote['pair'].replace('/', '').lower()
+
+    # print(quote)
+    return topic, quote
+
+
+def make_sub(pairs: list[str], data: dict[str, Any]) -> dict[str, str]:
+    '''
+    Create a request subscription packet dict.
+
+    https://docs.kraken.com/websockets/#message-subscribe
+
+    '''
+    # eg. specific logic for this in kraken's sync client:
+    # https://github.com/krakenfx/kraken-wsclient-py/blob/master/kraken_wsclient_py/kraken_wsclient_py.py#L188
+    return {
+        'pair': pairs,
+        'event': 'subscribe',
+        'subscription': data,
+    }
+
+
+@acm
+async def open_history_client(
+    symbol: str,
+
+) -> tuple[Callable, int]:
+
+    # TODO implement history getter for the new storage layer.
+    async with open_cached_client('kraken') as client:
+
+        # lol, kraken won't send any more then the "last"
+        # 720 1m bars.. so we have to just ignore further
+        # requests of this type..
+        queries: int = 0
+
+        async def get_ohlc(
+            end_dt: Optional[datetime] = None,
+            start_dt: Optional[datetime] = None,
+
+        ) -> tuple[
+            np.ndarray,
+            datetime,  # start
+            datetime,  # end
+        ]:
+
+            nonlocal queries
+            if queries > 0:
+                raise DataUnavailable
+
+            count = 0
+            while count <= 3:
+                try:
+                    array = await client.bars(
+                        symbol,
+                        since=end_dt,
+                    )
+                    count += 1
+                    queries += 1
+                    break
+                except DataThrottle:
+                    log.warning(f'kraken OHLC throttle for {symbol}')
+                    await trio.sleep(1)
+
+            start_dt = pendulum.from_timestamp(array[0]['time'])
+            end_dt = pendulum.from_timestamp(array[-1]['time'])
+            return array, start_dt, end_dt
+
+        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,
+    symbols: list[str],
+    feed_is_live: trio.Event,
+    loglevel: str = None,
+
+    # backend specific
+    sub_type: str = 'ohlc',
+
+    # startup sync
+    task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,
+
+) -> None:
+    '''
+    Subscribe for ohlc stream of quotes for ``pairs``.
+
+    ``pairs`` must be formatted <crypto_symbol>/<fiat_symbol>.
+
+    '''
+    # XXX: required to propagate ``tractor`` loglevel to piker logging
+    get_console_log(loglevel or tractor.current_actor().loglevel)
+
+    ws_pairs = {}
+    sym_infos = {}
+
+    async with open_cached_client('kraken') as client, send_chan as send_chan:
+
+        # keep client cached for real-time section
+        for sym in symbols:
+
+            # transform to upper since piker style is always lower
+            sym = sym.upper()
+
+            si = Pair(**await client.symbol_info(sym))  # validation
+            syminfo = si.dict()
+            syminfo['price_tick_size'] = 1 / 10**si.pair_decimals
+            syminfo['lot_tick_size'] = 1 / 10**si.lot_decimals
+            syminfo['asset_type'] = 'crypto'
+            sym_infos[sym] = syminfo
+            ws_pairs[sym] = si.wsname
+
+        symbol = symbols[0].lower()
+
+        init_msgs = {
+            # pass back token, and bool, signalling if we're the writer
+            # and that history has been written
+            symbol: {
+                'symbol_info': sym_infos[sym],
+                'shm_write_opts': {'sum_tick_vml': False},
+                'fqsn': sym,
+            },
+        }
+
+        @acm
+        async def subscribe(ws: wsproto.WSConnection):
+            # XXX: setup subs
+            # https://docs.kraken.com/websockets/#message-subscribe
+            # specific logic for this in kraken's shitty sync client:
+            # https://github.com/krakenfx/kraken-wsclient-py/blob/master/kraken_wsclient_py/kraken_wsclient_py.py#L188
+            ohlc_sub = make_sub(
+                list(ws_pairs.values()),
+                {'name': 'ohlc', 'interval': 1}
+            )
+
+            # TODO: we want to eventually allow unsubs which should
+            # be completely fine to request from a separate task
+            # since internally the ws methods appear to be FIFO
+            # locked.
+            await ws.send_msg(ohlc_sub)
+
+            # trade data (aka L1)
+            l1_sub = make_sub(
+                list(ws_pairs.values()),
+                {'name': 'spread'}  # 'depth': 10}
+            )
+
+            # pull a first quote and deliver
+            await ws.send_msg(l1_sub)
+
+            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()
+
+        # see the tips on reconnection logic:
+        # https://support.kraken.com/hc/en-us/articles/360044504011-WebSocket-API-unexpected-disconnections-from-market-data-feeds
+        ws: NoBsWs
+        async with open_autorecon_ws(
+            'wss://ws.kraken.com/',
+            fixture=subscribe,
+        ) as ws:
+
+            # pull a first quote and deliver
+            msg_gen = process_data_feed_msgs(ws)
+
+            # TODO: use ``anext()`` when it lands in 3.10!
+            typ, ohlc_last = await msg_gen.__anext__()
+
+            topic, quote = normalize(ohlc_last)
+
+            task_status.started((init_msgs,  quote))
+
+            # lol, only "closes" when they're margin squeezing clients ;P
+            feed_is_live.set()
+
+            # keep start of last interval for volume tracking
+            last_interval_start = ohlc_last.etime
+
+            # start streaming
+            async for typ, ohlc in msg_gen:
+
+                if typ == 'ohlc':
+
+                    # TODO: can get rid of all this by using
+                    # ``trades`` subscription...
+
+                    # generate tick values to match time & sales pane:
+                    # https://trade.kraken.com/charts/KRAKEN:BTC-USD?period=1m
+                    volume = ohlc.volume
+
+                    # new OHLC sample interval
+                    if ohlc.etime > last_interval_start:
+                        last_interval_start = ohlc.etime
+                        tick_volume = volume
+
+                    else:
+                        # this is the tick volume *within the interval*
+                        tick_volume = volume - ohlc_last.volume
+
+                    ohlc_last = ohlc
+                    last = ohlc.close
+
+                    if tick_volume:
+                        ohlc.ticks.append({
+                            'type': 'trade',
+                            'price': last,
+                            'size': tick_volume,
+                        })
+
+                    topic, quote = normalize(ohlc)
+
+                elif typ == 'l1':
+                    quote = ohlc
+                    topic = quote['symbol'].lower()
+
+                await send_chan.send({topic: quote})
+
+
+@tractor.context
+async def open_symbol_search(
+    ctx: tractor.Context,
+
+) -> Client:
+    async with open_cached_client('kraken') as client:
+
+        # load all symbols locally for fast search
+        cache = await client.cache_symbols()
+        await ctx.started(cache)
+
+        async with ctx.open_stream() as stream:
+
+            async for pattern in stream:
+
+                matches = fuzzy.extractBests(
+                    pattern,
+                    cache,
+                    score_cutoff=50,
+                )
+                # repack in dict form
+                await stream.send(
+                    {item[0]['altname']: item[0]
+                     for item in matches}
+                )

piker/clearing/_allocate.py

@@ -23,53 +23,10 @@ from typing import Optional
 
 from bidict import bidict
 from pydantic import BaseModel, validator
+# from msgspec import Struct
 
 from ..data._source import Symbol
-from ._messages import BrokerdPosition, Status
+from ..pp import Position
-
-
-class Position(BaseModel):
-    '''
-    Basic pp (personal position) model with attached fills history.
-
-    This type should be IPC wire ready?
-
-    '''
-    symbol: Symbol
-
-    # last size and avg entry price
-    size: float
-    avg_price: float  # TODO: contextual pricing
-
-    # ordered record of known constituent trade messages
-    fills: list[Status] = []
-
-    def update_from_msg(
-        self,
-        msg: BrokerdPosition,
-
-    ) -> None:
-
-        # XXX: better place to do this?
-        symbol = self.symbol
-
-        lot_size_digits = symbol.lot_size_digits
-        avg_price, size = (
-            round(msg['avg_price'], ndigits=symbol.tick_size_digits),
-            round(msg['size'], ndigits=lot_size_digits),
-        )
-
-        self.avg_price = avg_price
-        self.size = size
-
-    @property
-    def dsize(self) -> float:
-        '''
-        The "dollar" size of the pp, normally in trading (fiat) unit
-        terms.
-
-        '''
-        return self.avg_price * self.size
 
 
 _size_units = bidict({
@@ -173,7 +130,7 @@ class Allocator(BaseModel):
             l_sub_pp = self.units_limit - abs_live_size
 
         elif size_unit == 'currency':
-            live_cost_basis = abs_live_size * live_pp.avg_price
+            live_cost_basis = abs_live_size * live_pp.be_price
             slot_size = currency_per_slot / price
             l_sub_pp = (self.currency_limit - live_cost_basis) / price
 
@@ -205,7 +162,7 @@ class Allocator(BaseModel):
             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.avg_price
+                slot_size = currency_per_slot / live_pp.be_price
 
             else:
                 slot_size = u_per_slot
@@ -244,7 +201,12 @@ class Allocator(BaseModel):
         if order_size < slot_size:
             # compute a fractional slots size to display
             slots_used = self.slots_used(
-                Position(symbol=sym, size=order_size, avg_price=price)
+                Position(
+                    symbol=sym,
+                    size=order_size,
+                    be_price=price,
+                    bsuid=sym,
+                )
             )
 
         return {
@@ -271,8 +233,8 @@ class Allocator(BaseModel):
         abs_pp_size = abs(pp.size)
 
         if self.size_unit == 'currency':
-            # live_currency_size = size or (abs_pp_size * pp.avg_price)
+            # live_currency_size = size or (abs_pp_size * pp.be_price)
-            live_currency_size = abs_pp_size * pp.avg_price
+            live_currency_size = abs_pp_size * pp.be_price
             prop = live_currency_size / self.currency_limit
 
         else:
@@ -342,7 +304,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.avg_price
+        startup_size = startup_pp.size * startup_pp.be_price
 
         if startup_size > alloc.currency_limit:
             alloc.currency_limit = round(startup_size, ndigits=2)

piker/clearing/_ems.py

@@ -20,6 +20,7 @@ In da suit parlances: "Execution management systems"
 """
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
+from math import isnan
 from pprint import pformat
 import time
 from typing import AsyncIterator, Callable
@@ -289,7 +290,11 @@ class TradesRelay:
     brokerd_dialogue: tractor.MsgStream
 
     # map of symbols to dicts of accounts to pp msgs
-    positions: dict[str, dict[str, BrokerdPosition]]
+    positions: dict[
+        # brokername, acctid
+        tuple[str, str],
+        list[BrokerdPosition],
+    ]
 
     # allowed account names
     accounts: tuple[str]
@@ -461,18 +466,24 @@ async def open_brokerd_trades_dialogue(
                 # normalizing them to EMS messages and relaying back to
                 # the piker order client set.
 
-                # locally cache and track positions per account.
+                # locally cache and track positions per account with
+                # a table of (brokername, acctid) -> `BrokerdPosition`
+                # msgs.
                 pps = {}
                 for msg in positions:
                     log.info(f'loading pp: {msg}')
 
                     account = msg['account']
+
+                    # TODO: better value error for this which
+                    # dumps the account and message and states the
+                    # mismatch..
                     assert account in accounts
 
                     pps.setdefault(
-                        f'{msg["symbol"]}.{broker}',
+                        (broker, account),
-                        {}
+                        [],
-                    )[account] = msg
+                    ).append(msg)
 
                 relay = TradesRelay(
                     brokerd_dialogue=brokerd_trades_stream,
@@ -578,11 +589,9 @@ async def translate_and_relay_brokerd_events(
 
             relay.positions.setdefault(
                 # NOTE: translate to a FQSN!
-                f'{sym}.{broker}',
+                (broker, sym),
-                {}
+                []
-            ).setdefault(
+            ).append(pos_msg)
-                pos_msg['account'], {}
-            ).update(pos_msg)
 
             # fan-out-relay position msgs immediately by
             # broadcasting updates on all client streams
@@ -635,8 +644,8 @@ async def translate_and_relay_brokerd_events(
                 # something is out of order, we don't have an oid for
                 # this broker-side message.
                 log.error(
-                    'Unknown oid:{oid} for msg:\n'
+                    f'Unknown oid: {oid} for msg:\n'
-                    f'{pformat(brokerd_msg)}'
+                    f'{pformat(brokerd_msg)}\n'
                     'Unable to relay message to client side!?'
                 )
 
@@ -943,6 +952,12 @@ async def process_client_order_cmds(
                 # like every other shitty tina platform that makes
                 # the user choose the predicate operator.
                 last = dark_book.lasts[fqsn]
+
+                # sometimes the real-time feed hasn't come up
+                # so just pull from the latest history.
+                if isnan(last):
+                    last = feed.shm.array[-1]['close']
+
                 pred = mk_check(trigger_price, last, action)
 
                 spread_slap: float = 5
@@ -1088,15 +1103,12 @@ async def _emsd_main(
 
             brokerd_stream = relay.brokerd_dialogue  # .clone()
 
-            # flatten out collected pps from brokerd for delivery
-            pp_msgs = {
-                fqsn: list(pps.values())
-                for fqsn, pps in relay.positions.items()
-            }
-
             # signal to client that we're started and deliver
             # all known pps and accounts for this ``brokerd``.
-            await ems_ctx.started((pp_msgs, list(relay.accounts)))
+            await ems_ctx.started((
+                relay.positions,
+                list(relay.accounts),
+            ))
 
             # establish 2-way stream with requesting order-client and
             # begin handling inbound order requests and updates
@@ -1133,8 +1145,14 @@ async def _emsd_main(
                     )
 
                 finally:
-                    # remove client from "registry"
+                    # try to remove client from "registry"
+                    try:
                         _router.clients.remove(ems_client_order_stream)
+                    except KeyError:
+                        log.warning(
+                            f'Stream {ems_client_order_stream._ctx.chan.uid}'
+                            ' was already dropped?'
+                        )
 
                     dialogues = _router.dialogues
 

piker/clearing/_messages.py

@@ -258,6 +258,6 @@ class BrokerdPosition(BaseModel):
     broker: str
     account: str
     symbol: str
-    currency: str
     size: float
     avg_price: float
+    currency: str = ''

piker/clearing/_paper_engine.py

@@ -31,6 +31,8 @@ import tractor
 from dataclasses import dataclass
 
 from .. import data
+from ..data._source import Symbol
+from ..pp import Position
 from ..data._normalize import iterticks
 from ..data._source import unpack_fqsn
 from ..log import get_logger
@@ -257,29 +259,14 @@ class PaperBoi:
             )
         )
 
-        # "avg position price" calcs
+        # delegate update to `.pp.Position.lifo_update()`
-        # TODO: eventually it'd be nice to have a small set of routines
+        pp = Position(
-        # to do this stuff from a sequence of cleared orders to enable
+            Symbol(key=symbol),
-        # so called "contextual positions".
+            size=pp_msg.size,
-        new_size = size + pp_msg.size
+            be_price=pp_msg.avg_price,
-
+            bsuid=symbol,
-        # old size minus the new size gives us size differential with
+        )
-        # +ve -> increase in pp size
+        pp_msg.size, pp_msg.avg_price = pp.lifo_update(size, price)
-        # -ve -> decrease in pp size
-        size_diff = abs(new_size) - abs(pp_msg.size)
-
-        if new_size == 0:
-            pp_msg.avg_price = 0
-
-        elif size_diff > 0:
-            # only update the "average position price" when the position
-            # size increases not when it decreases (i.e. the position is
-            # being made smaller)
-            pp_msg.avg_price = (
-                abs(size) * price + pp_msg.avg_price * abs(pp_msg.size)
-            ) / abs(new_size)
-
-        pp_msg.size = new_size
 
         await self.ems_trades_stream.send(pp_msg.dict())
 
@@ -390,7 +377,8 @@ async def handle_order_requests(
             account = request_msg['account']
             if account != 'paper':
                 log.error(
-                    'This is a paper account, only a `paper` selection is valid'
+                    'This is a paper account,'
+                    ' only a `paper` selection is valid'
                 )
                 await ems_order_stream.send(BrokerdError(
                     oid=request_msg['oid'],
@@ -464,7 +452,7 @@ async def trades_dialogue(
         # TODO: load paper positions per broker from .toml config file
         # and pass as symbol to position data mapping: ``dict[str, dict]``
         # await ctx.started(all_positions)
-        await ctx.started(({}, {'paper',}))
+        await ctx.started(({}, ['paper']))
 
         async with (
             ctx.open_stream() as ems_stream,

piker/cli/__init__.py

@@ -83,9 +83,9 @@ def pikerd(loglevel, host, tl, pdb, tsdb):
 
                 )
                 log.info(
-                    f'`marketstore` up!\n'
+                    f'`marketstored` up!\n'
-                    f'`marketstored` pid: {pid}\n'
+                    f'pid: {pid}\n'
-                    f'docker container id: {cid}\n'
+                    f'container id: {cid[:12]}\n'
                     f'config: {pformat(config)}'
                 )
 

piker/config.py

@@ -21,6 +21,7 @@ Broker configuration mgmt.
 import platform
 import sys
 import os
+from os import path
 from os.path import dirname
 import shutil
 from typing import Optional
@@ -111,6 +112,7 @@ if _parent_user:
 
 _conf_names: set[str] = {
     'brokers',
+    'pps',
     'trades',
     'watchlists',
 }
@@ -147,19 +149,21 @@ def get_conf_path(
     conf_name: str = 'brokers',
 
 ) -> str:
-    """Return the default config path normally under
+    '''
-    ``~/.config/piker`` on linux.
+    Return the top-level default config path normally under
+    ``~/.config/piker`` on linux for a given ``conf_name``, the config
+    name.
 
     Contains files such as:
     - brokers.toml
+    - pp.toml
     - watchlists.toml
-    - trades.toml
 
     # maybe coming soon ;)
     - signals.toml
     - strats.toml
 
-    """
+    '''
     assert conf_name in _conf_names
     fn = _conf_fn_w_ext(conf_name)
     return os.path.join(
@@ -173,7 +177,7 @@ def repodir():
     Return the abspath to the repo directory.
 
     '''
-    dirpath = os.path.abspath(
+    dirpath = path.abspath(
         # we're 3 levels down in **this** module file
         dirname(dirname(os.path.realpath(__file__)))
     )
@@ -182,7 +186,9 @@ def repodir():
 
 def load(
     conf_name: str = 'brokers',
-    path: str = None
+    path: str = None,
+
+    **tomlkws,
 
 ) -> (dict, str):
     '''
@@ -190,6 +196,7 @@ def load(
 
     '''
     path = path or get_conf_path(conf_name)
+
     if not os.path.isfile(path):
         fn = _conf_fn_w_ext(conf_name)
 
@@ -202,8 +209,11 @@ def load(
         # if one exists.
         if os.path.isfile(template):
             shutil.copyfile(template, path)
+        else:
+            with open(path, 'w'):
+                pass  # touch
 
-    config = toml.load(path)
+    config = toml.load(path, **tomlkws)
     log.debug(f"Read config file {path}")
     return config, path
 
@@ -212,6 +222,7 @@ def write(
     config: dict,  # toml config as dict
     name: str = 'brokers',
     path: str = None,
+    **toml_kwargs,
 
 ) -> None:
     ''''
@@ -235,11 +246,14 @@ def write(
         f"{path}"
     )
     with open(path, 'w') as cf:
-        return toml.dump(config, cf)
+        return toml.dump(
+            config,
+            cf,
+            **toml_kwargs,
+        )
 
 
 def load_accounts(
-
     providers: Optional[list[str]] = None
 
 ) -> bidict[str, Optional[str]]:

piker/data/_source.py

@@ -23,7 +23,7 @@ import decimal
 
 from bidict import bidict
 import numpy as np
-from pydantic import BaseModel
+from msgspec import Struct
 # from numba import from_dtype
 
 
@@ -126,7 +126,7 @@ def unpack_fqsn(fqsn: str) -> tuple[str, str, str]:
     )
 
 
-class Symbol(BaseModel):
+class Symbol(Struct):
     '''
     I guess this is some kinda container thing for dealing with
     all the different meta-data formats from brokers?
@@ -152,9 +152,7 @@ class Symbol(BaseModel):
         info: dict[str, Any],
         suffix: str = '',
 
-    # XXX: like wtf..
+    ) -> Symbol:
-    # ) -> 'Symbol':
-    ) -> None:
 
         tick_size = info.get('price_tick_size', 0.01)
         lot_tick_size = info.get('lot_tick_size', 0.0)
@@ -175,9 +173,7 @@ class Symbol(BaseModel):
         fqsn: str,
         info: dict[str, Any],
 
-    # XXX: like wtf..
+    ) -> Symbol:
-    # ) -> 'Symbol':
-    ) -> None:
         broker, key, suffix = unpack_fqsn(fqsn)
         return cls.from_broker_info(
             broker,
@@ -240,7 +236,7 @@ class Symbol(BaseModel):
 
         '''
         tokens = self.tokens()
-        fqsn = '.'.join(tokens)
+        fqsn = '.'.join(map(str.lower, tokens))
         return fqsn
 
     def iterfqsns(self) -> list[str]:

piker/data/_web_bs.py

@@ -53,13 +53,11 @@ class NoBsWs:
     def __init__(
         self,
         url: str,
-        token: str,
         stack: AsyncExitStack,
         fixture: Callable,
         serializer: ModuleType = json,
     ):
         self.url = url
-        self.token = token
         self.fixture = fixture
         self._stack = stack
         self._ws: 'WebSocketConnection' = None  # noqa
@@ -83,14 +81,9 @@ class NoBsWs:
                     trio_websocket.open_websocket_url(self.url)
                 )
                 # rerun user code fixture
-                if self.token == '':
                 ret = await self._stack.enter_async_context(
                     self.fixture(self)
                 )
-                else:
-                    ret = await self._stack.enter_async_context(
-                        self.fixture(self, self.token)
-                    )
 
                 assert ret is None
 
@@ -135,14 +128,13 @@ async def open_autorecon_ws(
 
     # TODO: proper type annot smh
     fixture: Callable,
-    # used for authenticated websockets
+
-    token: str = '',
 ) -> AsyncGenerator[tuple[...],  NoBsWs]:
     """Apparently we can QoS for all sorts of reasons..so catch em.
 
     """
     async with AsyncExitStack() as stack:
-        ws = NoBsWs(url, token, stack, fixture=fixture)
+        ws = NoBsWs(url, stack, fixture=fixture)
         await ws._connect()
 
         try:

piker/fsp/_engine.py

@@ -114,7 +114,7 @@ async def fsp_compute(
         dict[str, np.ndarray],  # multi-output case
         np.ndarray,  # single output case
     ]
-    history_output = await out_stream.__anext__()
+    history_output = await anext(out_stream)
 
     func_name = func.__name__
     profiler(f'{func_name} generated history')
@@ -374,7 +374,8 @@ async def cascade(
                             'key': dst_shm_token,
                             'first': dst._first.value,
                             'last': dst._last.value,
-                    }})
+                        }
+                    })
                     return tracker, index
 
                 def is_synced(

piker/pp.py

@@ -0,0 +1,788 @@
+# piker: trading gear for hackers
+# Copyright (C) Tyler Goodlet (in stewardship for pikers)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+'''
+Personal/Private position parsing, calculating, summarizing in a way
+that doesn't try to cuk most humans who prefer to not lose their moneys..
+(looking at you `ib` and dirt-bird friends)
+
+'''
+from collections import deque
+from contextlib import contextmanager as cm
+# from pprint import pformat
+import os
+from os import path
+from math import copysign
+import re
+import time
+from typing import (
+    Any,
+    Optional,
+    Union,
+)
+
+from msgspec import Struct
+import pendulum
+from pendulum import datetime, now
+import tomli
+import toml
+
+from . import config
+from .brokers import get_brokermod
+from .clearing._messages import BrokerdPosition, Status
+from .data._source import Symbol
+from .log import get_logger
+
+log = get_logger(__name__)
+
+
+@cm
+def open_trade_ledger(
+    broker: str,
+    account: str,
+
+) -> str:
+    '''
+    Indempotently create and read in a trade log file from the
+    ``<configuration_dir>/ledgers/`` directory.
+
+    Files are named per broker account of the form
+    ``<brokername>_<accountname>.toml``. The ``accountname`` here is the
+    name as defined in the user's ``brokers.toml`` config.
+
+    '''
+    ldir = path.join(config._config_dir, 'ledgers')
+    if not path.isdir(ldir):
+        os.makedirs(ldir)
+
+    fname = f'trades_{broker}_{account}.toml'
+    tradesfile = path.join(ldir, fname)
+
+    if not path.isfile(tradesfile):
+        log.info(
+            f'Creating new local trades ledger: {tradesfile}'
+        )
+        with open(tradesfile, 'w') as cf:
+            pass  # touch
+    with open(tradesfile, 'rb') as cf:
+        start = time.time()
+        ledger = tomli.load(cf)
+        print(f'Ledger load took {time.time() - start}s')
+        cpy = ledger.copy()
+    try:
+        yield cpy
+    finally:
+        if cpy != ledger:
+            # TODO: show diff output?
+            # https://stackoverflow.com/questions/12956957/print-diff-of-python-dictionaries
+            print(f'Updating ledger for {tradesfile}:\n')
+            ledger.update(cpy)
+
+            # we write on close the mutated ledger data
+            with open(tradesfile, 'w') as cf:
+                return toml.dump(ledger, cf)
+
+
+class Transaction(Struct):
+    # TODO: should this be ``.to`` (see below)?
+    fqsn: str
+
+    tid: Union[str, int]  # unique transaction id
+    size: float
+    price: float
+    cost: float  # commisions or other additional costs
+    dt: datetime
+    expiry: Optional[datetime] = None
+
+    # optional key normally derived from the broker
+    # backend which ensures the instrument-symbol this record
+    # is for is truly unique.
+    bsuid: Optional[Union[str, int]] = None
+
+    # optional fqsn for the source "asset"/money symbol?
+    # from: Optional[str] = None
+
+
+class Position(Struct):
+    '''
+    Basic pp (personal/piker position) model with attached clearing
+    transaction history.
+
+    '''
+    symbol: Symbol
+
+    # can be +ve or -ve for long/short
+    size: float
+
+    # "breakeven price" above or below which pnl moves above and below
+    # zero for the entirety of the current "trade state".
+    be_price: float
+
+    # unique backend symbol id
+    bsuid: str
+
+    # ordered record of known constituent trade messages
+    clears: dict[
+        Union[str, int, Status],  # trade id
+        dict[str, Any],  # transaction history summaries
+    ] = {}
+
+    expiry: Optional[datetime] = None
+
+    def to_dict(self) -> dict:
+        return {
+            f: getattr(self, f)
+            for f in self.__struct_fields__
+        }
+
+    def to_pretoml(self) -> dict:
+        '''
+        Prep this position's data contents for export to toml including
+        re-structuring of the ``.clears`` table to an array of
+        inline-subtables for better ``pps.toml`` compactness.
+
+        '''
+        d = self.to_dict()
+        clears = d.pop('clears')
+        expiry = d.pop('expiry')
+
+        if expiry:
+            d['expiry'] = str(expiry)
+
+        clears_list = []
+
+        for tid, data in clears.items():
+            inline_table = toml.TomlDecoder().get_empty_inline_table()
+            inline_table['tid'] = tid
+
+            for k, v in data.items():
+                inline_table[k] = v
+
+            clears_list.append(inline_table)
+
+        d['clears'] = clears_list
+
+        return d
+
+    def update_from_msg(
+        self,
+        msg: BrokerdPosition,
+
+    ) -> None:
+
+        # XXX: better place to do this?
+        symbol = self.symbol
+
+        lot_size_digits = symbol.lot_size_digits
+        be_price, size = (
+            round(
+                msg['avg_price'],
+                ndigits=symbol.tick_size_digits
+            ),
+            round(
+                msg['size'],
+                ndigits=lot_size_digits
+            ),
+        )
+
+        self.be_price = be_price
+        self.size = size
+
+    @property
+    def dsize(self) -> float:
+        '''
+        The "dollar" size of the pp, normally in trading (fiat) unit
+        terms.
+
+        '''
+        return self.be_price * self.size
+
+    def update(
+        self,
+        t: Transaction,
+
+    ) -> None:
+        self.clears[t.tid] = {
+            'cost': t.cost,
+            'price': t.price,
+            'size': t.size,
+            'dt': str(t.dt),
+        }
+
+    def lifo_update(
+        self,
+        size: float,
+        price: float,
+        cost: float = 0,
+
+        # TODO: idea: "real LIFO" dynamic positioning.
+        # - when a trade takes place where the pnl for
+        # the (set of) trade(s) is below the breakeven price
+        # it may be that the trader took a +ve pnl on a short(er)
+        # term trade in the same account.
+        # - in this case we could recalc the be price to
+        # be reverted back to it's prior value before the nearest term
+        # trade was opened.?
+        # dynamic_breakeven_price: bool = False,
+
+    ) -> (float, float):
+        '''
+        Incremental update using a LIFO-style weighted mean.
+
+        '''
+        # "avg position price" calcs
+        # TODO: eventually it'd be nice to have a small set of routines
+        # to do this stuff from a sequence of cleared orders to enable
+        # so called "contextual positions".
+        new_size = self.size + size
+
+        # old size minus the new size gives us size diff with
+        # +ve -> increase in pp size
+        # -ve -> decrease in pp size
+        size_diff = abs(new_size) - abs(self.size)
+
+        if new_size == 0:
+            self.be_price = 0
+
+        elif size_diff > 0:
+            # XXX: LOFI incremental update:
+            # only update the "average price" when
+            # the size increases not when it decreases (i.e. the
+            # position is being made smaller)
+            self.be_price = (
+                # weight of current exec = (size * price) + cost
+                (abs(size) * price)
+                +
+                (copysign(1, new_size) * cost)  # transaction cost
+                +
+                # weight of existing be price
+                self.be_price * abs(self.size)  # weight of previous pp
+            ) / abs(new_size)  # normalized by the new size: weighted mean.
+
+        self.size = new_size
+
+        return new_size, self.be_price
+
+    def minimize_clears(
+        self,
+
+    ) -> dict[str, dict]:
+        '''
+        Minimize the position's clears entries by removing
+        all transactions before the last net zero size to avoid
+        unecessary history irrelevant to the current pp state.
+
+        '''
+        size: float = self.size
+        clears_since_zero: deque[tuple(str, dict)] = deque()
+
+        # scan for the last "net zero" position by
+        # iterating clears in reverse.
+        for tid, clear in reversed(self.clears.items()):
+            size -= clear['size']
+            clears_since_zero.appendleft((tid, clear))
+
+            if size == 0:
+                break
+
+        self.clears = dict(clears_since_zero)
+        return self.clears
+
+
+def update_pps(
+    records: dict[str, Transaction],
+    pps: Optional[dict[str, Position]] = None
+
+) -> dict[str, Position]:
+    '''
+    Compile a set of positions from a trades ledger.
+
+    '''
+    pps: dict[str, Position] = pps or {}
+
+    # lifo update all pps from records
+    for r in records:
+
+        pp = pps.setdefault(
+            r.bsuid,
+
+            # if no existing pp, allocate fresh one.
+            Position(
+                Symbol.from_fqsn(
+                    r.fqsn,
+                    info={},
+                ),
+                size=0.0,
+                be_price=0.0,
+                bsuid=r.bsuid,
+                expiry=r.expiry,
+            )
+        )
+
+        # don't do updates for ledger records we already have
+        # included in the current pps state.
+        if r.tid in pp.clears:
+            # NOTE: likely you'll see repeats of the same
+            # ``Transaction`` passed in here if/when you are restarting
+            # a ``brokerd.ib`` where the API will re-report trades from
+            # the current session, so we need to make sure we don't
+            # "double count" these in pp calculations.
+            continue
+
+        # lifo style "breakeven" price calc
+        pp.lifo_update(
+            r.size,
+            r.price,
+
+            # include transaction cost in breakeven price
+            # and presume the worst case of the same cost
+            # to exit this transaction (even though in reality
+            # it will be dynamic based on exit stratetgy).
+            cost=2*r.cost,
+        )
+
+        # track clearing data
+        pp.update(r)
+
+    return pps
+
+
+def load_pps_from_ledger(
+
+    brokername: str,
+    acctname: str,
+
+    # post normalization filter on ledger entries to be processed
+    filter_by: Optional[list[dict]] = None,
+
+) -> dict[str, Position]:
+    '''
+    Open a ledger file by broker name and account and read in and
+    process any trade records into our normalized ``Transaction``
+    form and then pass these into the position processing routine
+    and deliver the two dict-sets of the active and closed pps.
+
+    '''
+    with open_trade_ledger(
+        brokername,
+        acctname,
+    ) as ledger:
+        if not ledger:
+            # null case, no ledger file with content
+            return {}
+
+    brokermod = get_brokermod(brokername)
+    src_records = brokermod.norm_trade_records(ledger)
+
+    if filter_by:
+        bsuids = set(filter_by)
+        records = list(filter(lambda r: r.bsuid in bsuids, src_records))
+    else:
+        records = src_records
+
+    return update_pps(records)
+
+
+def get_pps(
+    brokername: str,
+    acctids: Optional[set[str]] = set(),
+
+) -> dict[str, dict[str, Position]]:
+    '''
+    Read out broker-specific position entries from
+    incremental update file: ``pps.toml``.
+
+    '''
+    conf, path = config.load(
+        'pps',
+        # load dicts as inlines to preserve compactness
+        # _dict=toml.decoder.InlineTableDict,
+    )
+
+    all_active = {}
+    all_closed = {}
+
+    # try to load any ledgers if no section found
+    bconf, path = config.load('brokers')
+    accounts = bconf[brokername]['accounts']
+    for account in accounts:
+
+        # TODO: instead of this filter we could
+        # always send all known pps but just not audit
+        # them since an active client might not be up?
+        if (
+            acctids and
+            f'{brokername}.{account}' not in acctids
+        ):
+            continue
+
+        active, closed = update_pps_conf(brokername, account)
+        all_active.setdefault(account, {}).update(active)
+        all_closed.setdefault(account, {}).update(closed)
+
+    return all_active, all_closed
+
+
+# TODO: instead see if we can hack tomli and tomli-w to do the same:
+# - https://github.com/hukkin/tomli
+# - https://github.com/hukkin/tomli-w
+class PpsEncoder(toml.TomlEncoder):
+    '''
+    Special "styled" encoder that makes a ``pps.toml`` redable and
+    compact by putting `.clears` tables inline and everything else
+    flat-ish.
+
+    '''
+    separator = ','
+
+    def dump_list(self, v):
+        '''
+        Dump an inline list with a newline after every element and
+        with consideration for denoted inline table types.
+
+        '''
+        retval = "[\n"
+        for u in v:
+            if isinstance(u, toml.decoder.InlineTableDict):
+                out = self.dump_inline_table(u)
+            else:
+                out = str(self.dump_value(u))
+
+            retval += " " + out + "," + "\n"
+        retval += "]"
+        return retval
+
+    def dump_inline_table(self, section):
+        """Preserve inline table in its compact syntax instead of expanding
+        into subsection.
+        https://github.com/toml-lang/toml#user-content-inline-table
+        """
+        val_list = []
+        for k, v in section.items():
+            # if isinstance(v, toml.decoder.InlineTableDict):
+            if isinstance(v, dict):
+                val = self.dump_inline_table(v)
+            else:
+                val = str(self.dump_value(v))
+
+            val_list.append(k + " = " + val)
+
+        retval = "{ " + ", ".join(val_list) + " }"
+        return retval
+
+    def dump_sections(self, o, sup):
+        retstr = ""
+        if sup != "" and sup[-1] != ".":
+            sup += '.'
+        retdict = self._dict()
+        arraystr = ""
+        for section in o:
+            qsection = str(section)
+            value = o[section]
+
+            if not re.match(r'^[A-Za-z0-9_-]+$', section):
+                qsection = toml.encoder._dump_str(section)
+
+            # arrayoftables = False
+            if (
+                self.preserve
+                and isinstance(value, toml.decoder.InlineTableDict)
+            ):
+                retstr += (
+                    qsection
+                    +
+                    " = "
+                    +
+                    self.dump_inline_table(o[section])
+                    +
+                    '\n'  # only on the final terminating left brace
+                )
+
+            # XXX: this code i'm pretty sure is just blatantly bad
+            # and/or wrong..
+            # if isinstance(o[section], list):
+            #     for a in o[section]:
+            #         if isinstance(a, dict):
+            #             arrayoftables = True
+            # if arrayoftables:
+            #     for a in o[section]:
+            #         arraytabstr = "\n"
+            #         arraystr += "[[" + sup + qsection + "]]\n"
+            #         s, d = self.dump_sections(a, sup + qsection)
+            #         if s:
+            #             if s[0] == "[":
+            #                 arraytabstr += s
+            #             else:
+            #                 arraystr += s
+            #         while d:
+            #             newd = self._dict()
+            #             for dsec in d:
+            #                 s1, d1 = self.dump_sections(d[dsec], sup +
+            #                                             qsection + "." +
+            #                                             dsec)
+            #                 if s1:
+            #                     arraytabstr += ("[" + sup + qsection +
+            #                                     "." + dsec + "]\n")
+            #                     arraytabstr += s1
+            #                 for s1 in d1:
+            #                     newd[dsec + "." + s1] = d1[s1]
+            #             d = newd
+            #         arraystr += arraytabstr
+
+            elif isinstance(value, dict):
+                retdict[qsection] = o[section]
+
+            elif o[section] is not None:
+                retstr += (
+                    qsection
+                    +
+                    " = "
+                    +
+                    str(self.dump_value(o[section]))
+                )
+
+                # if not isinstance(value, dict):
+                if not isinstance(value, toml.decoder.InlineTableDict):
+                    # inline tables should not contain newlines:
+                    # https://toml.io/en/v1.0.0#inline-table
+                    retstr += '\n'
+
+            else:
+                raise ValueError(value)
+
+        retstr += arraystr
+        return (retstr, retdict)
+
+
+def load_pps_from_toml(
+    brokername: str,
+    acctid: str,
+
+    # XXX: there is an edge case here where we may want to either audit
+    # the retrieved ``pps.toml`` output or reprocess it since there was
+    # an error on write on the last attempt to update the state file
+    # even though the ledger *was* updated. For this cases we allow the
+    # caller to pass in a symbol set they'd like to reload from the
+    # underlying ledger to be reprocessed in computing pps state.
+    reload_records: Optional[dict[str, str]] = None,
+    update_from_ledger: bool = False,
+
+) -> tuple[dict, dict[str, Position]]:
+    '''
+    Load and marshal to objects all pps from either an existing
+    ``pps.toml`` config, or from scratch from a ledger file when
+    none yet exists.
+
+    '''
+    conf, path = config.load('pps')
+    brokersection = conf.setdefault(brokername, {})
+    pps = brokersection.setdefault(acctid, {})
+    pp_objs = {}
+
+    # no pps entry yet for this broker/account so parse any available
+    # ledgers to build a brand new pps state.
+    if not pps or update_from_ledger:
+        pp_objs = load_pps_from_ledger(
+            brokername,
+            acctid,
+        )
+
+    # Reload symbol specific ledger entries if requested by the
+    # caller **AND** none exist in the current pps state table.
+    elif (
+        pps and reload_records
+    ):
+        # no pps entry yet for this broker/account so parse
+        # any available ledgers to build a pps state.
+        pp_objs = load_pps_from_ledger(
+            brokername,
+            acctid,
+            filter_by=reload_records,
+        )
+
+    if not pps:
+        log.warning(
+            f'No `pps.toml` positions could be loaded {brokername}:{acctid}'
+        )
+
+    # unmarshal/load ``pps.toml`` config entries into object form.
+    for fqsn, entry in pps.items():
+        bsuid = entry['bsuid']
+
+        # convert clears sub-tables (only in this form
+        # for toml re-presentation) back into a master table.
+        clears_list = entry['clears']
+
+        # index clears entries in "object" form by tid in a top
+        # level dict instead of a list (as is presented in our
+        # ``pps.toml``).
+        pp = pp_objs.get(bsuid)
+        if pp:
+            clears = pp.clears
+        else:
+            clears = {}
+
+        for clears_table in clears_list:
+            tid = clears_table.pop('tid')
+            clears[tid] = clears_table
+
+        size = entry['size']
+
+        # TODO: an audit system for existing pps entries?
+        # if not len(clears) == abs(size):
+        #     pp_objs = load_pps_from_ledger(
+        #         brokername,
+        #         acctid,
+        #         filter_by=reload_records,
+        #     )
+        #     reason = 'size <-> len(clears) mismatch'
+        #     raise ValueError(
+        #         '`pps.toml` entry is invalid:\n'
+        #         f'{fqsn}\n'
+        #         f'{pformat(entry)}'
+        #     )
+
+        expiry = entry.get('expiry')
+        if expiry:
+            expiry = pendulum.parse(expiry)
+
+        pp_objs[bsuid] = Position(
+            Symbol.from_fqsn(fqsn, info={}),
+            size=size,
+            be_price=entry['be_price'],
+            expiry=expiry,
+            bsuid=entry['bsuid'],
+
+            # XXX: super critical, we need to be sure to include
+            # all pps.toml clears to avoid reusing clears that were
+            # already included in the current incremental update
+            # state, since today's records may have already been
+            # processed!
+            clears=clears,
+        )
+
+    return conf, pp_objs
+
+
+def update_pps_conf(
+    brokername: str,
+    acctid: str,
+
+    trade_records: Optional[list[Transaction]] = None,
+    ledger_reload: Optional[dict[str, str]] = None,
+
+) -> tuple[
+    dict[str, Position],
+    dict[str, Position],
+]:
+
+    # this maps `.bsuid` values to positions
+    pp_objs: dict[Union[str, int], Position]
+
+    if trade_records and ledger_reload:
+        for r in trade_records:
+            ledger_reload[r.bsuid] = r.fqsn
+
+    conf, pp_objs = load_pps_from_toml(
+        brokername,
+        acctid,
+        reload_records=ledger_reload,
+    )
+
+    # update all pp objects from any (new) trade records which
+    # were passed in (aka incremental update case).
+    if trade_records:
+        pp_objs = update_pps(
+            trade_records,
+            pps=pp_objs,
+        )
+
+    pp_entries = {}  # dict-serialize all active pps
+    # NOTE: newly closed position are also important to report/return
+    # since a consumer, like an order mode UI ;), might want to react
+    # based on the closure.
+    closed_pp_objs: dict[str, Position] = {}
+
+    for bsuid in list(pp_objs):
+        pp = pp_objs[bsuid]
+
+        # XXX: debug hook for size mismatches
+        # if bsuid == 447767096:
+        #     breakpoint()
+
+        pp.minimize_clears()
+
+        if (
+            pp.size == 0
+
+            # drop time-expired positions (normally derivatives)
+            or (pp.expiry and pp.expiry < now())
+        ):
+            # if expired the position is closed
+            pp.size = 0
+
+            # position is already closed aka "net zero"
+            closed_pp = pp_objs.pop(bsuid, None)
+            if closed_pp:
+                closed_pp_objs[bsuid] = closed_pp
+
+        else:
+            # serialize to pre-toml form
+            asdict = pp.to_pretoml()
+
+            if pp.expiry is None:
+                asdict.pop('expiry', None)
+
+            # TODO: we need to figure out how to have one top level
+            # listing venue here even when the backend isn't providing
+            # it via the trades ledger..
+            # drop symbol obj in serialized form
+            s = asdict.pop('symbol')
+            fqsn = s.front_fqsn()
+            log.info(f'Updating active pp: {fqsn}')
+
+            # XXX: ugh, it's cuz we push the section under
+            # the broker name.. maybe we need to rethink this?
+            brokerless_key = fqsn.removeprefix(f'{brokername}.')
+
+            pp_entries[brokerless_key] = asdict
+
+    conf[brokername][acctid] = pp_entries
+
+    # TODO: why tf haven't they already done this for inline tables smh..
+    enc = PpsEncoder(preserve=True)
+    # table_bs_type = type(toml.TomlDecoder().get_empty_inline_table())
+    enc.dump_funcs[toml.decoder.InlineTableDict] = enc.dump_inline_table
+
+    config.write(
+        conf,
+        'pps',
+        encoder=enc,
+    )
+
+    # deliver object form of all pps in table to caller
+    return pp_objs, closed_pp_objs
+
+
+if __name__ == '__main__':
+    import sys
+
+    args = sys.argv
+    assert len(args) > 1, 'Specifiy account(s) from `brokers.toml`'
+    args = args[1:]
+    for acctid in args:
+        broker, name = acctid.split('.')
+        update_pps_conf(broker, name)

piker/ui/_chart.py

@@ -230,18 +230,19 @@ class GodWidget(QWidget):
             # - we'll probably want per-instrument/provider state here?
             #   change the order config form over to the new chart
 
-            # XXX: since the pp config is a singleton widget we have to
-            # also switch it over to the new chart's interal-layout
-            # self.linkedsplits.chart.qframe.hbox.removeWidget(self.pp_pane)
-            chart = linkedsplits.chart
-
             # chart is already in memory so just focus it
             linkedsplits.show()
             linkedsplits.focus()
             linkedsplits.graphics_cycle()
             await trio.sleep(0)
 
+            # XXX: since the pp config is a singleton widget we have to
+            # also switch it over to the new chart's interal-layout
+            # self.linkedsplits.chart.qframe.hbox.removeWidget(self.pp_pane)
+            chart = linkedsplits.chart
+
             # resume feeds *after* rendering chart view asap
+            if chart:
                 chart.resume_all_feeds()
 
                 # TODO: we need a check to see if the chart
@@ -760,9 +761,18 @@ class ChartPlotWidget(pg.PlotWidget):
 
         self.pi_overlay: PlotItemOverlay = PlotItemOverlay(self.plotItem)
 
+        # indempotent startup flag for auto-yrange subsys
+        # to detect the "first time" y-domain graphics begin
+        # to be shown in the (main) graphics view.
+        self._on_screen: bool = False
+
     def resume_all_feeds(self):
+        try:
             for feed in self._feeds.values():
                 self.linked.godwidget._root_n.start_soon(feed.resume)
+        except RuntimeError:
+            # TODO: cancel the qtractor runtime here?
+            raise
 
     def pause_all_feeds(self):
         for feed in self._feeds.values():
@@ -859,7 +869,8 @@ class ChartPlotWidget(pg.PlotWidget):
 
     def default_view(
         self,
-        bars_from_y: int = 3000,
+        bars_from_y: int = 616,
+        do_ds: bool = True,
 
     ) -> None:
         '''
@@ -920,8 +931,11 @@ class ChartPlotWidget(pg.PlotWidget):
             max=end,
             padding=0,
         )
+
+        if do_ds:
             self.view.maybe_downsample_graphics()
             view._set_yrange()
+
         try:
             self.linked.graphics_cycle()
         except IndexError:
@@ -1255,7 +1269,6 @@ class ChartPlotWidget(pg.PlotWidget):
         If ``bars_range`` is provided use that range.
 
         '''
-        # print(f'Chart[{self.name}].maxmin()')
         profiler = pg.debug.Profiler(
             msg=f'`{str(self)}.maxmin(name={name})`: `{self.name}`',
             disabled=not pg_profile_enabled(),
@@ -1287,11 +1300,18 @@ class ChartPlotWidget(pg.PlotWidget):
 
             key = round(lbar), round(rbar)
             res = flow.maxmin(*key)
-            if res == (None, None):
+
-                log.error(
+            if (
+                res is None
+            ):
+                log.warning(
                     f"{flow_key} no mxmn for bars_range => {key} !?"
                 )
                 res = 0, 0
+                if not self._on_screen:
+                    self.default_view(do_ds=False)
+                    self._on_screen = True
 
         profiler(f'yrange mxmn: {key} -> {res}')
+        # print(f'{flow_key} yrange mxmn: {key} -> {res}')
         return res

piker/ui/_compression.py

@@ -223,14 +223,20 @@ def ds_m4(
     assert frames >= (xrange / uppx)
 
     # call into ``numba``
-    nb, i_win, y_out = _m4(
+    (
+        nb,
+        x_out,
+        y_out,
+        ymn,
+        ymx,
+    ) = _m4(
         x,
         y,
 
         frames,
 
         # TODO: see func below..
-        # i_win,
+        # x_out,
         # y_out,
 
         # first index in x data to start at
@@ -243,10 +249,11 @@ def ds_m4(
     # filter out any overshoot in the input allocation arrays by
     # removing zero-ed tail entries which should start at a certain
     # index.
-    i_win = i_win[i_win != 0]
+    x_out = x_out[x_out != 0]
-    y_out = y_out[:i_win.size]
+    y_out = y_out[:x_out.size]
 
-    return nb, i_win, y_out
+    # print(f'M4 output ymn, ymx: {ymn},{ymx}')
+    return nb, x_out, y_out, ymn, ymx
 
 
 @jit(
@@ -260,8 +267,8 @@ def _m4(
 
     frames: int,
 
-    # TODO: using this approach by having the ``.zeros()`` alloc lines
+    # TODO: using this approach, having the ``.zeros()`` alloc lines
-    # below, in put python was causing segs faults and alloc crashes..
+    # below in pure python, there were segs faults and alloc crashes..
     # we might need to see how it behaves with shm arrays and consider
     # allocating them once at startup?
 
@@ -274,14 +281,22 @@ def _m4(
     x_start: int,
     step: float,
 
-) -> int:
+) -> tuple[
-    # nbins = len(i_win)
+    int,
-    # count = len(xs)
+    np.ndarray,
+    np.ndarray,
+    float,
+    float,
+]:
+    '''
+    Implementation of the m4 algorithm in ``numba``:
+    http://www.vldb.org/pvldb/vol7/p797-jugel.pdf
 
+    '''
     # these are pre-allocated and mutated by ``numba``
     # code in-place.
     y_out = np.zeros((frames, 4), ys.dtype)
-    i_win = np.zeros(frames, xs.dtype)
+    x_out = np.zeros(frames, xs.dtype)
 
     bincount = 0
     x_left = x_start
@@ -295,24 +310,34 @@ def _m4(
 
     # set all bins in the left-most entry to the starting left-most x value
     # (aka a row broadcast).
-    i_win[bincount] = x_left
+    x_out[bincount] = x_left
     # set all y-values to the first value passed in.
     y_out[bincount] = ys[0]
 
+    # full input y-data mx and mn
+    mx: float = -np.inf
+    mn: float = np.inf
+
+    # compute OHLC style max / min values per window sized x-frame.
     for i in range(len(xs)):
+
         x = xs[i]
         y = ys[i]
+
         if x < x_left + step:   # the current window "step" is [bin, bin+1)
-            y_out[bincount, 1] = min(y, y_out[bincount, 1])
+            ymn = y_out[bincount, 1] = min(y, y_out[bincount, 1])
-            y_out[bincount, 2] = max(y, y_out[bincount, 2])
+            ymx = y_out[bincount, 2] = max(y, y_out[bincount, 2])
             y_out[bincount, 3] = y
+            mx = max(mx, ymx)
+            mn = min(mn, ymn)
+
         else:
             # Find the next bin
             while x >= x_left + step:
                 x_left += step
 
             bincount += 1
-            i_win[bincount] = x_left
+            x_out[bincount] = x_left
             y_out[bincount] = y
 
-    return bincount, i_win, y_out
+    return bincount, x_out, y_out, mn, mx

piker/ui/_display.py

@@ -105,6 +105,10 @@ def chart_maxmin(
     mn, mx = out
 
     mx_vlm_in_view = 0
+
+    # TODO: we need to NOT call this to avoid a manual
+    # np.max/min trigger and especially on the vlm_chart
+    # flows which aren't shown.. like vlm?
     if vlm_chart:
         out = vlm_chart.maxmin()
         if out:
@@ -222,33 +226,9 @@ async def graphics_update_loop(
     tick_margin = 3 * tick_size
 
     chart.show()
-    # view = chart.view
     last_quote = time.time()
     i_last = ohlcv.index
 
-    # async def iter_drain_quotes():
-    #     # NOTE: all code below this loop is expected to be synchronous
-    #     # and thus draw instructions are not picked up jntil the next
-    #     # wait / iteration.
-    #     async for quotes in stream:
-    #         while True:
-    #             try:
-    #                 moar = stream.receive_nowait()
-    #             except trio.WouldBlock:
-    #                 yield quotes
-    #                 break
-    #             else:
-    #                 for sym, quote in moar.items():
-    #                     ticks_frame = quote.get('ticks')
-    #                     if ticks_frame:
-    #                         quotes[sym].setdefault(
-    #                             'ticks', []).extend(ticks_frame)
-    #                     print('pulled extra')
-
-    #                 yield quotes
-
-    # async for quotes in iter_drain_quotes():
-
     ds = linked.display_state = DisplayState(**{
         'quotes': {},
         'linked': linked,
@@ -293,6 +273,7 @@ async def graphics_update_loop(
 
         # chart isn't active/shown so skip render cycle and pause feed(s)
         if chart.linked.isHidden():
+            print('skipping update')
             chart.pause_all_feeds()
             continue
 
@@ -416,10 +397,8 @@ def graphics_update_cycle(
             )
             or trigger_all
         ):
-            # TODO: we should track and compute whether the last
-            # pixel in a curve should show new data based on uppx
-            # and then iff update curves and shift?
             chart.increment_view(steps=i_diff)
+            # chart.increment_view(steps=i_diff + round(append_diff - uppx))
 
             if vlm_chart:
                 vlm_chart.increment_view(steps=i_diff)
@@ -477,7 +456,6 @@ def graphics_update_cycle(
         ):
             chart.update_graphics_from_flow(
                 chart.name,
-                # do_append=uppx < update_uppx,
                 do_append=do_append,
             )
 

piker/ui/_flows.py

@@ -337,6 +337,7 @@ class Flow(msgspec.Struct):  # , frozen=True):
     name: str
     plot: pg.PlotItem
     graphics: Union[Curve, BarItems]
+    yrange: tuple[float, float] = None
 
     # in some cases a flow may want to change its
     # graphical "type" or, "form" when downsampling,
@@ -386,10 +387,11 @@ class Flow(msgspec.Struct):  # , frozen=True):
         lbar: int,
         rbar: int,
 
-    ) -> tuple[float, float]:
+    ) -> Optional[tuple[float, float]]:
         '''
         Compute the cached max and min y-range values for a given
-        x-range determined by ``lbar`` and ``rbar``.
+        x-range determined by ``lbar`` and ``rbar`` or ``None``
+        if no range can be determined (yet).
 
         '''
         rkey = (lbar, rbar)
@@ -399,9 +401,8 @@ class Flow(msgspec.Struct):  # , frozen=True):
 
         shm = self.shm
         if shm is None:
-            mxmn = None
+            return None
 
-        else:  # new block for profiling?..
         arr = shm.array
 
         # build relative indexes into shm array
@@ -414,7 +415,11 @@ class Flow(msgspec.Struct):  # , frozen=True):
         ]
 
         if not slice_view.size:
-                mxmn = None
+            return None
+
+        elif self.yrange:
+            mxmn = self.yrange
+            # print(f'{self.name} M4 maxmin: {mxmn}')
 
         else:
             if self.is_ohlc:
@@ -427,9 +432,10 @@ class Flow(msgspec.Struct):  # , frozen=True):
                 yhigh = np.max(view)
 
             mxmn = ylow, yhigh
+            # print(f'{self.name} MANUAL maxmin: {mxmin}')
 
-            if mxmn is not None:
+        # cache result for input range
-                # cache new mxmn result
+        assert mxmn
         self._mxmns[rkey] = mxmn
 
         return mxmn
@@ -628,10 +634,13 @@ class Flow(msgspec.Struct):  # , frozen=True):
             # source data so we clear our path data in prep
             # to generate a new one from original source data.
             new_sample_rate = True
-            showing_src_data = True
             should_ds = False
             should_redraw = True
 
+            showing_src_data = True
+            # reset yrange to be computed from source data
+            self.yrange = None
+
         # MAIN RENDER LOGIC:
         # - determine in view data and redraw on range change
         # - determine downsampling ops if needed
@@ -657,6 +666,10 @@ class Flow(msgspec.Struct):  # , frozen=True):
 
             **rkwargs,
         )
+        if showing_src_data:
+            # print(f"{self.name} SHOWING SOURCE")
+            # reset yrange to be computed from source data
+            self.yrange = None
 
         if not out:
             log.warning(f'{self.name} failed to render!?')
@@ -664,6 +677,9 @@ class Flow(msgspec.Struct):  # , frozen=True):
 
         path, data, reset = out
 
+        # if self.yrange:
+        #     print(f'flow {self.name} yrange from m4: {self.yrange}')
+
         # XXX: SUPER UGGGHHH... without this we get stale cache
         # graphics that don't update until you downsampler again..
         if reset:
@@ -1058,6 +1074,7 @@ class Renderer(msgspec.Struct):
         # xy-path data transform: convert source data to a format
         # able to be passed to a `QPainterPath` rendering routine.
         if not len(hist):
+            # XXX: this might be why the profiler only has exits?
             return
 
         x_out, y_out, connect = self.format_xy(
@@ -1144,11 +1161,14 @@ class Renderer(msgspec.Struct):
 
             elif should_ds and uppx > 1:
 
-                x_out, y_out = xy_downsample(
+                x_out, y_out, ymn, ymx = xy_downsample(
                     x_out,
                     y_out,
                     uppx,
                 )
+                self.flow.yrange = ymn, ymx
+                # print(f'{self.flow.name} post ds: ymn, ymx: {ymn},{ymx}')
+
                 reset = True
                 profiler(f'FULL PATH downsample redraw={should_ds}')
                 self._in_ds = True

piker/ui/_fsp.py

@@ -639,20 +639,25 @@ async def open_vlm_displays(
             names: list[str],
 
         ) -> tuple[float, float]:
+            '''
+            Flows "group" maxmin loop; assumes all named flows
+            are in the same co-domain and thus can be sorted
+            as one set.
 
+            Iterates all the named flows and calls the chart
+            api to find their range values and return.
+
+            TODO: really we should probably have a more built-in API
+            for this?
+
+            '''
             mx = 0
             for name in names:
-
+                ymn, ymx = chart.maxmin(name=name)
-                mxmn = chart.maxmin(name=name)
+                mx = max(mx, ymx)
-                if mxmn:
-                    ymax = mxmn[1]
-                    if ymax > mx:
-                        mx = ymax
 
             return 0, mx
 
-        chart.view.maxmin = partial(multi_maxmin, names=['volume'])
-
         # TODO: fix the x-axis label issue where if you put
         # the axis on the left it's totally not lined up...
         # show volume units value on LHS (for dinkus)
@@ -776,6 +781,7 @@ async def open_vlm_displays(
 
             ) -> None:
                 for name in names:
+
                     if 'dark' in name:
                         color = dark_vlm_color
                     elif 'rate' in name:

piker/ui/_interaction.py

@@ -923,6 +923,7 @@ class ChartView(ViewBox):
                     # XXX: super important to be aware of this.
                     # or not flow.graphics.isVisible()
                 ):
+                    # print(f'skipping {flow.name}')
                     continue
 
                 # pass in no array which will read and render from the last

piker/ui/_pathops.py

@@ -49,12 +49,17 @@ def xy_downsample(
 
     x_spacer: float = 0.5,
 
-) -> tuple[np.ndarray, np.ndarray]:
+) -> tuple[
+    np.ndarray,
+    np.ndarray,
+    float,
+    float,
+]:
 
     # downsample whenever more then 1 pixels per datum can be shown.
     # always refresh data bounds until we get diffing
     # working properly, see above..
-    bins, x, y = ds_m4(
+    bins, x, y, ymn, ymx = ds_m4(
         x,
         y,
         uppx,
@@ -67,7 +72,7 @@ def xy_downsample(
     )).flatten()
     y = y.flatten()
 
-    return x, y
+    return x, y, ymn, ymx
 
 
 @njit(

piker/ui/_position.py

@@ -19,6 +19,7 @@ Position info and display
 
 """
 from __future__ import annotations
+from copy import copy
 from dataclasses import dataclass
 from functools import partial
 from math import floor, copysign
@@ -105,8 +106,8 @@ async def update_pnl_from_feed(
                             # compute and display pnl status
                             order_mode.pane.pnl_label.format(
                                 pnl=copysign(1, size) * pnl(
-                                    # live.avg_price,
+                                    # live.be_price,
-                                    order_mode.current_pp.live_pp.avg_price,
+                                    order_mode.current_pp.live_pp.be_price,
                                     tick['price'],
                                 ),
                             )
@@ -356,7 +357,7 @@ class SettingsPane:
             # last historical close price
             last = feed.shm.array[-1][['close']][0]
             pnl_value = copysign(1, size) * pnl(
-                tracker.live_pp.avg_price,
+                tracker.live_pp.be_price,
                 last,
             )
 
@@ -476,7 +477,7 @@ class PositionTracker:
 
         self.alloc = alloc
         self.startup_pp = startup_pp
-        self.live_pp = startup_pp.copy()
+        self.live_pp = copy(startup_pp)
 
         view = chart.getViewBox()
 
@@ -556,7 +557,7 @@ class PositionTracker:
         pp = position or self.live_pp
 
         self.update_line(
-            pp.avg_price,
+            pp.be_price,
             pp.size,
             self.chart.linked.symbol.lot_size_digits,
         )
@@ -570,7 +571,7 @@ class PositionTracker:
             self.hide()
 
         else:
-            self._level_marker.level = pp.avg_price
+            self._level_marker.level = pp.be_price
 
             # these updates are critical to avoid lag on view/scene changes
             self._level_marker.update()  # trigger paint

piker/ui/order_mode.py

@@ -33,10 +33,10 @@ import trio
 from PyQt5.QtCore import Qt
 
 from .. import config
+from ..pp import Position
 from ..clearing._client import open_ems, OrderBook
 from ..clearing._allocate import (
     mk_allocator,
-    Position,
 )
 from ._style import _font
 from ..data._source import Symbol
@@ -59,7 +59,8 @@ log = get_logger(__name__)
 
 
 class OrderDialog(BaseModel):
-    '''Trade dialogue meta-data describing the lifetime
+    '''
+    Trade dialogue meta-data describing the lifetime
     of an order submission to ``emsd`` from a chart.
 
     '''
@@ -87,7 +88,8 @@ def on_level_change_update_next_order_info(
     tracker: PositionTracker,
 
 ) -> None:
-    '''A callback applied for each level change to the line
+    '''
+    A callback applied for each level change to the line
     which will recompute the order size based on allocator
     settings. this is assigned inside
     ``OrderMode.line_from_order()``
@@ -577,9 +579,9 @@ async def open_order_mode(
             providers=symbol.brokers
         )
 
-        # XXX: ``brokerd`` delivers a set of account names that it allows
+        # XXX: ``brokerd`` delivers a set of account names that it
-        # use of but the user also can define the accounts they'd like
+        # allows use of but the user also can define the accounts they'd
-        # to use, in order, in their `brokers.toml` file.
+        # like to use, in order, in their `brokers.toml` file.
         accounts = {}
         for name in brokerd_accounts:
             # ensure name is in ``brokers.toml``
@@ -592,10 +594,21 @@ async def open_order_mode(
             iter(accounts.keys())
         ) if accounts else 'paper'
 
+        # Pack position messages by account, should only be one-to-one.
         # NOTE: requires the backend exactly specifies
         # the expected symbol key in its positions msg.
-        pp_msgs = position_msgs.get(symkey, ())
+        pps_by_account = {}
-        pps_by_account = {msg['account']: msg for msg in pp_msgs}
+        for (broker, acctid), msgs in position_msgs.items():
+            for msg in msgs:
+
+                sym = msg['symbol']
+                if (
+                    sym == symkey or
+                    # mega-UGH, i think we need to fix the FQSN stuff sooner
+                    # then later..
+                    sym == symkey.removesuffix(f'.{broker}')
+                ):
+                    pps_by_account[acctid] = msg
 
         # update pp trackers with data relayed from ``brokerd``.
         for account_name in accounts:
@@ -604,7 +617,10 @@ async def open_order_mode(
             startup_pp = Position(
                 symbol=symbol,
                 size=0,
-                avg_price=0,
+                be_price=0,
+
+                # XXX: BLEH, do we care about this on the client side?
+                bsuid=symbol,
             )
             msg = pps_by_account.get(account_name)
             if msg:

setup.py

@@ -41,6 +41,7 @@ setup(
     },
     install_requires=[
         'toml',
+        'tomli',  # fastest pure py reader
         'click',
         'colorlog',
         'attrs',