Merge pull request 'binance: mk `AggTrade.nq` optional..'

Reviewed-on: #67
Reviewed-by: momo <dilarayalniz1@gmail.com>

.gitignore

@@ -103,11 +103,3 @@ ENV/
 # mypy
 .mypy_cache/
 .vscode/settings.json
-
-# macOS Finder metadata
-**/.DS_Store
-
-# LLM conversations that should remain private
-docs/conversations/
-
-

README.rst

@@ -93,27 +93,38 @@ bc why install with `python` when you can faster with `rust` ::
     # ^ astral's docs,
     # https://docs.astral.sh/uv/concepts/projects/sync/
 
-include all GUIs ::
+include all GUIs (ex. for charting)::
 
-    uv sync --extra uis
+    uv sync --group uis
 
-AND with all our hacking tools::
+AND with **all** our normal hacking tools::
 
-    uv sync --dev --extra uis
+    uv sync --dev
 
+AND if you want to try WIP integrations::
+
+    uv sync --all-groups
 
 Ensure you can run the root-daemon::
 
     uv run pikerd [-l info --pdb]
 
 
-hacky install on nixos
+install on nix(os)
-**********************
+******************
 ``NixOS`` is our core devs' distro of choice for which we offer
-a stringently defined development shell envoirment that can be loaded with::
+a stringently defined development shell envoirment that can currently
+be applied in one of 2 ways::
 
+    # ONLY if running on X11
     nix-shell default.nix
 
+Or if you prefer flakes style and a modern DE::
+
+    # ONLY if also running on Wayland
+    nix develop  # for default bash
+    nix develop -c uv run xonsh  # for @goodboy's preferred sh B)
+
 
 start a chart
 *************

config/brokers.toml

@@ -1,6 +1,5 @@
-################
 # ---- CEXY ----
-################
+
 [binance]
 accounts.paper = 'paper'
 
@@ -13,28 +12,41 @@ accounts.spot = 'spot'
 spot.use_testnet = false
 spot.api_key = ''
 spot.api_secret = ''
+# ------ binance ------
 
 
 [deribit]
+# std assets
 key_id = ''
 key_secret = ''
+# options
+accounts.option = 'option'
+option.use_testnet = false
+option.key_id = ''
+option.key_secret = ''
+# aux logging from `cryptofeed`
+option.log.filename = 'cryptofeed.log'
+option.log.level = 'DEBUG'
+option.log.disabled = true
+# ------ deribit ------
 
 
 [kraken]
 key_descr = ''
 api_key = ''
 secret = ''
+# ------ kraken ------
 
 
 [kucoin]
 key_id = ''
 key_secret = ''
 key_passphrase = ''
+# ------ kucoin ------
 
 
-################
 # -- BROKERZ ---
-################
+
 [questrade]
 refresh_token = ''
 access_token = ''
@@ -42,44 +54,55 @@ api_server = 'https://api06.iq.questrade.com/'
 expires_in = 1800
 token_type = 'Bearer'
 expires_at = 1616095326.355846
+# ------ questrade ------
 
 
 [ib]
+# define the (set of) host-port socketaddrs that
+# brokerd.ib will scan to connect to an API endpoint
+# (ib-gw or ib-tws listening instances)
 hosts = [
     '127.0.0.1',
 ]
-# XXX: the order in which ports will be scanned
-# (by the `brokerd` daemon-actor)
-# is determined # by the line order here.
-# TODO: when we eventually spawn gateways in our
-# container, we can just dynamically allocate these
-# using IBC.
 ports = [
     4002,  # gw
     7497,  # tws
 ]
 
-# XXX: for a paper account the flex web query service
+# When API endpoints are being scanned durin startup, the order
-# is not supported so you have to manually download
+# of user-defined-account "names" (as defined below) here
-# and XML report and put it in a location that can be
+# determines which py-client connection is given priority to be
-# accessed by the ``brokerd.ib`` backend code for parsing.
+# used for data-feed-requests by according to whichever client
-flex_token = ''
+# connected to an API endpoing which reported the equivalent
-flex_trades_query_id = ''  # live account
+# account number for that name.
-
-# when clients are being scanned this determines
-# which clients are preferred to be used for data
-# feeds based on the order of account names, if
-# detected as active on an API client.
 prefer_data_account = [
     'paper',
     'margin',
     'ira',
 ]
 
+# For long-term trades txn (transaction) history
+# processing (i.e your txn ledger with IB) you can
+# (automatically for live accounts) query the FLEX
+# report system for past history.
+#
+# (For paper accounts the web query service
+# is not supported so you have to manually download
+# an XML report and put it in a location that can be
+# accessed by our `brokerd.ib` backend code for parsing).
+#
+flex_token = ''
+flex_trades_query_id = ''  # live account
+
+# define "aliases" (names) for each account number
+# such that the names can be reffed and logged throughout
+# `piker.accounting` subsys and more easily
+# referred to  by the user.
+#
+# These keys will be the set exposed through the order-mode
+# account-selection UI so that numbers are never shown.
 [ib.accounts]
-# the order in which accounts will be selectable
+paper = 'DU0000000'  # <- literal account #
-# in the order mode UI (if found via clients during
+margin = 'U0000000'
-# API-app scanning)when a new symbol is loaded.
+ira = 'U0000000'
-paper = 'XX0000000'
+# ------ ib ------
-margin = 'X0000000'
-ira = 'X0000000'

default.nix

@@ -11,11 +11,12 @@ let
   libxkbcommonStorePath = lib.getLib libxkbcommon;
   xcbutilcursorStorePath = lib.getLib xcb-util-cursor;
 
-  qtpyStorePath = lib.getLib python312Packages.qtpy;
+  pypkgs = python313Packages;
-  pyqt6StorePath = lib.getLib python312Packages.pyqt6;
+  qtpyStorePath = lib.getLib pypkgs.qtpy;
-  pyqt6SipStorePath = lib.getLib python312Packages.pyqt6-sip;
+  pyqt6StorePath = lib.getLib pypkgs.pyqt6;
-  rapidfuzzStorePath = lib.getLib python312Packages.rapidfuzz;
+  pyqt6SipStorePath = lib.getLib pypkgs.pyqt6-sip;
-  qdarkstyleStorePath = lib.getLib python312Packages.qdarkstyle;
+  rapidfuzzStorePath = lib.getLib pypkgs.rapidfuzz;
+  qdarkstyleStorePath = lib.getLib pypkgs.qdarkstyle;
 
   xorgLibX11StorePath = lib.getLib xorg.libX11;
   xorgLibxcbStorePath = lib.getLib xorg.libxcb;
@@ -51,12 +52,12 @@ stdenv.mkDerivation {
     xorg.xcbutilrenderutil
 
     # Python requirements.
-    python312Full
+    python313
-    python312Packages.uv
+    uv
-    python312Packages.qdarkstyle
+    pypkgs.qdarkstyle
-    python312Packages.rapidfuzz
+    pypkgs.rapidfuzz
-    python312Packages.pyqt6
+    pypkgs.pyqt6
-    python312Packages.qtpy
+    pypkgs.qtpy
   ];
   src = null;
   shellHook = ''
@@ -113,11 +114,11 @@ stdenv.mkDerivation {
 
     export LD_LIBRARY_PATH
 
-    RPDFUZZ_PATH="${rapidfuzzStorePath}/lib/python3.12/site-packages"
+    RPDFUZZ_PATH="${rapidfuzzStorePath}/lib/python3.13/site-packages"
-    QDRKSTYLE_PATH="${qdarkstyleStorePath}/lib/python3.12/site-packages"
+    QDRKSTYLE_PATH="${qdarkstyleStorePath}/lib/python3.13/site-packages"
-    QTPY_PATH="${qtpyStorePath}/lib/python3.12/site-packages"
+    QTPY_PATH="${qtpyStorePath}/lib/python3.13/site-packages"
-    PYQT6_PATH="${pyqt6StorePath}/lib/python3.12/site-packages"
+    PYQT6_PATH="${pyqt6StorePath}/lib/python3.13/site-packages"
-    PYQT6_SIP_PATH="${pyqt6SipStorePath}/lib/python3.12/site-packages"
+    PYQT6_SIP_PATH="${pyqt6SipStorePath}/lib/python3.13/site-packages"
 
     PATCH="$PATCH:$RPDFUZZ_PATH"
     PATCH="$PATCH:$QDRKSTYLE_PATH"
@@ -127,8 +128,8 @@ stdenv.mkDerivation {
 
     export PATCH
 
-    # Install deps
+    # install all dev and extras
-    uv lock
+    uv sync --dev --all-extras
 
   '';
 }

dockering/ib/README.rst

@@ -24,9 +24,8 @@ here is an example using ``vncclient`` on ``linux``::
 
     vncviewer localhost:5900
 
-
+now enter the pw (password) you set via an (see second code blob)
-now enter the pw you set via an (see second code blob) `.env file`_
+`.env file`_ or pw-file according to the `credentials section`_.
-or pw-file according to the `credentials section`_.
 
 If you want to change away from their default config see the example
 `docker-compose.yml`-config issue and config-section of the readme,
@@ -39,6 +38,74 @@ If you want to change away from their default config see the example
 .. _credentials section: https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#credentials
 
 
+Connecting to the API from `piker`
+---------------------------------
+In order to expose the container's API endpoint to the
+`brokerd/datad/ib` actor, we need to add a section to the user's
+`brokers.toml` config (note the below is similar to the repo-shipped
+template file),
+
+.. code:: toml
+
+    [ib]
+    # define the (set of) host-port socketaddrs that
+    # brokerd.ib will scan to connect to an API endpoint
+    # (ib-gw or ib-tws listening instances)
+    hosts = [
+        '127.0.0.1',
+    ]
+    ports = [
+        4002,  # gw
+        7497,  # tws
+    ]
+
+    # When API endpoints are being scanned durin startup, the order
+    # of user-defined-account "names" (as defined below) here
+    # determines which py-client connection is given priority to be
+    # used for data-feed-requests by according to whichever client
+    # connected to an API endpoing which reported the equivalent
+    # account number for that name.
+    prefer_data_account = [
+        'paper',
+        'margin',
+        'ira',
+    ]
+
+    # define "aliases" (names) for each account number
+    # such that the names can be reffed and logged throughout
+    # `piker.accounting` subsys and more easily
+    # referred to  by the user.
+    #
+    # These keys will be the set exposed through the order-mode
+    # account-selection UI so that numbers are never shown.
+    [ib.accounts]
+    paper = 'XX0000000'
+    margin = 'X0000000'
+    ira = 'X0000000'
+
+
+the broker daemon can also connect to the container's VNC server for
+added functionalies including,
+
+- viewing the API endpoint program's GUI for manual interventions,
+- workarounds for historical data throttling using hotkey hacks,
+
+Add a further section to `brokers.toml` which maps each API-ep's
+port to a table of VNC server connection info like,
+
+.. code:: toml
+
+    [ib.vnc_addrs]
+    4002 = {host = 'localhost', port = 5900, pw = 'doggy'}
+
+The `pw = 'doggy'` here ^ should the same value as the particular
+container instances `.env` file setting (when it was run),
+
+.. code:: ini
+
+    VNC_SERVER_PASSWORD='doggy'
+
+
 IF you also want to run ``TWS``
 -------------------------------
 You can also run it containerized,

dockering/ib/docker-compose.yml

@@ -1,10 +1,15 @@
-# rework from the original @
+# a community maintained IB API container!
-# https://github.com/waytrade/ib-gateway-docker/blob/master/docker-compose.yml
+#
-version: "3.5"
+# https://github.com/gnzsnz/ib-gateway-docker
-
+#
+# For piker we (currently) include some minor deviations
+# for some config files in the `volumes` section.
+#
+# See full configuration settings @
+# - https://github.com/gnzsnz/ib-gateway-docker?tab=readme-ov-file#configuration
+# - https://github.com/gnzsnz/ib-gateway-docker/discussions/103
 
 services:
-
   ib_gw_paper:
 
     # apparently java is a mega cukc:
@@ -50,16 +55,22 @@ services:
         target: /root/scripts/run_x11_vnc.sh
         read_only: true
 
-    # NOTE:to fill these out, define an `.env` file in the same dir as
+    # NOTE: an alt method to fill these out is to
-    # this compose file which looks something like:
+    # define an `.env` file in the same dir as
-    # TWS_USERID='myuser'
+    # this compose file.
-    # TWS_PASSWORD='guest'
     environment:
       TWS_USERID: ${TWS_USERID}
+      # TWS_USERID: 'myuser'
       TWS_PASSWORD: ${TWS_PASSWORD}
-      TRADING_MODE: 'paper'
+      # TWS_PASSWORD: 'guest'
-      VNC_SERVER_PASSWORD: 'doggy'
+      TRADING_MODE: ${TRADING_MODE}
-      VNC_SERVER_PORT: '3003'
+      # TRADING_MODE: 'paper'
+      VNC_SERVER_PASSWORD: ${VNC_SERVER_PASSWORD}
+      # VNC_SERVER_PASSWORD: 'doggy'
+
+      # TODO, see if we can get this supported like it
+      # was on the old `waytrade` image?
+      # VNC_SERVER_PORT: '3003'
 
     # ports:
     #   - target: 4002
@@ -76,6 +87,9 @@ services:
       # - "127.0.0.1:4002:4002"
       # - "127.0.0.1:5900:5900"
 
+  # TODO, a masked but working example of dual paper + live
+  # ib-gw instances running in a single app run!
+  #
   # ib_gw_live:
   #   image: waytrade/ib-gateway:1012.2i
   #   restart: no

docs/macos/README.md

@@ -1,42 +0,0 @@
-# macOS Documentation
-
-This directory contains macOS-specific documentation for the piker project.
-
-## Contents
-
-- **[compatibility-fixes.md](compatibility-fixes.md)** - Comprehensive guide to macOS compatibility issues and their solutions
-
-## Quick Start
-
-If you're experiencing issues running piker on macOS, check the compatibility fixes guide:
-
-```bash
-cat docs/macos/compatibility-fixes.md
-```
-
-## Key Issues Addressed
-
-1. **Socket Credential Passing** - macOS uses different socket options than Linux
-2. **Shared Memory Name Limits** - macOS limits shm names to 31 characters
-3. **Cleanup Race Conditions** - Handling concurrent shared memory cleanup
-4. **Async Runtime Coordination** - Proper trio/asyncio shutdown on macOS
-
-## Platform Information
-
-- **Tested on**: macOS 15.0+ (Darwin 25.0.0)
-- **Python**: 3.13+
-- **Architecture**: ARM64 (Apple Silicon) and x86_64 (Intel)
-
-## Related Projects
-
-These fixes may also apply to:
-- [tractor](https://github.com/goodboy/tractor) - The actor runtime used by piker
-- Other projects using tractor on macOS
-
-## Contributing
-
-Found additional macOS issues? Please:
-1. Document the error and its cause
-2. Provide a solution with code examples
-3. Test on multiple macOS versions
-4. Submit a PR updating this documentation

docs/macos/compatibility-fixes.md

@@ -1,504 +0,0 @@
-# macOS Compatibility Fixes for Piker/Tractor
-
-This guide documents macOS-specific issues encountered when running `piker` on macOS and their solutions. These fixes address platform differences between Linux and macOS in areas like socket credentials, shared memory naming, and async runtime coordination.
-
-## Table of Contents
-
-1. [Socket Credential Passing](#1-socket-credential-passing)
-2. [Shared Memory Name Length Limits](#2-shared-memory-name-length-limits)
-3. [Shared Memory Cleanup Race Conditions](#3-shared-memory-cleanup-race-conditions)
-4. [Async Runtime (Trio/AsyncIO) Coordination](#4-async-runtime-trioasyncio-coordination)
-
----
-
-## 1. Socket Credential Passing
-
-### Problem
-
-On Linux, `tractor` uses `SO_PASSCRED` and `SO_PEERCRED` socket options for Unix domain socket credential passing. macOS doesn't support these constants, causing `AttributeError` when importing.
-
-```python
-# Linux code that fails on macOS
-from socket import SO_PASSCRED, SO_PEERCRED  # AttributeError on macOS
-```
-
-### Error Message
-
-```
-AttributeError: module 'socket' has no attribute 'SO_PASSCRED'
-```
-
-### Root Cause
-
-- **Linux**: Uses `SO_PASSCRED` (to enable credential passing) and `SO_PEERCRED` (to retrieve peer credentials)
-- **macOS**: Uses `LOCAL_PEERCRED` (value `0x0001`) instead, and doesn't require enabling credential passing
-
-### Solution
-
-Make the socket credential imports platform-conditional:
-
-**File**: `tractor/ipc/_uds.py` (or equivalent in `piker` if duplicated)
-
-```python
-import sys
-from socket import (
-    socket,
-    AF_UNIX,
-    SOCK_STREAM,
-)
-
-# Platform-specific credential passing constants
-if sys.platform == 'linux':
-    from socket import SO_PASSCRED, SO_PEERCRED
-elif sys.platform == 'darwin':  # macOS
-    # macOS uses LOCAL_PEERCRED instead of SO_PEERCRED
-    # and doesn't need SO_PASSCRED
-    LOCAL_PEERCRED = 0x0001
-    SO_PEERCRED = LOCAL_PEERCRED  # Alias for compatibility
-    SO_PASSCRED = None  # Not needed on macOS
-else:
-    # Other platforms - may need additional handling
-    SO_PASSCRED = None
-    SO_PEERCRED = None
-
-# When creating a socket
-if SO_PASSCRED is not None:
-    sock.setsockopt(SOL_SOCKET, SO_PASSCRED, 1)
-
-# When getting peer credentials
-if SO_PEERCRED is not None:
-    creds = sock.getsockopt(SOL_SOCKET, SO_PEERCRED, struct.calcsize('3i'))
-```
-
-### Implementation Notes
-
-- The `LOCAL_PEERCRED` value `0x0001` is specific to macOS (from `<sys/un.h>`)
-- macOS doesn't require explicitly enabling credential passing like Linux does
-- Consider using `ctypes` or `cffi` for a more robust solution if available
-
----
-
-## 2. Shared Memory Name Length Limits
-
-### Problem
-
-macOS limits POSIX shared memory names to **31 characters** (defined as `PSHMNAMLEN` in `<sys/posix_shm_internal.h>`). Piker generates long descriptive names that exceed this limit, causing `OSError`.
-
-```python
-# Long name that works on Linux but fails on macOS
-shm_name = "piker_quoter_tsla.nasdaq.ib_hist_1m"  # 39 chars - too long!
-```
-
-### Error Message
-
-```
-OSError: [Errno 63] File name too long: '/piker_quoter_tsla.nasdaq.ib_hist_1m'
-```
-
-### Root Cause
-
-- **Linux**: Supports shared memory names up to 255 characters
-- **macOS**: Limits to 31 characters (including leading `/`)
-
-### Solution
-
-Implement automatic name shortening for macOS while preserving the original key for lookups:
-
-**File**: `piker/data/_sharedmem.py`
-
-```python
-import hashlib
-import sys
-
-def _shorten_key_for_macos(key: str) -> str:
-    '''
-    macOS has a 31 character limit for POSIX shared memory names.
-    Hash long keys to fit within this limit while maintaining uniqueness.
-    '''
-    # macOS shm_open() has a 31 char limit (PSHMNAMLEN)
-    # Use format: /p_<hash16> where hash is first 16 hex chars of sha256
-    # This gives us: / + p_ + 16 hex chars = 19 chars, well under limit
-    # We keep the 'p' prefix to indicate it's from piker
-    if len(key) <= 31:
-        return key
-
-    # Create a hash of the full key
-    key_hash = hashlib.sha256(key.encode()).hexdigest()[:16]
-    short_key = f'p_{key_hash}'
-    return short_key
-
-
-class _Token(Struct, frozen=True):
-    '''
-    Internal representation of a shared memory "token"
-    which can be used to key a system wide post shm entry.
-    '''
-    shm_name: str  # actual OS-level name (may be shortened on macOS)
-    shm_first_index_name: str
-    shm_last_index_name: str
-    dtype_descr: tuple
-    size: int  # in struct-array index / row terms
-    key: str | None = None  # original descriptive key (for lookup)
-
-    def __eq__(self, other) -> bool:
-        '''
-        Compare tokens based on shm names and dtype, ignoring the key field.
-        The key field is only used for lookups, not for token identity.
-        '''
-        if not isinstance(other, _Token):
-            return False
-        return (
-            self.shm_name == other.shm_name
-            and self.shm_first_index_name == other.shm_first_index_name
-            and self.shm_last_index_name == other.shm_last_index_name
-            and self.dtype_descr == other.dtype_descr
-            and self.size == other.size
-        )
-
-    def __hash__(self) -> int:
-        '''Hash based on the same fields used in __eq__'''
-        return hash((
-            self.shm_name,
-            self.shm_first_index_name,
-            self.shm_last_index_name,
-            self.dtype_descr,
-            self.size,
-        ))
-
-
-def _make_token(
-    key: str,
-    size: int,
-    dtype: np.dtype | None = None,
-) -> _Token:
-    '''
-    Create a serializable token that uniquely identifies a shared memory segment.
-    '''
-    if dtype is None:
-        dtype = def_iohlcv_fields
-
-    # On macOS, shorten long keys to fit the 31-char limit
-    if sys.platform == 'darwin':
-        shm_name = _shorten_key_for_macos(key)
-        shm_first = _shorten_key_for_macos(key + "_first")
-        shm_last = _shorten_key_for_macos(key + "_last")
-    else:
-        shm_name = key
-        shm_first = key + "_first"
-        shm_last = key + "_last"
-
-    return _Token(
-        shm_name=shm_name,
-        shm_first_index_name=shm_first,
-        shm_last_index_name=shm_last,
-        dtype_descr=tuple(np.dtype(dtype).descr),
-        size=size,
-        key=key,  # Store original key for lookup
-    )
-```
-
-### Key Design Decisions
-
-1. **Hash-based shortening**: Uses SHA256 to ensure uniqueness and avoid collisions
-2. **Preserve original key**: Store the original descriptive key in the `_Token` for debugging and lookups
-3. **Custom equality**: The `__eq__` and `__hash__` methods ignore the `key` field to ensure tokens are compared by their actual shm properties
-4. **Platform detection**: Only applies shortening on macOS (`sys.platform == 'darwin'`)
-
-### Edge Cases to Consider
-
-- Token serialization across processes (the `key` field must survive IPC)
-- Token lookup in dictionaries and caches
-- Debugging output (use `key` field for human-readable names)
-
----
-
-## 3. Shared Memory Cleanup Race Conditions
-
-### Problem
-
-During teardown, shared memory segments may be unlinked by one process while another is still trying to clean them up, causing `FileNotFoundError` to crash the application.
-
-### Error Message
-
-```
-FileNotFoundError: [Errno 2] No such file or directory: '/p_74c86c7228dd773b'
-```
-
-### Root Cause
-
-In multi-process architectures like `tractor`, multiple processes may attempt to clean up shared resources simultaneously. Race conditions during shutdown can cause:
-
-1. Process A unlinks the shared memory
-2. Process B tries to unlink the same memory → `FileNotFoundError`
-3. Uncaught exception crashes Process B
-
-### Solution
-
-Add defensive error handling to catch and log cleanup races:
-
-**File**: `piker/data/_sharedmem.py`
-
-```python
-class ShmArray:
-    # ... existing code ...
-
-    def destroy(self) -> None:
-        '''
-        Destroy the shared memory segment and cleanup OS resources.
-        '''
-        if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            shm = self._shm
-            name = shm.name
-            try:
-                shm_unlink(name)
-            except FileNotFoundError:
-                # Might be a teardown race where another process
-                # already unlinked it - this is fine, just log it
-                log.warning(f'Shm for {name} already unlinked?')
-
-        # Also cleanup the index counters
-        if hasattr(self, '_first'):
-            try:
-                self._first.destroy()
-            except FileNotFoundError:
-                log.warning(f'First index shm already unlinked?')
-
-        if hasattr(self, '_last'):
-            try:
-                self._last.destroy()
-            except FileNotFoundError:
-                log.warning(f'Last index shm already unlinked?')
-
-
-class SharedInt:
-    # ... existing code ...
-
-    def destroy(self) -> None:
-        if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            name = self._shm.name
-            try:
-                shm_unlink(name)
-            except FileNotFoundError:
-                # might be a teardown race here?
-                log.warning(f'Shm for {name} already unlinked?')
-```
-
-### Implementation Notes
-
-- This fix is platform-agnostic but particularly important on macOS where the shortened names make debugging harder
-- The warnings help identify cleanup races during development
-- Consider adding metrics/counters if cleanup races become frequent
-
----
-
-## 4. Async Runtime (Trio/AsyncIO) Coordination
-
-### Problem
-
-The `TrioTaskExited` error occurs when trio tasks are cancelled while asyncio tasks are still running, indicating improper coordination between the two async runtimes.
-
-### Error Message
-
-```
-tractor._exceptions.TrioTaskExited: but the child `asyncio` task is still running?
->>
- |_<Task pending name='Task-2' coro=<wait_on_coro_final_result()> ...>
-```
-
-### Root Cause
-
-`tractor` uses "guest mode" to run trio as a guest in asyncio's event loop (or vice versa). The error occurs when:
-
-1. A trio task is cancelled (e.g., user closes the UI)
-2. The cancellation propagates to cleanup handlers
-3. Cleanup tries to exit while asyncio tasks are still running
-4. The `translate_aio_errors` context manager detects this inconsistent state
-
-### Current State
-
-This issue is **partially resolved** by the other fixes (socket credentials and shared memory), which eliminate the underlying errors that trigger premature cancellation. However, it may still occur in edge cases.
-
-### Potential Solutions
-
-#### Option 1: Improve Cancellation Propagation (Tractor-level)
-
-**File**: `tractor/to_asyncio.py`
-
-```python
-async def translate_aio_errors(
-    chan,
-    wait_on_aio_task: bool = False,
-    suppress_graceful_exits: bool = False,
-):
-    '''
-    Context manager to translate asyncio errors to trio equivalents.
-    '''
-    try:
-        yield
-    except trio.Cancelled:
-        # When trio is cancelled, ensure asyncio tasks are also cancelled
-        if wait_on_aio_task:
-            # Give asyncio tasks a chance to cleanup
-            await trio.lowlevel.checkpoint()
-
-            # Check if asyncio task is still running
-            if aio_task and not aio_task.done():
-                # Cancel it gracefully
-                aio_task.cancel()
-
-                # Wait briefly for cancellation
-                with trio.move_on_after(0.5):  # 500ms timeout
-                    await wait_for_aio_task_completion(aio_task)
-
-        raise  # Re-raise the cancellation
-```
-
-#### Option 2: Proper Shutdown Sequence (Application-level)
-
-**File**: `piker/brokers/ib/api.py` (or similar broker modules)
-
-```python
-async def load_clients_for_trio(
-    client: Client,
-    ...
-) -> None:
-    '''
-    Load asyncio client and keep it running for trio.
-    '''
-    try:
-        # Setup client
-        await client.connect()
-
-        # Keep alive - but make it cancellable
-        await trio.sleep_forever()
-
-    except trio.Cancelled:
-        # Explicit cleanup before propagating cancellation
-        log.info("Shutting down asyncio client gracefully")
-
-        # Disconnect client
-        if client.isConnected():
-            await client.disconnect()
-
-        # Small delay to let asyncio cleanup
-        await trio.sleep(0.1)
-
-        raise  # Now safe to propagate
-```
-
-#### Option 3: Detection and Warning (Current Approach)
-
-The current code detects the issue and raises a clear error. This is acceptable if:
-1. The error is rare (only during abnormal shutdown)
-2. It doesn't cause data loss
-3. Logs provide enough info for debugging
-
-### Recommended Approach
-
-For **piker**: Implement Option 2 (proper shutdown sequence) in broker modules where asyncio is used.
-
-For **tractor**: Consider Option 1 (improved cancellation propagation) as a library-level enhancement.
-
-### Testing
-
-Test the fix by:
-
-```python
-# Test graceful shutdown
-async def test_asyncio_trio_shutdown():
-    async with open_channel_from(...) as (first, chan):
-        # Do some work
-        await chan.send(msg)
-
-        # Trigger cancellation
-        raise KeyboardInterrupt
-
-    # Should cleanup without TrioTaskExited error
-```
-
----
-
-## Summary of Changes
-
-### Files Modified in Piker
-
-1. **`piker/data/_sharedmem.py`**
-   - Added `_shorten_key_for_macos()` function
-   - Modified `_Token` class to store original `key`
-   - Modified `_make_token()` to use shortened names on macOS
-   - Added `FileNotFoundError` handling in `destroy()` methods
-
-2. **`piker/ui/_display.py`**
-   - Removed assertion that checked for 'hist' in shm name (incompatible with shortened names)
-
-### Files to Modify in Tractor (Recommended)
-
-1. **`tractor/ipc/_uds.py`**
-   - Make socket credential imports platform-conditional
-   - Handle macOS-specific `LOCAL_PEERCRED`
-
-2. **`tractor/to_asyncio.py`** (Optional)
-   - Improve cancellation propagation between trio and asyncio
-   - Add graceful shutdown timeout for asyncio tasks
-
-### Platform Detection Pattern
-
-Use this pattern consistently:
-
-```python
-import sys
-
-if sys.platform == 'darwin':  # macOS
-    # macOS-specific code
-    pass
-elif sys.platform == 'linux':  # Linux
-    # Linux-specific code
-    pass
-else:
-    # Other platforms / fallback
-    pass
-```
-
-### Testing Checklist
-
-- [ ] Test on macOS (Darwin)
-- [ ] Test on Linux
-- [ ] Test shared memory with names > 31 chars
-- [ ] Test multi-process cleanup race conditions
-- [ ] Test graceful shutdown (Ctrl+C)
-- [ ] Test abnormal shutdown (kill signal)
-- [ ] Verify no memory leaks (check `/dev/shm` on Linux, `ipcs -m` on macOS)
-
----
-
-## Additional Resources
-
-- **macOS System Headers**:
-  - `/usr/include/sys/un.h` - Unix domain socket constants
-  - `/usr/include/sys/posix_shm_internal.h` - Shared memory limits
-
-- **Python Documentation**:
-  - [`socket` module](https://docs.python.org/3/library/socket.html)
-  - [`multiprocessing.shared_memory`](https://docs.python.org/3/library/multiprocessing.shared_memory.html)
-
-- **Trio/AsyncIO**:
-  - [Trio Guest Mode](https://trio.readthedocs.io/en/stable/reference-lowlevel.html#using-guest-mode-to-run-trio-on-top-of-other-event-loops)
-  - [Tractor Documentation](https://github.com/goodboy/tractor)
-
----
-
-## Contributing
-
-When implementing these fixes in your own project:
-
-1. **Test thoroughly** on both macOS and Linux
-2. **Add platform guards** to prevent cross-platform breakage
-3. **Document platform-specific behavior** in code comments
-4. **Consider CI/CD** testing on multiple platforms
-5. **Handle edge cases** gracefully with proper logging
-
-If you find additional macOS-specific issues, please contribute to this guide!

examples/max_pain.py

@@ -1,338 +0,0 @@
-#!/usr/bin/env python
-from decimal import (
-    Decimal,
-)
-from pathlib import Path
-
-import numpy as np
-# import polars as pl
-import trio
-import tractor
-from datetime import datetime
-# from pprint import pformat
-from piker.brokers.deribit.api import (
-    get_client,
-    maybe_open_oi_feed,
-)
-from piker.storage import open_storage_client, StorageClient
-from piker.log import get_logger
-import sys
-import pyqtgraph as pg
-from PyQt6 import QtCore
-from pyqtgraph import ScatterPlotItem, InfiniteLine
-from PyQt6.QtWidgets import QApplication
-from cryptofeed.symbols import Symbol
-
-
-log = get_logger(__name__)
-# XXX, use 2 newlines between top level LOC (even between these
-# imports and the next function line ;)
-
-def check_if_complete(
-        oi: dict[str, dict[str, Decimal | None]]
-    ) -> bool:
-    return all(
-        oi[strike]['C'] is not None
-        and
-        oi[strike]['P'] is not None for strike in oi
-    )
-
-
-async def max_pain_daemon(
-) -> None:
-    oi_by_strikes: dict[str, dict[str, Decimal | None]]
-    instruments: list[Symbol] = []
-    expiry_dates: list[str]
-    expiry_date: str
-    currency: str = 'btc'
-    kind: str = 'option'
-
-    async with get_client(
-    ) as client:
-        expiry_dates: list[str] = await client.get_expiration_dates(
-            currency=currency,
-            kind=kind
-        )
-
-        log.info(
-            f'Available expiries for {currency!r}-{kind}:\n'
-            f'{expiry_dates}\n'
-        )
-        expiry_date: str = input(
-            'Please enter a valid expiration date: '
-        ).upper()
-        print('Starting little daemon...')
-
-        # maybe move this type annot down to the assignment line?
-        oi_by_strikes: dict[str, dict[str, Decimal]]
-        instruments = await client.get_instruments(
-            expiry_date=expiry_date,
-        )
-        oi_by_strikes = client.get_strikes_dict(instruments)
-
-
-    def get_total_intrinsic_values(
-        oi_by_strikes: dict[str, dict[str, Decimal]]
-    ) -> dict[str, dict[str, Decimal]]:
-        call_cash: Decimal = Decimal(0)
-        put_cash: Decimal = Decimal(0)
-        intrinsic_values: dict[str, dict[str, Decimal]] = {}
-        closes: list = sorted(Decimal(close) for close in oi_by_strikes)
-
-        for strike, oi in oi_by_strikes.items():
-            s = Decimal(strike)
-            call_cash = sum(max(0, (s - c) * oi_by_strikes[str(c)]['C']) for c in closes)
-            put_cash = sum(max(0, (c - s) * oi_by_strikes[str(c)]['P']) for c in closes)
-
-            intrinsic_values[strike] = {
-                'C': call_cash,
-                'P': put_cash,
-                'total': call_cash + put_cash,
-            }
-
-        return intrinsic_values
-
-    def get_intrinsic_value_and_max_pain(
-        intrinsic_values: dict[str, dict[str, Decimal]]
-        ):
-        # We meed to find the lowest value, so we start at
-        # infinity to ensure that, and the max_pain must be
-        # an amount greater than zero.
-        total_intrinsic_value: Decimal = Decimal('Infinity')
-        max_pain: Decimal = Decimal(0)
-
-        for strike, oi in oi_by_strikes.items():
-            s = Decimal(strike)
-            if intrinsic_values[strike]['total'] < total_intrinsic_value:
-                total_intrinsic_value = intrinsic_values[strike]['total']
-                max_pain = s
-
-        return total_intrinsic_value, max_pain
-
-    def plot_graph(
-        oi_by_strikes: dict[str, dict[str, Decimal]],
-        plot,
-    ):
-        """Update the bar graph with new open interest data."""
-        plot.clear()
-
-        intrinsic_values = get_total_intrinsic_values(oi_by_strikes)
-
-        for strike_str in sorted(oi_by_strikes, key=lambda x: int(x)):
-            strike = int(strike_str)
-            calls_val = float(oi_by_strikes[strike_str]['C'])
-            puts_val  = float(oi_by_strikes[strike_str]['P'])
-
-            bar_c = pg.BarGraphItem(
-                x=[strike - 100],
-                height=[calls_val],
-                width=200,
-                pen='w',
-                brush=(0, 0, 255, 150)
-            )
-            plot.addItem(bar_c)
-
-            bar_p = pg.BarGraphItem(
-                x=[strike + 100],
-                height=[puts_val],
-                width=200,
-                pen='w',
-                brush=(255, 0, 0, 150)
-            )
-            plot.addItem(bar_p)
-
-            total_val = float(intrinsic_values[strike_str]['total']) / 100000
-
-            scatter_iv = ScatterPlotItem(
-                x=[strike],
-                y=[total_val],
-                pen=pg.mkPen(color=(0, 255, 0), width=2),
-                brush=pg.mkBrush(0, 255, 0, 150),
-                size=3,
-                symbol='o'
-            )
-            plot.addItem(scatter_iv)
-
-        _, max_pain = get_intrinsic_value_and_max_pain(intrinsic_values)
-
-        vertical_line = InfiniteLine(
-            pos=max_pain,
-            angle=90,
-            pen=pg.mkPen(color='yellow', width=1, style=QtCore.Qt.PenStyle.DotLine),
-            label=f'Max pain: {max_pain:,.0f}',
-            labelOpts={
-                'position': 0.85,
-                'color': 'yellow',
-                'movable': True
-            }
-        )
-        plot.addItem(vertical_line)
-
-    def update_oi_by_strikes(msg: tuple):
-        nonlocal oi_by_strikes
-        if 'oi' == msg[0]:
-            strike_price = msg[1]['strike_price']
-            option_type = msg[1]['option_type']
-            open_interest = msg[1]['open_interest']
-            oi_by_strikes.setdefault(
-                strike_price, {}
-            ).update(
-                {option_type: open_interest}
-            )
-
-    # Define the structured dtype
-    dtype = np.dtype([
-        ('time', int),
-        ('oi', float),
-        ('oi_calc', float),
-    ])
-    async def write_open_interest_on_file(msg: tuple, client: StorageClient):
-        if 'oi' == msg[0]:
-            nonlocal expiry_date
-            timestamp = msg[1]['timestamp']
-            strike_price = msg[1]["strike_price"]
-            option_type = msg[1]['option_type'].lower()
-            col_sym_key = f'btc-{expiry_date.lower()}-{strike_price}-{option_type}'
-
-            # Create the numpy array with sample data
-            data = np.array([
-                (
-                    int(timestamp),
-                    float(msg[1]['open_interest']),
-                    np.nan,
-                ),
-            ], dtype=dtype)
-
-            path: Path = await client.write_oi(
-                col_sym_key,
-                data,
-            )
-            # TODO, use std logging like this throughout for status
-            # emissions on console!
-            log.info(f'Wrote OI history to {path}')
-
-    def get_max_pain(
-        oi_by_strikes: dict[str, dict[str, Decimal]]
-    ) -> dict[str, str | Decimal]:
-        '''
-        This method requires only the strike_prices and oi for call
-        and puts, the closes list are the same as the strike_prices
-        the idea is to sum all the calls and puts cash for each strike
-        and the ITM strikes from that strike, the lowest value is what we
-        are looking for the intrinsic value.
-
-        '''
-
-        nonlocal timestamp
-
-        intrinsic_values = get_total_intrinsic_values(oi_by_strikes)
-
-        total_intrinsic_value, max_pain = get_intrinsic_value_and_max_pain(intrinsic_values)
-
-        return {
-            'timestamp': timestamp,
-            'expiry_date': expiry_date,
-            'total_intrinsic_value': total_intrinsic_value,
-            'max_pain': max_pain,
-        }
-
-    async with (
-        open_storage_client() as (_, storage),
-
-        maybe_open_oi_feed(
-            instruments,
-        ) as oi_feed,
-    ):
-        # Initialize QApplication
-        app = QApplication(sys.argv)
-
-        win = pg.GraphicsLayoutWidget(show=True)
-        win.setWindowTitle('Calls (blue) vs Puts (red)')
-
-        plot = win.addPlot(title='OI by Strikes')
-        plot.showGrid(x=True, y=True)
-        print('Plot initialized...')
-
-        async for msg in oi_feed:
-
-            # In memory oi_by_strikes dict, all message are filtered here
-            # and the dict is updated with the open interest data
-            update_oi_by_strikes(msg)
-
-            # Write on file using storage client
-            await write_open_interest_on_file(msg, storage)
-
-            # Max pain calcs, before start we must gather all the open interest for
-            # all the strike prices and option types available for a expiration date
-            if check_if_complete(oi_by_strikes):
-                if 'oi' == msg[0]:
-                    # Here we must read for the filesystem all the latest open interest value for
-                    # each instrument for that specific expiration date, that means look up for the
-                    # last update got the instrument btc-{expity_date}-*oi1s.parquet (1s because is
-                    # hardcoded to something, sorry.)
-                    timestamp = msg[1]['timestamp']
-                    max_pain = get_max_pain(oi_by_strikes)
-                    # intrinsic_values = get_total_intrinsic_values(oi_by_strikes)
-
-                    # graph here
-                    plot_graph(oi_by_strikes, plot)
-
-                    # TODO, use a single multiline string with `()`
-                    # and drop the multiple `print()` calls (this
-                    # should be done elsewhere in this file as well!
-                    #
-                    # As per the docs,
-                    # https://docs.python.org/3/reference/lexical_analysis.html#string-literal-concatenation
-                    # you could instead do,
-                    # print(
-                    #   '-----------------------------------------------\n'
-                    #   f'timestamp: {datetime.fromtimestamp(max_pain['timestamp'])}\n'
-                    # )
-                    # WHY?
-                    # |_ less ctx-switches/calls to `print()`
-                    # |_ the `str` can then be modified / passed
-                    #   around as a variable more easily if needed in
-                    #   the future ;)
-                    #
-                    # ALSO, i believe there already is a stdlib
-                    # module to do "alignment" of text which you
-                    # could try for doing the right-side alignment,
-                    # https://docs.python.org/3/library/textwrap.html#textwrap.indent
-                    #
-                    print('-----------------------------------------------')
-                    print(f'timestamp:             {datetime.fromtimestamp(max_pain['timestamp'])}')
-                    print(f'expiry_date:           {max_pain['expiry_date']}')
-                    print(f'max_pain:              {max_pain['max_pain']:,.0f}')
-                    print(f'total intrinsic value: {max_pain['total_intrinsic_value']:,.0f}')
-                    print('-----------------------------------------------')
-
-            # Process GUI events to keep the window responsive
-            app.processEvents()
-
-
-async def main():
-
-    async with tractor.open_nursery(
-        debug_mode=True,
-        loglevel='info',
-    ) as an:
-        from tractor import log
-        log.get_console_log(level='info')
-
-        ptl: tractor.Portal = await an.start_actor(
-            'max_pain_daemon',
-            enable_modules=[__name__],
-            infect_asyncio=True,
-            # ^TODO, we can actually run this in the root-actor now
-            # if needed as per 2nd "section" in,
-            # https://pikers.dev/goodboy/tractor/pulls/2
-            #
-            # NOTE, will first require us porting to modern
-            # `tractor:main` though ofc!
-
-        )
-        await ptl.run(max_pain_daemon)
-
-
-if __name__ == '__main__':
-    trio.run(main)

examples/max_pain_readme.rst

@@ -1,29 +0,0 @@
-## Max Pain Calculation for Deribit Options
-
-This feature, which calculates the max pain point for options traded
-on the Deribit exchange using cryptofeed library.
-
-- Functions in the api module for fetching options data from Deribit.
-  [commit](https://pikers.dev/pikers/piker/commit/da55856dd2876291f55a06eb0561438a912d8241)
-
-- Compute the max pain point based on open interest data using
-  deribit's api.
-  [commit](https://pikers.dev/pikers/piker/commit/0d9d6e15ba0edeb662ec97f7599dd66af3046b94)
-
-### How to test it?
-
-**Before start:** in order to get this working with `uv`,  you
-**must** use my [`tractor` fork](https://pikers.dev/ntorres/tractor/src/branch/aio_abandons)
-and this branch: `aio_abandons`, the reason is that I cherry-pick the
-`uv_migration` that guille made, for some reason that a didn't dive
-into, in my system y need tractor using `uv` too. quite hacky
-I guess.
-
-1. `uv lock`
-
-2. `uv run --no-dev python examples/max_pain.py`
-
-3. A message should be display, enter one of the expiration date
-   available.
-
-4. The script should be up and running.

flake.lock

@@ -1,135 +1,24 @@
 {
   "nodes": {
-    "flake-utils": {
-      "inputs": {
-        "systems": "systems"
-      },
-      "locked": {
-        "lastModified": 1689068808,
-        "narHash": "sha256-6ixXo3wt24N/melDWjq70UuHQLxGV8jZvooRanIHXw0=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "919d646de7be200f3bf08cb76ae1f09402b6f9b4",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
-    "flake-utils_2": {
-      "inputs": {
-        "systems": "systems_2"
-      },
-      "locked": {
-        "lastModified": 1689068808,
-        "narHash": "sha256-6ixXo3wt24N/melDWjq70UuHQLxGV8jZvooRanIHXw0=",
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "rev": "919d646de7be200f3bf08cb76ae1f09402b6f9b4",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "flake-utils",
-        "type": "github"
-      }
-    },
-    "nix-github-actions": {
-      "inputs": {
-        "nixpkgs": [
-          "poetry2nix",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1688870561,
-        "narHash": "sha256-4UYkifnPEw1nAzqqPOTL2MvWtm3sNGw1UTYTalkTcGY=",
-        "owner": "nix-community",
-        "repo": "nix-github-actions",
-        "rev": "165b1650b753316aa7f1787f3005a8d2da0f5301",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-community",
-        "repo": "nix-github-actions",
-        "type": "github"
-      }
-    },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1692174805,
+        "lastModified": 1765779637,
-        "narHash": "sha256-xmNPFDi/AUMIxwgOH/IVom55Dks34u1g7sFKKebxUm0=",
+        "narHash": "sha256-KJ2wa/BLSrTqDjbfyNx70ov/HdgNBCBBSQP3BIzKnv4=",
-        "owner": "NixOS",
+        "owner": "nixos",
         "repo": "nixpkgs",
-        "rev": "caac0eb6bdcad0b32cb2522e03e4002c8975c62e",
+        "rev": "1306659b587dc277866c7b69eb97e5f07864d8c4",
         "type": "github"
       },
       "original": {
-        "owner": "NixOS",
+        "owner": "nixos",
         "ref": "nixos-unstable",
         "repo": "nixpkgs",
         "type": "github"
       }
     },
-    "poetry2nix": {
-      "inputs": {
-        "flake-utils": "flake-utils_2",
-        "nix-github-actions": "nix-github-actions",
-        "nixpkgs": [
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1692048894,
-        "narHash": "sha256-cDw03rso2V4CDc3Mll0cHN+ztzysAvdI8pJ7ybbz714=",
-        "ref": "refs/heads/pyqt6",
-        "rev": "b059ad4c3051f45d6c912e17747aae37a9ec1544",
-        "revCount": 2276,
-        "type": "git",
-        "url": "file:///home/lord_fomo/repos/poetry2nix"
-      },
-      "original": {
-        "type": "git",
-        "url": "file:///home/lord_fomo/repos/poetry2nix"
-      }
-    },
     "root": {
       "inputs": {
-        "flake-utils": "flake-utils",
+        "nixpkgs": "nixpkgs"
-        "nixpkgs": "nixpkgs",
-        "poetry2nix": "poetry2nix"
-      }
-    },
-    "systems": {
-      "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-systems",
-        "repo": "default",
-        "type": "github"
-      }
-    },
-    "systems_2": {
-      "locked": {
-        "lastModified": 1681028828,
-        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
-        "owner": "nix-systems",
-        "repo": "default",
-        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-systems",
-        "repo": "default",
-        "type": "github"
       }
     }
   },

flake.nix

@@ -1,180 +1,103 @@
-# NOTE: to convert to a poetry2nix env like this here are the
+# An "impure" template thx to `pyproject.nix`,
-# steps:
+# https://pyproject-nix.github.io/pyproject.nix/templates.html#impure
-# - install poetry in your system nix config
+# https://github.com/pyproject-nix/pyproject.nix/blob/master/templates/impure/flake.nix
-# - convert the repo to use poetry using `poetry init`:
-#   https://python-poetry.org/docs/basic-usage/#initialising-a-pre-existing-project
-# - then manually ensuring all deps are converted over:
-# - add this file to the repo and commit it
-# - 
-
-# GROKin tips:
-# - CLI eps are (ostensibly) added via an `entry_points.txt`:
-#   - https://packaging.python.org/en/latest/specifications/entry-points/#file-format
-#   - https://github.com/nix-community/poetry2nix/blob/master/editable.nix#L49
 {
-  description = "piker: trading gear for hackers (pkged with poetry2nix)";
+  description = "An impure `piker` overlay using `uv` with Nix(OS)";
 
-  inputs.flake-utils.url = "github:numtide/flake-utils";
+  inputs = {
-  inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+    nixpkgs.url = "github:nixos/nixpkgs/nixos-unstable";
-
-  # see https://github.com/nix-community/poetry2nix/tree/master#api
-  inputs.poetry2nix = {
-    # url = "github:nix-community/poetry2nix";
-    # url = "github:K900/poetry2nix/qt5-explicit-deps";
-    url = "/home/lord_fomo/repos/poetry2nix";
-
-    inputs.nixpkgs.follows = "nixpkgs";
   };
 
-  outputs = {
+  outputs =
-    self,
+    { nixpkgs, ... }:
-    nixpkgs,
-    flake-utils,
-    poetry2nix,
-  }:
-    # TODO: build cross-OS and use the `${system}` var thingy..
-    flake-utils.lib.eachDefaultSystem (system:
     let
-        # use PWD as sources
+      inherit (nixpkgs) lib;
-        projectDir = ./.;
+      forAllSystems = lib.genAttrs lib.systems.flakeExposed;
-        pyproject = ./pyproject.toml;
-        poetrylock = ./poetry.lock;
-
-        # TODO: port to 3.11 and support both versions?
-        python = "python3.10";
-
-        # for more functions and examples.
-        # inherit
-        # (poetry2nix.legacyPackages.${system})
-        # mkPoetryApplication;
-        # pkgs = nixpkgs.legacyPackages.${system};
-
-        pkgs = nixpkgs.legacyPackages.x86_64-linux;
-        lib = pkgs.lib;
-        p2npkgs = poetry2nix.legacyPackages.x86_64-linux;
-
-        # define all pkg overrides per dep, see edgecases.md:
-        # https://github.com/nix-community/poetry2nix/blob/master/docs/edgecases.md
-        # TODO: add these into the json file:
-        # https://github.com/nix-community/poetry2nix/blob/master/overrides/build-systems.json
-        pypkgs-build-requirements = {
-          asyncvnc = [ "setuptools" ];
-          eventkit = [ "setuptools" ];
-          ib-insync = [ "setuptools" "flake8" ];
-          msgspec = [ "setuptools"];
-          pdbp = [ "setuptools" ];
-          pyqt6-sip = [ "setuptools" ];
-          tabcompleter = [ "setuptools" ];
-          tractor = [ "setuptools" ];
-          tricycle = [ "setuptools" ];
-          trio-typing = [ "setuptools" ];
-          trio-util = [ "setuptools" ];
-          xonsh = [ "setuptools" ];
-        };
-
-        # auto-generate override entries
-        p2n-overrides = p2npkgs.defaultPoetryOverrides.extend (self: super:
-          builtins.mapAttrs (package: build-requirements:
-            (builtins.getAttr package super).overridePythonAttrs (old: {
-              buildInputs = (
-                old.buildInputs or [ ]
-              ) ++ (
-                builtins.map (
-                  pkg: if builtins.isString pkg then builtins.getAttr pkg super else pkg
-                  ) build-requirements
-              );
-            })
-          ) pypkgs-build-requirements
-        );
-
-        # override some ahead-of-time compiled extensions
-        # to be built with their wheels.
-        ahot_overrides = p2n-overrides.extend(
-          final: prev: {
-
-            # llvmlite = prev.llvmlite.override {
-            #   preferWheel = false;
-            # };
-
-            # TODO: get this workin with p2n and nixpkgs..
-            # pyqt6 = prev.pyqt6.override {
-            #   preferWheel = true;
-            # };
-
-            # NOTE: this DOESN'T work atm but after a fix
-            # to poetry2nix, it will and actually this line
-            # won't be needed - thanks @k900:
-            # https://github.com/nix-community/poetry2nix/pull/1257
-            pyqt5 = prev.pyqt5.override {
-              # withWebkit = false;
-              preferWheel = true;
-            };
-
-            # see PR from @k900:
-            # https://github.com/nix-community/poetry2nix/pull/1257
-            # pyqt5-qt5 = prev.pyqt5-qt5.override {
-            #   withWebkit = false;
-            #   preferWheel = true;
-            # };
-
-            # TODO: patch in an override for polars to build
-            # from src! See the details likely needed from
-            # the cryptography entry:
-            # https://github.com/nix-community/poetry2nix/blob/master/overrides/default.nix#L426-L435
-            polars = prev.polars.override {
-              preferWheel = true;
-            };
-          }
-      );
-
-      # WHY!? -> output-attrs that `nix develop` scans for:
-      # https://nixos.org/manual/nix/stable/command-ref/new-cli/nix3-develop.html#flake-output-attributes
     in
-      rec {
+    {
-        packages = {
+      devShells = forAllSystems (
-          # piker = poetry2nix.legacyPackages.x86_64-linux.mkPoetryEditablePackage {
+        system:
-          #   editablePackageSources = { piker = ./piker; };
+        let
+          pkgs = nixpkgs.legacyPackages.${system};
 
-          piker = p2npkgs.mkPoetryApplication {
+          # do store-path extractions
-            projectDir = projectDir;
+          qt6baseStorePath = lib.getLib pkgs.qt6.qtbase;
+          # ?TODO? can remove below since manual linking not needed?
+          # qt6QtWaylandStorePath = lib.getLib pkgs.qt6.qtwayland;
 
-            # SEE ABOVE for auto-genned input set, override
+          # XXX NOTE XXX, for now we overlay specific pkgs via
-            # buncha deps with extras.. like `setuptools` mostly.
+          # a major-version-pinned-`cpython`
-            # TODO: maybe propose a patch to p2n to show that you
+          cpython = "python313";
-            # can even do this in the edgecases docs?
+          pypkgs = pkgs."${cpython}Packages";
-            overrides = ahot_overrides;
+        in
+        {
+          default = pkgs.mkShell {
 
-            # XXX: won't work on llvmlite..
+            packages = with pkgs; [
-            # preferWheels = true;
+              # XXX, ensure sh completions active!
-          };
+              bashInteractive
-        };
+              bash-completion
 
-        # devShells.default = pkgs.mkShell {
+              # dev utils
-        #   projectDir = projectDir;
+              ruff
-        #   python = "python3.10";
+              pypkgs.ruff
-        #   overrides = ahot_overrides;
-        #   inputsFrom = [ self.packages.x86_64-linux.piker ];
-        #   packages = packages;
-        #   # packages = [ poetry2nix.packages.${system}.poetry ];
-        # };
 
-        # TODO: grok the difference here..
+              qt6.qtwayland
-        # - avoid re-cloning git repos on every develop entry..
+              qt6.qtbase
-        # - ideally allow hacking on the src code of some deps
+
-        #   (tractor, pyqtgraph, tomlkit, etc.) WITHOUT having to
+              uv
-        #   re-install them every time a change is made.
+              python313  # ?TODO^ how to set from `cpython` above?
-        # - boot a usable xonsh inside the poetry virtualenv when
+              pypkgs.pyqt6
-        #   defined via a custom entry point?
+              pypkgs.pyqt6-sip
-        devShells.default = p2npkgs.mkPoetryEnv {
+              pypkgs.qtpy
-        # env = p2npkgs.mkPoetryEnv {
+              pypkgs.qdarkstyle
-            projectDir = projectDir;
+              pypkgs.rapidfuzz
-            python = pkgs.python310;
+            ];
-            overrides = ahot_overrides;
+
-            editablePackageSources = packages;
+            shellHook = ''
-              # piker = "./";
+              # unmask to debug **this** dev-shell-hook
-              # tractor = "../tractor/";
+              # set -e
-            # };  # wut?
+
+              # set qt-base/plugin path(s)
+              QTBASE_PATH="${qt6baseStorePath}/lib"
+              QT_PLUGIN_PATH="${qt6baseStorePath}/lib/qt-6/plugins"
+              QT_QPA_PLATFORM_PLUGIN_PATH="$QT_PLUGIN_PATH/platforms"
+
+              # link in Qt cc lib paths from <nixpkgs>
+              LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QTBASE_PATH"
+              LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QT_PLUGIN_PATH"
+              LD_LIBRARY_PATH="$LD_LIBRARY_PATH:$QT_QPA_PLATFORM_PLUGIN_PATH"
+
+              # link-in c++ stdlib for various AOT-ext-pkgs (numpy, etc.)
+              LD_LIBRARY_PATH="${pkgs.stdenv.cc.cc.lib}/lib:$LD_LIBRARY_PATH"
+
+              export LD_LIBRARY_PATH
+
+              # RUNTIME-SETTINGS
+              #
+              # ------ Qt ------
+              # XXX, unmask to debug qt .so linking/loading deats
+              # export QT_DEBUG_PLUGINS=1
+              #
+              # ALSO, for *modern linux* DEs,
+              # - maybe set wayland-mode (TODO, parametrtize this!)
+              #  * a chosen wayland-mode shell-integration
+              export QT_QPA_PLATFORM="wayland"
+              export QT_WAYLAND_SHELL_INTEGRATION="xdg-shell"
+
+              # ------ uv ------
+              # - always use the ./py313/ venv-subdir
+              export UV_PROJECT_ENVIRONMENT="py313"
+              # sync project-env with all extras
+              uv sync --dev --all-extras --no-group lint
+
+              # ------ TIPS ------
+              # NOTE, to launch the py-venv installed `xonsh` (like @goodboy)
+              # run the `nix develop` cmd with,
+              # >> nix develop -c uv run xonsh
+            '';
           };
         }
-    );  # end of .outputs scope
+      );
+    };
 }

piker/accounting/_ledger.py

@@ -40,7 +40,7 @@ import tomli_w  # for fast ledger writing
 
 from piker.types import Struct
 from piker import config
-from ..log import get_logger
+from piker.log import get_logger
 from .calc import (
     iter_by_dt,
 )
@@ -239,7 +239,9 @@ class TransactionLedger(UserDict):
 
         symcache: SymbologyCache = self._symcache
         towrite: dict[str, Any] = {}
-        for tid, txdict in self.tx_sort(self.data.copy()):
+        for tid, txdict in self.tx_sort(
+            self.data.copy()
+        ):
             # write blank-str expiry for non-expiring assets
             if (
                 'expiry' in txdict
@@ -377,7 +379,7 @@ def open_trade_ledger(
         account,
         dirpath=_fp,
     )
-    cpy = ledger_dict.copy()
+    cpy: dict = ledger_dict.copy()
 
     # XXX NOTE: if not provided presume we are being called from
     # sync code and need to maybe run `trio` to generate..
@@ -406,7 +408,13 @@ def open_trade_ledger(
         account=account,
         mod=mod,
         symcache=symcache,
-        tx_sort=getattr(mod, 'tx_sort', tx_sort),
+
+        # NOTE: allow backends to provide custom ledger sorting
+        tx_sort=getattr(
+            mod,
+            'tx_sort',
+            tx_sort,
+        ),
     )
     try:
         yield ledger

piker/accounting/_mktinfo.py

@@ -305,8 +305,8 @@ class MktPair(Struct, frozen=True):
     # config right?
     # src_type: AssetTypeName
 
-    # for derivs, info describing contract, egs.
+    # for derivs, info describing contract, egs. strike price, call
-    # strike price, call or put, swap type, exercise model, etc.
+    # or put, swap type, exercise model, etc.
     contract_info: list[str] | None = None
 
     # TODO: rename to sectype since all of these can

piker/accounting/_pos.py

@@ -30,7 +30,8 @@ from types import ModuleType
 from typing import (
     Any,
     Iterator,
-    Generator
+    Generator,
+    TYPE_CHECKING,
 )
 
 import pendulum
@@ -59,8 +60,10 @@ from ..clearing._messages import (
     BrokerdPosition,
 )
 from piker.types import Struct
+from piker.log import get_logger
+
+if TYPE_CHECKING:
     from piker.data._symcache import SymbologyCache
-from ..log import get_logger
 
 log = get_logger(__name__)
 
@@ -502,6 +505,17 @@ class Account(Struct):
 
         _mktmap_table: dict[str, MktPair] | None = None,
 
+        only_require: list[str]|True = True,
+        # ^list of fqmes that are "required" to be processed from
+        # this ledger pass; we often don't care about others and
+        # definitely shouldn't always error in such cases.
+        # (eg. broker backend loaded that doesn't yet supsport the
+        # symcache but also, inside the paper engine we don't ad-hoc
+        # request `get_mkt_info()` for every symbol in the ledger,
+        # only the one for which we're simulating against).
+        # TODO, not sure if there's a better soln for this, ideally
+        # all backends get symcache support afap i guess..
+
     ) -> dict[str, Position]:
         '''
         Update the internal `.pps[str, Position]` table from input
@@ -544,11 +558,32 @@ class Account(Struct):
                 if _mktmap_table is None:
                     raise
 
+                required: bool = (
+                    only_require is True
+                    or (
+                        only_require is not True
+                        and
+                        fqme in only_require
+                    )
+                )
                 # XXX: caller is allowed to provide a fallback
                 # mktmap table for the case where a new position is
                 # being added and the preloaded symcache didn't
                 # have this entry prior (eg. with frickin IB..)
-                mkt = _mktmap_table[fqme]
+                if (
+                    not (mkt := _mktmap_table.get(fqme))
+                    and
+                    required
+                ):
+                    raise
+
+                elif not required:
+                    continue
+
+                else:
+                    # should be an entry retreived somewhere
+                    assert mkt
+
 
             if not (pos := pps.get(bs_mktid)):
 
@@ -665,7 +700,7 @@ class Account(Struct):
     def write_config(self) -> None:
         '''
         Write the current account state to the user's account TOML file, normally
-        something like ``pps.toml``.
+        something like `pps.toml`.
 
         '''
         # TODO: show diff output?

piker/accounting/calc.py

@@ -268,9 +268,6 @@ def iter_by_dt(
                     (v := tx.get(k))
                 )
             ):
-                # TODO? remove yah?
-                # v = tx[k] if isdict else tx.dt
-
                 # only call parser on the value if not None from
                 # the `parsers` table above (when NOT using
                 # `.get()`), otherwise pass through the value and
@@ -287,24 +284,50 @@ def iter_by_dt(
                 return ret
 
             else:
+                log.debug(
+                    f'Parser-field not found in txn\n'
+                    f'\n'
+                    f'parser-field: {k!r}\n'
+                    f'txn: {tx!r}\n'
+                    f'\n'
+                    f'Trying next..\n'
+                )
                 continue
 
-        # XXX: should never get here..
+        # XXX: we should never really get here bc it means some kinda
+        # bad txn-record (field) data..
+        #
+        # -> set the `debug_mode = True` if you want to trace such
+        # cases from REPL ;)
         else:
-            with maybe_open_crash_handler(pdb=True):
+            # XXX: we should really never get here..
-                raise ValueError(
+            # only if a ledger record has no expected sort(able)
-                    f'Invalid txn time ??\n'
+            # field will we likely hit this.. like with ze IB.
-                    f'txn-id: {k!r}\n'
+            # if no sortable field just deliver epoch?
-                    f'{k!r}: {v!r}\n'
+            log.warning(
+                'No (time) sortable field for TXN:\n'
+                f'{tx!r}\n'
             )
-                    # assert v is not None, f'No valid value for `{k}`!?'
+            report: str = (
+                f'No supported time-field found in txn !?\n'
+                f'\n'
+                f'supported-time-fields: {parsers!r}\n'
+                f'\n'
+                f'txn: {tx!r}\n'
+            )
+            if debug:
+                with maybe_open_crash_handler(
+                    pdb=debug,
+                    raise_on_exit=False,
+                ):
+                    raise ValueError(report)
+            else:
+                log.error(report)
 
             if _invalid is not None:
                 _invalid.append(tx)
                 return from_timestamp(0.)
 
-                # breakpoint()
-
     entry: tuple[str, dict]|Transaction
     invalid: list = []
     for entry in sorted(
@@ -318,8 +341,6 @@ def iter_by_dt(
             log.warning(
                 f'Ignoring txn w invalid timestamp ??\n'
                 f'{pformat(entry)}\n'
-                # f'txn-id: {k!r}\n'
-                # f'{k!r}: {v!r}\n'
             )
             continue
 
@@ -400,7 +421,10 @@ def open_ledger_dfs(
     can update the ledger on exit.
 
     '''
-    with maybe_open_crash_handler(pdb=debug_mode):
+    with maybe_open_crash_handler(
+        pdb=debug_mode,
+        # raise_on_exit=False,
+    ):
         if not ledger:
             import time
             from ._ledger import open_trade_ledger

piker/accounting/cli.py

@@ -300,7 +300,8 @@ def disect(
         assert not df.is_empty()
 
         # muck around in pdbp REPL
-        breakpoint()
+        # tractor.devx.mk_pdb().set_trace()
+        # breakpoint()
 
         # TODO: we REALLY need a better console REPL for this
         # kinda thing..

piker/brokers/__init__.py

@@ -51,7 +51,6 @@ __brokers__: list[str] = [
     'ib',
     'kraken',
     'kucoin',
-    'deribit',
 
     # broken but used to work
     # 'questrade',
@@ -62,6 +61,7 @@ __brokers__: list[str] = [
     # wstrade
     # iex
 
+    # deribit
     # bitso
 ]
 
@@ -98,13 +98,14 @@ async def open_cached_client(
     If one has not been setup do it and cache it.
 
     '''
-    brokermod = get_brokermod(brokername)
+    brokermod: ModuleType = get_brokermod(brokername)
+
+    # TODO: make abstract or `typing.Protocol`
+    # client: Client
     async with maybe_open_context(
         acm_func=brokermod.get_client,
         kwargs=kwargs,
-
     ) as (cache_hit, client):
-
         if cache_hit:
             log.runtime(f'Reusing existing {client}')
 

piker/brokers/binance/feed.py

@@ -94,18 +94,21 @@ class L1(Struct):
 
 
 # validation type
+# https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams/Aggregate-Trade-Streams#response-example
 class AggTrade(Struct, frozen=True):
     e: str  # Event type
     E: int  # Event time
     s: str  # Symbol
     a: int  # Aggregate trade ID
     p: float  # Price
-    q: float  # Quantity
+    q: float  # Quantity with all the market trades
     f: int  # First trade ID
     l: int  # noqa Last trade ID
     T: int  # Trade time
     m: bool  # Is the buyer the market maker?
     M: bool|None = None  # Ignore
+    nq: float|None = None # Normal quantity without the trades involving RPI orders
+    # ^XXX https://developers.binance.com/docs/derivatives/change-log#2025-12-29
 
 
 async def stream_messages(

piker/brokers/binance/venues.py

@@ -104,6 +104,9 @@ class Pair(Struct, frozen=True, kw_only=True):
     # https://developers.binance.com/docs/binance-spot-api-docs#future-changes
     pegInstructionsAllowed: bool = False
 
+    # https://developers.binance.com/docs/binance-spot-api-docs#2025-12-02
+    opoAllowed: bool = False
+
     filters: dict[
         str,
         str | int | float,
@@ -220,7 +223,10 @@ class FutesPair(Pair):
                 assert pair == self.pair  # sanity
                 return f'{expiry}'
 
-            case 'PERPETUAL':
+            case (
+                'PERPETUAL'
+                | 'TRADIFI_PERPETUAL'
+            ):
                 return 'PERP'
 
             case '':
@@ -249,7 +255,10 @@ class FutesPair(Pair):
         margin: str = self.marginAsset
 
         match ctype:
-            case 'PERPETUAL':
+            case (
+                'PERPETUAL'
+                | 'TRADIFI_PERPETUAL'
+            ):
                 return f'{margin}M'
 
             case (

piker/brokers/cli.py

@@ -471,11 +471,15 @@ def search(
 
     '''
     # global opts
-    brokermods = list(config['brokermods'].values())
+    brokermods: list[ModuleType] = list(config['brokermods'].values())
+
+    # TODO: this is coming from the `search --pdb` NOT from
+    # the `piker --pdb` XD ..
+    # -[ ] pull from the parent click ctx's values..dumdum
+    # assert pdb
 
     # define tractor entrypoint
     async def main(func):
-
         async with maybe_open_pikerd(
             loglevel=config['loglevel'],
             debug_mode=pdb,

piker/brokers/core.py

@@ -22,7 +22,9 @@ routines should be primitive data types where possible.
 """
 import inspect
 from types import ModuleType
-from typing import List, Dict, Any, Optional
+from typing import (
+    Any,
+)
 
 import trio
 
@@ -34,8 +36,10 @@ from ..accounting import MktPair
 
 
 async def api(brokername: str, methname: str, **kwargs) -> dict:
-    """Make (proxy through) a broker API call by name and return its result.
+    '''
-    """
+    Make (proxy through) a broker API call by name and return its result.
+
+    '''
     brokermod = get_brokermod(brokername)
     async with brokermod.get_client() as client:
         meth = getattr(client, methname, None)
@@ -62,10 +66,14 @@ async def api(brokername: str, methname: str, **kwargs) -> dict:
 
 async def stocks_quote(
     brokermod: ModuleType,
-    tickers: List[str]
+    tickers: list[str]
-) -> Dict[str, Dict[str, Any]]:
+
-    """Return quotes dict for ``tickers``.
+) -> dict[str, dict[str, Any]]:
-    """
+    '''
+    Return a `dict` of snapshot quotes for the provided input
+    `tickers`: a `list` of fqmes.
+
+    '''
     async with brokermod.get_client() as client:
         return await client.quote(tickers)
 
@@ -74,13 +82,15 @@ async def stocks_quote(
 async def option_chain(
     brokermod: ModuleType,
     symbol: str,
-    date: Optional[str] = None,
+    date: str|None = None,
-) -> Dict[str, Dict[str, Dict[str, Any]]]:
+) -> dict[str, dict[str, dict[str, Any]]]:
-    """Return option chain for ``symbol`` for ``date``.
+    '''
+    Return option chain for ``symbol`` for ``date``.
 
     By default all expiries are returned. If ``date`` is provided
     then contract quotes for that single expiry are returned.
-    """
+
+    '''
     async with brokermod.get_client() as client:
         if date:
             id = int((await client.tickers2ids([symbol]))[symbol])
@@ -98,7 +108,7 @@ async def option_chain(
 # async def contracts(
 #     brokermod: ModuleType,
 #     symbol: str,
-# ) -> Dict[str, Dict[str, Dict[str, Any]]]:
+# ) -> dict[str, dict[str, dict[str, Any]]]:
 #     """Return option contracts (all expiries) for ``symbol``.
 #     """
 #     async with brokermod.get_client() as client:
@@ -110,15 +120,24 @@ async def bars(
     brokermod: ModuleType,
     symbol: str,
     **kwargs,
-) -> Dict[str, Dict[str, Dict[str, Any]]]:
+) -> dict[str, dict[str, dict[str, Any]]]:
-    """Return option contracts (all expiries) for ``symbol``.
+    '''
-    """
+    Return option contracts (all expiries) for ``symbol``.
+
+    '''
     async with brokermod.get_client() as client:
         return await client.bars(symbol, **kwargs)
 
 
-async def search_w_brokerd(name: str, pattern: str) -> dict:
+async def search_w_brokerd(
+    name: str,
+    pattern: str,
+) -> dict:
 
+    # TODO: WHY NOT WORK!?!
+    # when we `step` through the next block?
+    # import tractor
+    # await tractor.pause()
     async with open_cached_client(name) as client:
 
         # TODO: support multiple asset type concurrent searches.
@@ -130,12 +149,12 @@ async def symbol_search(
     pattern: str,
     **kwargs,
 
-) -> Dict[str, Dict[str, Dict[str, Any]]]:
+) -> dict[str, dict[str, dict[str, Any]]]:
     '''
     Return symbol info from broker.
 
     '''
-    results = []
+    results: list[str] = []
 
     async def search_backend(
         brokermod: ModuleType
@@ -143,6 +162,13 @@ async def symbol_search(
 
         brokername: str = mod.name
 
+        # TODO: figure this the FUCK OUT
+        # -> ok so obvi in the root actor any async task that's
+        # spawned outside the main tractor-root-actor task needs to
+        # call this..
+        # await tractor.devx._debug.maybe_init_greenback()
+        # tractor.pause_from_sync()
+
         async with maybe_spawn_brokerd(
             mod.name,
             infect_asyncio=getattr(
@@ -162,7 +188,6 @@ async def symbol_search(
             ))
 
     async with trio.open_nursery() as n:
-
         for mod in brokermods:
             n.start_soon(search_backend, mod.name)
 
@@ -172,11 +197,13 @@ async def symbol_search(
 async def mkt_info(
     brokermod: ModuleType,
     fqme: str,
+
     **kwargs,
 
 ) -> MktPair:
     '''
-    Return MktPair info from broker including src and dst assets.
+    Return the `piker.accounting.MktPair` info struct from a given
+    backend broker tradable src/dst asset pair.
 
     '''
     async with open_cached_client(brokermod.name) as client:

piker/brokers/deribit/__init__.py

@@ -25,7 +25,6 @@ from .api import (
     get_client,
 )
 from .feed import (
-    get_mkt_info,
     open_history_client,
     open_symbol_search,
     stream_quotes,
@@ -35,20 +34,15 @@ from .feed import (
     # open_trade_dialog,
     # norm_trade_records,
 # )
-from .venues import (
-    OptionPair,
-)
 
 log = get_logger(__name__)
 
 __all__ = [
     'get_client',
 #    'trades_dialogue',
-    'get_mkt_info',
     'open_history_client',
     'open_symbol_search',
     'stream_quotes',
-    'OptionPair',
 #    'norm_trade_records',
 ]
 

piker/brokers/deribit/feed.py

@@ -18,59 +18,38 @@
 Deribit backend.
 
 '''
-from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
-from typing import (
+from typing import Any, Optional, Callable
-    # Any,
-    # Optional,
-    Callable,
-)
-# from pprint import pformat
 import time
 
-import cryptofeed
 import trio
 from trio_typing import TaskStatus
-from pendulum import (
+import pendulum
-    from_timestamp,
+from rapidfuzz import process as fuzzy
-)
 import numpy as np
 import tractor
 
-from piker.accounting import (
+from piker.brokers import open_cached_client
-    Asset,
+from piker.log import get_logger, get_console_log
-    MktPair,
+from piker.data import ShmArray
-    unpack_fqme,
+from piker.brokers._util import (
-)
+    BrokerError,
-from piker.brokers import (
-    open_cached_client,
-    NoData,
     DataUnavailable,
 )
-from piker._cacheables import (
-    async_lifo_cache,
-)
-from piker.log import (
-    get_logger,
-    mk_repr,
-)
-from piker.data.validate import FeedInit
 
+from cryptofeed import FeedHandler
+from cryptofeed.defines import (
+    DERIBIT, L1_BOOK, TRADES, OPTION, CALL, PUT
+)
+from cryptofeed.symbols import Symbol
 
 from .api import (
-    Client,
+    Client, Trade,
-    # get_config,
+    get_config,
-    piker_sym_to_cb_sym,
+    str_to_cb_sym, piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
-    cb_sym_to_deribit_inst,
-    str_to_cb_sym,
     maybe_open_price_feed
 )
-from .venues import (
-    Pair,
-    OptionPair,
-    Trade,
-)
 
 _spawn_kwargs = {
     'infect_asyncio': True,
@@ -85,215 +64,90 @@ async def open_history_client(
     mkt: MktPair,
 ) -> tuple[Callable, int]:
 
+    fnstrument: str = mkt.bs_fqme
     # TODO implement history getter for the new storage layer.
     async with open_cached_client('deribit') as client:
 
-        pair: OptionPair = client._pairs[mkt.dst.name]
-        # XXX NOTE, the cuckers use ms !!!
-        creation_time_s: int = pair.creation_timestamp/1000
-
         async def get_ohlc(
-            timeframe: float,
+            end_dt: Optional[datetime] = None,
-            end_dt: datetime | None = None,
+            start_dt: Optional[datetime] = None,
-            start_dt: datetime | None = None,
 
         ) -> tuple[
             np.ndarray,
             datetime,  # start
             datetime,  # end
         ]:
-            if timeframe != 60:
-                raise DataUnavailable('Only 1m bars are supported')
 
-            array: np.ndarray = await client.bars(
+            array = await client.bars(
-                mkt,
+                instrument,
                 start_dt=start_dt,
                 end_dt=end_dt,
             )
             if len(array) == 0:
-                if (
+                raise DataUnavailable
-                    end_dt is None
-                ):
-                    raise DataUnavailable(
-                        'No history seems to exist yet?\n\n'
-                        f'{mkt}'
-                    )
-                elif (
-                    end_dt
-                    and
-                    end_dt.timestamp() < creation_time_s
-                ):
-                    # the contract can't have history
-                    # before it was created.
-                    pair_type_str: str = type(pair).__name__
-                    create_dt: datetime = from_timestamp(creation_time_s)
-                    raise DataUnavailable(
-                        f'No history prior to\n'
-                        f'`{pair_type_str}.creation_timestamp: int = '
-                        f'{pair.creation_timestamp}\n\n'
-                        f'------ deribit sux ------\n'
-                        f'WHICH IN "NORMAL PEOPLE WHO USE EPOCH TIME" form is,\n'
-                        f'creation_time_s: {creation_time_s}\n'
-                        f'create_dt: {create_dt}\n'
-                    )
-                raise NoData(
-                    f'No frame for {start_dt} -> {end_dt}\n'
-                )
 
-            start_dt = from_timestamp(array[0]['time'])
+            start_dt = pendulum.from_timestamp(array[0]['time'])
-            end_dt = from_timestamp(array[-1]['time'])
+            end_dt = pendulum.from_timestamp(array[-1]['time'])
-
-            times = array['time']
-            if not times.any():
-                raise ValueError(
-                    'Bad frame with null-times?\n\n'
-                    f'{times}'
-                )
-
-            if end_dt is None:
-                inow: int = round(time.time())
-                if (inow - times[-1]) > 60:
-                    await tractor.pause()
 
             return array, start_dt, end_dt
 
-        yield (
+        yield get_ohlc, {'erlangs': 3, 'rate': 3}
-            get_ohlc,
-            {  # backfill config
-                'erlangs': 3,
-                'rate': 3,
-            }
-        )
-
-
-@async_lifo_cache()
-async def get_mkt_info(
-    fqme: str,
-
-) -> tuple[MktPair, Pair|OptionPair] | None:
-
-    # uppercase since kraken bs_mktid is always upper
-    if 'deribit' not in fqme.lower():
-        fqme += '.deribit'
-
-    mkt_mode: str = ''
-    broker, mkt_ep, venue, expiry = unpack_fqme(fqme)
-
-    # NOTE: we always upper case all tokens to be consistent with
-    # binance's symbology style for pairs, like `BTCUSDT`, but in
-    # theory we could also just keep things lower case; as long as
-    # we're consistent and the symcache matches whatever this func
-    # returns, always!
-    expiry: str = expiry.upper()
-    venue: str = venue.upper()
-    # venue_lower: str = venue.lower()
-
-    mkt_mode: str = 'option'
-
-    async with open_cached_client(
-        'deribit',
-    ) as client:
-
-        assets: dict[str, Asset] = await client.get_assets()
-        pair_str: str = mkt_ep.lower()
-
-        pair: Pair = await client.exch_info(
-            sym=pair_str,
-        )
-        mkt_mode = pair.venue
-        client.mkt_mode = mkt_mode
-
-        dst: Asset | None = assets.get(pair.bs_dst_asset)
-        src: Asset | None = assets.get(pair.bs_src_asset)
-
-        mkt = MktPair(
-            dst=dst,
-            src=src,
-            price_tick=pair.price_tick,
-            size_tick=pair.size_tick,
-            bs_mktid=pair.symbol,
-            venue=mkt_mode,
-            broker='deribit',
-            _atype=mkt_mode,
-            _fqme_without_src=True,
-
-            # expiry=pair.expiry,
-            # XXX TODO, currently we don't use it since it's
-            # already "described" in the `OptionPair.symbol: str`
-            # and if we slap in the ISO repr it's kinda hideous..
-            # -[ ] figure out the best either std
-        )
-        return mkt, pair
 
 
 async def stream_quotes(
+
     send_chan: trio.abc.SendChannel,
     symbols: list[str],
     feed_is_live: trio.Event,
+    loglevel: str = None,
 
     # startup sync
     task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,
 
 ) -> None:
-    '''
+    # XXX: required to propagate ``tractor`` loglevel to piker logging
-    Open a live quote stream for the market set defined by `symbols`.
+    get_console_log(loglevel or tractor.current_actor().loglevel)
 
-    Internally this starts a `cryptofeed.FeedHandler` inside an `asyncio`-side
+    sym = symbols[0]
-    task and relays through L1 and `Trade` msgs here to our `trio.Task`.
-
-    '''
-    sym = symbols[0].split('.')[0]
-    init_msgs: list[FeedInit] = []
-
-    # multiline nested `dict` formatter (since rn quote-msgs are
-    # just that).
-    pfmt: Callable[[str], str] = mk_repr(
-        # so we can see `deribit`'s delightfully mega-long bs fields..
-        maxstring=100,
-    )
 
     async with (
         open_cached_client('deribit') as client,
         send_chan as send_chan
     ):
-        mkt: MktPair
-        pair: Pair
-        mkt, pair = await get_mkt_info(sym)
 
-        # build out init msgs according to latest spec
+        init_msgs = {
-        init_msgs.append(
+            # pass back token, and bool, signalling if we're the writer
-            FeedInit(
+            # and that history has been written
-                mkt_info=mkt,
+            sym: {
-            )
+                'symbol_info': {
-        )
+                    'asset_type': 'option',
-        # build `cryptofeed` feed-handle
+                    'price_tick_size': 0.0005
-        cf_sym: cryptofeed.Symbol = piker_sym_to_cb_sym(sym)
+                },
+                'shm_write_opts': {'sum_tick_vml': False},
+                'fqsn': sym,
+            },
+        }
 
-        from_cf: tractor.to_asyncio.LinkedTaskChannel
+        nsym = piker_sym_to_cb_sym(sym)
-        async with maybe_open_price_feed(sym) as from_cf:
 
-            # load the "last trades" summary
+        async with maybe_open_price_feed(sym) as stream:
-            last_trades_res: cryptofeed.LastTradesResult = await client.last_trades(
-                cb_sym_to_deribit_inst(cf_sym),
-                count=1,
-            )
-            last_trades: list[Trade] = last_trades_res.trades
 
-            # TODO, do we even need this or will the above always
+            cache = await client.cache_symbols()
-            # work?
-            # if not last_trades:
-            #     await tractor.pause()
-            #     async for typ, quote in from_cf:
-            #         if typ == 'trade':
-            #             last_trade = Trade(**(quote['data']))
-            #             break
 
-            # else:
+            last_trades = (await client.last_trades(
-            last_trade = Trade(
+                cb_sym_to_deribit_inst(nsym), count=1)).trades
-                **(last_trades[0])
-            )
 
-            first_quote: dict = {
+            if len(last_trades) == 0:
+                last_trade = None
+                async for typ, quote in stream:
+                    if typ == 'trade':
+                        last_trade = Trade(**(quote['data']))
+                        break
+
+            else:
+                last_trade = Trade(**(last_trades[0]))
+
+            first_quote = {
                 'symbol': sym,
                 'last': last_trade.price,
                 'brokerd_ts': last_trade.timestamp,
@@ -304,84 +158,13 @@ async def stream_quotes(
                     'broker_ts': last_trade.timestamp
                 }]
             }
-            task_status.started((
+            task_status.started((init_msgs,  first_quote))
-                init_msgs,
-                first_quote,
-            ))
 
             feed_is_live.set()
 
-            # NOTE XXX, static for now!
+            async for typ, quote in stream:
-            # => since this only handles ONE mkt feed at a time we
+                topic = quote['symbol']
-            # don't need a lookup table to map interleaved quotes
+                await send_chan.send({topic: quote})
-            # from multiple possible mkt-pairs
-            topic: str = mkt.bs_fqme
-
-            # deliver until cancelled
-            async for typ, ref in from_cf:
-                match typ:
-                    case 'trade':
-                        trade: cryptofeed.types.Trade = ref
-
-                        # TODO, re-impl this according to teh ideal
-                        # fqme for opts that we choose!!
-                        bs_fqme: str = cb_sym_to_deribit_inst(
-                            str_to_cb_sym(trade.symbol)
-                        ).lower()
-
-                        piker_quote: dict = {
-                            'symbol': bs_fqme,
-                            'last': trade.price,
-                            'broker_ts': time.time(),
-                            # ^TODO, name this `brokerd/datad_ts` and
-                            # use `time.time_ns()` ??
-                            'ticks': [{
-                                'type': 'trade',
-                                'price': float(trade.price),
-                                'size': float(trade.amount),
-                                'broker_ts': trade.timestamp,
-                            }],
-                        }
-                        log.info(
-                            f'deribit {typ!r} quote for {sym!r}\n\n'
-                            f'{trade}\n\n'
-                            f'{pfmt(piker_quote)}\n'
-                        )
-
-                    case 'l1':
-                        book: cryptofeed.types.L1Book = ref
-
-                        # TODO, so this is where we can possibly change things
-                        # and instead lever the `MktPair.bs_fqme: str` output?
-                        bs_fqme: str = cb_sym_to_deribit_inst(
-                            str_to_cb_sym(book.symbol)
-                        ).lower()
-
-                        piker_quote: dict = {
-                            'symbol': bs_fqme,
-                            'ticks': [
-
-                                {'type': 'bid',
-                                 'price': float(book.bid_price),
-                                 'size': float(book.bid_size)},
-
-                                {'type': 'bsize',
-                                 'price': float(book.bid_price),
-                                 'size': float(book.bid_size),},
-
-                                {'type': 'ask',
-                                 'price': float(book.ask_price),
-                                 'size': float(book.ask_size),},
-
-                                {'type': 'asize',
-                                 'price': float(book.ask_price),
-                                 'size': float(book.ask_size),}
-                            ]
-                        }
-
-                await send_chan.send({
-                    topic: piker_quote,
-                })
 
 
 @tractor.context
@@ -391,21 +174,12 @@ async def open_symbol_search(
     async with open_cached_client('deribit') as client:
 
         # load all symbols locally for fast search
-        # cache = client._pairs
+        cache = await client.cache_symbols()
         await ctx.started()
 
         async with ctx.open_stream() as stream:
-            pattern: str
+
             async for pattern in stream:
-
+                # repack in dict form
-                # NOTE: pattern fuzzy-matching is done within
+                await stream.send(
-                # the methd impl.
+                    await client.search_symbols(pattern))
-                pairs: dict[str, Pair] = await client.search_symbols(
-                    pattern,
-                )
-                # repack in fqme-keyed table
-                byfqme: dict[str, Pair] = {}
-                for pair in pairs.values():
-                    byfqme[pair.bs_fqme] = pair
-
-                await stream.send(byfqme)

piker/brokers/deribit/venues.py

@@ -1,196 +0,0 @@
-# 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/>.
-
-"""
-Per market data-type definitions and schemas types.
-
-"""
-from __future__ import annotations
-import pendulum
-from typing import (
-    Literal,
-    Optional,
-)
-from decimal import Decimal
-
-from piker.types import Struct
-
-
-# API endpoint paths by venue / sub-API
-_domain: str = 'deribit.com'
-_url = f'https://www.{_domain}'
-
-# WEBsocketz
-_ws_url: str = f'wss://www.{_domain}/ws/api/v2'
-
-# test nets
-_testnet_ws_url: str = f'wss://test.{_domain}/ws/api/v2'
-
-MarketType = Literal[
-    'option'
-]
-
-
-def get_api_eps(venue: MarketType) -> tuple[str, str]:
-    '''
-    Return API ep root paths per venue.
-
-    '''
-    return {
-        'option': (
-            _ws_url,
-        ),
-    }[venue]
-
-
-class Pair(Struct, frozen=True, kw_only=True):
-
-    symbol: str
-
-    # src
-    quote_currency: str # 'BTC'
-
-    # dst
-    base_currency: str # "BTC",
-
-    tick_size: float # 0.0001 # [{'above_price': 0.005, 'tick_size': 0.0005}]
-    tick_size_steps: list[dict[str, float]] 
-
-    @property
-    def price_tick(self) -> Decimal:
-        return Decimal(str(self.tick_size_steps[0]['above_price']))
-
-    @property
-    def size_tick(self) -> Decimal:
-        return Decimal(str(self.tick_size))
-
-    @property
-    def bs_fqme(self) -> str:
-        return f'{self.symbol}'
-
-    @property
-    def bs_mktid(self) -> str:
-        return f'{self.symbol}.{self.venue}'
-
-
-class OptionPair(Pair, frozen=True):
-
-    taker_commission: float # 0.0003
-    strike: float # 5000.0
-    settlement_period: str # 'day'
-    settlement_currency: str # "BTC",
-    rfq: bool # false
-    price_index: str # 'btc_usd'
-    option_type: str # 'call'
-    min_trade_amount: float # 0.1
-    maker_commission: float # 0.0003
-    kind: str # 'option'
-    is_active: bool # true
-    instrument_type: str # 'reversed'
-    instrument_name: str # 'BTC-1SEP24-55000-C'
-    instrument_id: int # 364671
-    expiration_timestamp: int # 1725177600000
-    creation_timestamp: int # 1724918461000
-    counter_currency: str # 'USD' 
-    contract_size: float # '1.0'
-    block_trade_tick_size: float # '0.0001'
-    block_trade_min_trade_amount: int # '25'
-    block_trade_commission: float # '0.003'
-
-    # NOTE: see `.data._symcache.SymbologyCache.load()` for why
-    ns_path: str = 'piker.brokers.deribit:OptionPair'
-
-    # TODO, impl this without the MM:SS part of
-    # the `'THH:MM:SS..'` etc..
-    @property
-    def expiry(self) -> str:
-        iso_date = pendulum.from_timestamp(
-            self.expiration_timestamp / 1000
-        ).isoformat()
-        return iso_date 
-
-    @property
-    def venue(self) -> str:
-        return f'{self.instrument_type}_option'
-
-    @property
-    def bs_fqme(self) -> str:
-        return f'{self.symbol}'
-
-    @property
-    def bs_src_asset(self) -> str:
-        return f'{self.quote_currency}'
-
-    @property
-    def bs_dst_asset(self) -> str:
-        return f'{self.symbol}'
-
-
-PAIRTYPES: dict[MarketType, Pair] = {
-    'option': OptionPair,
-}
-
-
-class JSONRPCResult(Struct):
-    id: int
-    usIn: int
-    usOut: int
-    usDiff: int
-    testnet: bool
-    jsonrpc: str = '2.0'
-    error: Optional[dict] = None
-    result: Optional[list[dict]] = None
-
-
-class JSONRPCChannel(Struct):
-    method: str
-    params: dict
-    jsonrpc: str = '2.0'
-
-
-class KLinesResult(Struct):
-    low: list[float]
-    cost: list[float]
-    high: list[float]
-    open: list[float]
-    close: list[float]
-    ticks: list[int]
-    status: str
-    volume: list[float]
-
-
-class Trade(Struct):
-    iv: float
-    price: float
-    amount: float
-    trade_id: str
-    contracts: float
-    direction: str
-    trade_seq: int
-    timestamp: int
-    mark_price: float
-    index_price: float
-    tick_direction: int
-    instrument_name: str
-    combo_id: Optional[str] = '',
-    combo_trade_id: Optional[int] = 0,
-    block_trade_id: Optional[str] = '',
-    block_trade_leg_count: Optional[int] = 0,
-
-
-class LastTradesResult(Struct):
-    trades: list[Trade]
-    has_more: bool

piker/brokers/ib/_util.py

@@ -38,7 +38,6 @@ from piker.brokers._util import get_logger
 
 if TYPE_CHECKING:
     from .api import Client
-    from ib_insync import IB
     import i3ipc
 
 log = get_logger('piker.brokers.ib')
@@ -62,7 +61,7 @@ no_setup_msg:str = (
 
 
 def try_xdo_manual(
-    vnc_sockaddr: str,
+    client: Client,
 ):
     '''
     Do the "manual" `xdo`-based screen switch + click
@@ -79,6 +78,7 @@ def try_xdo_manual(
         _reset_tech = 'i3ipc_xdotool'
         return True
     except OSError:
+        vnc_sockaddr: str = client.conf.vnc_addrs
         log.exception(
             no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
         )
@@ -86,7 +86,6 @@ def try_xdo_manual(
 
 
 async def data_reset_hack(
-    # vnc_host: str,
     client: Client,
     reset_type: Literal['data', 'connection'],
 
@@ -118,36 +117,24 @@ async def data_reset_hack(
           that need to be wrangle.
 
     '''
-    ib_client: IB = client.ib
-
     # look up any user defined vnc socket address mapped from
     # a particular API socket port.
-    api_port: str = str(ib_client.client.port)
+    vnc_addrs: tuple[str]|None = client.conf.get('vnc_addrs')
-    vnc_host: str
+    if not vnc_addrs:
-    vnc_port: int
-    vnc_sockaddr: tuple[str] | None = client.conf.get('vnc_addrs')
-
-    if not vnc_sockaddr:
         log.warning(
-            no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
+            no_setup_msg.format(vnc_sockaddr=client.conf)
             +
             'REQUIRES A `vnc_addrs: array` ENTRY'
         )
 
-    vnc_host, vnc_port = vnc_sockaddr.get(
-        api_port,
-        ('localhost', 3003)
-    )
     global _reset_tech
-
     match _reset_tech:
         case 'vnc':
             try:
                 await tractor.to_asyncio.run_task(
                     partial(
                         vnc_click_hack,
-                        host=vnc_host,
+                        client=client,
-                        port=vnc_port,
                     )
                 )
             except (
@@ -158,29 +145,31 @@ async def data_reset_hack(
                     import i3ipc  # noqa  (since a deps dynamic check)
                 except ModuleNotFoundError:
                     log.warning(
-                        no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
+                        no_setup_msg.format(vnc_sockaddr=client.conf)
                     )
                     return False
 
-                if vnc_host not in {
+                # XXX, Xorg only workaround..
-                    'localhost',
+                # TODO? remove now that we have `pyvnc`?
-                    '127.0.0.1',
+                # if vnc_host not in {
-                }:
+                #     'localhost',
-                    focussed, matches = i3ipc_fin_wins_titled()
+                #     '127.0.0.1',
-                    if not matches:
+                # }:
-                        log.warning(
+                #     focussed, matches = i3ipc_fin_wins_titled()
-                            no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
+                #     if not matches:
-                        )
+                #         log.warning(
-                        return False
+                #             no_setup_msg.format(vnc_sockaddr=vnc_sockaddr)
-                    else:
+                #         )
-                        try_xdo_manual(vnc_sockaddr)
+                #         return False
+                #     else:
+                #         try_xdo_manual(vnc_sockaddr)
 
                 # localhost but no vnc-client or it borked..
                 else:
-                    try_xdo_manual(vnc_sockaddr)
+                    try_xdo_manual(client)
 
         case 'i3ipc_xdotool':
-            try_xdo_manual(vnc_sockaddr)
+            try_xdo_manual(client)
             # i3ipc_xdotool_manual_click_hack()
 
         case _ as tech:
@@ -191,15 +180,55 @@ async def data_reset_hack(
 
 
 async def vnc_click_hack(
-    host: str,
+    client: Client,
-    port: int,
+    reset_type: str = 'data',
-    reset_type: str = 'data'
+    pw: str|None = None,
+
 ) -> None:
     '''
     Reset the data or network connection for the VNC attached
     ib-gateway using a (magic) keybinding combo.
 
+    A vnc-server password can be set either by an input `pw` param or
+    set in the client's config with the latter loaded from the user's
+    `brokers.toml` in a vnc-addrs-port-mapping section,
+
+    .. code:: toml
+
+        [ib.vnc_addrs]
+        4002 = {host = 'localhost', port = 5900, pw = 'doggy'}
+
     '''
+    api_port: str = str(client.ib.client.port)
+    conf: dict = client.conf
+    vnc_addrs: dict[int, tuple] = conf.get('vnc_addrs')
+    if not vnc_addrs:
+        return None
+
+    addr_entry: dict|tuple = vnc_addrs.get(
+        api_port,
+        ('localhost', 5900)  # a typical default
+    )
+    if pw is None:
+        match addr_entry:
+            case (
+                host,
+                port,
+            ):
+                pass
+
+            case {
+                'host': host,
+                'port': port,
+                'pw': pw
+            }:
+                pass
+
+            case _:
+                raise ValueError(
+                    f'Invalid `ib.vnc_addrs` entry ?\n'
+                    f'{addr_entry!r}\n'
+                )
     try:
         from pyvnc import (
             AsyncVNCClient,
@@ -221,11 +250,12 @@ async def vnc_click_hack(
         'connection': 'r'
     }[reset_type]
 
+    with tractor.devx.open_crash_handler():
         client = await AsyncVNCClient.connect(
             VNCConfig(
                 host=host,
                 port=port,
-            password='doggy',
+                password=pw,
             )
         )
         async with client:

piker/brokers/ib/api.py

@@ -97,10 +97,6 @@ from ._util import (
     get_logger,
 )
 
-# ?TODO? this can now be removed since it was originally to extend
-# with a `bar_vwap` field that we removed from the default ohlcv
-# dtype since it's better calculated in an FSP func
-#
 _bar_load_dtype: list[tuple[str, type]] = [
     # NOTE XXX: only part that's diff
     # from our default fields where
@@ -948,6 +944,7 @@ class Client:
                     )
                     if tkr:
                         break
+
                 except TimeoutError as err:
                     timeouterr = err
                     await asyncio.sleep(0.01)
@@ -956,7 +953,9 @@ class Client:
                 else:
                     if not warnset:
                         log.warning(
-                            f'Quote req timed out..maybe venue is closed?\n'
+                            f'Quote req timed out..\n'
+                            f'Maybe the venue is closed?\n'
+                            f'\n'
                             f'{asdict(contract)}'
                         )
                         warnset = True
@@ -968,9 +967,11 @@ class Client:
                 )
                 break
         else:
-            if timeouterr and raise_on_timeout:
+            if (
-                import pdbp
+                timeouterr
-                pdbp.set_trace()
+                and
+                raise_on_timeout
+            ):
                 raise timeouterr
 
             if not warnset:

piker/brokers/ib/broker.py

@@ -117,7 +117,11 @@ def pack_position(
             symbol=fqme,
             currency=con.currency,
             size=float(pos.position),
-            avg_price=float(pos.avgCost) / float(con.multiplier or 1.0),
+            avg_price=(
+                float(pos.avgCost)
+                /
+                float(con.multiplier or 1.0)
+            ),
         ),
     )
 
@@ -563,7 +567,7 @@ async def open_trade_dialog(
         ledgers: dict[str, TransactionLedger] = {}
         tables: dict[str, Account] = {}
         order_msgs: list[Status] = []
-        conf = get_config()
+        conf: dict = get_config()
         accounts_def_inv: bidict[str, str] = bidict(
             conf['accounts']
         ).inverse

piker/brokers/ib/feed.py

@@ -613,7 +613,7 @@ async def get_bars(
                         data_cs.cancel()
 
                     # spawn new data reset task
-                    data_cs, reset_done = await nurse.start(
+                    data_cs, reset_done = await tn.start(
                         partial(
                             wait_on_data_reset,
                             proxy,
@@ -635,12 +635,12 @@ async def get_bars(
     unset_resetter: bool = False
     async with (
         tractor.trionics.collapse_eg(),
-        trio.open_nursery() as nurse
+        trio.open_nursery() as tn
     ):
 
         # start history request that we allow
         # to run indefinitely until a result is acquired
-        nurse.start_soon(query)
+        tn.start_soon(query)
 
         # start history reset loop which waits up to the timeout
         # for a result before triggering a data feed reset.
@@ -660,7 +660,7 @@ async def get_bars(
                 unset_resetter: bool = True
 
             # spawn new data reset task
-            data_cs, reset_done = await nurse.start(
+            data_cs, reset_done = await tn.start(
                 partial(
                     wait_on_data_reset,
                     proxy,
@@ -896,7 +896,10 @@ async def open_aio_quote_stream(
     symbol: str,
     contract: Contract|None = None,
 
-) -> trio.abc.ReceiveStream:
+) -> (
+    trio.abc.Channel|  # iface
+    tractor.to_asyncio.LinkedTaskChannel  # actually
+):
     '''
     Open a real-time `Ticker` quote stream from an `asyncio.Task`
     spawned via `tractor.to_asyncio.open_channel_from()`, deliver the
@@ -919,6 +922,7 @@ async def open_aio_quote_stream(
             yield from_aio
             return
 
+    from_aio: tractor.to_asyncio.LinkedTaskChannel
     async with tractor.to_asyncio.open_channel_from(
         _setup_quote_stream,
         symbol=symbol,
@@ -1079,7 +1083,8 @@ async def stream_quotes(
         con: Contract = details.contract
         first_ticker: Ticker|None = None
 
-        with trio.move_on_after(1.6) as quote_cs:
+        timeout: float = 1.6
+        with trio.move_on_after(timeout) as quote_cs:
             first_ticker: Ticker = await proxy.get_quote(
                 contract=con,
                 raise_on_timeout=False,
@@ -1088,7 +1093,9 @@ async def stream_quotes(
         # XXX should never happen with this ep right?
         # but if so then, more then likely mkt is closed?
         if quote_cs.cancelled_caught:
-            await tractor.pause()
+            log.warning(
+                f'First quote req timed out after {timeout!r}s'
+            )
 
         if first_ticker:
             first_quote: dict = normalize(first_ticker)
@@ -1161,6 +1168,7 @@ async def stream_quotes(
         )
         cs: trio.CancelScope|None = None
         startup: bool = True
+        iter_quotes: trio.abc.Channel
         while (
             startup
             or
@@ -1169,11 +1177,11 @@ async def stream_quotes(
             with trio.CancelScope() as cs:
                 async with (
                     tractor.trionics.collapse_eg(),
-                    trio.open_nursery() as nurse,
+                    trio.open_nursery() as tn,
                     open_aio_quote_stream(
                         symbol=sym,
                         contract=con,
-                    ) as stream,
+                    ) as iter_quotes,
                 ):
                     # ?TODO? can we rm this - particularly for `ib_async`?
                     # ugh, clear ticks since we've consumed them
@@ -1202,9 +1210,9 @@ async def stream_quotes(
                         await rt_ev.wait()
                         cs.cancel()  # cancel called should now be set
 
-                    nurse.start_soon(reset_on_feed)
+                    tn.start_soon(reset_on_feed)
 
-                    async with aclosing(stream):
+                    async with aclosing(iter_quotes):
                         # if syminfo.get('no_vlm', False):
                         if not init_msg.shm_write_opts['has_vlm']:
 
@@ -1219,19 +1227,21 @@ async def stream_quotes(
                             # wait for real volume on feed (trading might be
                             # closed)
                             while True:
-                                ticker = await stream.receive()
+                                ticker = await iter_quotes.receive()
 
                                 # for a real volume contract we rait for
                                 # the first "real" trade to take place
                                 if (
                                     # not calc_price
                                     # and not ticker.rtTime
-                                    not ticker.rtTime
+                                    False
+                                    # not ticker.rtTime
                                 ):
                                     # spin consuming tickers until we
                                     # get a real market datum
                                     log.debug(f"New unsent ticker: {ticker}")
                                     continue
+
                                 else:
                                     log.debug("Received first volume tick")
                                     # ugh, clear ticks since we've
@@ -1247,13 +1257,18 @@ async def stream_quotes(
                             log.debug(f"First ticker received {quote}")
 
                         # tell data-layer spawner-caller that live
-                        # quotes are now streaming.
+                        # quotes are now active desptie not having
+                        # necessarily received a first vlm/clearing
+                        # tick.
+                        ticker = await iter_quotes.receive()
                         feed_is_live.set()
+                        fqme: str = quote['fqme']
+                        await send_chan.send({fqme: quote})
 
                         # last = time.time()
-                        async for ticker in stream:
+                        async for ticker in iter_quotes:
                             quote = normalize(ticker)
-                            fqme = quote['fqme']
+                            fqme: str = quote['fqme']
                             log.debug(
                                 f'Sending quote\n'
                                 f'{quote}'

piker/brokers/kraken/broker.py

@@ -549,7 +549,7 @@ async def open_trade_dialog(
             # to be reloaded.
             balances: dict[str, float] = await client.get_balances()
 
-            verify_balances(
+            await verify_balances(
                 acnt,
                 src_fiat,
                 balances,

piker/brokers/questrade.py

@@ -37,6 +37,12 @@ import tractor
 from async_generator import asynccontextmanager
 import numpy as np
 import wrapt
+
+# TODO, port to `httpx`/`trio-websocket` whenver i get back to
+# writing a proper ws-api streamer for this backend (since the data
+# feeds are free now) as per GH feat-req:
+# https://github.com/pikers/piker/issues/509
+#
 import asks
 
 from ..calc import humanize, percent_change

piker/clearing/_client.py

@@ -171,7 +171,6 @@ class OrderClient(Struct):
 
 
 async def relay_orders_from_sync_code(
-
     client: OrderClient,
     symbol_key: str,
     to_ems_stream: tractor.MsgStream,
@@ -245,6 +244,11 @@ async def open_ems(
 
     async with maybe_open_emsd(
         broker,
+        # XXX NOTE, LOL so this determines the daemon `emsd` loglevel
+        # then FYI.. that's kinda wrong no?
+        # -[ ] shouldn't it be set by `pikerd -l` or no?
+        # -[ ] would make a lot more sense to have a subsys ctl for
+        #     levels.. like `-l emsd.info` or something?
         loglevel=loglevel,
     ) as portal:
 

piker/clearing/_ems.py

@@ -655,7 +655,11 @@ class Router(Struct):
             flume = feed.flumes[fqme]
             first_quote: dict = flume.first_quote
             book: DarkBook = self.get_dark_book(broker)
-            book.lasts[fqme]: float = float(first_quote['last'])
+
+            if not (last := first_quote.get('last')):
+                last: float = flume.rt_shm.array[-1]['close']
+
+            book.lasts[fqme]: float = float(last)
 
             async with self.maybe_open_brokerd_dialog(
                 brokermod=brokermod,
@@ -718,7 +722,7 @@ class Router(Struct):
             subs = self.subscribers[sub_key]
 
         sent_some: bool = False
-        for client_stream in subs:
+        for client_stream in subs.copy():
             try:
                 await client_stream.send(msg)
                 sent_some = True
@@ -1014,6 +1018,10 @@ async def translate_and_relay_brokerd_events(
                 status_msg.brokerd_msg = msg
                 status_msg.src = msg.broker_details['name']
 
+                if not status_msg.req:
+                    # likely some order change state?
+                    await tractor.pause()
+                else:
                     await router.client_broadcast(
                         status_msg.req.symbol,
                         status_msg,

piker/clearing/_paper_engine.py

@@ -297,6 +297,8 @@ class PaperBoi(Struct):
 
         # transmit pp msg to ems
         pp: Position = self.acnt.pps[bs_mktid]
+        # TODO, this will break if `require_only=True` was passed to
+        # `.update_from_ledger()`
 
         pp_msg = BrokerdPosition(
             broker=self.broker,
@@ -653,6 +655,7 @@ async def open_trade_dialog(
                 # in) use manually constructed table from calling
                 # the `.get_mkt_info()` provider EP above.
                 _mktmap_table=mkt_by_fqme,
+                only_require=list(mkt_by_fqme),
             )
 
             pp_msgs: list[BrokerdPosition] = []

piker/clearing/_util.py

@@ -30,6 +30,7 @@ subsys: str = 'piker.clearing'
 
 log = get_logger(subsys)
 
+# TODO, oof doesn't this ignore the `loglevel` then???
 get_console_log = partial(
     get_console_log,
     name=subsys,

piker/cli/__init__.py

@@ -183,8 +183,8 @@ def pikerd(
                 registry_addrs=regaddrs,
                 loglevel=loglevel,
                 debug_mode=pdb,
-                enable_transports=['uds'],
+                # enable_transports=['uds'],
-                # enable_transports=['tcp'],
+                enable_transports=['tcp'],
             ) as service_mngr,
         ):
             assert service_mngr

piker/config.py

@@ -41,10 +41,13 @@ from .log import get_logger
 log = get_logger('broker-config')
 
 
-# XXX NOTE: taken from ``click`` since apparently they have some
+# XXX NOTE: taken from `click`
-# super weirdness with sigint and sudo..no clue
+# |_https://github.com/pallets/click/blob/main/src/click/utils.py#L449
-# we're probably going to slowly just modify it to our own version over
+#
-# time..
+# (since apparently they have some super weirdness with SIGINT and
+# sudo.. no clue we're probably going to slowly just modify it to our
+# own version over time..)
+#
 def get_app_dir(
     app_name: str,
     roaming: bool = True,
@@ -261,7 +264,7 @@ def load(
         MutableMapping,
     ] = tomllib.loads,
 
-    touch_if_dne: bool = False,
+    touch_if_dne: bool = True,
 
     **tomlkws,
 
@@ -270,7 +273,7 @@ def load(
     Load config file by name.
 
     If desired config is not in the top level piker-user config path then
-    pass the ``path: Path`` explicitly.
+    pass the `path: Path` explicitly.
 
     '''
     # create the $HOME/.config/piker dir if dne
@@ -285,7 +288,8 @@ def load(
 
     if (
         not path.is_file()
-        and touch_if_dne
+        and
+        touch_if_dne
     ):
         # only do a template if no path provided,
         # just touch an empty file with same name.

piker/data/_sampling.py

@@ -95,6 +95,12 @@ class Sampler:
     # history loading.
     incr_task_cs: trio.CancelScope | None = None
 
+    bcast_errors: tuple[Exception] = (
+        trio.BrokenResourceError,
+        trio.ClosedResourceError,
+        trio.EndOfChannel,
+    )
+
     # holds all the ``tractor.Context`` remote subscriptions for
     # a particular sample period increment event: all subscribers are
     # notified on a step.
@@ -258,14 +264,15 @@ class Sampler:
         subs: set
         last_ts, subs = pair
 
-        task = trio.lowlevel.current_task()
+        # NOTE, for debugging pub-sub issues
-        log.debug(
+        # task = trio.lowlevel.current_task()
-            f'SUBS {self.subscribers}\n'
+        # log.debug(
-            f'PAIR {pair}\n'
+        #     f'AlL-SUBS@{period_s!r}: {self.subscribers}\n'
-            f'TASK: {task}: {id(task)}\n'
+        #     f'PAIR: {pair}\n'
-            f'broadcasting {period_s} -> {last_ts}\n'
+        #     f'TASK: {task}: {id(task)}\n'
+        #     f'broadcasting {period_s} -> {last_ts}\n'
         #     f'consumers: {subs}'
-        )
+        # )
         borked: set[MsgStream] = set()
         sent: set[MsgStream] = set()
         while True:
@@ -282,13 +289,11 @@ class Sampler:
                         await stream.send(msg)
                         sent.add(stream)
 
-                    except (
+                    except self.bcast_errors as err:
-                        trio.BrokenResourceError,
-                        trio.ClosedResourceError,
-                        trio.EndOfChannel,
-                    ):
                         log.error(
-                            f'{stream._ctx.chan.uid} dropped connection'
+                            f'Connection dropped for IPC ctx\n'
+                            f'{stream._ctx}\n\n'
+                            f'Due to {type(err)}'
                         )
                         borked.add(stream)
                 else:
@@ -395,7 +400,8 @@ async def register_with_sampler(
                 finally:
                     if (
                         sub_for_broadcasts
-                        and subs
+                        and
+                        subs
                     ):
                         try:
                             subs.remove(stream)
@@ -562,8 +568,7 @@ async def open_sample_stream(
 
 
 async def sample_and_broadcast(
-
+    bus: _FeedsBus,
-    bus: _FeedsBus,  # noqa
     rt_shm: ShmArray,
     hist_shm: ShmArray,
     quote_stream: trio.abc.ReceiveChannel,
@@ -583,11 +588,33 @@ async def sample_and_broadcast(
 
     overruns = Counter()
 
+    # NOTE, only used for debugging live-data-feed issues, though
+    # this should be resolved more correctly in the future using the
+    # new typed-msgspec feats of `tractor`!
+    #
+    # XXX, a multiline nested `dict` formatter (since rn quote-msgs
+    # are just that).
+    # pfmt: Callable[[str], str] = mk_repr()
+
     # iterate stream delivered by broker
     async for quotes in quote_stream:
         # print(quotes)
 
-        # TODO: ``numba`` this!
+        # XXX WARNING XXX only enable for debugging bc ow can cost
+        # ALOT of perf with HF-feedz!!!
+        #
+        # log.info(
+        #     'Rx live quotes:\n'
+        #     f'{pfmt(quotes)}'
+        # )
+
+        # TODO,
+        # -[ ] `numba` or `cython`-nize this loop possibly?
+        #  |_alternatively could we do it in rust somehow by upacking
+        #    arrow msgs instead of using `msgspec`?
+        # -[ ] use `msgspec.Struct` support in new typed-msging from
+        #     `tractor` to ensure only allowed msgs are transmitted?
+        #
         for broker_symbol, quote in quotes.items():
             # TODO: in theory you can send the IPC msg *before* writing
             # to the sharedmem array to decrease latency, however, that
@@ -660,6 +687,21 @@ async def sample_and_broadcast(
             sub_key: str = broker_symbol.lower()
             subs: set[Sub] = bus.get_subs(sub_key)
 
+            # TODO, figure out how to make this useful whilst
+            # incoporating feed "pausing" ..
+            #
+            # if not subs:
+            #     all_bs_fqmes: list[str] = list(
+            #         bus._subscribers.keys()
+            #     )
+            #     log.warning(
+            #         f'No subscribers for {brokername!r} live-quote ??\n'
+            #         f'broker_symbol: {broker_symbol}\n\n'
+
+            #         f'Maybe the backend-sys symbol does not match one of,\n'
+            #         f'{pfmt(all_bs_fqmes)}\n'
+            #     )
+
             # NOTE: by default the broker backend doesn't append
             # it's own "name" into the fqme schema (but maybe it
             # should?) so we have to manually generate the correct
@@ -729,18 +771,14 @@ async def sample_and_broadcast(
                         if lags > 10:
                             await tractor.pause()
 
-                except (
+                except Sampler.bcast_errors as ipc_err:
-                    trio.BrokenResourceError,
-                    trio.ClosedResourceError,
-                    trio.EndOfChannel,
-                ):
                     ctx: Context = ipc._ctx
                     chan: Channel = ctx.chan
                     if ctx:
                         log.warning(
-                            'Dropped `brokerd`-quotes-feed connection:\n'
+                            f'Dropped `brokerd`-feed for {broker_symbol!r} due to,\n'
-                            f'{broker_symbol}:'
+                            f'x>) {ctx.cid}@{chan.uid}'
-                            f'{ctx.cid}@{chan.uid}'
+                            f'|_{ipc_err!r}\n\n'
                         )
                     if sub.throttle_rate:
                         assert ipc._closed
@@ -757,12 +795,11 @@ async def sample_and_broadcast(
 
 
 async def uniform_rate_send(
-
     rate: float,
     quote_stream: trio.abc.ReceiveChannel,
     stream: MsgStream,
 
-    task_status: TaskStatus = trio.TASK_STATUS_IGNORED,
+    task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
 
 ) -> None:
     '''
@@ -780,13 +817,16 @@ async def uniform_rate_send(
     https://gist.github.com/njsmith/7ea44ec07e901cb78ebe1dd8dd846cb9
 
     '''
-    # TODO: compute the approx overhead latency per cycle
+    # ?TODO? dynamically compute the **actual** approx overhead latency per cycle
-    left_to_sleep = throttle_period = 1/rate - 0.000616
+    # instead of this magic # bidinezz?
+    throttle_period: float = 1/rate - 0.000616
+    left_to_sleep: float = throttle_period
 
     # send cycle state
+    first_quote: dict|None
     first_quote = last_quote = None
-    last_send = time.time()
+    last_send: float = time.time()
-    diff = 0
+    diff: float = 0
 
     task_status.started()
     ticks_by_type: dict[
@@ -797,22 +837,28 @@ async def uniform_rate_send(
     clear_types = _tick_groups['clears']
 
     while True:
-
         # compute the remaining time to sleep for this throttled cycle
-        left_to_sleep = throttle_period - diff
+        left_to_sleep: float = throttle_period - diff
 
         if left_to_sleep > 0:
+            cs: trio.CancelScope
             with trio.move_on_after(left_to_sleep) as cs:
+                sym: str
+                last_quote: dict
                 try:
                     sym, last_quote = await quote_stream.receive()
                 except trio.EndOfChannel:
-                    log.exception(f"feed for {stream} ended?")
+                    log.exception(
+                        f'Live stream for feed for ended?\n'
+                        f'<=c\n'
+                        f'  |_[{stream!r}\n'
+                    )
                     break
 
-                diff = time.time() - last_send
+                diff: float = time.time() - last_send
 
                 if not first_quote:
-                    first_quote = last_quote
+                    first_quote: float = last_quote
                     # first_quote['tbt'] = ticks_by_type
 
                 if (throttle_period - diff) > 0:
@@ -873,11 +919,12 @@ async def uniform_rate_send(
         # TODO: now if only we could sync this to the display
         # rate timing exactly lul
         try:
-            await stream.send({sym: first_quote})
+            await stream.send({
+                sym: first_quote
+            })
         except tractor.RemoteActorError as rme:
             if rme.type is not tractor._exceptions.StreamOverrun:
                 raise
-
             ctx = stream._ctx
             chan = ctx.chan
             log.warning(
@@ -885,20 +932,28 @@ async def uniform_rate_send(
                 f'{sym}:{ctx.cid}@{chan.uid}'
             )
 
-        except (
+        # NOTE: any of these can be raised by `tractor`'s IPC
-            # NOTE: any of these can be raised by ``tractor``'s IPC
         # transport-layer and we want to be highly resilient
         # to consumers which crash or lose network connection.
         # I.e. we **DO NOT** want to crash and propagate up to
         # ``pikerd`` these kinds of errors!
-            trio.ClosedResourceError,
+        except (
-            trio.BrokenResourceError,
             ConnectionResetError,
-            trio.EndOfChannel,
+        ) + Sampler.bcast_errors as ipc_err:
-        ):
+            match ipc_err:
+                case trio.EndOfChannel():
+                    log.info(
+                        f'{stream} terminated by peer,\n'
+                        f'{ipc_err!r}'
+                    )
+                case _:
                     # if the feed consumer goes down then drop
                     # out of this rate limiter
-            log.warning(f'{stream} closed')
+                    log.warning(
+                        f'{stream} closed due to,\n'
+                        f'{ipc_err!r}'
+                    )
+
             await stream.aclose()
             return
 

piker/data/_sharedmem.py

@@ -19,11 +19,7 @@ NumPy compatible shared memory buffers for real-time IPC streaming.
 
 """
 from __future__ import annotations
-import hashlib
+from sys import byteorder
-from sys import (
-    byteorder,
-    platform,
-)
 import time
 from typing import Optional
 from multiprocessing.shared_memory import SharedMemory, _USE_POSIX
@@ -109,12 +105,11 @@ class _Token(Struct, frozen=True):
     which can be used to key a system wide post shm entry.
 
     '''
-    shm_name: str  # actual OS-level name (may be shortened on macOS)
+    shm_name: str  # this servers as a "key" value
     shm_first_index_name: str
     shm_last_index_name: str
     dtype_descr: tuple
     size: int  # in struct-array index / row terms
-    key: str | None = None  # original descriptive key (for lookup)
 
     @property
     def dtype(self) -> np.dtype:
@@ -123,31 +118,6 @@ class _Token(Struct, frozen=True):
     def as_msg(self):
         return self.to_dict()
 
-    def __eq__(self, other) -> bool:
-        '''
-        Compare tokens based on shm names and dtype, ignoring the key field.
-        The key field is only used for lookups, not for token identity.
-        '''
-        if not isinstance(other, _Token):
-            return False
-        return (
-            self.shm_name == other.shm_name
-            and self.shm_first_index_name == other.shm_first_index_name
-            and self.shm_last_index_name == other.shm_last_index_name
-            and self.dtype_descr == other.dtype_descr
-            and self.size == other.size
-        )
-
-    def __hash__(self) -> int:
-        '''Hash based on the same fields used in __eq__'''
-        return hash((
-            self.shm_name,
-            self.shm_first_index_name,
-            self.shm_last_index_name,
-            self.dtype_descr,
-            self.size,
-        ))
-
     @classmethod
     def from_msg(cls, msg: dict) -> _Token:
         if isinstance(msg, _Token):
@@ -178,31 +148,6 @@ def get_shm_token(key: str) -> _Token:
     return _known_tokens.get(key)
 
 
-def _shorten_key_for_macos(key: str) -> str:
-    '''
-    macOS has a 31 character limit for POSIX shared memory names.
-    Hash long keys to fit within this limit while maintaining uniqueness.
-
-    '''
-    # macOS shm_open() has a 31 char limit (PSHMNAMLEN)
-    # Use format: /p_<hash16> where hash is first 16 hex chars of sha256
-    # This gives us: / + p_ + 16 hex chars = 19 chars, well under limit
-    # We keep the 'p' prefix to indicate it's from piker
-    if len(key) <= 31:
-        return key
-
-    # Create a hash of the full key
-    key_hash = hashlib.sha256(key.encode()).hexdigest()[:16]
-    short_key = f'p_{key_hash}'
-
-    log.debug(
-        f'Shortened shm key for macOS:\n'
-        f'  original: {key} ({len(key)} chars)\n'
-        f'  shortened: {short_key} ({len(short_key)} chars)'
-    )
-    return short_key
-
-
 def _make_token(
     key: str,
     size: int,
@@ -214,24 +159,12 @@ def _make_token(
 
     '''
     dtype = def_iohlcv_fields if dtype is None else dtype
-
-    # On macOS, shorten keys that exceed the 31 character limit
-    if platform == 'darwin':
-        shm_name = _shorten_key_for_macos(key)
-        shm_first = _shorten_key_for_macos(key + "_first")
-        shm_last = _shorten_key_for_macos(key + "_last")
-    else:
-        shm_name = key
-        shm_first = key + "_first"
-        shm_last = key + "_last"
-
     return _Token(
-        shm_name=shm_name,
+        shm_name=key,
-        shm_first_index_name=shm_first,
+        shm_first_index_name=key + "_first",
-        shm_last_index_name=shm_last,
+        shm_last_index_name=key + "_last",
         dtype_descr=tuple(np.dtype(dtype).descr),
         size=size,
-        key=key,  # Store original key for lookup
     )
 
 
@@ -488,12 +421,7 @@ class ShmArray:
         if _USE_POSIX:
             # We manually unlink to bypass all the "resource tracker"
             # nonsense meant for non-SC systems.
-            name = self._shm.name
+            shm_unlink(self._shm.name)
-            try:
-                shm_unlink(name)
-            except FileNotFoundError:
-                # might be a teardown race here?
-                log.warning(f'Shm for {name} already unlinked?')
 
         self._first.destroy()
         self._last.destroy()
@@ -522,15 +450,8 @@ def open_shm_array(
     a = np.zeros(size, dtype=dtype)
     a['index'] = np.arange(len(a))
 
-    # Create token first to get the (possibly shortened) shm name
-    token = _make_token(
-        key=key,
-        size=size,
-        dtype=dtype,
-    )
-
     shm = SharedMemory(
-        name=token.shm_name,  # Use shortened name from token
+        name=key,
         create=True,
         size=a.nbytes
     )
@@ -542,6 +463,12 @@ def open_shm_array(
     array[:] = a[:]
     array.setflags(write=int(not readonly))
 
+    token = _make_token(
+        key=key,
+        size=size,
+        dtype=dtype,
+    )
+
     # create single entry arrays for storing an first and last indices
     first = SharedInt(
         shm=SharedMemory(
@@ -614,11 +541,10 @@ def attach_shm_array(
 
     '''
     token = _Token.from_msg(token)
-    # Use original key for _known_tokens lookup, shm_name for OS calls
+    key = token.shm_name
-    lookup_key = token.key if token.key else token.shm_name
 
-    if lookup_key in _known_tokens:
+    if key in _known_tokens:
-        assert _Token.from_msg(_known_tokens[lookup_key]) == token, "WTF"
+        assert _Token.from_msg(_known_tokens[key]) == token, "WTF"
 
     # XXX: ugh, looks like due to the ``shm_open()`` C api we can't
     # actually place files in a subdir, see discussion here:
@@ -629,7 +555,7 @@ def attach_shm_array(
     for _ in range(3):
         try:
             shm = SharedMemory(
-                name=token.shm_name,  # Use (possibly shortened) OS name
+                name=key,
                 create=False,
             )
             break
@@ -677,8 +603,8 @@ def attach_shm_array(
     # Stash key -> token knowledge for future queries
     # via `maybe_opepn_shm_array()` but only after we know
     # we can attach.
-    if lookup_key not in _known_tokens:
+    if key not in _known_tokens:
-        _known_tokens[lookup_key] = token
+        _known_tokens[key] = token
 
     # "close" attached shm on actor teardown
     tractor.current_actor().lifetime_stack.callback(sha.close)

piker/data/_symcache.py

@@ -31,6 +31,7 @@ from pathlib import Path
 from pprint import pformat
 from typing import (
     Any,
+    Callable,
     Sequence,
     Hashable,
     TYPE_CHECKING,
@@ -56,7 +57,7 @@ from piker.brokers import (
 )
 
 if TYPE_CHECKING:
-    from ..accounting import (
+    from piker.accounting import (
         Asset,
         MktPair,
     )
@@ -161,19 +162,36 @@ class SymbologyCache(Struct):
                     'Implement `Client.get_assets()`!'
                 )
 
-            if get_mkt_pairs := getattr(client, 'get_mkt_pairs', None):
+            get_mkt_pairs: Callable|None = getattr(
+                client,
+                'get_mkt_pairs',
+                None,
+            )
+            if not get_mkt_pairs:
+                log.warning(
+                    'No symbology cache `Pair` support for `{provider}`..\n'
+                    'Implement `Client.get_mkt_pairs()`!'
+                )
+                return self
 
             pairs: dict[str, Struct] = await get_mkt_pairs()
-                for bs_fqme, pair in pairs.items():
+            if not pairs:
+                log.warning(
+                    'No pairs from intial {provider!r} sym-cache request?\n\n'
+                    '`Client.get_mkt_pairs()` -> {pairs!r} ?'
+                )
+                return self
 
-                    # NOTE: every backend defined pair should
+            for bs_fqme, pair in pairs.items():
-                    # declare it's ns path for roundtrip
-                    # serialization lookup.
                 if not getattr(pair, 'ns_path', None):
+                    # XXX: every backend defined pair must declare
+                    # a `.ns_path: tractor.NamespacePath` to enable
+                    # roundtrip serialization lookup from a local
+                    # cache file.
                     raise TypeError(
                         f'Pair-struct for {self.mod.name} MUST define a '
-                            '`.ns_path: str`!\n'
+                        '`.ns_path: str`!\n\n'
-                            f'{pair}'
+                        f'{pair!r}'
                     )
 
                 entry = await self.mod.get_mkt_info(pair.bs_fqme)
@@ -207,12 +225,6 @@ class SymbologyCache(Struct):
                 pair,
             )
 
-            else:
-                log.warning(
-                    'No symbology cache `Pair` support for `{provider}`..\n'
-                    'Implement `Client.get_mkt_pairs()`!'
-                )
-
         return self
 
     @classmethod

piker/data/feed.py

@@ -357,7 +357,9 @@ async def allocate_persistent_feed(
 
     # yield back control to starting nursery once we receive either
     # some history or a real-time quote.
-    log.info(f'loading OHLCV history: {fqme}')
+    log.info(
+        f'loading OHLCV history: {fqme!r}\n'
+    )
     await some_data_ready.wait()
 
     flume = Flume(
@@ -794,7 +796,6 @@ async def install_brokerd_search(
 
 @acm
 async def maybe_open_feed(
-
     fqmes: list[str],
     loglevel: str | None = None,
 
@@ -848,7 +849,6 @@ async def maybe_open_feed(
 
 @acm
 async def open_feed(
-
     fqmes: list[str],
 
     loglevel: str|None = None,

piker/data/flows.py

@@ -36,10 +36,10 @@ from ._sharedmem import (
     ShmArray,
     _Token,
 )
+from piker.accounting import MktPair
 
 if TYPE_CHECKING:
-    from ..accounting import MktPair
+    from piker.data.feed import Feed
-    from .feed import Feed
 
 
 class Flume(Struct):

piker/data/validate.py

@@ -113,9 +113,9 @@ def validate_backend(
             )
             if ep is None:
                 log.warning(
-                    f'Provider backend {mod.name} is missing '
+                    f'Provider backend {mod.name!r} is missing '
-                    f'{daemon_name} support :(\n'
+                    f'{daemon_name!r} support?\n'
-                    f'The following endpoint is missing: {name}'
+                    f'|_module endpoint-func missing: {name!r}\n'
                 )
 
     inits: list[

piker/log.py

@@ -18,8 +18,8 @@
 Log like a forester!
 """
 import logging
-import reprlib
 import json
+import reprlib
 from typing import (
     Callable,
 )
@@ -90,6 +90,8 @@ def colorize_json(
     )
 
 
+# TODO, eventually defer to the version in `modden` once
+# it becomes a dep!
 def mk_repr(
     **repr_kws,
 ) -> Callable[[str], str]:

piker/storage/__init__.py

@@ -138,16 +138,6 @@ class StorageClient(
     ) -> None:
         ...
 
-    async def write_oi(
-        self,
-        fqme: str,
-        oi: np.ndarray,
-        append_and_duplicate: bool = True,
-        limit: int = int(800e3),
-
-    ) -> None:
-        ...
-
 
 class TimeseriesNotFound(Exception):
     '''

piker/storage/nativedb.py

@@ -111,24 +111,6 @@ def mk_ohlcv_shm_keyed_filepath(
     return path
 
 
-def mk_oi_shm_keyed_filepath(
-    fqme: str,
-    period: float | int,
-    datadir: Path,
-
-) -> Path:
-
-    if period < 1.:
-        raise ValueError('Sample period should be >= 1.!?')
-
-    path: Path = (
-        datadir
-        /
-        f'{fqme}.oi{int(period)}s.parquet'
-    )
-    return path
-
-
 def unpack_fqme_from_parquet_filepath(path: Path) -> str:
 
     filename: str = str(path.name)
@@ -190,11 +172,7 @@ class NativeStorageClient:
 
             key: str = path.name.rstrip('.parquet')
             fqme, _, descr = key.rpartition('.')
-            if 'ohlcv' in descr: 
             prefix, _, suffix = descr.partition('ohlcv')
-            elif 'oi' in descr:
-                prefix, _, suffix = descr.partition('oi')
-
             period: int = int(suffix.strip('s'))
 
             # cache description data
@@ -391,61 +369,6 @@ class NativeStorageClient:
             timeframe,
         )
 
-    def _write_oi(
-        self,
-        fqme: str,
-        oi: np.ndarray,
-
-    ) -> Path:
-        '''
-        Sync version of the public interface meth, since we don't
-        currently actually need or support an async impl.
-
-        '''
-        path: Path = mk_oi_shm_keyed_filepath(
-            fqme=fqme,
-            period=1,
-            datadir=self._datadir,
-        )
-        if isinstance(oi, np.ndarray):
-            new_df: pl.DataFrame = tsp.np2pl(oi)
-        else:
-            new_df = oi
-
-        if path.exists():
-            old_df = pl.read_parquet(path)
-            df = pl.concat([old_df, new_df])
-        else:
-            df = new_df
-
-        start = time.time()
-        df.write_parquet(path)
-        delay: float = round(
-            time.time() - start,
-            ndigits=6,
-        )
-        log.info(
-            f'parquet write took {delay} secs\n'
-            f'file path: {path}'
-        )
-        return path
-
-    async def write_oi(
-        self,
-        fqme: str,
-        oi: np.ndarray,
-
-    ) -> Path:
-        '''
-        Write input oi time series for fqme and sampling period
-        to (local) disk.
-
-        '''
-        return self._write_oi(
-            fqme,
-            oi,
-        )
-
     async def delete_ts(
         self,
         key: str,

piker/ui/_display.py

@@ -212,12 +212,7 @@ async def increment_history_view(
     hist_chart: ChartPlotWidget = ds.hist_chart
     hist_viz: Viz = ds.hist_viz
     # viz: Viz = ds.viz
-    # NOTE: On macOS, shm names are shortened to fit the 31-char limit,
+    assert 'hist' in hist_viz.shm.token['shm_name']
-    # so we can't reliably check for 'hist' in the name anymore.
-    # The important thing is that hist_viz is correctly assigned from ds.
-    # token = hist_viz.shm.token
-    # shm_key = token.get('key') or token['shm_name']
-    # assert 'hist' in shm_key
     # name: str = hist_viz.name
 
     # TODO: seems this is more reliable at keeping the slow

piker/ui/_window.py

@@ -37,7 +37,6 @@ from piker.ui.qt import (
     QStatusBar,
     QScreen,
     QCloseEvent,
-    QSettings,
 )
 from ..log import get_logger
 from ._style import _font_small, hcolor
@@ -182,13 +181,6 @@ class MainWindow(QMainWindow):
         self._status_label: QLabel = None
         self._size: tuple[int, int] | None = None
 
-        # restore window geometry from previous session
-        settings = QSettings('pikers', 'piker')
-        geometry = settings.value('windowGeometry')
-        if geometry is not None:
-            self.restoreGeometry(geometry)
-            log.debug('Restored window geometry from previous session')
-
     @property
     def mode_label(self) -> QLabel:
 
@@ -225,11 +217,6 @@ class MainWindow(QMainWindow):
         '''Cancel the root actor asap.
 
         '''
-        # save window geometry for next session
-        settings = QSettings('pikers', 'piker')
-        settings.setValue('windowGeometry', self.saveGeometry())
-        log.debug('Saved window geometry for next session')
-
         # raising KBI seems to get intercepted by by Qt so just use the system.
         os.kill(os.getpid(), signal.SIGINT)
 

piker/ui/qt.py

@@ -44,7 +44,6 @@ from PyQt6.QtCore import (
     QItemSelectionModel,
     pyqtBoundSignal,
     pyqtRemoveInputHook,
-    QSettings,
 )
 
 align_flag: EnumType = Qt.AlignmentFlag

pyproject.toml

@@ -63,10 +63,8 @@ dependencies = [
     "trio-util >=0.7.0, <0.8.0",
     "trio-websocket >=0.10.3, <0.11.0",
     "typer >=0.9.0, <1.0.0",
-    "rapidfuzz >=3.5.2, <4.0.0",
-    "pdbp >=1.5.0, <2.0.0",
     "trio >=0.27",
-    "pendulum >=3.0.0, <4.0.0",
+    "pendulum",
     "httpx >=0.27.0, <0.28.0",
     "cryptofeed >=2.4.0, <3.0.0",
     "pyarrow>=18.0.0",
@@ -78,48 +76,89 @@ dependencies = [
     "numba>=0.61.0",
     "pyvnc",
 ]
+# ------ dependencies ------
+# NOTE, by default we ship only a "headless" deps set bc
+# the `uis` group is not listed in the optional set.
 
-[project.optional-dependencies]
+# [optional-dependencies]
-uis = [
+# uis = []
+# ?TODO? really we should be able to mv this `uis` group
+# to be under [optional-dependencies] and then include
+# it in the dev deps?
 # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
-    # TODO: make sure the levenshtein shit compiles on nix..
+# -> uis should be included in pubbed pkgs.
-    # rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
+# [ ] uv seems to have no way to do this though?
-    "rapidfuzz >=3.2.0, <4.0.0",
-    "qdarkstyle >=3.0.2, <4.0.0",
-    "pyqt6 >=6.7.0, <7.0.0",
-    "pyqtgraph",
 
-    # for consideration,
+# TODO? move to a `uv.toml`?
-    # - 'visidata'
+[tool.uv]
-
+# https://docs.astral.sh/uv/reference/settings/#python-preference
-    # TODO: add an `--only daemon` group for running non-ui / pikerd
+python-preference = 'system'
-    # service tree in distributed mode B)
+# https://docs.astral.sh/uv/reference/settings/#python-downloads
-    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
+python-downloads = 'manual'
+# https://docs.astral.sh/uv/concepts/projects/dependencies/#default-groups
+default-groups = [
+  'uis',
 ]
+# ------ tool.uv ------
 
 [dependency-groups]
-# TODO: a toolset that makes debugging a `pikerd` service (tree) easy
+uis = [
-# to hack on directly using more or less the local env:
+  "pyqtgraph",
-# - xonsh + xxh
+  "qdarkstyle >=3.0.2, <4.0.0",
-# - rsyscall + pdbp
+  "pyqt6 >=6.7.0, <7.0.0",
-# - actor runtime control console like BEAM/OTP
-#
-# console ehancements and eventually remote debugging extras/helpers.
-# use `uv --dev` to enable
-dev = [
-    "pytest",
-    "elasticsearch >=8.9.0, <9.0.0",
-    "prompt-toolkit ==3.0.40",
-    "cython >=3.0.0, <4.0.0",
-    "greenback >=1.1.1, <2.0.0",
-    "ruff>=0.9.6",
-    "pyperclip>=1.9.0",
-    "i3ipc>=2.2.1",
 
-    # ?from git, see below.
+  # fuzzy search
-    "xonsh",
+  "rapidfuzz >=3.2.0, <4.0.0",
 ]
 
+# dev deps enabled by `uv --dev`
+# https://docs.astral.sh/uv/concepts/projects/dependencies/#development-dependencies
+dev = [
+  # https://docs.astral.sh/uv/concepts/projects/dependencies/#development-dependencies
+  "cython >=3.0.0, <4.0.0",
+
+  # nested deps-groups
+  # https://docs.astral.sh/uv/concepts/projects/dependencies/#nesting-groups
+  {include-group = 'uis'},
+  {include-group = 'repl'},
+  {include-group = 'testing'},
+  {include-group = 'de'},
+]
+repl = [
+  # `tractor`'s debugger
+  "pdbp >=1.8.2, <2.0.0",
+    "greenback >=1.1.1, <2.0.0",
+
+  # @goodboy's preferred console toolz
+  "xonsh",
+    "prompt-toolkit ==3.0.40",
+    "pyperclip>=1.9.0",
+
+  # ?TODO, new stuff to consider..
+  # "visidata"  # console numerics
+  # "xxh"  # for remote `xonsh`-ing
+  # "rsyscall"  # (eventual) optional `tractor` backend
+  # - an actor-runtime-ctl console like BEAM/OTP
+]
+testing = [
+  "pytest",
+]
+de = [ # (linux) specific DEs
+  "i3ipc>=2.2.1",
+]
+lint = [
+  # XXX, with flake.nix needs to be from nixpkgs
+  "ruff>=0.9.6"
+  # ; os_name != 'nixos' and platform_system != 'NixOS'",
+  # ?TODO? since ^ markers won't work, use a deps-flags to toggle for
+  # now.
+]
+dbs = [
+  "elasticsearch >=8.9.0, <9.0.0",
+]
+# ------ dependency-groups ------
+
+
 [tool.pytest.ini_options]
 # https://docs.pytest.org/en/stable/reference/reference.html#configuration-options
 testpaths = [
@@ -131,24 +170,20 @@ console_output_style = 'progress'
 # https://docs.pytest.org/en/stable/how-to/plugins.html#disabling-plugins-from-autoloading
 # https://docs.pytest.org/en/stable/how-to/plugins.html#deactivating-unregistering-a-plugin-by-name
 addopts = '-p no:xonsh'
+# ------ tool.pytest ------
 
 
 [project.scripts]
 piker = "piker.cli:cli"
 pikerd = "piker.cli:pikerd"
 ledger = "piker.accounting.cli:ledger"
+# ------ project.scripts ------
 
 [tool.hatch.build.targets.sdist]
 include = ["piker"]
-
 [tool.hatch.build.targets.wheel]
 include = ["piker"]
-
+# ------ tool.hatch ------
-
-# TODO? move to a `uv.toml`?
-[tool.uv]
-python-preference = 'system'
-python-downloads = 'manual'
 
 
 [tool.uv.sources]
@@ -156,12 +191,10 @@ pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
 tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
 pyvnc = { git = "https://github.com/regulad/pyvnc.git" }
 
-# TODO, long term we should be synced to upstream `main` branch!
+# XXX since, we're like, always hacking new shite all-the-time. Bp
-# tractor = { git = "https://github.com/goodboy/tractor.git", branch ="piker_pin" }
+tractor = { git = "https://github.com/goodboy/tractor.git", branch ="piker_pin" }
-tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "piker_pin" }
+# tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "piker_pin" }
-
+# tractor = { git = "https://pikers.dev/goodboy/tractor", branch = "main" }
-# goodboy's dev-env
+# ------ goodboy ------
-# XXX for @goodboy's hackin dev env, usually there's something new in
+# hackin dev-envs, usually there's something new he's hackin in..
-# the runtime being seriously tested here Bp
+# tractor = { path = "../tractor", editable = true }
-# tractor = { path = "../tractor/", editable = true }
-# xonsh = { path = "../xonsh", editable = true }