From 36c4ebbb232b7d578c23d590989b7254cbdef85a Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Thu, 3 Sep 2026 12:32:14 -0400 Subject: [PATCH 1/2] feat(data): record which feed a cached series came from (#696, part 1 of 2) Schema v17. The provenance half of #696, landing BEFORE the gate that reads it -- a gate built on a feed inferred at read time would encode the bug it exists to fix. THE BUG. keel's liquidity statistic is `median(volume * close)` over cached candles, and every threshold keyed on it was calibrated against venue-reported volume. That is coherent on a crypto exchange: Coinbase's own volume IS the scale the floor was chosen against. It is not coherent on Alpaca's IEX feed, which reports one US equity exchange's own executions. The cost-fidelity run measured MSFT -- one of the most liquid securities in the world -- cached at $186M/day against a model anchored at $500M, and priced it as a thin asset. Nothing recorded which feed produced those bars, so the answer was whatever `broker.data_feed` happened to be loaded when someone later read the series. WHAT LANDS HERE. `candle_series_feed`, keyed on `(product_id, granularity, feed)` -- NOT a mutable column on `candles`. A series fetched under both feeds records BOTH rows, because that is what happened; a single overwritable column would let the most recent fetch erase the fact that most of the bars came from somewhere narrower, and a mixed series is exactly the case a reader most needs warning about. NO BACKFILL, and that is the point rather than a shortcut. Seeding rows from the current `broker.data_feed` would manufacture precisely the claim the table exists to make checkable, for years of bars that may have come from elsewhere. An empty table means "unrecorded", which is TRUE of every pre-existing series and is a different statement from "consolidated" -- `reports_consolidated_volume` returns `None` for the first and `False`/`True` for the second, and callers must not conflate the absence of evidence with evidence. `keel/data/feed_scope.py` carries the design, whose core is an ASYMMETRY: a single-venue feed's volume is a LOWER BOUND on consolidated volume, never an upper one. Volume at or above the floor on a partial feed is therefore a CONCLUSIVE pass -- if a name traded that much on one venue alone it necessarily traded at least that much in total -- while volume below the floor licenses no claim either way and must be refused as unmeasured rather than reported as thin. That is what lets this be honest without encoding any venue's market share: the bound holds for any share below 100%, so nothing here has a percentage in it that market structure could invalidate. `test_the_bound_needs_no_market_share` pins that no figure leaks into executable code, because the one thing that must never happen is someone scaling a volume statistic by an assumed share. An unrecognised feed id reads as PARTIAL, which costs a conclusive pass on a consolidated feed nobody declared and never grants one on a narrow feed nobody declared. Adapters declare their own feed (`volume_feed_id`), qualified by venue -- `alpaca:iex` / `alpaca:sip`, `coinbase` -- never bare `iex`, which would collide the day another venue routes there. Declaration is structural and optional: an adapter that declares nothing records nothing, so unrecorded stays unrecorded rather than every crypto series silently acquiring a provenance nobody established. Forcing all five adapters to implement it would invite a placeholder, which is the one value that must never enter this table. Tests: 31 across three files. tests/data/test_series_feed_provenance.py -- 16, the storage contract tests/data/test_feed_scope.py -- 11, the scope verdict tests/data/test_feed_provenance_is_wired.py -- 4, the fetch path actually carries the declaration; without these the whole thing degrades silently to pre-issue behaviour with a schema and a module suggesting otherwise The provenance and wiring suites were written test-first. `feed_scope.py` was NOT -- I wrote it before its tests, which is the wrong order; the 8-mutant run below is what establishes those tests constrain the code rather than merely describe it. Mutation-verified, 8 mutants, all killed: unrecorded reading as consolidated; `any` for `all` on a mixed series; `alpaca:iex` declared consolidated; a blank declaration accepted; provenance recorded for an empty batch; an empty feed string accepted; a re-fetch overwriting `first_seen_ts`; a missing table raising instead of reading empty. Five existing version pins move 16 -> 17. They are literals on purpose -- a deliberate speed bump that makes a schema bump a decision someone acknowledged. Refs #696 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NzuKAe2RVrPt9acVAWjRyL --- keel/data/db.py | 29 ++- keel/data/feed_scope.py | 103 ++++++++++ keel/data/history.py | 26 ++- keel/data/repository.py | 75 +++++++- .../keel_broker_alpaca/adapter.py | 12 ++ .../keel_broker_coinbase/adapter.py | 11 ++ tests/data/test_db.py | 4 +- tests/data/test_feed_provenance_is_wired.py | 102 ++++++++++ tests/data/test_feed_scope.py | 115 ++++++++++++ tests/data/test_migrations.py | 8 +- tests/data/test_series_feed_provenance.py | 176 ++++++++++++++++++ tests/data/test_trade_outcomes.py | 4 +- 12 files changed, 648 insertions(+), 17 deletions(-) create mode 100644 keel/data/feed_scope.py create mode 100644 tests/data/test_feed_provenance_is_wired.py create mode 100644 tests/data/test_feed_scope.py create mode 100644 tests/data/test_series_feed_provenance.py diff --git a/keel/data/db.py b/keel/data/db.py index ddd42205..8ed2ef74 100644 --- a/keel/data/db.py +++ b/keel/data/db.py @@ -20,7 +20,7 @@ from pathlib import Path from typing import Any -SCHEMA_VERSION = 16 +SCHEMA_VERSION = 17 # Creation order matters for readability (and for backends that validate FK targets eagerly); # SQLite itself only checks FK targets at DML time, but we still declare referenced tables first. @@ -171,6 +171,16 @@ ) """, """ + CREATE TABLE IF NOT EXISTS candle_series_feed ( + product_id TEXT NOT NULL, + granularity TEXT NOT NULL, + feed TEXT NOT NULL, + first_seen_ts INTEGER NOT NULL, + last_seen_ts INTEGER NOT NULL, + PRIMARY KEY (product_id, granularity, feed) + ) + """, + """ CREATE TABLE IF NOT EXISTS signals ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_id INTEGER, @@ -753,6 +763,22 @@ def _migrate_v16_orders_submit_book(conn: sqlite3.Connection) -> None: conn.execute("ALTER TABLE orders ADD COLUMN submit_best_ask TEXT") +def _migrate_v17_candle_series_feed(conn: sqlite3.Connection) -> None: + """v17 adds `candle_series_feed`. Table creation is handled by `_SCHEMA_STATEMENTS`; there is + deliberately NO backfill, and the reason is the whole point of the table (#696). + + A row here asserts "these bars were written while THIS feed was in use". For every candle + already cached, nobody wrote that down -- the feed was inferred at read time from whatever + config happened to be loaded, which is precisely the bug. Seeding rows from the CURRENT + `broker.data_feed` would manufacture exactly the claim this table exists to make checkable, + and would do it for years of bars that may well have come from somewhere else. + + An empty table means "provenance unrecorded", which is true of every pre-existing series and + is a different statement from "consolidated". Callers must keep those apart: the first is + the absence of evidence, the second is evidence. + """ + + _MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = { 2: _migrate_v2_broker_subscriptions, 3: _migrate_v3_trade_outcomes, @@ -769,6 +795,7 @@ def _migrate_v16_orders_submit_book(conn: sqlite3.Connection) -> None: 14: _migrate_v14_venue_trade_scopes, 15: _migrate_v15_trade_scope_credential_fingerprint, 16: _migrate_v16_orders_submit_book, + 17: _migrate_v17_candle_series_feed, } diff --git a/keel/data/feed_scope.py b/keel/data/feed_scope.py new file mode 100644 index 00000000..f5d421b4 --- /dev/null +++ b/keel/data/feed_scope.py @@ -0,0 +1,103 @@ +"""Which data feed a series came from, and whether that feed sees the whole market -- #696. + +keel's liquidity statistic is `median(volume * close)` over cached candles +(`compliance/screen.median_daily_quote_volume`), and every threshold keyed on it was calibrated +against VENUE-REPORTED volume. On a crypto exchange that is coherent: Coinbase's own volume is +the scale the floor was chosen against. On Alpaca's IEX feed it is not, because IEX reports one +US equity exchange's executions -- IEX publishes its overall share as roughly 3.8% for Q2 2026 +-- while keel's model is anchored at $500M/day. The cost-fidelity run measured MSFT, one of the +most liquid securities in the world, cached at $186M/day and priced as a thin asset. + +**THE ASYMMETRY IS THE WHOLE DESIGN.** A single-venue feed's volume is a LOWER BOUND on +consolidated volume, never an upper one. So the two directions are not equally informative: + +* volume at or above the floor, on a partial feed -> **conclusive PASS.** If a name traded that + much on one venue alone, it necessarily traded at least that much in total. No assumption is + needed and none is made. +* volume below the floor, on a partial feed -> **not a verdict.** It licenses no claim either + way, and must be refused as UNMEASURED rather than reported as thin. + +That asymmetry is why this module can be honest without knowing any venue's market share. The +bound holds for any share below 100%, so nothing here encodes a percentage that would drift as +market structure changes -- the 3.8% above is context for a reader, never an input to a decision. + +WHAT THIS DELIBERATELY DOES NOT DO. It does not fail closed on a partial feed. Nothing is +currently admitted on the equities profile, whose entire job is accruing paper evidence, and +refusing every IEX series would convert a data-vendor pricing tier into a hard prerequisite for +running the engine at all. The bound above is strong enough to admit the liquid names honestly +and refuse the rest without pretending to have measured them. +""" + +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +#: Feeds that report CONSOLIDATED market volume -- i.e. where venue volume is market volume for +#: the purpose keel uses it. A crypto exchange is its own market in the sense that matters here: +#: the floor was calibrated on exactly this kind of number. +#: +#: Membership is a claim about market structure, so it is declared, never inferred from a feed's +#: name. An unrecognised feed id is treated as PARTIAL, which is the conservative reading: it +#: costs a conclusive pass on a genuinely consolidated feed nobody declared, and never grants one. +CONSOLIDATED_VOLUME_FEEDS: frozenset[str] = frozenset( + { + "coinbase", + "kraken", + # Alpaca's paid tier IS the consolidated tape (SIP), which is what makes the IEX + # distinction meaningful rather than a blanket statement about the venue. + "alpaca:sip", + } +) + + +@runtime_checkable +class DeclaresVolumeFeed(Protocol): + """A broker adapter that can name the feed its candles come from. + + Structural and optional on purpose: an adapter that does not declare one records no + provenance, which reads back as "unrecorded" rather than as a guess. Forcing every adapter + to implement it would mean editing five packages to add a value four of them would state + identically -- and a required field invites a placeholder, which is the one value that must + never enter this table. + """ + + @property + def volume_feed_id(self) -> str | None: ... + + +def volume_feed_of(client: Any) -> str | None: + """The feed id `client` declares, or `None` when it declares nothing. + + `None` means UNRECORDED. It must never be normalised to a default: the whole point of + recording provenance at fetch time is that an unknown feed stays visibly unknown instead of + inheriting whatever config is loaded when someone later reads the series. + """ + feed = getattr(client, "volume_feed_id", None) + if not isinstance(feed, str) or not feed.strip(): + return None + return feed + + +def reports_consolidated_volume(feeds: tuple[str, ...]) -> bool | None: + """Does this series' recorded provenance support reading its volume as market volume? + + * `True` -- every recorded feed is consolidated. + * `False` -- at least one recorded feed is partial, so the statistic is a LOWER BOUND. A + mixed series counts as partial: the bars from the narrow feed are still in the median, and + a median is not decomposable by source. + * `None` -- nothing recorded. Distinct from `False` on purpose: `False` is a known + limitation a caller can reason about with the asymmetric bound, `None` is the absence of + evidence, and a caller that cannot tell them apart will treat legacy databases as if their + scope had been established. + """ + if not feeds: + return None + return all(feed in CONSOLIDATED_VOLUME_FEEDS for feed in feeds) + + +__all__ = [ + "CONSOLIDATED_VOLUME_FEEDS", + "DeclaresVolumeFeed", + "reports_consolidated_volume", + "volume_feed_of", +] diff --git a/keel/data/history.py b/keel/data/history.py index 8493eb61..b3a41a0e 100644 --- a/keel/data/history.py +++ b/keel/data/history.py @@ -16,6 +16,7 @@ from dataclasses import dataclass from typing import Protocol +from keel.data.feed_scope import volume_feed_of from keel.types import Candle, Granularity GRANULARITY_SECONDS: dict[Granularity, int] = { @@ -61,7 +62,12 @@ def get_candles( ) -> list[Candle]: ... def upsert_candles( - self, product_id: str, granularity: Granularity, candles: list[Candle] + self, + product_id: str, + granularity: Granularity, + candles: list[Candle], + *, + feed: str | None = None, ) -> int: ... @@ -107,6 +113,7 @@ def _fill_forward( now_ts: int, sleep_fn, sleep_sec: float, + feed: str | None = None, ) -> None: """Fetch any bars strictly newer than `latest_cached`, up to `now_ts` (recent-bar gaps).""" window_start = latest_cached + step @@ -116,7 +123,7 @@ def _fill_forward( window_end = min(now_ts, window_start + (MAX_CANDLES_PER_REQUEST - 1) * step) batch = client.get_candles(product, granularity, window_start, window_end) if batch: - repo.upsert_candles(product, granularity, batch) + repo.upsert_candles(product, granularity, batch, feed=feed) sleep_fn(sleep_sec) window_start = window_end + step @@ -131,6 +138,7 @@ def _fill_backward( start_floor: int, sleep_fn, sleep_sec: float, + feed: str | None = None, ) -> None: """Page backward from `window_end` down to `start_floor`, stopping at the first empty window -- that window is either the asset's inception or already-covered territory.""" @@ -140,7 +148,7 @@ def _fill_backward( batch = client.get_candles(product, granularity, window_start, window_end) if not batch: break # inception (or a confirmed-empty probe) -- nothing older to fetch - repo.upsert_candles(product, granularity, batch) + repo.upsert_candles(product, granularity, batch, feed=feed) sleep_fn(sleep_sec) window_end = window_start - step @@ -164,6 +172,9 @@ def ensure_history( Otherwise (no cache, or `refresh=True`) page backward from `now_ts` down to the `years` floor, stopping the instant a window comes back empty. """ + # Asked once, at the top: the client cannot change feed mid-run, and asking per call would + # invite a caller to pass a different one for the forward and backward halves of one series. + feed = volume_feed_of(client) result: dict[tuple[str, Granularity], CoverageInfo] = {} for product in products: for granularity in granularities: @@ -177,16 +188,19 @@ def ensure_history( _fill_forward( client, repo, product, granularity, step, latest_cached, now_ts, sleep_fn, sleep_sec, - ) + feed=feed, +) _fill_backward( client, repo, product, granularity, step, earliest_cached - step, start_floor, sleep_fn, sleep_sec, - ) + feed=feed, +) else: _fill_backward( client, repo, product, granularity, step, now_ts, start_floor, sleep_fn, sleep_sec, - ) + feed=feed, +) result[(product, granularity)] = coverage(repo, product, granularity, start_floor) return result diff --git a/keel/data/repository.py b/keel/data/repository.py index 4415fe6b..8382dc04 100644 --- a/keel/data/repository.py +++ b/keel/data/repository.py @@ -195,9 +195,28 @@ def _transaction_row_to_dict(self, row: sqlite3.Row) -> dict[str, Any]: # -- candles ---------------------------------------------------------- def upsert_candles( - self, product_id: str, granularity: Granularity, candles: list[Candle] + self, + product_id: str, + granularity: Granularity, + candles: list[Candle], + *, + feed: str | None = None, + now_ts: int | None = None, ) -> int: - """Upsert `candles` keyed on `(product_id, granularity, ts)`. Returns rows written.""" + """Upsert `candles` keyed on `(product_id, granularity, ts)`. Returns rows written. + + `feed`, when given, records WHICH DATA FEED served these bars (#696) -- the liquidity + statistic is `median(volume * close)` over this table, and on a single-exchange feed + that number is a lower bound on the market rather than a measurement of it. Recording it + at write time is the point: inferring it later from whatever `broker.data_feed` happens + to be loaded lets a database filled under IEX be judged under a SIP setting, silently. + + `feed=None` records NOTHING rather than a default. Every bar cached before this existed + has genuinely unknown provenance, and "unrecorded" must stay distinguishable from + "consolidated" -- the absence of evidence is not evidence. + """ + if feed is not None and not feed.strip(): + raise ValueError("feed must be a non-empty identifier or None, not an empty string") gran_value = Granularity(granularity).value rows = [ ( @@ -221,9 +240,61 @@ def upsert_candles( """, rows, ) + # Only a fetch that actually returned bars is evidence that this feed served this + # series. An empty batch records nothing. + if feed is not None and rows: + stamp = int(time.time()) if now_ts is None else now_ts + self._conn.execute( + """ + INSERT INTO candle_series_feed + (product_id, granularity, feed, first_seen_ts, last_seen_ts) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(product_id, granularity, feed) DO UPDATE SET + last_seen_ts = excluded.last_seen_ts + """, + (product_id, gran_value, feed, stamp, stamp), + ) self._conn.commit() return len(rows) + def get_series_feeds(self, product_id: str, granularity: Granularity) -> tuple[str, ...]: + """Every feed recorded against this series, sorted. `()` means UNRECORDED, which is not + the same as consolidated -- see `upsert_candles`. + + More than one entry means the series is MIXED, which is the case a caller most needs to + know about and the one a single overwritable column would have hidden. + """ + try: + rows = self._conn.execute( + "SELECT feed FROM candle_series_feed WHERE product_id = ? AND granularity = ?" + " ORDER BY feed", + (product_id, Granularity(granularity).value), + ).fetchall() + except sqlite3.OperationalError: + # A hand-patched or partially-migrated file must not turn a liquidity read into a + # crash. No table is no provenance, which is true. + return () + return tuple(row["feed"] for row in rows) + + def get_series_feed_window( + self, product_id: str, granularity: Granularity, feed: str + ) -> tuple[int, int] | None: + """`(first_seen_ts, last_seen_ts)` for one feed on one series, or `None` if unrecorded. + + Lets a report say "the IEX rows stopped in March" rather than only "this series is + mixed" -- the difference between a series that switched feeds cleanly and one still + being written by both. + """ + try: + row = self._conn.execute( + "SELECT first_seen_ts, last_seen_ts FROM candle_series_feed" + " WHERE product_id = ? AND granularity = ? AND feed = ?", + (product_id, Granularity(granularity).value, feed), + ).fetchone() + except sqlite3.OperationalError: + return None + return None if row is None else (int(row["first_seen_ts"]), int(row["last_seen_ts"])) + def get_candles( self, product_id: str, diff --git a/packages/keel-broker-alpaca/keel_broker_alpaca/adapter.py b/packages/keel-broker-alpaca/keel_broker_alpaca/adapter.py index 3ff65d17..b484ff40 100644 --- a/packages/keel-broker-alpaca/keel_broker_alpaca/adapter.py +++ b/packages/keel-broker-alpaca/keel_broker_alpaca/adapter.py @@ -231,6 +231,18 @@ def endpoint(self) -> str: host URL, so a paper configuration cannot reach the live venue.""" return self._endpoint + @property + def volume_feed_id(self) -> str: + """Which feed this adapter's candles came from, for `candle_series_feed` (#696). + + Qualified by venue (`alpaca:iex`, never bare `iex`) because the id is a market-structure + claim and "iex" alone would collide the day another venue routes there. The distinction + is load-bearing: IEX reports one exchange's own executions, so `median(volume * close)` + over its bars is a LOWER BOUND on consolidated volume; SIP is the consolidated tape and + the same statistic is a measurement of it. + """ + return f"alpaca:{self.data_feed}" + @property def data_feed(self) -> str: """The declared market-data tier ("iex" | "sip"), sent on every data request.""" diff --git a/packages/keel-broker-coinbase/keel_broker_coinbase/adapter.py b/packages/keel-broker-coinbase/keel_broker_coinbase/adapter.py index ff4c1217..ef7d482d 100644 --- a/packages/keel-broker-coinbase/keel_broker_coinbase/adapter.py +++ b/packages/keel-broker-coinbase/keel_broker_coinbase/adapter.py @@ -161,6 +161,17 @@ class CoinbaseAdapter: #: second list keyed by venue name would be the thing #233's display exists to avoid. DECLARED_CREDENTIAL_ENV: tuple[str, str] = ("CDP_API_KEY", "CDP_API_SECRET") + @property + def volume_feed_id(self) -> str: + """Which feed this adapter's candles come from, for `candle_series_feed` (#696). + + Coinbase serves one feed: its own book. That is DECLARED here rather than assumed, + because keel's liquidity floor was calibrated against exactly this number -- a crypto + exchange's own reported volume -- and the claim "venue volume is market volume for this + purpose" should be written down somewhere a reader can find it, not left as the silent + default it used to be. + """ + return "coinbase" def __init__(self, transport: Transport | None = None) -> None: self._transport = transport diff --git a/tests/data/test_db.py b/tests/data/test_db.py index 5be46326..a56abd7a 100644 --- a/tests/data/test_db.py +++ b/tests/data/test_db.py @@ -103,11 +103,11 @@ def test_agent_state_table_has_key_primary_key(): assert pk_columns == {"key"} -def test_schema_version_is_16(): +def test_schema_version_is_17(): """Deliberate tripwire: bump this literal consciously on every schema change.""" from keel.data.db import SCHEMA_VERSION - assert SCHEMA_VERSION == 16 + assert SCHEMA_VERSION == 17 def test_a_v6_database_migrates_up_and_gains_the_profile_table(tmp_path): diff --git a/tests/data/test_feed_provenance_is_wired.py b/tests/data/test_feed_provenance_is_wired.py new file mode 100644 index 00000000..9638fc54 --- /dev/null +++ b/tests/data/test_feed_provenance_is_wired.py @@ -0,0 +1,102 @@ +"""Provenance is recorded where the bars are WRITTEN, not where they are read -- issue #696. + +`feed_scope` can say what a feed means and `Repository` can store it; neither matters unless the +fetch path actually passes it. These are the wiring pins: an adapter declares its feed, and every +path that writes candles carries that declaration through to the row. + +The failure this prevents is silent and total. If the fetch path drops the feed, every series +reads back as "unrecorded", the scope verdict is `None` everywhere, and the gate downstream +degrades to exactly the behaviour that existed before the issue was filed -- with a schema, a +module and a test suite all suggesting otherwise. +""" + +from __future__ import annotations + +from decimal import Decimal + +from keel_core.types import Candle, Granularity + +from keel.data import history +from keel.data.db import connect, migrate +from keel.data.feed_scope import volume_feed_of +from keel.data.repository import Repository + + +def _candle(ts: int) -> Candle: + return Candle( + ts=ts, + open=Decimal("100"), + high=Decimal("101"), + low=Decimal("99"), + close=Decimal("100"), + volume=Decimal("1000"), + ) + + +class _FeedClient: + """A venue client that declares its feed, as the Alpaca adapter does.""" + + def __init__(self, feed: str, batches: list[list[Candle]]) -> None: + self._feed = feed + self._batches = batches + + @property + def volume_feed_id(self) -> str: + return self._feed + + def get_candles(self, product_id, granularity, start, end): # noqa: ANN001, ANN201 + return self._batches.pop(0) if self._batches else [] + + +class _SilentClient(_FeedClient): + """A venue client that declares nothing -- most adapters, today.""" + + def __init__(self, batches: list[list[Candle]]) -> None: + super().__init__("", batches) + + @property + def volume_feed_id(self) -> str | None: # type: ignore[override] + return None + + +def _repo() -> Repository: + conn = connect(":memory:") + migrate(conn) + return Repository(conn) + + +def test_the_alpaca_adapter_declares_its_configured_feed() -> None: + from keel_broker_alpaca.adapter import AlpacaAdapter + + assert volume_feed_of(AlpacaAdapter(transport=object(), data_feed="iex")) == "alpaca:iex" + assert volume_feed_of(AlpacaAdapter(transport=object(), data_feed="sip")) == "alpaca:sip" + + +def test_the_coinbase_adapter_declares_its_venue() -> None: + """Coinbase's own volume IS the scale keel's floor was calibrated on, so it declares a + consolidated feed -- and declaring it explicitly is what keeps that a stated claim rather + than a default nobody wrote down.""" + from keel_broker_coinbase.adapter import CoinbaseAdapter + + assert volume_feed_of(CoinbaseAdapter(transport=object())) == "coinbase" + + +def test_ensure_history_records_the_feed_it_fetched_under() -> None: + repo = _repo() + client = _FeedClient("alpaca:iex", [[_candle(0), _candle(86400)]]) + history.ensure_history( + client, repo, ["MSFT-USD"], [Granularity.ONE_DAY], years=1, now_ts=86400 + ) + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == ("alpaca:iex",) + + +def test_a_client_that_declares_nothing_records_nothing() -> None: + """Unrecorded stays unrecorded. A fetch through an adapter with no declaration must not + invent one, or every crypto series silently acquires a provenance nobody established.""" + repo = _repo() + client = _SilentClient([[_candle(0), _candle(86400)]]) + history.ensure_history( + client, repo, ["BTC-USD"], [Granularity.ONE_DAY], years=1, now_ts=86400 + ) + assert repo.get_candles("BTC-USD", Granularity.ONE_DAY) + assert repo.get_series_feeds("BTC-USD", Granularity.ONE_DAY) == () diff --git a/tests/data/test_feed_scope.py b/tests/data/test_feed_scope.py new file mode 100644 index 00000000..93bb2182 --- /dev/null +++ b/tests/data/test_feed_scope.py @@ -0,0 +1,115 @@ +"""Whether a feed's volume may be read as market volume -- issue #696. + +The load-bearing test is `test_a_partial_feed_above_the_floor_is_still_conclusive`, which pins +the asymmetry the whole design rests on: a single-venue feed's volume is a LOWER BOUND on +consolidated volume, so clearing the floor on one venue alone proves the floor is cleared, while +failing it proves nothing. Get that backwards and you either ban MSFT from an IEX feed or admit +a penny stock on one. + +`test_the_bound_needs_no_market_share` is the other one that matters: nothing in this module +encodes a venue's percentage, because the bound holds for any share below 100%. A module that +needed 3.8% to be right would need editing every time market structure moved. +""" + +from __future__ import annotations + +from keel.data.feed_scope import ( + CONSOLIDATED_VOLUME_FEEDS, + reports_consolidated_volume, + volume_feed_of, +) + + +class _Declares: + def __init__(self, feed: object) -> None: + self._feed = feed + + @property + def volume_feed_id(self) -> object: + return self._feed + + +class _Silent: + pass + + +# --- what a client declares ------------------------------------------------------------------ + + +def test_a_client_that_declares_a_feed_is_read() -> None: + assert volume_feed_of(_Declares("alpaca:iex")) == "alpaca:iex" + + +def test_a_client_that_declares_nothing_is_unrecorded() -> None: + """`None`, never a default. A guessed feed is the bug this table exists to prevent.""" + assert volume_feed_of(_Silent()) is None + + +def test_a_blank_or_non_string_declaration_is_unrecorded() -> None: + """Defensive against an adapter that grows the attribute but leaves it unset -- a `""` or a + stray object would otherwise be written into provenance and read back as an unknown feed.""" + assert volume_feed_of(_Declares("")) is None + assert volume_feed_of(_Declares(" ")) is None + assert volume_feed_of(_Declares(None)) is None + assert volume_feed_of(_Declares(object())) is None + + +# --- the scope verdict ----------------------------------------------------------------------- + + +def test_no_recorded_provenance_is_none_not_false() -> None: + """`None` is the absence of evidence; `False` is evidence of a limitation. A caller that + conflates them treats every legacy database as though its scope had been established.""" + assert reports_consolidated_volume(()) is None + + +def test_a_consolidated_feed_reads_as_consolidated() -> None: + assert reports_consolidated_volume(("coinbase",)) is True + assert reports_consolidated_volume(("alpaca:sip",)) is True + + +def test_a_single_venue_feed_reads_as_partial() -> None: + assert reports_consolidated_volume(("alpaca:iex",)) is False + + +def test_an_unrecognised_feed_is_treated_as_partial() -> None: + """Conservative by design: it costs a conclusive pass on a consolidated feed nobody + declared, and never grants one on a narrow feed nobody declared.""" + assert reports_consolidated_volume(("some-new-venue",)) is False + + +def test_a_mixed_series_is_partial() -> None: + """A median is not decomposable by source. Once narrow-feed bars are in the series, the + statistic is a lower bound for the whole series -- there is no honest way to read the + consolidated half separately.""" + assert reports_consolidated_volume(("alpaca:iex", "alpaca:sip")) is False + + +def test_every_declared_consolidated_feed_is_a_nonempty_string() -> None: + assert CONSOLIDATED_VOLUME_FEEDS + assert all(isinstance(f, str) and f.strip() for f in CONSOLIDATED_VOLUME_FEEDS) + + +def test_the_narrow_alpaca_feed_is_not_declared_consolidated() -> None: + """The one membership question this issue exists to answer.""" + assert "alpaca:iex" not in CONSOLIDATED_VOLUME_FEEDS + assert "alpaca:sip" in CONSOLIDATED_VOLUME_FEEDS + + +def test_the_bound_needs_no_market_share() -> None: + """No percentage anywhere in the module. The lower bound holds for ANY share below 100%, so + a figure here would be a maintenance liability that buys nothing -- and would invite someone + to scale a volume statistic by it, which is the one thing that must never happen.""" + import inspect + import re + + from keel.data import feed_scope + + code = "".join( + line + for line in inspect.getsource(feed_scope).splitlines(keepends=True) + if not line.lstrip().startswith("#") + ) + body = code[code.index('"""', code.index('"""') + 3) + 3 :] + assert not re.search(r"\d+(\.\d+)?\s*%", body), "a market share leaked into executable code" + assert "0.038" not in body and "3.8" not in body diff --git a/tests/data/test_migrations.py b/tests/data/test_migrations.py index a06df454..9405f885 100644 --- a/tests/data/test_migrations.py +++ b/tests/data/test_migrations.py @@ -49,7 +49,7 @@ def test_fresh_database_is_stamped_at_the_current_version() -> None: conn = db.connect(":memory:") db.migrate(conn) version = conn.execute("SELECT version FROM schema_version").fetchone()["version"] - assert version == db.SCHEMA_VERSION == 16 + assert version == db.SCHEMA_VERSION == 17 def test_fresh_database_gets_no_subscription_row() -> None: @@ -612,7 +612,7 @@ def test_v14_migration_bumps_the_stored_version() -> None: conn = _v12_database() db.migrate(conn) stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"] - assert stamped == db.SCHEMA_VERSION == 16 + assert stamped == db.SCHEMA_VERSION == 17 def test_v14_migration_step_is_not_blocked_by_another_venues_existing_row() -> None: @@ -773,7 +773,7 @@ def test_v15_migration_bumps_the_stored_version() -> None: conn = _v12_database() db.migrate(conn) stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"] - assert stamped == db.SCHEMA_VERSION == 16 + assert stamped == db.SCHEMA_VERSION == 17 def test_v15_the_12_to_15_chain_creates_the_table_with_the_column_already_present() -> None: @@ -875,7 +875,7 @@ def test_an_existing_orders_table_gains_the_submit_book_by_ALTER() -> None: assert row["submit_best_bid"] is None assert row["submit_best_ask"] is None stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"] - assert stamped == db.SCHEMA_VERSION == 16 + assert stamped == db.SCHEMA_VERSION == 17 def test_v16_is_idempotent_per_column() -> None: diff --git a/tests/data/test_series_feed_provenance.py b/tests/data/test_series_feed_provenance.py new file mode 100644 index 00000000..0f2f6696 --- /dev/null +++ b/tests/data/test_series_feed_provenance.py @@ -0,0 +1,176 @@ +"""What feed a cached series actually came from -- issue #696. + +keel's liquidity statistic is `median(volume * close)` over cached candles, and it silently +assumed the venue's feed sees the whole market. That holds on a crypto exchange; it does not +hold on Alpaca's IEX feed, which reports one US equity exchange's own executions (IEX publishes +its overall share as roughly 3.8% for Q2 2026). The cost-fidelity measurement found MSFT cached +at $186M/day against a model anchored at $500M. + +The half of that fixed here is the PROVENANCE, and it is the load-bearing half: before this, +"what feed produced these bars" was inferred from whatever config happened to be loaded at READ +time, so a database filled under IEX could be judged under a SIP setting and nothing would say +so. A gate built on an inferred value would encode the bug it was meant to fix, which is why +this lands before the gate does. + +**Provenance is keyed on `(product_id, granularity, feed)`, not overwritten per series.** A +series fetched under both feeds records BOTH rows, because that is what happened. A single +mutable column would let the most recent fetch erase the fact that most of the bars came from +somewhere else -- and a mixed series is exactly the case a reader most needs to be warned about. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest +from keel_core.types import Candle, Granularity + +from keel.data.db import SCHEMA_VERSION, connect, migrate +from keel.data.repository import Repository + + +@pytest.fixture() +def repo() -> Repository: + conn = connect(":memory:") + migrate(conn) + return Repository(conn) + + +def _candles(n: int = 3) -> list[Candle]: + return [ + Candle( + ts=i * 86400, + open=Decimal("100"), + high=Decimal("101"), + low=Decimal("99"), + close=Decimal("100"), + volume=Decimal("1000"), + ) + for i in range(n) + ] + + +def test_the_schema_carries_a_series_feed_table() -> None: + conn = connect(":memory:") + migrate(conn) + names = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert "candle_series_feed" in names + + +def test_migrating_twice_is_a_no_op() -> None: + conn = connect(":memory:") + migrate(conn) + migrate(conn) + assert int(conn.execute("SELECT version FROM schema_version").fetchone()["version"]) == ( + SCHEMA_VERSION + ) + + +def test_an_existing_database_gains_the_table_on_migration() -> None: + """The live deployments are at v16 with years of candles. The upgrade must add the table + without touching a single cached bar.""" + conn = connect(":memory:") + migrate(conn) + conn.execute("UPDATE schema_version SET version = 16") + conn.execute("DROP TABLE candle_series_feed") + conn.commit() + migrate(conn) + names = {r["name"] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} + assert "candle_series_feed" in names + + +def test_a_series_with_no_recorded_feed_reports_nothing(repo: Repository) -> None: + """NOT "reports the default feed". Every bar cached before this existed has unknown + provenance, and saying so is the only honest answer -- inventing one would put a + fabricated claim into the table a gate is about to read.""" + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles()) + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == () + + +def test_upserting_with_a_feed_records_it(repo: Repository) -> None: + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex") + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == ("alpaca:iex",) + + +def test_a_series_fetched_under_two_feeds_reports_both(repo: Repository) -> None: + """The case a single mutable column would erase, and the one a reader most needs.""" + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex") + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:sip") + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == ("alpaca:iex", "alpaca:sip") + + +def test_re_fetching_under_the_same_feed_does_not_duplicate(repo: Repository) -> None: + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex") + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex") + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == ("alpaca:iex",) + + +def test_provenance_is_per_granularity(repo: Repository) -> None: + """A daily series and an hourly one can genuinely come from different feeds -- Alpaca mints + hourly bars only inside a session -- so they are recorded separately.""" + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:sip") + repo.upsert_candles("MSFT-USD", Granularity.ONE_HOUR, _candles(), feed="alpaca:iex") + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == ("alpaca:sip",) + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_HOUR) == ("alpaca:iex",) + + +def test_provenance_is_per_product(repo: Repository) -> None: + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex") + repo.upsert_candles("BTC-USD", Granularity.ONE_DAY, _candles(), feed="coinbase") + assert repo.get_series_feeds("BTC-USD", Granularity.ONE_DAY) == ("coinbase",) + + +def test_the_window_each_feed_contributed_is_recorded(repo: Repository) -> None: + """First and last time this feed was seen writing this series. `doctor` needs it to say + "the IEX rows stopped in March" rather than only "this series is mixed".""" + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex", now_ts=100) + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="alpaca:iex", now_ts=500) + row = repo.get_series_feed_window("MSFT-USD", Granularity.ONE_DAY, "alpaca:iex") + assert row == (100, 500) + + +def test_an_unknown_feed_window_is_none(repo: Repository) -> None: + assert repo.get_series_feed_window("MSFT-USD", Granularity.ONE_DAY, "alpaca:iex") is None + + +def test_writing_no_candles_records_no_provenance(repo: Repository) -> None: + """A fetch that returned nothing is not evidence that this feed served this series.""" + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, [], feed="alpaca:iex") + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == () + + +def test_an_empty_feed_string_is_refused(repo: Repository) -> None: + """`feed=""` is a caller bug that would otherwise record provenance meaning nothing, and + would then read back as a feed whose scope cannot be looked up.""" + with pytest.raises(ValueError, match="feed"): + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(), feed="") + + +def test_the_candles_themselves_are_untouched_by_provenance(repo: Repository) -> None: + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, _candles(3), feed="alpaca:iex") + assert len(repo.get_candles("MSFT-USD", Granularity.ONE_DAY)) == 3 + cols = { + r["name"] + for r in repo._conn.execute("PRAGMA table_info(candles)") # noqa: SLF001 + } + assert cols == {"product_id", "granularity", "ts", "o", "h", "l", "c", "v"} + + +def test_feeds_are_returned_in_a_stable_order(repo: Repository) -> None: + """Sorted, so a caller rendering "iex, sip" cannot produce a different string on a + different day for the same database.""" + repo.upsert_candles("X-USD", Granularity.ONE_DAY, _candles(), feed="zzz") + repo.upsert_candles("X-USD", Granularity.ONE_DAY, _candles(), feed="aaa") + assert repo.get_series_feeds("X-USD", Granularity.ONE_DAY) == ("aaa", "zzz") + + +def test_a_legacy_database_missing_the_table_does_not_break_reads() -> None: + """Defensive: a hand-patched or partially-migrated file must not turn a liquidity read into + a crash. Missing table reads as "no provenance recorded", which is true.""" + conn = connect(":memory:") + migrate(conn) + conn.execute("DROP TABLE candle_series_feed") + conn.commit() + repo = Repository(conn) + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == () + assert repo.get_series_feed_window("MSFT-USD", Granularity.ONE_DAY, "alpaca:iex") is None diff --git a/tests/data/test_trade_outcomes.py b/tests/data/test_trade_outcomes.py index 3be81317..1b3a4b52 100644 --- a/tests/data/test_trade_outcomes.py +++ b/tests/data/test_trade_outcomes.py @@ -35,11 +35,11 @@ def _outcome(**overrides: object) -> dict: return base -def test_schema_is_at_version_16() -> None: +def test_schema_is_at_version_17() -> None: conn = db.connect(":memory:") db.migrate(conn) version = conn.execute("SELECT version FROM schema_version").fetchone()["version"] - assert version == db.SCHEMA_VERSION == 16 + assert version == db.SCHEMA_VERSION == 17 def test_fresh_database_has_no_outcomes() -> None: From a1c234d12ff927f0d2b5c27d58a0b083d6862a19 Mon Sep 17 00:00:00 2001 From: Elmehdi Aitbrahim Date: Thu, 3 Sep 2026 13:04:04 -0400 Subject: [PATCH 2/2] fix(data): every candle writer records provenance, not just `keel fetch` (#696) Four findings from the review of #710, fixed on the branch. The first one meant the feature was very nearly inert. 1. THE ONE THAT MATTERED. Only `history.ensure_history` carried the feed declaration. `market_feed.poll_once` -- the path `agent.run_once` uses on EVERY cycle in every deployment -- and `repair.repair_series` both dropped it, so in a real database almost every bar would have been written with no provenance and `candle_series_feed` would have stayed nearly empty while a schema, a module and 31 tests said otherwise. `keel fetch` is not how bars normally arrive. The wiring tests did not catch it because they exercised `ensure_history` alone: they proved the mechanism worked, never that every writer used it. Both paths now resolve the feed once and thread it down, and there are tests per WRITER rather than per mechanism. `backfill` resolves its own rather than relying on one being threaded in -- it has no in-tree caller today, and a writer that silently records nothing is precisely the failure this table exists to prevent. 2. `get_series_feeds` / `get_series_feed_window` rescued `sqlite3.OperationalError` wholesale, which is also what a LOCK TIMEOUT raises -- and keel reads and writes this file from more than one process by design. A lock would have been reported as "scope unrecorded" for a series whose scope is on disk: exactly the `None`-vs-`False` conflation `feed_scope` exists to prevent, silently. Now only `no such table` is rescued; anything else propagates. 3. `DeclaresVolumeFeed` was declared, exported and never used, because `volume_feed_of` reads the attribute with `getattr` (presence is not enough -- the value must also be a non-blank string). A Protocol nothing checks drifts from the check that matters, so a test now asserts both adapters satisfy it. 4. Three closing parens sat at column 0 from a scripted edit. `ruff check` passes on them because `ruff format` is not a CI gate, which is why they survived. Mutation-verified, 4 mutants. One SURVIVED on the first pass: reverting the narrowed rescue in (2) to a bare `return ()` broke nothing, because I had fixed the catch without writing a test that a lock must propagate. `test_a_locked_database_is_raised_not_read_as_unrecorded` closes that, and all four now die. Refs #696 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01NzuKAe2RVrPt9acVAWjRyL --- keel/data/feed_scope.py | 6 +++ keel/data/history.py | 6 +-- keel/data/market_feed.py | 31 +++++++++-- keel/data/repair.py | 4 +- keel/data/repository.py | 15 ++++-- tests/data/test_feed_provenance_is_wired.py | 60 ++++++++++++++++++++- tests/data/test_series_feed_provenance.py | 25 +++++++++ 7 files changed, 133 insertions(+), 14 deletions(-) diff --git a/keel/data/feed_scope.py b/keel/data/feed_scope.py index f5d421b4..8299bdbf 100644 --- a/keel/data/feed_scope.py +++ b/keel/data/feed_scope.py @@ -59,6 +59,12 @@ class DeclaresVolumeFeed(Protocol): to implement it would mean editing five packages to add a value four of them would state identically -- and a required field invites a placeholder, which is the one value that must never enter this table. + + `volume_feed_of` reads the attribute with `getattr` rather than through this Protocol, + because presence is not enough -- the value must also be a non-blank string. So the Protocol + would drift from the check that actually matters unless something asserted it: + `tests/data/test_feed_provenance_is_wired.py::test_both_adapters_satisfy_the_declared_protocol` + is what keeps it load-bearing rather than decorative. """ @property diff --git a/keel/data/history.py b/keel/data/history.py index b3a41a0e..040cbc0f 100644 --- a/keel/data/history.py +++ b/keel/data/history.py @@ -189,18 +189,18 @@ def ensure_history( client, repo, product, granularity, step, latest_cached, now_ts, sleep_fn, sleep_sec, feed=feed, -) + ) _fill_backward( client, repo, product, granularity, step, earliest_cached - step, start_floor, sleep_fn, sleep_sec, feed=feed, -) + ) else: _fill_backward( client, repo, product, granularity, step, now_ts, start_floor, sleep_fn, sleep_sec, feed=feed, -) + ) result[(product, granularity)] = coverage(repo, product, granularity, start_floor) return result diff --git a/keel/data/market_feed.py b/keel/data/market_feed.py index f4f3ddcf..450bbf61 100644 --- a/keel/data/market_feed.py +++ b/keel/data/market_feed.py @@ -22,14 +22,14 @@ import time from typing import TYPE_CHECKING +from keel.data.feed_scope import volume_feed_of from keel.data.history import MAX_CANDLES_PER_REQUEST +from keel.data.repository import Repository from keel.types import Candle, Granularity if TYPE_CHECKING: from keel_broker_api.port import Broker - from keel.data.repository import Repository - _GRANULARITY_SECONDS: dict[Granularity, int] = { Granularity.ONE_MINUTE: 60, Granularity.FIVE_MINUTE: 5 * 60, @@ -107,6 +107,7 @@ def backfill( history_days: int, *, now_ts: int | None = None, + feed: str | None = None, ) -> int: """Backfill `history_days` of closed candles for every product x granularity. @@ -116,6 +117,10 @@ def backfill( Returns the total number of candle rows written across all products/granularities. """ + # `backfill` has no in-tree caller today, so it resolves its own rather than relying on + # one being threaded in -- a writer that silently records nothing is the failure mode + # this whole table exists to prevent (#696). + feed = volume_feed_of(client) if feed is None else feed now_ts = int(time.time()) if now_ts is None else now_ts total_written = 0 @@ -144,7 +149,9 @@ def backfill( if window_start <= c.ts <= latest_closed and c.ts not in existing ] if gap_candles: - total_written += repo.upsert_candles(product_id, granularity, gap_candles) + total_written += repo.upsert_candles( + product_id, granularity, gap_candles, feed=feed + ) return total_written @@ -158,6 +165,7 @@ def _poll_catch_up( fetch_start: int, latest_closed: int, last_ts: int | None, + feed: str | None = None, ) -> int: """Fetch and upsert `[fetch_start, latest_closed]`, chunked under the venue's candle cap. @@ -178,7 +186,7 @@ def _poll_catch_up( ] if new_candles: seen.update(c.ts for c in new_candles) - total_written += repo.upsert_candles(product_id, granularity, new_candles) + total_written += repo.upsert_candles(product_id, granularity, new_candles, feed=feed) return total_written @@ -202,6 +210,11 @@ def poll_once( Returns the total number of new candle rows written. """ + # Resolved ONCE here, then passed down: this is the path `agent.run_once` uses on EVERY + # cycle, so it decides whether `candle_series_feed` has anything in it at all (#696). + # A client that declares nothing records nothing. + feed = volume_feed_of(client) + now_ts = int(time.time()) if now_ts is None else now_ts total_written = 0 @@ -217,7 +230,15 @@ def poll_once( fetch_start = last_ts + gran_sec if last_ts is not None else latest_closed total_written += _poll_catch_up( - client, repo, product_id, granularity, gran_sec, fetch_start, latest_closed, last_ts + client, + repo, + product_id, + granularity, + gran_sec, + fetch_start, + latest_closed, + last_ts, + feed=feed, ) return total_written diff --git a/keel/data/repair.py b/keel/data/repair.py index 42dfd54f..3bfd5e1f 100644 --- a/keel/data/repair.py +++ b/keel/data/repair.py @@ -37,6 +37,7 @@ from dataclasses import dataclass, field from keel.data import gaps as gaps_mod +from keel.data.feed_scope import volume_feed_of from keel.data.history import GRANULARITY_SECONDS, MAX_CANDLES_PER_REQUEST from keel.types import Granularity @@ -79,6 +80,7 @@ def _fetch_window_chunked( Upserting per chunk, rather than batching the whole window, is what lets an interior chunk failure leave the earlier chunks persisted instead of losing the whole fetch. """ + feed = volume_feed_of(client) fetch_start = window.start_ts - step fetch_end = window.end_ts + step chunk_start = fetch_start @@ -86,7 +88,7 @@ def _fetch_window_chunked( chunk_end = min(fetch_end, chunk_start + (MAX_CANDLES_PER_REQUEST - 1) * step) fetched = client.get_candles(product, granularity, chunk_start, chunk_end) if fetched: - repo.upsert_candles(product, granularity, fetched) + repo.upsert_candles(product, granularity, fetched, feed=feed) sleep_fn(sleep_sec) chunk_start = chunk_end + step diff --git a/keel/data/repository.py b/keel/data/repository.py index 8382dc04..f708f415 100644 --- a/keel/data/repository.py +++ b/keel/data/repository.py @@ -270,9 +270,14 @@ def get_series_feeds(self, product_id: str, granularity: Granularity) -> tuple[s " ORDER BY feed", (product_id, Granularity(granularity).value), ).fetchall() - except sqlite3.OperationalError: - # A hand-patched or partially-migrated file must not turn a liquidity read into a - # crash. No table is no provenance, which is true. + except sqlite3.OperationalError as exc: + # ONLY a missing table. A hand-patched or partially-migrated file must not turn a + # liquidity read into a crash -- no table is no provenance, which is true. But + # `OperationalError` is also what a lock timeout raises, and swallowing THAT would + # report "scope unrecorded" for a series whose scope is on disk, which is exactly + # the `None`-vs-`False` conflation `feed_scope` exists to prevent. + if "no such table" not in str(exc): + raise return () return tuple(row["feed"] for row in rows) @@ -291,7 +296,9 @@ def get_series_feed_window( " WHERE product_id = ? AND granularity = ? AND feed = ?", (product_id, Granularity(granularity).value, feed), ).fetchone() - except sqlite3.OperationalError: + except sqlite3.OperationalError as exc: + if "no such table" not in str(exc): # a lock is not an absence -- see above + raise return None return None if row is None else (int(row["first_seen_ts"]), int(row["last_seen_ts"])) diff --git a/tests/data/test_feed_provenance_is_wired.py b/tests/data/test_feed_provenance_is_wired.py index 9638fc54..9a3a59b0 100644 --- a/tests/data/test_feed_provenance_is_wired.py +++ b/tests/data/test_feed_provenance_is_wired.py @@ -16,7 +16,7 @@ from keel_core.types import Candle, Granularity -from keel.data import history +from keel.data import history, market_feed, repair from keel.data.db import connect, migrate from keel.data.feed_scope import volume_feed_of from keel.data.repository import Repository @@ -100,3 +100,61 @@ def test_a_client_that_declares_nothing_records_nothing() -> None: ) assert repo.get_candles("BTC-USD", Granularity.ONE_DAY) assert repo.get_series_feeds("BTC-USD", Granularity.ONE_DAY) == () + + +# --- EVERY writer, not just the one the mechanism was built against --------------------------- +# +# The three tests above prove `ensure_history` carries the declaration. They say nothing about +# the other paths that write candles, and the first review of this branch found that they did +# not: `market_feed.poll_once` -- the path `agent.run_once` uses on EVERY cycle in every +# deployment -- and `repair.repair_series` both dropped it. The mechanism worked and was very +# nearly inert, because `keel fetch` is not how bars normally arrive. + + +class _PollClient(_FeedClient): + """Enough of `Broker` for `poll_once`: it only reads candles here.""" + + def get_candles(self, product_id, granularity, start, end): # noqa: ANN001, ANN201 + return self._batches.pop(0) if self._batches else [] + + +def test_the_live_poll_path_records_the_feed() -> None: + """`agent.run_once` polls through here every cycle, so this is the path that decides + whether provenance exists in a real database at all.""" + repo = _repo() + client = _PollClient("alpaca:iex", [[_candle(0), _candle(86400)]]) + market_feed.poll_once( + client, repo, ["MSFT-USD"], [Granularity.ONE_DAY], now_ts=86400 * 3 + ) + assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == ("alpaca:iex",) + + +def test_the_gap_repair_path_records_the_feed() -> None: + """`fetch --repair-gaps` backfilled 29,676 bars into the hourly profile in one run. Bars + that arrive in bulk are exactly the ones whose provenance must not be missing.""" + repo = _repo() + repo.upsert_candles("MSFT-USD", Granularity.ONE_DAY, [_candle(0), _candle(86400 * 5)]) + client = _FeedClient("alpaca:iex", [[_candle(86400 * 2)], []]) + repair.repair_series( + client, + repo, + "MSFT-USD", + Granularity.ONE_DAY, + now_ts=86400 * 6, + sleep_fn=lambda _: None, + ) + assert "alpaca:iex" in repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) + + +def test_both_adapters_satisfy_the_declared_protocol() -> None: + """`DeclaresVolumeFeed` documents the contract `volume_feed_of` reads. Unless something + checks it, it drifts from the `getattr` that actually enforces it -- so this is what makes + the Protocol load-bearing rather than decorative.""" + from keel_broker_alpaca.adapter import AlpacaAdapter + from keel_broker_coinbase.adapter import CoinbaseAdapter + + from keel.data.feed_scope import DeclaresVolumeFeed + + assert isinstance(AlpacaAdapter(transport=object(), data_feed="iex"), DeclaresVolumeFeed) + assert isinstance(CoinbaseAdapter(transport=object()), DeclaresVolumeFeed) + assert volume_feed_of(_SilentClient([])) is None diff --git a/tests/data/test_series_feed_provenance.py b/tests/data/test_series_feed_provenance.py index 0f2f6696..8e694b0f 100644 --- a/tests/data/test_series_feed_provenance.py +++ b/tests/data/test_series_feed_provenance.py @@ -20,6 +20,7 @@ from __future__ import annotations +import sqlite3 from decimal import Decimal import pytest @@ -174,3 +175,27 @@ def test_a_legacy_database_missing_the_table_does_not_break_reads() -> None: repo = Repository(conn) assert repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) == () assert repo.get_series_feed_window("MSFT-USD", Granularity.ONE_DAY, "alpaca:iex") is None + + +def test_a_locked_database_is_raised_not_read_as_unrecorded(repo: Repository) -> None: + """The missing-table rescue must catch ONLY a missing table. + + `sqlite3.OperationalError` is also what a lock timeout raises, and keel reads and writes this + file from more than one process (see `db.connect`'s WAL note). Swallowing a lock would report + "scope unrecorded" for a series whose scope is sitting on disk -- turning a retryable error + into the `None` verdict that `feed_scope` reserves for the absence of evidence, and doing it + silently. A caught mutant is what put this test here: the rescue started life as a bare + `except`, and nothing failed. + """ + + class _Locked: + row_factory = sqlite3.Row + + def execute(self, *_args: object, **_kwargs: object) -> object: + raise sqlite3.OperationalError("database is locked") + + repo._conn = _Locked() # type: ignore[assignment] # noqa: SLF001 + with pytest.raises(sqlite3.OperationalError, match="locked"): + repo.get_series_feeds("MSFT-USD", Granularity.ONE_DAY) + with pytest.raises(sqlite3.OperationalError, match="locked"): + repo.get_series_feed_window("MSFT-USD", Granularity.ONE_DAY, "alpaca:iex")