`accouning.calc`: enable crash handlers on `debug_mode` input (via test harness)

To start this is just a shell for the test, there's no checking logic
yet.. put it as `test_accounting.test_ib_account_with_duplicated_mktids()`.
The test is composed for now to be completely runtime-free using only
the offline txn-ledger / symcache / account loading APIs, ideally we
fill in the activated symbology-data-runtime cases once we figure a sane
way to handle incremental symcache updates for backends like IB..

To actually fill the test out with real checks we still need to,
- extract the problem account file from my ib.algopape into the test
  harness data.
- pick some contracts with multiple fqmes despite a single bs_mktid and
  ensure they're aggregated as a single `Position` as well as,
  * ideally de-duplicating txns from the account file section for the
    mkt..
  * warning appropriately about greater-then-one fqme for the bs_mktid
    and providing a way for the ledger re-writing to choose the
    appropriate `<venue>` as the "primary" when the
    data-symbology-runtime is up and possibly use it to incrementally
    update the IB symcache and store offline for next use?

piker/accounting/calc.py

@@ -32,6 +32,7 @@ from typing import (
     TYPE_CHECKING,
 )
 
+from tractor.devx import maybe_open_crash_handler
 import polars as pl
 from pendulum import (
     DateTime,
@@ -290,9 +291,7 @@ def iter_by_dt(
 
         # XXX: should never get here..
         else:
-            if debug:
-                import tractor
-                with tractor.devx.maybe_open_crash_handler():
+            with maybe_open_crash_handler(pdb=True):
                 raise ValueError(
                     f'Invalid txn time ??\n'
                     f'txn-id: {k!r}\n'
@@ -386,6 +385,7 @@ def open_ledger_dfs(
     acctname: str,
 
     ledger: TransactionLedger | None = None,
+    debug_mode: bool = False,
 
     **kwargs,
 
@@ -400,8 +400,7 @@ def open_ledger_dfs(
     can update the ledger on exit.
 
     '''
-    from piker.toolz import open_crash_handler
-    with open_crash_handler():
+    with maybe_open_crash_handler(pdb=debug_mode):
         if not ledger:
             import time
             from ._ledger import open_trade_ledger

piker/data/_symcache.py

@@ -90,6 +90,18 @@ class SymbologyCache(Struct):
     # provided by the backend pkg.
     mktmaps: dict[str, MktPair] = field(default_factory=dict)
 
+    def pformat(self) -> str:
+        return (
+            f'<{type(self).__name__}(\n'
+            f' .mod: {self.mod!r}\n'
+            f' .assets: {len(self.assets)!r}\n'
+            f' .pairs: {len(self.pairs)!r}\n'
+            f' .mktmaps: {len(self.mktmaps)!r}\n'
+            f')>'
+        )
+
+    __repr__ = pformat
+
     def write_config(self) -> None:
 
         # put the backend's pair-struct type ref at the top

tests/conftest.py

@@ -15,6 +15,12 @@ from piker.service import (
 from piker.log import get_console_log
 
 
+# include `tractor`'s built-in fixtures!
+pytest_plugins: tuple[str] = (
+    "tractor._testing.pytest",
+)
+
+
 def pytest_addoption(parser):
     parser.addoption("--ll", action="store", dest='loglevel',
                      default=None, help="logging level to set when testing")

tests/test_accounting.py

@@ -12,12 +12,14 @@ from piker import config
 from piker.accounting import (
     Account,
     calc,
-    Position,
-    TransactionLedger,
-    open_trade_ledger,
+    open_account,
     load_account,
     load_account_from_ledger,
+    open_trade_ledger,
+    Position,
+    TransactionLedger,
 )
+import tractor
 
 
 def test_root_conf_networking_section(
@@ -53,12 +55,17 @@ def test_account_file_default_empty(
 )
 def test_paper_ledger_position_calcs(
     fq_acnt: tuple[str, str],
+    debug_mode: bool,
 ):
     broker: str
     acnt_name: str
     broker, acnt_name = fq_acnt
 
-    accounts_path: Path = config.repodir() / 'tests' / '_inputs'
+    accounts_path: Path = (
+        config.repodir()
+        / 'tests'
+        / '_inputs'  # tests-local-subdir
+    )
 
     ldr: TransactionLedger
     with (
@@ -77,6 +84,7 @@ def test_paper_ledger_position_calcs(
             ledger=ldr,
 
             _fp=accounts_path,
+            debug_mode=debug_mode,
 
         ) as (dfs, ledger),
 
@@ -102,3 +110,87 @@ def test_paper_ledger_position_calcs(
         df = dfs[xrp]
         assert df['cumsize'][-1] == 0
         assert pos.cumsize == 0
+
+
+
+@pytest.mark.parametrize(
+    'fq_acnt',
+    [
+        ('ib', 'algopaper'),
+    ],
+)
+def test_ib_account_with_duplicated_mktids(
+    fq_acnt: tuple[str, str],
+    debug_mode: bool,
+):
+    # ?TODO, once we start symcache-incremental-update-support?
+    # from piker.data import (
+    #     open_symcache,
+    # )
+    #
+    # async def main():
+    #     async with (
+    #         # TODO: do this as part of `open_account()`!?
+    #         open_symcache(
+    #             'ib',
+    #             only_from_memcache=True,
+    #         ) as symcache,
+    #     ):
+
+
+    from piker.brokers.ib.ledger import (
+        tx_sort,
+
+        # ?TODO, once we want to pull lowlevel txns and process them?
+        # norm_trade_records,
+        # update_ledger_from_api_trades,
+    )
+
+    broker: str
+    acnt_id: str = 'algopaper'
+    broker, acnt_id = fq_acnt
+    accounts_def = config.load_accounts([broker])
+    assert accounts_def[f'{broker}.{acnt_id}']
+
+    ledger: TransactionLedger
+    acnt: Account
+    with (
+        tractor.devx.maybe_open_crash_handler(pdb=debug_mode),
+
+        open_trade_ledger(
+            'ib',
+            acnt_id,
+            tx_sort=tx_sort,
+
+            # TODO, eventually incrementally updated for IB..
+            # symcache=symcache,
+            symcache=None,
+            allow_from_sync_code=True,
+
+        ) as ledger,
+
+        open_account(
+            'ib',
+            acnt_id,
+            write_on_exit=True,
+        ) as acnt,
+    ):
+        # per input params
+        symcache = ledger.symcache
+        assert not (
+            symcache.pairs
+            or
+            symcache.pairs
+            or
+            symcache.mktmaps
+        )
+        # re-compute all positions that have changed state.
+        # TODO: likely we should change the API to return the
+        # position updates from `.update_from_ledger()`?
+        active, closed = acnt.dump_active()
+
+        # breakpoint()
+
+        # TODO, (see above imports as well) incremental update from
+        #      (updated) ledger?
+        # -[ ] pull some code from `.ib.broker` content.