feat(executor): the increment read goes through the port, and stops fetching 900 products for one

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).
 

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:

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) ------------------------------------------------
 
 

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