feat(data): record which feed a cached series came from (#696, part 1…

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,
 }
 
 

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",
+]

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)

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

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,14 +80,15 @@ 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
     while chunk_start <= fetch_end:
         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