diff --git a/keel/data/cb_client.py b/keel/data/cb_client.py index e624d8f..ed12dc1 100644 --- a/keel/data/cb_client.py +++ b/keel/data/cb_client.py @@ -26,7 +26,7 @@ from decimal import Decimal from typing import Any, Protocol -from keel_broker_api.results import Balance, CancelOutcome +from keel_broker_api.results import Balance, CancelOutcome, Instrument from keel_core.telemetry import log_exception, log_venue_failure from keel.types import Candle, Granularity, Side @@ -233,6 +233,35 @@ def get_accounts(self) -> list[dict]: ) return accounts + def get_instrument(self, product_id: str) -> Instrument | None: + """One product's `base_increment`, in the PORT's shape (#524). + + The same bridge `get_balances` is: this client predates `keel-broker-api`, and + `executor._base_increment_for` had to read `list_products()` and pick through raw dicts + because that was the only catalogue read this client offered. Answering `Instrument` here + means the executor asks one question whether it holds this client or a real adapter, and + the flip needs no further change on this path. + + `get_product`, not `get_products`. The caller needs ONE product; `list_products` returns + about 900 and stays where it belongs -- `keel assets discover`, which genuinely wants the + catalogue. + + `None` for a product this venue does not list, or whose increment is missing, unparseable + or non-positive: all four are the same fact to a caller, and none is worth raising on. + """ + response = self._transport.get_product(product_id=product_id) + raw = _field(response, "product", response) + increment = _field(raw, "base_increment") + if increment is None: + return None + try: + value = Decimal(str(increment)) + except (ArithmeticError, TypeError, ValueError): + return None + if value <= 0: + return None + return Instrument(product_id=product_id, base_increment=value) + def get_balances(self) -> list[Balance]: """The same read as `get_accounts`, in the PORT's shape (#524). diff --git a/keel/execution/executor.py b/keel/execution/executor.py index 5ff4d24..8834ec0 100644 --- a/keel/execution/executor.py +++ b/keel/execution/executor.py @@ -364,12 +364,12 @@ def _base_increment_for( (no broker in paper mode, a venue error, a malformed or absent field) returns `None`, and the exit proceeds exactly as it did before #516. - **Exactly ONE row is written per miss, deliberately, even though the response carries every - product.** `Repository.set_state` commits per call, so caching all ~900 would mean ~900 - commits -- 900 fsyncs -- inside the order-placement path, which is the most latency-sensitive - moment in the engine. The alternative it buys is a handful of extra `list_products` calls: - one per allowlisted product per TTL, so roughly six a week at the current allowlist. Trading - 900 disk syncs during an order to save six network calls a week is the wrong way round. + **The venue is asked for ONE product, and that is a change from how this read began (#524).** + It called `list_products()` -- about 900 rows on Coinbase -- and cached exactly one of them, + because `Repository.set_state` commits per call and caching all of them would have meant ~900 + fsyncs inside the order-placement path, the most latency-sensitive moment in the engine. The + argument was sound and the shape was not: the port's `get_instrument` asks the venue for the + product the caller actually wants, so there is no longer a catalogue to decline to cache. """ key = f"{BASE_INCREMENT_PREFIX}{product_id}" cached = repo.get_state(key) @@ -384,19 +384,19 @@ def _base_increment_for( # `_fetch_available_quote`). return None try: - products = broker.list_products() + instrument = broker.get_instrument(product_id) except Exception: + # Every failure is the same answer here -- unknown, send the quantity unquantized, never + # refuse the exit. `NotImplementedError` from an adapter that has not written the read + # (keel-broker-alpaca) lands here too, and correctly: it is unknown to THIS deployment. log_venue_failure(logger, "executor.base_increment_fetch_failed", product=product_id) return None - for product in products or []: - if not isinstance(product, dict) or product.get("product_id") != product_id: - continue - increment = _coerce_increment(product.get("base_increment")) - if increment is not None: - repo.set_state(key, {"increment": str(increment), "fetched_at": now_ts}) - return increment - return None + if instrument is None: + return None + increment = instrument.base_increment + repo.set_state(key, {"increment": str(increment), "fetched_at": now_ts}) + return increment def _coerce_increment(raw: object) -> Decimal | None: diff --git a/tests/data/test_cb_client.py b/tests/data/test_cb_client.py index 2ae19bf..2caec64 100644 --- a/tests/data/test_cb_client.py +++ b/tests/data/test_cb_client.py @@ -14,7 +14,7 @@ from typing import Any import pytest -from keel_broker_api.results import Balance, CancelOutcome +from keel_broker_api.results import Balance, CancelOutcome, Instrument from keel_core import telemetry from keel.data.cb_client import CoinbaseClient @@ -205,6 +205,35 @@ def test_get_spot_returns_decimal() -> None: assert transport.calls["get_product"] == {"product_id": "BTC-USD"} +# --- get_instrument (the port's shape, #524) ---------------------------------------------- + + +def test_get_instrument_answers_the_ports_type_from_the_per_product_endpoint() -> None: + """`get_product`, not `get_products`. + + `executor._base_increment_for` needs ONE product. This client's `list_products` returns about + 900 and stays where it belongs -- `keel assets discover`, which genuinely wants the catalogue. + """ + transport = FakeTransport(product={"product_id": "XLM-USD", "base_increment": "0.000001"}) + + instrument = CoinbaseClient(transport).get_instrument("XLM-USD") + + assert isinstance(instrument, Instrument) + assert instrument.base_increment == Decimal("0.000001") + + +@pytest.mark.parametrize( + "payload", + [{}, {"base_increment": None}, {"base_increment": "nope"}, {"base_increment": "0"}], +) +def test_get_instrument_answers_none_for_anything_unusable(payload: dict) -> None: + """Missing, unparseable and non-positive are one fact to the caller: no usable granularity. + + Zero matters most -- the exit path quantizes against this value, so a zero crossing the + boundary is a division error or a silent zero size on a SELL.""" + assert CoinbaseClient(FakeTransport(product=payload)).get_instrument("XLM-USD") is None + + # --- get_balances (the port's shape, #524) ------------------------------------------------ diff --git a/tests/execution/test_sell_precision.py b/tests/execution/test_sell_precision.py index 2ffa7ed..cb89dd0 100644 --- a/tests/execution/test_sell_precision.py +++ b/tests/execution/test_sell_precision.py @@ -12,6 +12,7 @@ from decimal import Decimal import pytest +from keel_broker_api.results import Instrument from keel.execution.executor import ( _base_increment_for, @@ -150,16 +151,26 @@ def set_state(self, key, value) -> None: # noqa: ANN001 class _Broker: + """The port's catalogue read (#524), not the pre-port `list_products`. + + `products` is still a mapping of the whole catalogue so the fixtures below read the same, but + the broker now answers ONE product per call -- which is the point of the change: the executor + asks for what it wants rather than fetching ~900 rows to use one field of one of them. + """ + def __init__(self, products=None, raises: bool = False) -> None: # noqa: ANN001 - self._products = products or [] + self._products = {p["product_id"]: p["base_increment"] for p in (products or [])} self._raises = raises self.calls = 0 - def list_products(self): # noqa: ANN202 + def get_instrument(self, product_id: str) -> Instrument | None: self.calls += 1 if self._raises: raise RuntimeError("venue unreachable") - return self._products + raw = self._products.get(product_id) + if raw is None: + return None + return Instrument(product_id=product_id, base_increment=Decimal(str(raw))) PRODUCTS = [ @@ -186,9 +197,13 @@ def test_a_second_call_for_the_same_product_is_served_from_cache() -> None: def test_a_miss_writes_exactly_one_row_not_one_per_product() -> None: """The review finding this test exists for. - `set_state` commits per call, so caching all ~900 products would mean ~900 fsyncs inside the - order-placement path -- the most latency-sensitive moment in the engine -- to save a handful - of `list_products` calls per week. One row per miss is the right way round. + The finding this test was written for: `set_state` commits per call, so caching all ~900 + products would have meant ~900 fsyncs inside the order-placement path to save a handful of + `list_products` calls per week. + + #524 removed the temptation rather than resisting it -- `get_instrument` returns the one + product asked for, so there is no catalogue to decline to cache. The assertion still holds + and is still worth holding: one venue answer must still write one row. """ repo, broker = _Repo(), _Broker(PRODUCTS) _base_increment_for(broker, repo, "XLM-USD", NOW) @@ -219,3 +234,20 @@ def test_a_missing_or_malformed_increment_is_unknown() -> None: repo, broker = _Repo(), _Broker(PRODUCTS) assert _base_increment_for(broker, repo, "BAD-USD", NOW) is None assert _base_increment_for(broker, repo, "NOT-LISTED", NOW) is None + + +def test_an_adapter_without_a_catalogue_read_is_unknown_not_a_crash() -> None: + """`keel-broker-alpaca` raises `NotImplementedError` from `get_instrument` -- deliberately, + so a caller is never told a symbol is unlisted when the truth is that nobody wrote the read. + + On this path that must land as UNKNOWN, not as an exception. `_base_increment_for` never + raises by contract: unknown means send the quantity unquantized, and an exit that refused to + place because a catalogue lookup was unimplemented would be a protective order withheld over + a missing convenience. + """ + + class _Unwritten: + def get_instrument(self, product_id: str) -> Instrument | None: + raise NotImplementedError("no catalogue read on this adapter") + + assert _base_increment_for(_Unwritten(), _Repo(), "XLM-USD", NOW) is None