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..8299bdbf --- /dev/null +++ b/keel/data/feed_scope.py @@ -0,0 +1,109 @@ +"""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. + + `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 + 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..040cbc0f 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,15 +188,18 @@ 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) 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 4415fe6b..f708f415 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,68 @@ 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 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) + + 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 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"])) + 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..9a3a59b0 --- /dev/null +++ b/tests/data/test_feed_provenance_is_wired.py @@ -0,0 +1,160 @@ +"""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, 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 + + +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) == () + + +# --- 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_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..8e694b0f --- /dev/null +++ b/tests/data/test_series_feed_provenance.py @@ -0,0 +1,201 @@ +"""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 + +import sqlite3 +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 + + +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") 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: