Allow ledger passes to ignore (symcache) unknown fqmes

For example in the paper-eng, if you have a backend that doesn't fully support a symcache (yet) it's handy to be able to ignore processing other paper-eng txns when all you care about at the moment is the simulated symbol. NOTE, that currently this will still result in a key-error when you load more then one mkt with the paper engine (for which the backend does not have the symcache implemented) since no fqme ad-hoc query was made for the 2nd symbol (and i'm not sure we should support that kinda hackery over just encouraging the sym-cache being added?). Def needs a little more thought depending on how many backends are never going to be able to (easily) support caching..
.clearing._ems: Don't require `first_quote['last']`
2025-02-03 18:55:47 -03:00 · 2025-02-03 18:55:47 -03:00 · 2025-02-03 18:55:47 -03:00 · 2025-02-03 18:55:47 -03:00 · 2025-02-03 18:55:47 -03:00 · 2025-02-03 18:55:47 -03:00
6 changed files with 221 additions and 135 deletions
--- a/piker/accounting/_pos.py
+++ b/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.data._symcache import SymbologyCache
-from ..log import get_logger
+from piker.log import get_logger
+
+if TYPE_CHECKING:
+    from piker.data._symcache import SymbologyCache

 log = get_logger(__name__)

@ -493,6 +496,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
@ -535,11 +549,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)):

@ -656,7 +691,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?
--- a/piker/brokers/binance/venues.py
+++ b/piker/brokers/binance/venues.py
@ -181,7 +181,6 @@ class FutesPair(Pair):
    quoteAsset: str  # 'USDT',
    quotePrecision: int  # 8,
    requiredMarginPercent: float  # '5.0000',
-    settlePlan: int  # 0,
    timeInForce: list[str]  # ['GTC', 'IOC', 'FOK', 'GTX'],
    triggerProtect: float  # '0.0500',
    underlyingSubType: list[str]  # ['PoW'],
--- a/piker/brokers/kucoin.py
+++ b/piker/brokers/kucoin.py
@ -111,6 +111,10 @@ class KucoinMktPair(Struct, frozen=True):
    quoteMaxSize: float
    quoteMinSize: float
    symbol: str  # our bs_mktid, kucoin's internal id
+    feeCategory: int
+    makerFeeCoefficient: float
+    takerFeeCoefficient: float
+    st: bool


 class AccountTrade(Struct, frozen=True):
@ -593,7 +597,7 @@ async def get_client() -> AsyncGenerator[Client, None]:
    '''
    async with (
        httpx.AsyncClient(
-            base_url=f'https://api.kucoin.com/api',
+            base_url='https://api.kucoin.com/api',
        ) as trio_client,
    ):
        client = Client(httpx_client=trio_client)
@ -637,7 +641,7 @@ async def open_ping_task(
                await trio.sleep((ping_interval - 1000) / 1000)
                await ws.send_msg({'id': connect_id, 'type': 'ping'})

-        log.info('Starting ping task for kucoin ws connection')
+        log.warning('Starting ping task for kucoin ws connection')
        n.start_soon(ping_server)

        yield
@ -649,9 +653,14 @@ async def open_ping_task(
 async def get_mkt_info(
    fqme: str,

-) -> tuple[MktPair, KucoinMktPair]:
+) -> tuple[
+    MktPair,
+    KucoinMktPair,
+]:
    '''
-    Query for and return a `MktPair` and `KucoinMktPair`.
+    Query for and return both a `piker.accounting.MktPair` and
+    `KucoinMktPair` from provided `fqme: str`
+    (fully-qualified-market-endpoint).

    '''
    async with open_cached_client('kucoin') as client:
@ -726,6 +735,8 @@ async def stream_quotes(

        log.info(f'Starting up quote stream(s) for {symbols}')
        for sym_str in symbols:
+            mkt: MktPair
+            pair: KucoinMktPair
            mkt, pair = await get_mkt_info(sym_str)
            init_msgs.append(
                FeedInit(mkt_info=mkt)
@ -733,7 +744,11 @@ async def stream_quotes(

        ws: NoBsWs
        token, ping_interval = await client._get_ws_token()
-        connect_id = str(uuid4())
+        log.info('API reported ping_interval: {ping_interval}\n')
+
+        connect_id: str = str(uuid4())
+        typ: str
+        quote: dict
        async with (
            open_autorecon_ws(
                (
@ -747,20 +762,37 @@ async def stream_quotes(
                ),
            ) as ws,
            open_ping_task(ws, ping_interval, connect_id),
-            aclosing(stream_messages(ws, sym_str)) as msg_gen,
+            aclosing(
+                iter_normed_quotes(
+                    ws, sym_str
+                )
+            ) as iter_quotes,
        ):
-            typ, quote = await anext(msg_gen)
+            typ, quote = await anext(iter_quotes)

-            while typ != 'trade':
-                # take care to not unblock here until we get a real
-                # trade quote
-                typ, quote = await anext(msg_gen)
+            # take care to not unblock here until we get a real
+            # trade quote?
+            # ^TODO, remove this right?
+            # -[ ] what often blocks chart boot/new-feed switching
+            #   since we'ere waiting for a live quote instead of just
+            #   loading history afap..
+            #  |_ XXX, not sure if we require a bit of rework to core
+            #    feed init logic or if backends justg gotta be
+            #    changed up.. feel like there was some causality
+            #    dilema prolly only seen with IB too..
+            # while typ != 'trade':
+            #     typ, quote = await anext(iter_quotes)

            task_status.started((init_msgs, quote))
            feed_is_live.set()

-            async for typ, msg in msg_gen:
-                await send_chan.send({sym_str: msg})
+            # XXX NOTE, DO NOT include the `.<backend>` suffix!
+            # OW the sampling loop will not broadcast correctly..
+            # since `bus._subscribers.setdefault(bs_fqme, set())`
+            # is used inside `.data.open_feed_bus()` !!!
+            topic: str = mkt.bs_fqme
+            async for typ, quote in iter_quotes:
+                await send_chan.send({topic: quote})


@acm
@ -815,7 +847,7 @@ async def subscribe(
            )


-async def stream_messages(
+async def iter_normed_quotes(
    ws: NoBsWs,
    sym: str,

@ -846,6 +878,9 @@ async def stream_messages(

                yield 'trade', {
                    'symbol': sym,
+                    # TODO, is 'last' even used elsewhere/a-good
+                    # semantic? can't we just read the ticks with our
+                    # .data.ticktools.frame_ticks()`/
                    'last': trade_data.price,
                    'brokerd_ts': last_trade_ts,
                    'ticks': [
@ -938,7 +973,7 @@ async def open_history_client(
            if end_dt is None:
                inow = round(time.time())

-                print(
+                log.debug(
                    f'difference in time between load and processing'
                    f'{inow - times[-1]}'
                )
--- a/piker/clearing/_ems.py
+++ b/piker/clearing/_ems.py
@ -653,7 +653,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,
@ -716,7 +720,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
@ -1010,10 +1014,14 @@ async def translate_and_relay_brokerd_events(
                status_msg.brokerd_msg = msg
                status_msg.src = msg.broker_details['name']

-                await router.client_broadcast(
-                    status_msg.req.symbol,
-                    status_msg,
-                )
+                if not status_msg.req:
+                    # likely some order change state?
+                    await tractor.pause()
+                else:
+                    await router.client_broadcast(
+                        status_msg.req.symbol,
+                        status_msg,
+                    )

                if status == 'closed':
                    log.info(f'Execution for {oid} is complete!')
--- a/piker/clearing/_paper_engine.py
+++ b/piker/clearing/_paper_engine.py
@ -653,6 +653,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] = []
--- a/pyproject.toml
+++ b/pyproject.toml
@ -15,8 +15,8 @@
 # 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/>.
 [build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

 # ------ - ------

@ -34,114 +34,122 @@ ignore = []

 # ------ - ------

-[project]
+[tool.poetry]
 name = "piker"
-version = "0.1.0a0dev0"
+version = "0.1.0.alpha0.dev0"
 description = "trading gear for hackers"
-authors = [{ name = "Tyler Goodlet", email = "goodboy_foss@protonmail.com" }]
-requires-python = ">=3.12, <3.13"
-license = "AGPL-3.0-or-later"
+authors = ["Tyler Goodlet <goodboy_foss@protonmail.com>"]
+license = "AGPLv3"
 readme = "README.rst"
-keywords = [
-    "async",
-    "trading",
-    "finance",
-    "quant",
-    "charting",
+
+# ------ - ------
+
+[tool.poetry.dependencies]
+async-generator = "^1.10"
+attrs = "^23.1.0"
+bidict = "^0.22.1"
+colorama = "^0.4.6"
+colorlog = "^6.7.0"
+cython = "^3.0.0"
+greenback = "^1.1.1"
+ib-insync = "^0.9.86"
+msgspec = "^0.18.0"
+numba = "^0.59.0"
+numpy = "^1.25"
+polars = "^0.18.13"
+pygments = "^2.16.1"
+python = ">=3.11, <3.13"
+rich = "^13.5.2"
+# setuptools = "^68.0.0"
+tomli = "^2.0.1"
+tomli-w = "^1.0.0"
+trio-util = "^0.7.0"
+trio-websocket = "^0.10.3"
+typer = "^0.9.0"
+rapidfuzz = "^3.5.2"
+pdbp = "^1.5.0"
+trio = "^0.24"
+pendulum = "^3.0.0"
+httpx = "^0.27.0"
+
+[tool.poetry.dependencies.tractor]
+develop = true
+git = 'https://github.com/goodboy/tractor.git'
+branch = 'asyncio_debugger_support'
+# path = "../tractor"
+
+[tool.poetry.dependencies.asyncvnc]
+git = 'https://github.com/pikers/asyncvnc.git'
+branch = 'main'
+
+[tool.poetry.dependencies.tomlkit]
+develop = true
+git = 'https://github.com/pikers/tomlkit.git'
+branch = 'piker_pin'
+# path = "../tomlkit/"
+
+[tool.poetry.group.uis]
+optional = true
+[tool.poetry.group.uis.dependencies]
+# https://python-poetry.org/docs/managing-dependencies/#dependency-groups
+# TODO: make sure the levenshtein shit compiles on nix..
+# rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
+rapidfuzz = "^3.2.0"
+qdarkstyle = ">=3.0.2"
+pyqtgraph = { git = 'https://github.com/pikers/pyqtgraph.git' }
+
+# ------ - ------
+pyqt6 = "^6.7.0"
+
+[tool.poetry.group.dev]
+optional = true
+[tool.poetry.group.dev.dependencies]
+# testing / CI
+pytest = "^6.0.0"
+elasticsearch = "^8.9.0"
+xonsh = "^0.14.2"
+prompt-toolkit = "3.0.40"
+
+# console ehancements and eventually remote debugging
+# extras/helpers.
+# TODO: add a toolset that makes debugging a `pikerd` service
+# (tree) easy to hack on directly using more or less the local env:
+# - xonsh + xxh
+# - rsyscall + pdbp
+# - actor runtime control console like BEAM/OTP
+
+# ------ - ------
+
+# TODO: add an `--only daemon` group for running non-ui / pikerd
+# service tree in distributed mode B)
+# https://python-poetry.org/docs/managing-dependencies/#installing-group-dependencies
+# [tool.poetry.group.daemon.dependencies]
+
+[tool.poetry.scripts]
+piker = 'piker.cli:cli'
+pikerd = 'piker.cli:pikerd'
+ledger = 'piker.accounting.cli:ledger'
+
+
+[project]
+name="piker"
+keywords=[
+  "async",
+  "trading",
+  "finance",
+  "quant",
+  "charting",
 ]
-classifiers = [
-    "Development Status :: 3 - Alpha",
-    "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
-    "Operating System :: POSIX :: Linux",
-    "Programming Language :: Python :: Implementation :: CPython",
-    "Programming Language :: Python :: 3 :: Only",
-    "Programming Language :: Python :: 3.11",
-    "Programming Language :: Python :: 3.12",
-    "Intended Audience :: Financial and Insurance Industry",
-    "Intended Audience :: Science/Research",
-    "Intended Audience :: Developers",
-    "Intended Audience :: Education",
+classifiers=[
+  'Development Status :: 3 - Alpha',
+  "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
+  'Operating System :: POSIX :: Linux',
+  "Programming Language :: Python :: Implementation :: CPython",
+  "Programming Language :: Python :: 3 :: Only",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  'Intended Audience :: Financial and Insurance Industry',
+  'Intended Audience :: Science/Research',
+  'Intended Audience :: Developers',
+  'Intended Audience :: Education',
 ]
-dependencies = [
-    "async-generator >=1.10, <2.0.0",
-    "attrs >=23.1.0, <24.0.0",
-    "bidict >=0.22.1, <0.23.0",
-    "colorama >=0.4.6, <0.5.0",
-    "colorlog >=6.7.0, <7.0.0",
-    "ib-insync >=0.9.86, <0.10.0",
-    "numba >=0.59.0, <0.60.0",
-    "numpy >=1.25, <2.0",
-    "polars >=0.18.13, <0.19.0",
-    "pygments >=2.16.1, <3.0.0",
-    "rich >=13.5.2, <14.0.0",
-    "tomli >=2.0.1, <3.0.0",
-    "tomli-w >=1.0.0, <2.0.0",
-    "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.24, <0.25",
-    "pendulum >=3.0.0, <4.0.0",
-    "httpx >=0.27.0, <0.28.0",
-    "cryptofeed >=2.4.0, <3.0.0",
-    "pyarrow >=17.0.0, <18.0.0",
-    "websockets ==12.0",
-    "msgspec",
-    "tractor",
-    "asyncvnc",
-    "tomlkit",
-]
-
-[project.optional-dependencies]
-uis = [
-    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
-    # TODO: make sure the levenshtein shit compiles on nix..
-    # rapidfuzz = {extras = ["speedup"], version = "^0.18.0"}
-    "rapidfuzz >=3.2.0, <4.0.0",
-    "qdarkstyle >=3.0.2, <4.0.0",
-    "pyqt6 >=6.7.0, <7.0.0",
-    "pyqtgraph",
-
-    # ------ - ------
-
-    # TODO: add an `--only daemon` group for running non-ui / pikerd
-    # service tree in distributed mode B)
-    # https://docs.astral.sh/uv/concepts/projects/dependencies/#optional-dependencies
-    # [project.optional-dependencies]
-]
-
-[dependency-groups]
-dev = [
-    "pytest >=6.0.0, <7.0.0",
-    "elasticsearch >=8.9.0, <9.0.0",
-    "xonsh >=0.14.2, <0.15.0",
-    "prompt-toolkit ==3.0.40",
-    "cython >=3.0.0, <4.0.0",
-    "greenback >=1.1.1, <2.0.0",
-    # console ehancements and eventually remote debugging
-    # extras/helpers.
-    # TODO: add a toolset that makes debugging a `pikerd` service
-    # (tree) easy to hack on directly using more or less the local env:
-    # - xonsh + xxh
-    # - rsyscall + pdbp
-    # - actor runtime control console like BEAM/OTP
-]
-
-[project.scripts]
-piker = "piker.cli:cli"
-pikerd = "piker.cli:pikerd"
-ledger = "piker.accounting.cli:ledger"
-
-[tool.hatch.build.targets.sdist]
-include = ["piker"]
-
-[tool.hatch.build.targets.wheel]
-include = ["piker"]
-
-[tool.uv.sources]
-pyqtgraph = { git = "https://github.com/pikers/pyqtgraph.git" }
-asyncvnc = { git = "https://github.com/pikers/asyncvnc.git", branch = "main" }
-tomlkit = { git = "https://github.com/pikers/tomlkit.git", branch ="piker_pin" }
-msgspec = { git = "https://github.com/jcrist/msgspec.git" }
-tractor = { path = "../tractor" }
Author	SHA1	Message	Date
Tyler Goodlet	fe59ffc63f	Allow ledger passes to ignore (symcache) unknown fqmes For example in the paper-eng, if you have a backend that doesn't fully support a symcache (yet) it's handy to be able to ignore processing other paper-eng txns when all you care about at the moment is the simulated symbol. NOTE, that currently this will still result in a key-error when you load more then one mkt with the paper engine (for which the backend does not have the symcache implemented) since no fqme ad-hoc query was made for the 2nd symbol (and i'm not sure we should support that kinda hackery over just encouraging the sym-cache being added?). Def needs a little more thought depending on how many backends are never going to be able to (easily) support caching..	2025-02-03 18:55:47 -03:00
Tyler Goodlet	01081e1d80	.clearing._ems: Don't require `first_quote['last']` Instead just check for the field (which i'm not huge on the key-name for anyway) and if not found get the "last price" from the real-time shm buffer's latest 'close' sample. Unrelatedly, use a `subs.copy()` in the `Router.client_broadcast()` loop such that if a `client_stream` is popped on connection failure, we don't RTE for the "size changed on iteration".	2025-02-03 18:55:47 -03:00
Tyler Goodlet	a30c6d0ea1	`kucoin`: repair live quotes streaming.. This must have broke at some point during the new `MktPair` and thus `.fqme: str` updates; mas-o-menos the symbol key in the quote-msg-`dict` was NOT set to the `MktPair.bs_fqme: str` value and thus wasn't being processed by the downstream sampling and feed subsys. So fix that as well as a few other refinements, - set the `topic: mkt.bs_fqme` in quote msgs obvi. - drop the "wait for first clearing vlm" quote poll loop; going to fix the sampler to handle a `first_quote` without a `'last'` key. - add some typing around calls to `get_mkt_info()`. - rename `stream_messages()` -> `iter_normed_quotes()`.	2025-02-03 18:55:47 -03:00
Nelson Torres	7091a80a1d	Deleted settlePlan field from binance FutesPair.	2025-02-03 18:55:47 -03:00
Nelson Torres	5db98ef531	Added missing fields for kucoin. feeCategory, makerFeeCoefficient, takerFeeCoefficient and st.	2025-02-03 18:55:47 -03:00
Nelson Torres	3e999b164a	Default.nix fix: Since the last `poetry` update the command `poetry shell` have been moved to a plugin, this affects the entire instalation, so now we need to add more deps to the `buildInputs` and added to the `LD_LIBRARY_PATH` To run a command now you need to do something like this: `poetry run piker ...`	2025-02-03 18:55:47 -03:00