From 42be9793f58dc0d9689c488bd62caa52b3430eec Mon Sep 17 00:00:00 2001 From: Max Date: Tue, 21 Jul 2026 07:32:59 +0200 Subject: [PATCH] refactor: derive feed mapping; add startup preflight MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Delete feed_instrument — it was a persisted cache of a pure function. candidates() + to_feed_symbol() fully determine it, so derive the subscription set at subscribe time instead. Removes two failure modes (stale rows, non-retroactive) rather than managing them, and drops the map-feed step and its ordering rule. Add preflight: refuse to boot on an empty catalog/FK targets; log loud (but boot) when a held position cannot be priced or routed. Make sync-broker exit non-zero on FK skips unless --allow-skips. Stop logging the DB password on boot. --- .env.example | 6 +- Makefile | 4 +- cockpit/src/pages/Architecture.tsx | 157 +++++++--- cockpit/src/pages/DataFeeds.tsx | 6 +- cockpit/src/pages/Instruments.tsx | 8 +- crates/dataprovider/src/provider.rs | 4 +- crates/dataprovider/src/quote.rs | 68 +++++ db/access/ods.sql | 4 +- .../ods/public/0020_DROP_FEED_INSTRUMENT.sql | 26 ++ db/scripts/seed.sh | 2 +- db/scripts/seed_live.sh | 12 +- readme.md | 1 - scripts/fixtures/minimal_seed.sql | 12 +- src/admin.rs | 52 +++- src/feeds.rs | 56 ++++ src/main.rs | 20 +- src/preflight.rs | 272 ++++++++++++++++++ src/quote_feed.rs | 74 ++--- src/setup/brokers.rs | 17 ++ src/setup/catalog.rs | 3 +- src/setup/feeds.rs | 137 --------- src/setup/mod.rs | 1 - src/symbology_resolver.rs | 2 +- 23 files changed, 676 insertions(+), 268 deletions(-) create mode 100644 db/migrations/ods/public/0020_DROP_FEED_INSTRUMENT.sql create mode 100644 src/feeds.rs create mode 100644 src/preflight.rs delete mode 100644 src/setup/feeds.rs diff --git a/.env.example b/.env.example index d1b92dd..94c8922 100644 --- a/.env.example +++ b/.env.example @@ -34,9 +34,9 @@ KAFKA_TOPIC=orders KAFKA_CLIENT_ID=rust-oms KAFKA_PROJECTOR_GROUP_ID=position-projector-v1 -# --- Optional: instrument seeding (`make sync-broker` = instruments + broker mapping; -# `make map-feed` = data-feed pricing). On-demand, NOT scheduled. Run in-process, -# write as DB_USER (oms_user). --- +# --- Optional: instrument seeding (`make sync-broker` = instruments + broker +# mapping). On-demand, NOT scheduled. Runs in-process, writes as DB_USER +# (oms_user). Data feeds need no seeding — each derives what it can price. --- # DATABENTO_API_KEY= ALPACA_ENV=PAPER ALPACA_PAPER_API_KEY= diff --git a/Makefile b/Makefile index bd19670..17c84ed 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ .DEFAULT_GOAL := help -.PHONY: help db-provision db-migrate db-access db-seed db-fixtures db-setup db-reset sync-broker map-feed seed-live +.PHONY: help db-provision db-migrate db-access db-seed db-fixtures db-setup db-reset sync-broker seed-live # Load .env into the recipe shell (one shell per recipe line, so chain with &&). ENV := set -a && . ./.env && set +a @@ -45,8 +45,6 @@ sync-broker: ## Seed instruments + broker mapping from a broker (BROKER=alpaca|b @$(ENV) && cargo run --quiet -- setup sync-broker \ --broker "$${BROKER:-alpaca}" $${UNDERLYINGS:+--underlyings $$UNDERLYINGS} $${DRY_RUN:+--dry-run} $$ARGS -map-feed: ## Map a data feed's symbols onto seeded instruments (FEED=databento|binance|bybit; DRY_RUN=1) - @$(ENV) && cargo run --quiet -- setup map-feed --feed "$${FEED:?set FEED=databento|binance|bybit}" $${DRY_RUN:+--dry-run} $$ARGS seed-live: ## Sync every configured broker + map every feed in one idempotent pass (OPTION_UNDERLYINGS=SPY,QQQ) @./db/scripts/seed_live.sh diff --git a/cockpit/src/pages/Architecture.tsx b/cockpit/src/pages/Architecture.tsx index 1c8f081..a5c1562 100644 --- a/cockpit/src/pages/Architecture.tsx +++ b/cockpit/src/pages/Architecture.tsx @@ -10,8 +10,7 @@ const OVERVIEW = `flowchart LR BNF["Binance
spot book"]:::feed BYF["Bybit
spot book"]:::feed - FS["FeedSymbology
to_feed_symbol()"]:::feed - FI[("feed_instrument")]:::feed + FS["FeedSymbology
candidates() + to_feed_symbol()"]:::feed INST[("instrument
symbol @ venue")]:::core @@ -24,8 +23,7 @@ const OVERVIEW = `flowchart LR DBN --> FS BNF --> FS BYF --> FS - FS -- "map-feed" --> FI - FI -- "n:1" --> INST + FS -- "derived at startup
(nothing stored)" --> INST INST -- "1:n" --> BI BI -- "sync-broker" --> IP IP --> ALP @@ -40,7 +38,6 @@ const ER = `erDiagram CURRENCY ||--o{ INSTRUMENT : "quoted in" INSTRUMENT ||--o| INSTRUMENT_DERIVATIVE : "option legs" INSTRUMENT ||--o{ BROKER_INSTRUMENT : "tradable via" - INSTRUMENT ||--o{ FEED_INSTRUMENT : "priced by" BROKER_CONNECTION ||--o{ ACCOUNT : "routing target" INSTRUMENT { bigint id PK @@ -66,12 +63,6 @@ const ER = `erDiagram bool is_tradeable numeric min_quantity } - FEED_INSTRUMENT { - text feed_code "DATABENTO|BINANCE|BYBIT" - text feed_symbol - bigint instrument_id FK - bool is_active - } VENUE { text code PK "XNAS, OPRA, BINANCE" } CURRENCY { text code PK "USD, USDT" } BROKER_CONNECTION { @@ -82,16 +73,28 @@ const ER = `erDiagram ACCOUNT { text code PK }`; const SEED = `flowchart LR - ALP["Alpaca adapter"]:::exec - BIN["Binance adapter"]:::exec - INST[("instrument
+ derivative")]:::core - BI[("broker_instrument")]:::exec - FI[("feed_instrument")]:::feed - ALP -- "sync-broker" --> INST - BIN -- "sync-broker" --> INST - ALP -- "sync-broker" --> BI - BIN -- "sync-broker" --> BI - INST -- "map-feed" --> FI + subgraph S0["Step 0 — make db-seed"] + direction TB + CUR[("currency")]:::ref + VEN[("venue")]:::ref + end + + subgraph S1["Step 1 — make sync-broker"] + direction TB + INST[("instrument
+ derivative")]:::core + BI[("broker_instrument")]:::exec + end + + subgraph S2["At every startup — no seeding step"] + FI["each feed derives
what it can price"]:::feed + end + + CUR -- "FK" --> INST + VEN -- "FK" --> INST + INST --> BI + INST -- "read, never written" --> FI + + classDef ref fill:#eceff2,stroke:#6b7885,color:#1d262e; classDef core fill:#e7edf3,stroke:#3a4a5a,color:#16202b; classDef exec fill:#f7ecd6,stroke:#b4700e,color:#3a2a06; classDef feed fill:#dcf0f6,stroke:#0e7490,color:#05323d;`; @@ -100,7 +103,7 @@ const RUNTIME = `flowchart TB subgraph PRICE["Pricing path (market data)"] direction LR LF["Live feeds
Databento · Binance · Bybit"]:::feed - QF["subscribe held
via feed_instrument"]:::feed + QF["subscribe held
via FeedSymbology"]:::feed MR["mark_router
ranked by provider_feed_policy"]:::feed MS[("MarkStore")]:::feed PL["positions · M2M P/L"]:::core @@ -171,6 +174,44 @@ function Diagram({ chart }: { chart: string }) { ); } +// Genuinely ordered — each step reads what the previous one wrote, so the numbering +// carries information rather than decorating. +const STEPS = [ + { + cmd: "make db-seed", + color: "#6b7885", + body: ( + <> + Reference data: ISO 4217 currency, the ISO 10383 MIC venue registry, + plus crypto exchange venues (BINANCE, BYBIT) that have no MIC. Both + are foreign-key targets for instrument. + + ), + }, + { + cmd: "make sync-broker BROKER=…", + color: "#b4700e", + body: ( + <> + The broker's catalog becomes the instrument set. Writes instrument (+{" "} + instrument_derivative) and the broker_instrument routing handle in + one pass. Anything the broker doesn't list, we don't know about. + + ), + }, + { + cmd: "(nothing — feeds derive)", + color: "#0e7490", + body: ( + <> + There is no third step. At startup each feed reads the catalog step 2 built and works out what + it calls those instruments, in memory. Nothing is written, so nothing can go stale or be + forgotten after the next sync-broker. + + ), + }, +]; + const CARDS = [ { tag: "Master", @@ -198,12 +239,12 @@ const CARDS = [ { tag: "Market data", color: "#0e7490", - title: "feed_instrument", + title: "FeedSymbology", body: ( <> - Pricing mapping, 1:n. A feed's feed_symbol → instrument(s); one symbol can price - the pair on several venues. Built by map-feed, using each feed's own{" "} - FeedSymbology. + Pricing mapping, 1:n — and a pure function, not a table. candidates() picks the + catalog slice a feed covers, to_feed_symbol() renames it; one symbol can price the + pair on several venues. Evaluated at subscribe time. ), }, @@ -217,9 +258,9 @@ export function ArchitecturePage() { Instrument model One canonical instrument catalog in the middle, two independent mappings off it: a - broker's tradable handle (broker_instrument) and a data feed's pricing symbol ( - feed_instrument). Priceable and tradable are independent — each is just the - existence of a row. + broker's tradable handle (broker_instrument, a table) and a data feed's + pricing symbol (FeedSymbology, derived in code). Priceable and tradable are + independent. @@ -262,9 +303,9 @@ export function ArchitecturePage() { Tables & relationships - Foreign keys shown. broker_instrument and feed_instrument each - reference the master instrument; neither references the other. Two soft links by code (no - FK): broker_code → broker_connection, and feed_code{" "} + Foreign keys shown. broker_instrument references the master instrument. There + is no feed table: the pricing mapping is derived in code. Two soft links by code (no FK): + broker_code → broker_connection, and feed_code{" "} is ranked for failover in provider_feed_policy. @@ -272,14 +313,51 @@ export function ArchitecturePage() {
- Seeding — where rows come from + Seeding — the order matters - Broker-first: sync-broker creates the master catalog + broker mapping;{" "} - map-feed then maps feeds onto it. make seed-live runs the whole - chain idempotently. + Two steps, and the order is a hard dependency rather than a convention. Each reads what the + one before it wrote, and nothing ever points backwards — a data feed never creates an + instrument, and an instrument never creates a venue. make seed-live runs the + chain idempotently. Feeds are not part of it: they derive their mapping at startup. + + + {STEPS.map((s, i) => ( + +
+ + {i} + +
+ + {s.cmd} + + + {s.body} + +
+
+ + ))} + + + + How this used to bite. A missing venue or currency row + silently dropped every instrument referencing it, so a broker sync could + “succeed” with thousands of rows missing — sync-broker now exits + non-zero instead (pass --allow-skips to accept a partial catalog). And the + old map-feed step was not retroactive: instruments added by a later sync had + no feed mapping until someone re-ran it. Deriving the mapping removed that failure rather + than fixing it. What remains is checked at startup by preflight, which + refuses to boot on an empty catalog and names any held position it cannot price or + route. +
@@ -287,14 +365,15 @@ export function ArchitecturePage() { Runtime — the two paths - Pricing reads feed_instrument; order routing reads broker_instrument. - They never cross — a feed can price an instrument no broker trades, and vice versa. + Pricing derives its symbols through FeedSymbology; order routing reads{" "} + broker_instrument. They never cross — a feed can price an instrument no + broker trades, and vice versa.
- Reflects migrations 0016–0019 and the sync-broker / map-feed flow. + Reflects migrations 0016–0020 and the sync-broker flow. See also{" "} API docs diff --git a/cockpit/src/pages/DataFeeds.tsx b/cockpit/src/pages/DataFeeds.tsx index efbd245..d8f409c 100644 --- a/cockpit/src/pages/DataFeeds.tsx +++ b/cockpit/src/pages/DataFeeds.tsx @@ -6,8 +6,8 @@ import type { FeedSummary } from "../api/types"; // Market-data feeds — distinct from broker connections (execution). Shows live // feed stream health plus the configured ranked-failover policy and how many -// instruments each feed currently prices. Feeds are seeded/mapped from the CLI -// (`make map-feed FEED=…`), so this view is read-only. +// instruments each feed currently prices. The policy is configuration and the +// coverage count is derived from each feed's symbology, so this view is read-only. export function DataFeedsPage() { const { data, isLoading, error } = useQuery({ queryKey: ["/admin/feeds"], @@ -29,7 +29,7 @@ export function DataFeedsPage() { Ranked market-data sources per instrument class; lower rank wins, higher ranks are - failover. Mapped via make map-feed FEED=…. + failover. Coverage is derived from each feed's own symbology — there is no mapping table. {error && Failed to load feeds.} diff --git a/cockpit/src/pages/Instruments.tsx b/cockpit/src/pages/Instruments.tsx index ed05816..f9c7b29 100644 --- a/cockpit/src/pages/Instruments.tsx +++ b/cockpit/src/pages/Instruments.tsx @@ -6,8 +6,8 @@ import { api } from "../api/client"; import type { InstrumentSummary } from "../api/types"; // Read-only browser over the seeded master catalog. Instruments are seeded -// broker-first (`oms setup sync-broker`) and priced via feed mapping -// (`oms setup map-feed`) — there is no in-UI seeding flow. Server-side search +// broker-first (`oms setup sync-broker`); pricing coverage is derived per feed at +// startup, not seeded — there is no in-UI seeding flow. Server-side search // (debounced) against /admin/instruments. export function InstrumentsPage() { const [search, setSearch] = useState(""); @@ -27,8 +27,8 @@ export function InstrumentsPage() { {isLoading && } - Seeded from a broker's catalog (make sync-broker BROKER=alpaca) and priced via - a data feed (make map-feed FEED=databento). + Seeded from a broker's catalog (make sync-broker BROKER=alpaca). Which of these + a data feed can price is derived from the feed's own symbology at startup. &'static str; diff --git a/crates/dataprovider/src/quote.rs b/crates/dataprovider/src/quote.rs index 0a038eb..c953b91 100644 --- a/crates/dataprovider/src/quote.rs +++ b/crates/dataprovider/src/quote.rs @@ -79,6 +79,35 @@ pub struct InstrumentFilter { pub venue: Option<&'static str>, } +impl InstrumentFilter { + /// Append this filter to a `WHERE` clause already in progress, returning the + /// values to bind in the order they must be bound. + /// + /// Placeholders are numbered from `first_placeholder`, so a caller that already + /// binds parameters can append after them. Values are bound, never interpolated + /// — the column names are the only thing written into the SQL, and those are + /// `&'static str` chosen here rather than anything caller-supplied. + /// + /// Every condition is `AND`-ed, so the caller must supply a leading predicate + /// (`WHERE true`, or a real one) for the SQL to be well-formed. + pub fn push_conditions(&self, sql: &mut String, first_placeholder: usize) -> Vec<&'static str> { + use std::fmt::Write; + + let mut binds = Vec::new(); + for (column, value) in [ + ("instrument_class", self.instrument_class), + ("asset_class", self.asset_class), + ("venue", self.venue), + ] { + if let Some(v) = value { + binds.push(v); + let _ = write!(sql, " AND {column} = ${}", first_placeholder + binds.len() - 1); + } + } + binds + } +} + /// How a feed names the instruments it prices. /// /// The counterpart to [`LiveQuoteFeed`]: that trait moves a vendor's *data*, this one @@ -118,3 +147,42 @@ pub trait LiveQuoteFeed: DataProvider { health: &dyn FeedHealth, ) -> Result<(), ProviderError>; } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_filter_adds_nothing() { + let mut sql = String::from("WHERE true"); + let binds = InstrumentFilter::default().push_conditions(&mut sql, 1); + assert_eq!(sql, "WHERE true"); + assert!(binds.is_empty()); + } + + /// Placeholders must count only the conditions actually emitted, so a filter + /// that skips `asset_class` still numbers `venue` as `$2`, not `$3`. + #[test] + fn placeholders_are_contiguous_across_skipped_fields() { + let f = InstrumentFilter { + instrument_class: Some("OPTION"), + asset_class: None, + venue: Some("OPRA"), + }; + let mut sql = String::from("WHERE true"); + let binds = f.push_conditions(&mut sql, 1); + assert_eq!(sql, "WHERE true AND instrument_class = $1 AND venue = $2"); + assert_eq!(binds, vec!["OPTION", "OPRA"]); + } + + /// A caller that already bound `$1` appends starting at `$2`; bind order must + /// still match the returned vec. + #[test] + fn honours_a_placeholder_offset() { + let f = InstrumentFilter { instrument_class: Some("SPOT"), asset_class: Some("CRYPTO"), venue: None }; + let mut sql = String::from("WHERE x = $1"); + let binds = f.push_conditions(&mut sql, 2); + assert_eq!(sql, "WHERE x = $1 AND instrument_class = $2 AND asset_class = $3"); + assert_eq!(binds, vec!["SPOT", "CRYPTO"]); + } +} diff --git a/db/access/ods.sql b/db/access/ods.sql index c1807e6..c7b5622 100644 --- a/db/access/ods.sql +++ b/db/access/ods.sql @@ -12,12 +12,12 @@ GRANT SELECT ON ALL TABLES IN SCHEMA public TO oms_user; -- Instrument seeding + symbology: the OMS app seeds the master instrument catalog -- itself, broker-first. Broker sync (`oms setup sync-broker`) creates the master -- public.instrument (+ instrument_derivative) rows and the broker_instrument mapping --- in one pass; feed mapping (`oms setup map-feed`) writes feed_instrument; the +-- in one pass. Feed mapping is derived at runtime, not stored; the -- resolver stamps FIGI/CUSIP anchors from OpenFIGI. So oms_user needs write on the -- master catalog and both mapping tables. (SELECT on the FK targets venue/currency -- is covered by the blanket public SELECT above.) GRANT INSERT, UPDATE ON public.instrument, public.instrument_derivative TO oms_user; -GRANT INSERT, UPDATE, DELETE ON public.broker_instrument, public.feed_instrument TO oms_user; +GRANT INSERT, UPDATE, DELETE ON public.broker_instrument TO oms_user; -- Every future master table mdm_master creates is readable by oms_user. ALTER DEFAULT PRIVILEGES FOR ROLE mdm_master IN SCHEMA public diff --git a/db/migrations/ods/public/0020_DROP_FEED_INSTRUMENT.sql b/db/migrations/ods/public/0020_DROP_FEED_INSTRUMENT.sql new file mode 100644 index 0000000..86602b7 --- /dev/null +++ b/db/migrations/ods/public/0020_DROP_FEED_INSTRUMENT.sql @@ -0,0 +1,26 @@ +-- Drop feed_instrument: it was a persisted cache of a pure function. +-- +-- 0017 created it to hold "which feed prices this instrument, and what does that +-- feed call it". Both halves are now answered by the feed itself — `candidates()` +-- selects the catalog slice, `to_feed_symbol()` renames it — and both are pure +-- Rust with no I/O (see `crates/dataprovider/src/quote.rs`). The table's contents +-- were therefore fully determined by `instrument` plus those two functions. +-- +-- A cache of a pure function can only add failure modes, and it added two: +-- +-- * stale rows — a delisted instrument's mapping stayed forever, nothing +-- deactivated it; +-- * not retroactive — instruments seeded by a later `sync-broker` had no +-- mapping until someone remembered to re-run `map-feed`, and the symptom was +-- silence: no error, no prices. +-- +-- Deriving the mapping at subscribe time removes both states rather than managing +-- them, and removes the ordering rule (`sync-broker` *then* `map-feed`) that made +-- them possible. `oms setup map-feed` is gone with it. +-- +-- Irreversible in the sense that the rows are not recoverable, but nothing is +-- lost: re-deriving from the catalog reproduces them exactly. This was verified +-- against the live catalog before the drop — the derived subscription set matched +-- the stored one for all three feeds. + +DROP TABLE IF EXISTS feed_instrument; diff --git a/db/scripts/seed.sh b/db/scripts/seed.sh index 702f971..cc718fd 100755 --- a/db/scripts/seed.sh +++ b/db/scripts/seed.sh @@ -43,7 +43,7 @@ psql_db "$ODS_DB" -f scripts/seed_crypto_venues.sql # Instruments are seeded on demand, broker-first (not here, not scheduled): # `make sync-broker BROKER=alpaca` creates the master instrument + broker_instrument -# rows, then `make map-feed FEED=databento` maps a data feed onto them. Or +# rows; data feeds then derive what they can price from the catalog. Or # `make db-fixtures` for the no-creds minimal set. All depend on currency + venue. echo "post-deployment seeding complete." diff --git a/db/scripts/seed_live.sh b/db/scripts/seed_live.sh index 3a740ed..01fe7a5 100755 --- a/db/scripts/seed_live.sh +++ b/db/scripts/seed_live.sh @@ -51,15 +51,7 @@ else echo "· skip binance sync-broker (BINANCE_${binance_env}_API_KEY/PRIVATE_KEY_PATH not set)" fi -# --- Feeds: map onto the seeded instruments (no external creds needed). --- -# Databento only prices options that were seeded above; skip if no key was ever -# configured (the feed itself won't run without it either). -if have DATABENTO_API_KEY; then - run map-feed --feed databento -else - echo "· skip databento map-feed (DATABENTO_API_KEY not set)" -fi -run map-feed --feed binance -run map-feed --feed bybit +# Feeds need no seeding step: each feed derives the instruments it can price from +# the catalog above, at startup, from its own symbology. Nothing to run here. echo "✅ live seeding complete." diff --git a/readme.md b/readme.md index c140903..7d859e2 100644 --- a/readme.md +++ b/readme.md @@ -37,7 +37,6 @@ Optional live data (on-demand, needs vendor creds in `.env`): ```sh make sync-broker BROKER=alpaca # seed instruments + broker mapping from Alpaca (ALPACA_PAPER_*) make sync-broker BROKER=alpaca UNDERLYINGS=SPY,QQQ # also seed those option chains -make map-feed FEED=databento # price the seeded options via Databento OPRA (DATABENTO_API_KEY) ``` ## Run diff --git a/scripts/fixtures/minimal_seed.sql b/scripts/fixtures/minimal_seed.sql index c66a7ca..8d91eea 100644 --- a/scripts/fixtures/minimal_seed.sql +++ b/scripts/fixtures/minimal_seed.sql @@ -23,13 +23,11 @@ FROM instrument i WHERE i.symbol = 'SPY' AND i.venue = 'ARCX' ON CONFLICT (instrument_id, broker_code) DO NOTHING; --- Feed mapping (public.feed_instrument): Databento prices SPY on ARCX. Databento's --- raw symbol equals ours here, so feed_symbol = 'SPY'. -INSERT INTO feed_instrument (feed_code, feed_symbol, instrument_id) -SELECT 'DATABENTO', 'SPY', i.id -FROM instrument i -WHERE i.symbol = 'SPY' AND i.venue = 'ARCX' -ON CONFLICT (feed_code, feed_symbol, instrument_id) DO NOTHING; +-- No feed mapping row: each feed derives what it can price from its own symbology +-- at startup. Note that nothing prices SPY *equity* today — the Databento feed +-- carries OPRA options only — so a position in it will be reported as unmarkable +-- by preflight. That was equally true before, when a feed_instrument row here +-- claimed otherwise and the subscription silently matched nothing. -- A test broker connection so an account can be created and orders can route. -- Creds are resolved from env by (broker_code, environment); this is just the diff --git a/src/admin.rs b/src/admin.rs index 36b2d1e..8676d82 100644 --- a/src/admin.rs +++ b/src/admin.rs @@ -575,7 +575,8 @@ pub struct FeedSummary { /// Ranked failover preference within the class; lower wins. pub rank: i32, pub enabled: bool, - /// How many instruments this feed currently maps (feed_instrument), for the class. + /// How many active instruments this feed's symbology covers, for the class. + /// Derived from the feed's own `candidates()` filter at request time. pub mapped_instruments: i64, } @@ -589,22 +590,53 @@ pub struct FeedSummary { pub async fn list_feeds( State(state): State, ) -> Result>, AdminError> { - let rows = sqlx::query_as::<_, FeedSummary>( - "SELECT p.source_code AS feed_code, p.instrument_class, p.rank, p.enabled, \ - count(i.id) AS mapped_instruments \ - FROM oms.provider_feed_policy p \ - LEFT JOIN feed_instrument fi ON fi.feed_code = p.source_code AND fi.is_active \ - LEFT JOIN instrument i ON i.id = fi.instrument_id \ - AND i.instrument_class = p.instrument_class \ - GROUP BY p.source_code, p.instrument_class, p.rank, p.enabled \ - ORDER BY p.source_code, p.rank", + let policies = sqlx::query_as::<_, (String, String, i32, bool)>( + "SELECT source_code, instrument_class, rank, enabled \ + FROM oms.provider_feed_policy \ + ORDER BY source_code, rank", ) .fetch_all(state.pool()) .await .map_err(map_db_error)?; + + // `mapped_instruments` is derived, not stored: count the catalog slice this + // feed's symbology covers, narrowed to the policy row's instrument_class. A + // policy naming a feed this build doesn't ship counts zero rather than 404ing — + // the row is still real and the operator should see it. + let mut rows = Vec::with_capacity(policies.len()); + for (feed_code, instrument_class, rank, enabled) in policies { + let mapped_instruments = match crate::feeds::by_code(&feed_code) { + Some(feed) => count_priceable(state.pool(), feed, &instrument_class) + .await + .map_err(map_db_error)?, + None => 0, + }; + rows.push(FeedSummary { feed_code, instrument_class, rank, enabled, mapped_instruments }); + } Ok(Json(rows)) } +/// How many active instruments of `instrument_class` this feed's symbology covers. +/// +/// Counts what `candidates()` selects; it does not run `to_feed_symbol` over the +/// catalog, so an instrument the feed would decline is still counted. That keeps +/// this a single `COUNT` rather than a full scan into Rust, and the two only differ +/// for malformed symbols — which `preflight` reports separately. +async fn count_priceable( + pool: &sqlx::PgPool, + feed: &dyn dataprovider::FeedSymbology, + instrument_class: &str, +) -> Result { + let mut sql = String::from("SELECT count(*) FROM instrument WHERE status = 'ACTIVE' AND instrument_class = $1"); + let binds = feed.candidates().push_conditions(&mut sql, 2); + + let mut query = sqlx::query_scalar::<_, i64>(&sql).bind(instrument_class); + for b in &binds { + query = query.bind(*b); + } + query.fetch_one(pool).await +} + #[utoipa::path( get, path = "/admin/broker-connections", tag = "admin", responses( diff --git a/src/feeds.rs b/src/feeds.rs new file mode 100644 index 0000000..9536bfc --- /dev/null +++ b/src/feeds.rs @@ -0,0 +1,56 @@ +//! The market-data feeds this build ships, as a lookup by feed code. +//! +//! The live path does not need this: `QuoteFeedSession` is generic over its feed, +//! so it reaches [`dataprovider::FeedSymbology`] through the concrete type. This +//! registry exists for the callers that only hold a *code* — a `provider_feed_policy` +//! row, a preflight report — and must find the symbology behind it. +//! +//! Unit structs, so these are `'static` references to promoted constants: no +//! allocation, no state, nothing to initialise at boot. + +use dataprovider::FeedSymbology; + +/// Every feed with a symbology, whether or not it is currently spawned. Spawning is +/// conditional on credentials (see `serve`); coverage is not, so +/// [`crate::preflight`] can still report what a feed *would* price. +pub static ALL: &[&dyn FeedSymbology] = &[ + &crate::opra_stream::DatabentoOpraFeed, + &crate::binance_feed::BinanceFeed, + &crate::bybit_feed::BybitFeed, +]; + +/// The symbology for a feed code, or `None` if this build has no such feed — +/// which is the case for a `provider_feed_policy` row naming a feed that was +/// removed or never implemented. +pub fn by_code(code: &str) -> Option<&'static dyn FeedSymbology> { + ALL.iter().copied().find(|f| f.code() == code) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_feed_is_reachable_by_its_own_code() { + for feed in ALL { + let found = by_code(feed.code()).expect("registered feed resolves by code"); + assert_eq!(found.code(), feed.code()); + } + } + + /// Codes must be distinct or `by_code` silently resolves to the wrong feed, + /// which would subscribe one vendor's symbols on another's session. + #[test] + fn feed_codes_are_unique() { + let mut codes: Vec<&str> = ALL.iter().map(|f| f.code()).collect(); + codes.sort_unstable(); + let before = codes.len(); + codes.dedup(); + assert_eq!(codes.len(), before, "duplicate feed code in ALL"); + } + + #[test] + fn unknown_code_resolves_to_none() { + assert!(by_code("NOPE").is_none()); + } +} diff --git a/src/main.rs b/src/main.rs index a64dced..d312955 100644 --- a/src/main.rs +++ b/src/main.rs @@ -58,6 +58,8 @@ mod quote_feed; mod mark_router; mod binance_feed; mod bybit_feed; +mod feeds; +mod preflight; #[derive(OpenApi)] #[openapi( @@ -172,8 +174,6 @@ enum Command { enum SetupCmd { /// Seed the master instrument catalog + broker_instrument mapping from a broker. SyncBroker(setup::brokers::Args), - /// Map a data feed's symbols onto seeded instruments (feed_instrument rows). - MapFeed(setup::feeds::Args), } #[tokio::main] @@ -189,12 +189,6 @@ async fn main() { std::process::exit(1); } } - Some(Command::Setup(SetupCmd::MapFeed(args))) => { - if let Err(e) = setup::feeds::run(args).await { - error!("setup map-feed failed: {e}"); - std::process::exit(1); - } - } None => serve().await, } } @@ -213,7 +207,7 @@ async fn serve() { db_user, db_password, db_host, db_port, db_name ); - info!("Connecting to the database at {}", database_url); + info!("Connecting to the database at {}:{}/{} as {}", db_host, db_port, db_name, db_user); let pool = match PgPool::connect(&database_url).await { Ok(pool) => pool, Err(e) => { @@ -222,7 +216,13 @@ async fn serve() { } }; - // Schema is owned and migrated by the `mdm` repo; OMS only connects. + // Refuse to start on a catalog that cannot work, and name what is merely + // degraded. Before any feed spawns, so a broken catalog surfaces here rather + // than as a feed that quietly subscribes to nothing. + if let Err(e) = preflight::run(&pool).await { + error!("preflight failed: {e}"); + return; + } // Init kafka client from .env (optional — publishing is disabled if not configured) let kafka_client: Option = match kafka::KafkaConfig::from_env() { diff --git a/src/preflight.rs b/src/preflight.rs new file mode 100644 index 0000000..9f75ab2 --- /dev/null +++ b/src/preflight.rs @@ -0,0 +1,272 @@ +//! Startup checks: refuse to boot broken, and say so loudly when merely degraded. +//! +//! Runs once in `serve()` after the pool connects and before any feed spawns. +//! Database-only by design — it never calls a broker or vendor API, so it stays +//! fast, works with every venue offline, and cannot be the reason startup hangs. +//! +//! Two severities, and the split is deliberate: +//! +//! * **Fatal** — the catalog is empty or its FK targets are missing. Nothing can +//! work in these states and booting only hides them behind a later, stranger +//! error. +//! * **Degraded** — a specific held position cannot be priced or cannot be routed. +//! Logged at ERROR with the symbol, but the server still comes up: a position you +//! cannot price is exactly the position you need the application running to +//! manage. Refusing to boot would be the worse failure. + +use sqlx::PgPool; +use tracing::{error, info}; + +/// How many offending symbols to name before truncating. The point is to notice a +/// class of problem, not to dump the catalog into the log. +const SAMPLE: usize = 10; + +/// A condition that makes the process unable to do anything useful. +#[derive(Debug, PartialEq, Eq)] +pub struct Fatal(pub String); + +impl std::fmt::Display for Fatal { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +/// Run every check. `Err` means do not start. +pub async fn run(pool: &PgPool) -> Result<(), Fatal> { + check_catalog(pool).await?; + report_held(pool).await; + Ok(()) +} + +/// Fatal checks: the master catalog and the FK targets it depends on. +/// +/// An empty `venue` or `currency` table is the signature of a half-run `make +/// db-seed`; every subsequent `sync-broker` would skip every instrument on an FK +/// miss and report success over an empty catalog. +async fn check_catalog(pool: &PgPool) -> Result<(), Fatal> { + for (table, hint) in [ + ("venue", "run `make db-seed`"), + ("currency", "run `make db-seed`"), + ("instrument", "run `make sync-broker BROKER=alpaca`"), + ] { + let n: i64 = sqlx::query_scalar(&format!("SELECT count(*) FROM {table}")) + .fetch_one(pool) + .await + .map_err(|e| Fatal(format!("preflight: reading {table} failed: {e}")))?; + if n == 0 { + return Err(Fatal(format!("{table} is empty — {hint}"))); + } + } + Ok(()) +} + +/// Degraded checks, per held position. Logs; never fails the boot. +/// +/// Answers the two questions that otherwise fail silently at runtime: can we mark +/// this position, and can we close it? +async fn report_held(pool: &PgPool) { + // position.instrument_id is text holding the numeric instrument.id, so a held + // row whose instrument is gone shows up as a NULL join rather than an error. + let held = sqlx::query_as::<_, HeldRow>( + "SELECT DISTINCT p.instrument_id AS raw_id, i.id, i.symbol, \ + i.instrument_class, i.asset_class, i.venue \ + FROM position p \ + LEFT JOIN instrument i ON i.id::text = p.instrument_id \ + WHERE p.net_qty <> 0", + ) + .fetch_all(pool) + .await; + + let held = match held { + Ok(h) => h, + // Not fatal: the catalog checks passed, so this is a transient read + // failure, and refusing to boot over a report would be perverse. + Err(e) => { + error!("preflight: could not read held positions: {e}"); + return; + } + }; + + if held.is_empty() { + info!("preflight: no held positions"); + return; + } + + let orphans: Vec<&str> = held + .iter() + .filter(|r| r.id.is_none()) + .map(|r| r.raw_id.as_str()) + .collect(); + if !orphans.is_empty() { + error!( + "preflight: {} held position(s) reference a missing instrument: {}", + orphans.len(), + sample(&orphans) + ); + } + + let live: Vec<&HeldRow> = held.iter().filter(|r| r.id.is_some()).collect(); + + // Priceable: does any feed's symbology both select this instrument and accept + // its symbol? Uses the same two functions the live path uses, so this cannot + // drift from what actually gets subscribed. + let unpriceable: Vec<&str> = live + .iter() + .filter(|r| !crate::feeds::ALL.iter().any(|feed| covers(*feed, r))) + .filter_map(|r| r.symbol.as_deref()) + .collect(); + if !unpriceable.is_empty() { + error!( + "preflight: {} held instrument(s) no feed can price — they will sit unmarked: {}", + unpriceable.len(), + sample(&unpriceable) + ); + } + + // Routable: a held instrument with no broker_instrument row cannot be closed. + let unroutable = sqlx::query_scalar::<_, String>( + "SELECT DISTINCT i.symbol \ + FROM position p \ + JOIN instrument i ON i.id::text = p.instrument_id \ + LEFT JOIN broker_instrument bi ON bi.instrument_id = i.id AND bi.is_tradeable \ + WHERE p.net_qty <> 0 AND bi.instrument_id IS NULL \ + ORDER BY 1", + ) + .fetch_all(pool) + .await; + match unroutable { + Ok(rows) if !rows.is_empty() => { + let names: Vec<&str> = rows.iter().map(String::as_str).collect(); + error!( + "preflight: {} held instrument(s) have no tradeable broker mapping — cannot be closed: {}", + names.len(), + sample(&names) + ); + } + Ok(_) => {} + Err(e) => error!("preflight: could not check broker routing: {e}"), + } + + // One line per feed, so the healthy case is a glance rather than an absence. + for feed in crate::feeds::ALL { + let n = live.iter().filter(|r| covers(*feed, r)).count(); + info!("preflight: {} covers {} of {} held instrument(s)", feed.code(), n, live.len()); + } +} + +/// A held position joined to its instrument. `id`/`symbol`/… are `NULL` when the +/// position references an instrument that no longer exists. +#[derive(sqlx::FromRow)] +struct HeldRow { + raw_id: String, + id: Option, + symbol: Option, + instrument_class: Option, + asset_class: Option, + venue: Option, +} + +/// Whether `feed` would actually subscribe this instrument. +/// +/// Mirrors `load_subscribable` exactly: every `Some` field of `candidates()` must +/// match, *and* `to_feed_symbol` must accept the symbol. Checking only some of the +/// filter would over-report coverage and hide the very positions this exists to +/// surface. +fn covers(feed: &dyn dataprovider::FeedSymbology, row: &HeldRow) -> bool { + let f = feed.candidates(); + let matches = |want: Option<&str>, have: &Option| match (want, have) { + (None, _) => true, + (Some(w), Some(h)) => w == h, + // The filter constrains a column the row has no value for: not a match. + (Some(_), None) => false, + }; + + matches(f.instrument_class, &row.instrument_class) + && matches(f.asset_class, &row.asset_class) + && matches(f.venue, &row.venue) + && row.symbol.as_deref().is_some_and(|s| feed.to_feed_symbol(s).is_some()) +} + +/// Render up to [`SAMPLE`] names, noting how many were withheld. +fn sample(names: &[&str]) -> String { + let shown: Vec<&str> = names.iter().copied().take(SAMPLE).collect(); + let rest = names.len().saturating_sub(shown.len()); + if rest == 0 { + shown.join(", ") + } else { + format!("{} (+{rest} more)", shown.join(", ")) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sample_lists_everything_when_short() { + assert_eq!(sample(&["A", "B"]), "A, B"); + } + + /// A big unpriceable set must not dump the catalog into the log line. + #[test] + fn sample_truncates_and_says_how_many_it_hid() { + let names: Vec = (0..13).map(|i| format!("S{i}")).collect(); + let refs: Vec<&str> = names.iter().map(String::as_str).collect(); + let out = sample(&refs); + assert!(out.ends_with("(+3 more)"), "{out}"); + assert_eq!(out.matches(", ").count(), SAMPLE - 1); + } + + #[test] + fn sample_of_nothing_is_empty() { + assert_eq!(sample(&[]), ""); + } + + fn row(symbol: &str, class: &str, asset: &str, venue: &str) -> HeldRow { + HeldRow { + raw_id: "1".into(), + id: Some(1), + symbol: Some(symbol.into()), + instrument_class: Some(class.into()), + asset_class: Some(asset.into()), + venue: Some(venue.into()), + } + } + + /// The case already live in the fixture: SPY equity on ARCX. It is *not* + /// priceable by the OPRA feed — that feed carries options only — so preflight + /// must report it rather than let it sit silently unmarked. + #[test] + fn equity_is_not_covered_by_the_opra_feed() { + let held = row("SPY", "SPOT", "EQUITY", "ARCX"); + assert!(!covers(&crate::opra_stream::DatabentoOpraFeed, &held)); + assert!(!crate::feeds::ALL.iter().any(|f| covers(*f, &held))); + } + + #[test] + fn opra_feed_covers_a_held_option() { + let held = row("SPY260724P00739000", "OPTION", "EQUITY", "OPRA"); + assert!(covers(&crate::opra_stream::DatabentoOpraFeed, &held)); + } + + /// Bybit leaves `venue` open, so it prices the pair wherever it is listed — + /// including the Binance-seeded row. Binance does not: its filter pins the venue. + #[test] + fn open_venue_filter_covers_another_venues_listing() { + let held = row("BTCUSDT", "SPOT", "CRYPTO", "BINANCE"); + assert!(covers(&crate::bybit_feed::BybitFeed, &held)); + assert!(covers(&crate::binance_feed::BinanceFeed, &held)); + + let elsewhere = row("BTCUSDT", "SPOT", "CRYPTO", "XNAS"); + assert!(covers(&crate::bybit_feed::BybitFeed, &elsewhere)); + assert!(!covers(&crate::binance_feed::BinanceFeed, &elsewhere)); + } + + /// A symbol inside `candidates` that the symbology still declines is not + /// covered — the filter alone is not enough. + #[test] + fn declined_symbol_is_not_covered_despite_matching_the_filter() { + let malformed = row("SPY", "OPTION", "EQUITY", "OPRA"); + assert!(!covers(&crate::opra_stream::DatabentoOpraFeed, &malformed)); + } +} diff --git a/src/quote_feed.rs b/src/quote_feed.rs index e6fdfa0..522a8b9 100644 --- a/src/quote_feed.rs +++ b/src/quote_feed.rs @@ -10,8 +10,8 @@ use std::collections::HashMap; -use dataprovider::{LiveQuoteFeed, Quote, SymbolAdds}; -use sqlx::{PgPool, Row}; +use dataprovider::{FeedSymbology, LiveQuoteFeed, Quote, SymbolAdds}; +use sqlx::PgPool; use tokio::sync::mpsc; use tokio::time::{sleep, Duration}; use tracing::info; @@ -23,7 +23,7 @@ use crate::stream_supervisor::{Session, StreamResult}; /// the query, so this is just the poll interval for "did we buy anything yet". const IDLE_RECHECK_SECS: u64 = 60; -pub struct QuoteFeedSession { +pub struct QuoteFeedSession { feed: F, pool: PgPool, out: mpsc::Sender, @@ -31,7 +31,7 @@ pub struct QuoteFeedSession { health: StreamHandle, } -impl QuoteFeedSession { +impl QuoteFeedSession { pub fn new( feed: F, pool: PgPool, @@ -51,7 +51,7 @@ impl QuoteFeedSession { /// with a timer as the backstop for positions this process didn't see. async fn wait_for_subscribable(&mut self) -> Result, sqlx::Error> { loop { - let held = load_subscribable(&self.pool, self.feed.code()).await?; + let held = load_subscribable(&self.pool, &self.feed).await?; if !held.is_empty() { return Ok(held); } @@ -64,7 +64,7 @@ impl QuoteFeedSession { } #[async_trait::async_trait] -impl Session for QuoteFeedSession { +impl Session for QuoteFeedSession { async fn run_once(&mut self) -> StreamResult { let known = self.wait_for_subscribable().await?; info!(source = self.feed.code(), count = known.len(), "quote feed: subscribing held set"); @@ -75,7 +75,7 @@ impl Session for QuoteFeedSession { // them lets a fill be picked up without waiting for a reconnect. tokio::select! { r = self.feed.run_session(known.clone(), &self.out, &mut add_rx, &self.health) => r.map_err(Into::into), - r = watch_held(&self.pool, self.feed.code(), &mut self.position_changed_rx, &add_tx, known) => r, + r = watch_held(&self.pool, &self.feed, &mut self.position_changed_rx, &add_tx, known) => r, } } } @@ -86,7 +86,7 @@ impl Session for QuoteFeedSession { /// carries no data cannot carry stale data. async fn watch_held( pool: &PgPool, - source_code: &'static str, + feed: &dyn FeedSymbology, doorbell: &mut mpsc::Receiver<()>, add_tx: &mpsc::Sender, mut known: HashMap, @@ -94,7 +94,7 @@ async fn watch_held( loop { match doorbell.recv().await { Some(()) => { - let latest = load_subscribable(pool, source_code).await?; + let latest = load_subscribable(pool, feed).await?; let adds: SymbolAdds = latest .into_iter() .filter(|(sym, _)| !known.contains_key(sym)) @@ -103,7 +103,7 @@ async fn watch_held( continue; } info!( - source = source_code, + source = feed.code(), count = adds.len(), "quote feed: subscribing newly-held instruments" ); @@ -124,34 +124,42 @@ async fn watch_held( /// The held instruments this feed can price, as `feed_symbol -> instrument_id`. /// -/// Driven by `feed_instrument`: the feed's *own* symbol drives the subscription, -/// and the mapping is the single source of what this feed covers. An instrument -/// with no `feed_instrument` row for this feed is silently absent — that is the -/// "held but this feed can't price it" state, not an error; another feed may. +/// Derived, not stored. The feed declares the slice of the catalog it covers +/// ([`FeedSymbology::candidates`]) and how it names it +/// ([`FeedSymbology::to_feed_symbol`]); both are pure, so the mapping is a function +/// of the catalog and cannot go stale or miss instruments seeded later. A held +/// instrument outside `candidates`, or one `to_feed_symbol` declines, is silently +/// absent — that is the "held but this feed can't price it" state, not an error; +/// another feed may. [`crate::preflight`] is what makes it visible. /// -/// `feed_instrument` is 1:n (one feed symbol may price instruments on several -/// venues), so a `feed_symbol` collision keeps the last instrument id. In practice -/// held sets don't collide today (one venue per crypto pair); true fan-out to -/// multiple held instruments from one quote is a later change in the mark path. -async fn load_subscribable( +/// The mapping is 1:n (one feed symbol may price instruments on several venues), so +/// a `feed_symbol` collision keeps the last instrument id. In practice held sets +/// don't collide today (one venue per crypto pair); true fan-out to multiple held +/// instruments from one quote is a later change in the mark path. +pub(crate) async fn load_subscribable( pool: &PgPool, - feed_code: &str, + feed: &dyn FeedSymbology, ) -> Result, sqlx::Error> { - let rows = sqlx::query( - // position.instrument_id is text holding the numeric instrument.id. - "SELECT DISTINCT fi.feed_symbol, i.id \ + // Held only: this is the subscription set, not the catalog. Push the feed's + // filter down rather than scanning every instrument. + // position.instrument_id is text holding the numeric instrument.id. + let mut sql = String::from( + "SELECT DISTINCT i.id, i.symbol \ FROM position p \ JOIN instrument i ON i.id::text = p.instrument_id \ - JOIN feed_instrument fi ON fi.instrument_id = i.id \ - AND fi.feed_code = $1 AND fi.is_active \ - WHERE p.net_qty <> 0", - ) - .bind(feed_code) - .fetch_all(pool) - .await?; + WHERE p.net_qty <> 0 AND i.status = 'ACTIVE'", + ); + let binds = feed.candidates().push_conditions(&mut sql, 1); - Ok(rows - .iter() - .map(|r| (r.get::("feed_symbol"), r.get::("id"))) + let mut query = sqlx::query_as::<_, (i64, String)>(&sql); + for b in &binds { + query = query.bind(*b); + } + + Ok(query + .fetch_all(pool) + .await? + .into_iter() + .filter_map(|(id, symbol)| feed.to_feed_symbol(&symbol).map(|s| (s, id))) .collect()) } diff --git a/src/setup/brokers.rs b/src/setup/brokers.rs index 2120680..ef6ecf3 100644 --- a/src/setup/brokers.rs +++ b/src/setup/brokers.rs @@ -53,6 +53,11 @@ pub struct Args { /// Fetch + match + print counts, but write nothing. #[arg(long)] pub dry_run: bool, + /// Exit 0 even when instruments were skipped on a missing venue/currency. + /// Without this, any FK skip fails the run: a partial catalog that reports + /// success is how a seeding gap survives to become a missing price. + #[arg(long)] + pub allow_skips: bool, } pub async fn run(args: Args) -> Result<(), Box> { @@ -119,6 +124,18 @@ pub async fn run(args: Args) -> Result<(), Box> { } tx.commit().await?; info!("sync-broker done: upserted {n} {broker_code} broker_instrument row(s)"); + + // Loud by default. `upsert_catalog` already named the offending venues and + // currencies; make them consequential so a half-seeded catalog cannot pass for + // a finished one in a script or CI step. + let skipped = summary.skipped_fk(); + if skipped > 0 && !args.allow_skips { + return Err(format!( + "{skipped} instrument(s) skipped on a missing venue/currency — see the warnings above. \ + Seed the missing rows, or pass --allow-skips to accept a partial catalog." + ) + .into()); + } Ok(()) } diff --git a/src/setup/catalog.rs b/src/setup/catalog.rs index 0d63582..b84f14b 100644 --- a/src/setup/catalog.rs +++ b/src/setup/catalog.rs @@ -6,7 +6,8 @@ //! `InstrumentProvider` yields the definitions, this module writes the master rows //! and hands back the `(symbol, venue) -> id` map so the caller can attach the //! `broker_instrument` mapping. It does not touch any symbology bridge table — -//! broker mapping lives in `broker_instrument`, feed mapping in `feed_instrument`. +//! broker mapping lives in `broker_instrument`; feed mapping is derived at runtime +//! from each feed's own symbology, not stored. use std::collections::{BTreeSet, HashMap, HashSet}; diff --git a/src/setup/feeds.rs b/src/setup/feeds.rs deleted file mode 100644 index 71a2204..0000000 --- a/src/setup/feeds.rs +++ /dev/null @@ -1,137 +0,0 @@ -//! `oms setup map-feed` — map a data feed's own symbols onto seeded instruments. -//! -//! Populates `public.feed_instrument`, the market-data mapping. Feeds are -//! independent of brokers: this maps a feed's symbology onto whatever instruments -//! exist, and it is deliberately 1:n — a crypto feed maps its pair onto the pair on -//! every venue that trades it (e.g. the Bybit feed prices the Binance-seeded -//! BTCUSDT). Idempotent; safe to re-run after seeding more instruments. -//! -//! This module owns the *orchestration* only. The symbology itself — which -//! instruments a feed can price and what it calls them — lives on each feed struct -//! as a [`FeedSymbology`] impl, next to the code that speaks that vendor's protocol -//! (`opra_stream.rs`, `binance_feed.rs`, `bybit_feed.rs`). That keeps a vendor's -//! quirks in one place and makes them unit-testable without a database. - -use clap::{Args as ClapArgs, ValueEnum}; -use dataprovider::FeedSymbology; -use sqlx::PgPool; -use tracing::{info, warn}; - -/// Max rows per bulk statement, matching `setup::catalog`. -const BATCH: usize = 4000; - -#[derive(Copy, Clone, Debug, PartialEq, Eq, ValueEnum)] -pub enum Feed { - /// Databento OPRA consolidated options NBBO. Maps the space-padded OSI. - Databento, - /// Binance spot book ticker. Maps the pair on the Binance venue. - Binance, - /// Bybit spot orderbook. Maps the pair onto every crypto instrument for it. - Bybit, -} - -impl Feed { - /// The feed struct that owns this feed's symbology. Unit structs, so these are - /// `'static` references to promoted constants — no allocation, no state. - fn symbology(self) -> &'static dyn FeedSymbology { - match self { - Feed::Databento => &crate::opra_stream::DatabentoOpraFeed, - Feed::Binance => &crate::binance_feed::BinanceFeed, - Feed::Bybit => &crate::bybit_feed::BybitFeed, - } - } -} - -#[derive(ClapArgs, Debug, Clone)] -pub struct Args { - /// Which feed's symbol mapping to (re)build. - #[arg(long, value_enum)] - pub feed: Feed, - /// Count what would be mapped, but write nothing. - #[arg(long)] - pub dry_run: bool, -} - -pub async fn run(args: Args) -> Result<(), Box> { - let pool = PgPool::connect(&super::database_url()?).await?; - let feed = args.feed.symbology(); - let feed_code = feed.code(); - - // The feed declares which slice of the catalog it can price; push that down into - // the query rather than scanning every instrument. - let filter = feed.candidates(); - let mut sql = String::from("SELECT id, symbol FROM instrument WHERE status = 'ACTIVE'"); - let mut binds: Vec<&str> = Vec::new(); - for (column, value) in [ - ("instrument_class", filter.instrument_class), - ("asset_class", filter.asset_class), - ("venue", filter.venue), - ] { - if let Some(v) = value { - binds.push(v); - sql.push_str(&format!(" AND {column} = ${}", binds.len())); - } - } - - let mut query = sqlx::query_as::<_, (i64, String)>(&sql); - for b in &binds { - query = query.bind(*b); - } - let candidates = query.fetch_all(&pool).await?; - - // Translate each master symbol into the feed's own. `None` means the feed - // declines it — counted, not silently dropped. - let mut feed_symbols: Vec = Vec::with_capacity(candidates.len()); - let mut instrument_ids: Vec = Vec::with_capacity(candidates.len()); - let mut declined: Vec = Vec::new(); - for (id, symbol) in candidates { - match feed.to_feed_symbol(&symbol) { - Some(feed_symbol) => { - feed_symbols.push(feed_symbol); - instrument_ids.push(id); - } - None => declined.push(symbol), - } - } - - if !declined.is_empty() { - // Bounded sample: the point is to notice a symbology mismatch, not to dump - // the catalog. - let sample: Vec<&str> = declined.iter().take(5).map(String::as_str).collect(); - warn!( - "{feed_code}: {} candidate(s) declined by symbology, e.g. {sample:?}", - declined.len() - ); - } - - if args.dry_run { - info!( - "dry run: {feed_code} would map {} instrument(s) ({} declined), no write", - feed_symbols.len(), - declined.len() - ); - return Ok(()); - } - - let mut affected = 0u64; - for (symbol_chunk, id_chunk) in feed_symbols.chunks(BATCH).zip(instrument_ids.chunks(BATCH)) { - affected += sqlx::query( - "INSERT INTO feed_instrument (feed_code, feed_symbol, instrument_id) \ - SELECT $1, t.feed_symbol, t.instrument_id \ - FROM UNNEST($2::text[], $3::bigint[]) AS t(feed_symbol, instrument_id) \ - ON CONFLICT (feed_code, feed_symbol, instrument_id) DO NOTHING", - ) - .bind(feed_code) - .bind(symbol_chunk) - .bind(id_chunk) - .execute(&pool) - .await? - .rows_affected(); - } - - info!( - "map-feed done: {feed_code} mapped {} instrument(s), {affected} new feed_instrument row(s)", - feed_symbols.len() - ); - Ok(()) -} diff --git a/src/setup/mod.rs b/src/setup/mod.rs index f9c2874..b8cdcf0 100644 --- a/src/setup/mod.rs +++ b/src/setup/mod.rs @@ -2,7 +2,6 @@ pub mod brokers; pub mod catalog; -pub mod feeds; use std::env; diff --git a/src/symbology_resolver.rs b/src/symbology_resolver.rs index 48fbf0d..6d99368 100644 --- a/src/symbology_resolver.rs +++ b/src/symbology_resolver.rs @@ -3,7 +3,7 @@ //! `public.instrument`. //! //! This is enrichment, not mapping. The two mapping tables are owned elsewhere — -//! `broker_instrument` by broker sync, `feed_instrument` by feed mapping. The +//! `broker_instrument` by broker sync; feed symbology is derived, not stored. The //! resolver only answers "which master instrument is this, and what is its FIGI", //! and records the FIGI anchor so later lookups are cheap. Off the order/quote hot //! path.