From 3b93f1264d9d05a879113e18566e32d6359767a3 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 7 Aug 2026 11:48:42 +0200 Subject: [PATCH 1/7] fix(openapi): document bearer auth + add order-submit body examples MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The /orders/submit and other user endpoints go through auth_middleware, which accepts both Basic (key_id:secret) and a Bearer trading token (key_id.secret), but their utoipa annotations only declared basic_auth — so Scalar/Swagger showed username:password as the sole auth option. Declare bearer_token alongside basic_auth on all six user endpoints and add descriptions to both schemes. Also add #[schema(example = ...)] to SubmitOrderRequest fields so the generated docs show a runnable body (valid UUIDs, BIGINT instrument_id) instead of empty placeholder strings. --- src/handlers.rs | 25 +++++++++++++++++++------ src/main.rs | 19 ++++++++++++++++--- 2 files changed, 35 insertions(+), 9 deletions(-) diff --git a/src/handlers.rs b/src/handlers.rs index fc44ecf..88afd69 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -83,15 +83,28 @@ pub async fn health() -> &'static str { /// resolved from the portfolio's `default_account_id`; when present it overrides. #[derive(Debug, Deserialize, utoipa::ToSchema)] pub struct SubmitOrderRequest { + /// Client-generated UUID; also the idempotency key (a repeat is a 409). + #[schema(example = "3f6b1c2e-8a4d-4e5f-9b21-1c2d3e4f5a6b")] pub order_id: String, + /// Your own reference string, echoed back on updates. + #[schema(example = "my-ref-001")] pub client_order_id: String, + /// Portfolio UUID the order books against. + #[schema(example = "b2c3d4e5-6f70-4812-93a4-556677889900")] pub portfolio_id: String, + /// Optional account UUID; omit to use the portfolio's default account. + #[schema(example = json!(null))] pub account_id: Option, + /// Instrument surrogate key (BIGINT as string), not a UUID. + #[schema(example = "42")] pub instrument_id: String, pub side: OrderSide, pub order_type: OrderType, pub time_in_force: TimeInForce, + /// Required for `limit` orders; omit for `market`. + #[schema(example = json!(null))] pub limit_price: Option, + #[schema(example = 1.0)] pub quantity: f64, } @@ -124,7 +137,7 @@ impl SubmitOrderRequest { (status = 502, description = "Broker rejected the order"), (status = 503, description = "No broker adapter configured"), ), - security(("basic_auth" = [])) + security(("basic_auth" = []), ("bearer_token" = [])) )] pub async fn orders_submit( State(state): State, @@ -624,7 +637,7 @@ pub async fn orders_submit( (status = 404, description = "Order not found"), (status = 409, description = "Order state version mismatch"), ), - security(("basic_auth" = [])) + security(("basic_auth" = []), ("bearer_token" = [])) )] pub async fn orders_cancel( State(app_state): State, @@ -888,7 +901,7 @@ pub async fn orders_cancel( (status = 400, description = "Invalid UUID"), (status = 404, description = "Order not found"), ), - security(("basic_auth" = [])) + security(("basic_auth" = []), ("bearer_token" = [])) )] pub async fn get_order( State(state): State, @@ -958,7 +971,7 @@ pub async fn get_order( (status = 200, description = "OK", body = [Position]), (status = 403, description = "No view grant for principal/portfolio"), ), - security(("basic_auth" = [])) + security(("basic_auth" = []), ("bearer_token" = [])) )] pub async fn get_portfolio_positions( State(state): State, @@ -1076,7 +1089,7 @@ pub struct Allocation { (status = 404, description = "Order not found"), (status = 422, description = "Nothing filled, over-allocation, or invalid target"), ), - security(("basic_auth" = [])) + security(("basic_auth" = []), ("bearer_token" = [])) )] pub async fn create_allocations( State(state): State, @@ -1206,7 +1219,7 @@ pub async fn create_allocations( get, path = "/orders/{id}/allocations", tag = "orders", params(("id" = Uuid, Path, description = "Order ID")), responses((status = 200, description = "OK", body = [Allocation])), - security(("basic_auth" = [])) + security(("basic_auth" = []), ("bearer_token" = [])) )] pub async fn list_allocations( State(state): State, diff --git a/src/main.rs b/src/main.rs index 8acba7e..7953ae8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -45,7 +45,7 @@ use dotenvy::dotenv; use tracing::{error, info, warn, Level}; use tracing_subscriber; use utoipa::OpenApi; -use utoipa::openapi::security::{Http, HttpAuthScheme, SecurityScheme}; +use utoipa::openapi::security::{HttpAuthScheme, HttpBuilder, SecurityScheme}; mod kafka; mod execution; mod alpaca_stream; @@ -141,11 +141,24 @@ impl utoipa::Modify for SecurityAddon { let components = openapi.components.get_or_insert_default(); components.add_security_scheme( "basic_auth", - SecurityScheme::Http(Http::new(HttpAuthScheme::Basic)), + SecurityScheme::Http( + HttpBuilder::new() + .scheme(HttpAuthScheme::Basic) + .description(Some("Username = key_id, password = secret.")) + .build(), + ), ); components.add_security_scheme( "bearer_token", - SecurityScheme::Http(Http::new(HttpAuthScheme::Bearer)), + SecurityScheme::Http( + HttpBuilder::new() + .scheme(HttpAuthScheme::Bearer) + .description(Some( + "User endpoints: a trading token, `key_id.secret` (Databento-style). \ + Admin endpoints (/admin/*): the static admin token.", + )) + .build(), + ), ); } } From e05abdbc25dcc0a58261af297ab3f9514db166c1 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 14 Aug 2026 15:20:28 +0200 Subject: [PATCH 2/7] fix(opra): refresh playground symbols, add OPRA_SYMBOLS override The hardcoded SPY 260717 contracts expired on 2026-07-17, so the gateway answered SymbolResolutionFailed for both. Live symbology only resolves currently-listed contracts; the historical API still has them, which is why this looks like a data gap and is not one. Defaults now point at the Sep-2026 monthly ATM straddle, verified listed against the OPRA definition schema. They will go stale too, so symbols are overridable with OPRA_SYMBOLS and the header carries the curl that lists what is actually listed today. --- examples/opra_playground.rs | 43 ++++++++++++++++++++++++++++++------- 1 file changed, 35 insertions(+), 8 deletions(-) diff --git a/examples/opra_playground.rs b/examples/opra_playground.rs index 5847521..8477b98 100644 --- a/examples/opra_playground.rs +++ b/examples/opra_playground.rs @@ -6,11 +6,19 @@ //! export DATABENTO_API_KEY=db-... # your key //! cargo run --example opra_playground //! -//! Edit `SYMBOLS` below. These are OSI strings in Databento's space-padded -//! form: 6-char root left-justified + space-padded, then YYMMDD, C/P, strike*1000 -//! zero-padded to 8. The expiries below WILL go stale — pick a live contract -//! (e.g. from https://databento.com or the Alpaca option chain) or you'll just -//! see the symbol mapping and no quotes. +//! Symbols come from `OPRA_SYMBOLS` (comma-separated) or the `SYMBOLS` default +//! below. These are OSI strings in Databento's space-padded form: 6-char root +//! left-justified + space-padded, then YYMMDD, C/P, strike*1000 zero-padded to 8. +//! +//! The defaults WILL go stale: an expired contract is not in live symbology and +//! the gateway answers `SymbolResolutionFailed` (non-fatal — the rest of the +//! subscription still streams). To refresh, list what is actually listed: +//! +//! curl -s -u "$DATABENTO_API_KEY:" \ +//! -X POST https://hist.databento.com/v0/timeseries.get_range \ +//! -d dataset=OPRA.PILLAR -d symbols=SPY.OPT -d stype_in=parent \ +//! -d schema=definition -d encoding=csv -d start=$(date -v-1d +%F) \ +//! | cut -d, -f6 | sort -u use databento::{ dbn::{self, Record, Schema, SType, UNDEF_PRICE}, @@ -19,9 +27,11 @@ use databento::{ }; // Space-padded OSI. Root is 6 chars, so "SPY" → "SPY " (3 trailing spaces). +// ATM straddle on the Sep-2026 monthly; verified listed on 2026-08-13 with SPY +// around 772. Overridable with OPRA_SYMBOLS. const SYMBOLS: &[&str] = &[ - "SPY 260717C00750000", - "SPY 260717P00750000", + "SPY 260918C00770000", + "SPY 260918P00770000", ]; const DATASET: &str = "OPRA.PILLAR"; @@ -31,6 +41,18 @@ fn px(v: i64) -> Option { (v != UNDEF_PRICE).then_some(v as f64 / 1e9) } +/// Left-justify the root in 6 so compact OSI matches Databento's wire form. +/// Already-padded input is returned unchanged (tail is a fixed 15 chars). +fn pad_osi(s: &str) -> String { + match s.len().checked_sub(15).filter(|&n| n > 0) { + Some(split) => { + let (root, tail) = s.split_at(split); + format!("{:<6}{tail}", root.trim_end()) + } + None => s.to_string(), + } +} + /// Readable symbol for a dbn numeric instrument_id (falls back to the raw id). fn sym_of(map: &std::collections::HashMap, id: u32) -> String { map.get(&id).cloned().unwrap_or_else(|| format!("id={id}")) @@ -48,7 +70,12 @@ async fn main() -> Result<(), Box> { .build() .await?; - let symbols: Vec = SYMBOLS.iter().map(|s| s.to_string()).collect(); + // OPRA_SYMBOLS lets a stale default be swapped without a rebuild. Roots are + // padded here so a hand-typed "SPY260918C00770000" also resolves. + let symbols: Vec = match std::env::var("OPRA_SYMBOLS") { + Ok(v) => v.split(',').map(str::trim).filter(|s| !s.is_empty()).map(pad_osi).collect(), + Err(_) => SYMBOLS.iter().map(|s| s.to_string()).collect(), + }; // OPRA is multi-venue; use the consolidated top-of-book (cmbp-1) rather than // per-publisher mbp-1. (tcbbo — quote-on-trade — decodes to the same Cmbp1Msg.) println!("subscribing {} symbols (cmbp-1):", symbols.len()); From fcaf969acb56df72ac7fea9e5fe0827a48be8e2a Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 14 Aug 2026 15:20:36 +0200 Subject: [PATCH 3/7] fix(alpaca): request the full option chain via expiration_date_gte MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Without an explicit expiration_date_gte, /v2/options/contracts returns only the nearest one or two expiries *and* omits next_page_token — so the pagination loop saw a complete response and stopped. Every far-dated contract was silently missing. Masked until now because the retired Databento-sourced rows did carry the far expiries, so the catalog looked complete while the broker path had been truncating since July. A SPY sync goes from 720 contracts across 2 expiries to 14,368 across 36, out to 2028-12-15. --- src/adapters/alpaca.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/adapters/alpaca.rs b/src/adapters/alpaca.rs index 4a7fbaa..f49d797 100644 --- a/src/adapters/alpaca.rs +++ b/src/adapters/alpaca.rs @@ -161,15 +161,24 @@ impl AlpacaAdapter { /// (`GET /v2/options/contracts`). The compact OSI is both the master symbol and /// the order-entry handle; options route by symbol, so the routing native id is /// left None. Strike/expiry/kind come straight from Alpaca's contract fields. + /// + /// `expiration_date_gte` is not a filter here — it is what makes the endpoint + /// return the whole chain. Without it Alpaca answers with only the nearest one or + /// two expiries and no `next_page_token`, so the pagination loop below sees a + /// complete response and the far month/LEAPS contracts never arrive. Pinning it + /// to today asks for everything still live; already-expired contracts are dropped + /// again at ingest (`setup::catalog`), which uses the same UTC date. pub async fn list_option_contracts( &self, underlyings: &[String], ) -> Result, BrokerError> { let mut out = Vec::new(); let mut page_token: Option = None; + let today = chrono::Utc::now().date_naive(); loop { let mut url = format!( - "{}/v2/options/contracts?status=active&limit=10000&underlying_symbols={}", + "{}/v2/options/contracts?status=active&limit=10000\ + &expiration_date_gte={today}&underlying_symbols={}", self.base_url, underlyings.join(",") ); From 127591f0bcd4d8e8cac0a2a389673335b37444df Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 14 Aug 2026 15:21:23 +0200 Subject: [PATCH 4/7] feat(expiry): retire dated instruments on a real UTC expiry instant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit setup::catalog drops already-expired contracts at ingest, but nothing aged a row out afterwards: an option seeded while live stayed ACTIVE forever once its expiry passed, and re-running the sync could not help — it is an upsert, so a contract that stops being listed leaves its stale row untouched. The quote feed kept resubscribing to dead OSI symbols and the order path kept accepting them. expiry_date could not answer the question on its own. It is a bare DATE, and the value is a venue-local calendar date (Alpaca sends "YYYY-MM-DD"), so comparing it to current_date silently depends on the DB server's timezone — and an option trades until 16:00 local on that date, not until midnight. So the timezone lives on the venue's calendar, not per broker: instrument is keyed (symbol, venue) and shared across brokers, so a broker-scoped timezone would give one row two different UTC expiries. calendar.timezone was already declared the authority in 0001; it was simply never seeded or read. The conversion runs in Postgres, which carries the IANA database — 16:00 New York is 20:00Z in summer and 21:00Z in winter, so a stored offset would be wrong for half the year. expiry_date stays as the OSI truth; expires_at is derived beside it. The sweep then needs no timezone reasoning at all: expires_at < now(). Retiring an instrument is all this does. The existing status = 'ACTIVE' filters on the order and subscription paths do the rest, so nothing else became expiry-aware. What happens to an open order or a held position when its contract expires is deliberately left alone — the broker cancels the order and the custodian resolves the position, and neither is something the OMS should infer. Preflight reports a stranded position rather than acting on one. Backfill lives in the sweep rather than a migration because db-migrate runs before db-seed: a migration would join an empty calendar and update nothing. Recomputing on every pass also re-heals after a calendar is corrected. Verified against the live catalog: 28,990 contracts dated, 13,138 retired, DST boundary confirmed (2026-09-18 -> 20:00Z, 2026-12-18 -> 21:00Z), and a HALTED contract with a past expiry left untouched. --- .../0021_ALTER_CALENDAR_ADD_CLOSE_TIME.sql | 11 ++ ...R_INSTRUMENT_DERIVATIVE_ADD_EXPIRES_AT.sql | 12 ++ ...23_ALTER_INSTRUMENT_ADD_EXPIRED_STATUS.sql | 15 ++ db/scripts/seed.sh | 5 + db/scripts/seed_calendars.sql | 55 +++++++ src/admin.rs | 28 ++++ src/expiry.rs | 138 ++++++++++++++++++ src/main.rs | 10 ++ src/preflight.rs | 64 ++++++++ src/setup/brokers.rs | 9 +- src/setup/catalog.rs | 10 ++ 11 files changed, 355 insertions(+), 2 deletions(-) create mode 100644 db/migrations/ods/public/0021_ALTER_CALENDAR_ADD_CLOSE_TIME.sql create mode 100644 db/migrations/ods/public/0022_ALTER_INSTRUMENT_DERIVATIVE_ADD_EXPIRES_AT.sql create mode 100644 db/migrations/ods/public/0023_ALTER_INSTRUMENT_ADD_EXPIRED_STATUS.sql create mode 100644 db/scripts/seed_calendars.sql create mode 100644 src/expiry.rs diff --git a/db/migrations/ods/public/0021_ALTER_CALENDAR_ADD_CLOSE_TIME.sql b/db/migrations/ods/public/0021_ALTER_CALENDAR_ADD_CLOSE_TIME.sql new file mode 100644 index 0000000..e40d114 --- /dev/null +++ b/db/migrations/ods/public/0021_ALTER_CALENDAR_ADD_CLOSE_TIME.sql @@ -0,0 +1,11 @@ +-- The local time a venue's contracts stop trading on their expiry date. Together +-- with `calendar.timezone` this turns `instrument_derivative.expiry_date` — a bare +-- venue-local DATE — into a real UTC instant (`instrument_derivative.expires_at`), +-- computed by the OMS expiry job (src/expiry.rs). +-- +-- Nullable on purpose. A timezone alone cannot produce an instant, and a venue with +-- no close (crypto, 24/7) has no honest value here. NULL leaves `expires_at` NULL, +-- which preflight names out loud rather than guessing an hour and sweeping a live +-- contract. + +ALTER TABLE calendar ADD COLUMN close_time TIME; diff --git a/db/migrations/ods/public/0022_ALTER_INSTRUMENT_DERIVATIVE_ADD_EXPIRES_AT.sql b/db/migrations/ods/public/0022_ALTER_INSTRUMENT_DERIVATIVE_ADD_EXPIRES_AT.sql new file mode 100644 index 0000000..776ef73 --- /dev/null +++ b/db/migrations/ods/public/0022_ALTER_INSTRUMENT_DERIVATIVE_ADD_EXPIRES_AT.sql @@ -0,0 +1,12 @@ +-- The UTC instant a contract stops trading, derived from `expiry_date` plus the +-- venue calendar's timezone + close_time. Written by the OMS expiry job +-- (src/expiry.rs), which also re-derives it if a calendar is corrected. +-- +-- `expiry_date` stays: it is the OSI truth and the venue-local calendar date that +-- symbology and broker reconciliation speak in. This column is the derived instant +-- the sweep compares against now(), so the comparison never depends on the DB's +-- timezone setting. + +ALTER TABLE instrument_derivative ADD COLUMN expires_at TIMESTAMPTZ; + +CREATE INDEX idx_derivative_expires_at ON instrument_derivative(expires_at); diff --git a/db/migrations/ods/public/0023_ALTER_INSTRUMENT_ADD_EXPIRED_STATUS.sql b/db/migrations/ods/public/0023_ALTER_INSTRUMENT_ADD_EXPIRED_STATUS.sql new file mode 100644 index 0000000..5247547 --- /dev/null +++ b/db/migrations/ods/public/0023_ALTER_INSTRUMENT_ADD_EXPIRED_STATUS.sql @@ -0,0 +1,15 @@ +-- Allow an EXPIRED status: a dated contract whose expiry instant has passed. Set by +-- the OMS expiry sweep (src/expiry.rs), never by hand. +-- +-- Distinct from INACTIVE on purpose. INACTIVE is a judgement about a listing +-- (delisted, withdrawn); EXPIRED is a scheduled fact that was knowable from +-- `expires_at` the day the contract was created. Both are excluded by the +-- `status = 'ACTIVE'` filters on the order path and the quote-subscription path, so +-- the sweep needs no other code to take effect — but only one of them says why. +-- +-- The original CHECK was anonymous (0003_CREATE_INSTRUMENT_TABLE.sql), so Postgres +-- named it instrument_status_check. + +ALTER TABLE instrument DROP CONSTRAINT IF EXISTS instrument_status_check; +ALTER TABLE instrument ADD CONSTRAINT instrument_status_check + CHECK (status IN ('ACTIVE', 'INACTIVE', 'HALTED', 'EXPIRED')); diff --git a/db/scripts/seed.sh b/db/scripts/seed.sh index cc718fd..f0e13ba 100755 --- a/db/scripts/seed.sh +++ b/db/scripts/seed.sh @@ -41,6 +41,11 @@ python3 scripts/seed_venues.py --source data/ISO10383_MIC.csv echo "→ seeding crypto exchange venues (synthetic, non-MIC)" psql_db "$ODS_DB" -f scripts/seed_crypto_venues.sql +# After both venue seeders: calendars FK to venue, and the MIC registry has no +# timezone, so this is the hand-curated source for expiry/close-time conversion. +echo "→ seeding venue calendars (timezone + close time)" +psql_db "$ODS_DB" -f scripts/seed_calendars.sql + # Instruments are seeded on demand, broker-first (not here, not scheduled): # `make sync-broker BROKER=alpaca` creates the master instrument + broker_instrument # rows; data feeds then derive what they can price from the catalog. Or diff --git a/db/scripts/seed_calendars.sql b/db/scripts/seed_calendars.sql new file mode 100644 index 0000000..465edb6 --- /dev/null +++ b/db/scripts/seed_calendars.sql @@ -0,0 +1,55 @@ +-- Venue trading calendars: the timezone + close time authority. +-- +-- These cannot come from the automated venue source. seed_venues.py loads the ISO +-- 10383 MIC registry, and that file carries no timezone column — only country and +-- city (see the note in 0001_CREATE_VENUE_TABLE.sql, which omits timezone for that +-- reason and points here instead). So the mapping is hand-curated, and every venue +-- the OMS can actually book against needs a row. +-- +-- What depends on it: an option's `expiry_date` is a bare venue-local DATE (Alpaca +-- sends "YYYY-MM-DD", meaning an ET calendar date). The expiry job derives the real +-- UTC instant as `(expiry_date + close_time) AT TIME ZONE timezone`, so an option +-- whose venue has no calendar row never expires and never stops being subscribed. +-- Preflight names those at boot rather than letting them go quiet. +-- +-- Timezone is an IANA name, not an offset: 16:00 in New York is 20:00Z in summer and +-- 21:00Z in winter, and Postgres applies the right one per date. +-- +-- close_time is NULL for the crypto venues — they trade 24/7, and spot pairs carry +-- no expiry_date at all, so there is nothing to derive. NULL is the honest value; a +-- placeholder would invent an expiry instant for something that has none. +-- +-- The equity/option venues are the MICs alpaca_exchange_to_mic() can emit +-- (src/adapters/alpaca.rs), plus OPRA for the listed options themselves. All US, +-- all 16:00 ET. +-- +-- Run as mdm_master (owner of public), matching seed_currencies.sql. Depends on +-- venue being seeded first — a missing venue silently seeds no calendar, which +-- preflight then reports. + +SET ROLE mdm_master; +SET search_path TO public; + +INSERT INTO calendar (code, venue_id, timezone, close_time, description) +SELECT c.code, v.id, c.timezone, c.close_time, c.description +FROM (VALUES + ('OPRA', 'America/New_York', TIME '16:00', 'US listed options (OPRA consolidated)'), + ('XNAS', 'America/New_York', TIME '16:00', 'Nasdaq'), + ('XNYS', 'America/New_York', TIME '16:00', 'New York Stock Exchange'), + ('ARCX', 'America/New_York', TIME '16:00', 'NYSE Arca'), + ('XASE', 'America/New_York', TIME '16:00', 'NYSE American'), + ('BATS', 'America/New_York', TIME '16:00', 'Cboe BZX'), + ('IEXG', 'America/New_York', TIME '16:00', 'IEX'), + ('OTCM', 'America/New_York', TIME '16:00', 'OTC Markets'), + ('BINANCE', 'UTC', NULL, 'Binance — 24/7, no close'), + ('BYBIT', 'UTC', NULL, 'Bybit — 24/7, no close') +) AS c(code, timezone, close_time, description) +JOIN venue v ON v.code = c.code +ON CONFLICT (code) DO UPDATE SET + venue_id = EXCLUDED.venue_id, + timezone = EXCLUDED.timezone, + close_time = EXCLUDED.close_time, + description = EXCLUDED.description, + updated_at = now(); + +RESET ROLE; diff --git a/src/admin.rs b/src/admin.rs index 0c3c026..d2610af 100644 --- a/src/admin.rs +++ b/src/admin.rs @@ -1471,6 +1471,16 @@ pub struct BackfillResult { pub unresolved: i64, } +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct ExpirySweepResult { + /// Contracts whose UTC `expires_at` was derived or corrected from the venue + /// calendar. Non-zero on the first run after seeding a calendar, or after + /// correcting one; zero on a steady-state run. + pub recomputed: u64, + /// Instruments moved from ACTIVE to EXPIRED. + pub expired: u64, +} + fn map_resolve_error(err: ResolveError) -> AdminError { match err { ResolveError::Db(e) => AdminError { @@ -1552,6 +1562,24 @@ pub async fn backfill_symbology( Ok(Json(result)) } +/// Run the instrument expiry pass now, instead of waiting for the hourly tick. +/// +/// Same function the background job calls, so an on-demand run and a scheduled one +/// cannot diverge. Useful right after seeding a calendar (every dated contract gets +/// its instant immediately) and for confirming a correction took effect. +#[utoipa::path( + post, path = "/admin/instruments/expiry-sweep", tag = "admin", + responses((status = 200, description = "Expiry sweep summary", body = ExpirySweepResult)), + security(("bearer_token" = [])) +)] +pub async fn expiry_sweep( + State(state): State, +) -> Result, AdminError> { + info!("admin expiry sweep"); + let (recomputed, expired) = crate::expiry::run_once(state.pool()).await.map_err(map_db_error)?; + Ok(Json(ExpirySweepResult { recomputed, expired })) +} + // ───────────────────────────────────────────────────────────────────────────── #[derive(Debug)] diff --git a/src/expiry.rs b/src/expiry.rs new file mode 100644 index 0000000..9933ff6 --- /dev/null +++ b/src/expiry.rs @@ -0,0 +1,138 @@ +//! Instrument expiry lifecycle: derive each dated contract's UTC expiry instant, +//! then retire the ones that have passed. +//! +//! Two problems, one job. The catalog stores `instrument_derivative.expiry_date` as +//! a bare DATE because that is what the source gives it — Alpaca sends +//! `"YYYY-MM-DD"`, which is a *venue-local* calendar date, and an option trades +//! until 16:00 that day, not until midnight. Comparing that date to `current_date` +//! would make the answer depend on the database server's timezone. So the first half +//! of this module resolves the date into a real instant using the venue's calendar, +//! and everything downstream compares instants. +//! +//! The second half is the lifecycle gap that motivated it. [`crate::setup::catalog`] +//! filters already-expired contracts at *ingest*, so nothing expired ever enters the +//! catalog — but nothing ages a row out *afterwards*. An option seeded while live +//! stays `ACTIVE` forever once its expiry passes, and re-running the broker sync does +//! not help: it is an upsert, so a contract that simply stops being listed leaves its +//! stale row untouched. The visible symptom was Databento answering +//! `SymbolResolutionFailed` for a contract the quote feed kept resubscribing. +//! +//! Retiring an instrument is all this does. The existing `status = 'ACTIVE'` filters +//! on the order path ([`crate::handlers`]) and the subscription path +//! ([`crate::quote_feed::load_subscribable`]) then take effect on their own — no +//! expiry-awareness anywhere else. +//! +//! Deliberately *not* here: what happens to an open order or a held position when its +//! contract expires. The broker cancels the order and the custodian resolves the +//! position (worthless, or exercised into the underlying). Both are events the OMS +//! receives — order recon already reconciles to the broker's snapshot — and neither +//! is something it should infer. [`crate::preflight`] reports a position stranded in +//! an expired contract; it does not act on one. + +use std::time::Duration; + +use sqlx::PgPool; +use tracing::{error, info}; + +/// How long between sweeps. +/// +/// Hourly rather than daily: the work is two `UPDATE`s that match almost nothing on a +/// normal tick, so the only thing a longer period buys is a contract that stays +/// tradeable for most of a day after it stopped existing. Hourly also avoids +/// next-run-at-wall-clock arithmetic — this is a plain interval, and a restart at any +/// hour is equivalent to one at any other. +const SWEEP_INTERVAL: Duration = Duration::from_secs(3600); + +/// Fill in (or correct) `expires_at` from the venue calendar. Returns rows written. +/// +/// `(expiry_date + close_time) AT TIME ZONE timezone` is evaluated by Postgres, which +/// carries the IANA database and applies the offset in force *on that date* — 16:00 +/// New York is 20:00Z in summer and 21:00Z in winter, so a stored offset would be +/// wrong for half the year. Doing it in SQL also keeps the tz database in one place, +/// patched with the server, rather than pinned to a Rust crate version. +/// +/// The `IS DISTINCT FROM` guard makes this idempotent *and* self-healing: a normal +/// run writes nothing, while correcting a calendar's close time or timezone re-derives +/// every contract on that venue at the next tick. `IS DISTINCT FROM` rather than `<>` +/// so a row that has never been computed (NULL) also matches. +/// +/// A contract whose venue has no calendar row, or whose calendar has no `close_time`, +/// is skipped and keeps a NULL `expires_at` — it can never be swept, so preflight +/// reports it rather than this function inventing an hour for it. +pub async fn recompute_expires_at(pool: &PgPool) -> Result { + let n = sqlx::query( + "UPDATE instrument_derivative d \ + SET expires_at = (d.expiry_date + c.close_time) AT TIME ZONE c.timezone \ + FROM instrument i \ + JOIN venue v ON v.code = i.venue \ + JOIN calendar c ON c.venue_id = v.id \ + WHERE d.instrument_id = i.id \ + AND d.expiry_date IS NOT NULL \ + AND c.close_time IS NOT NULL \ + AND d.expires_at IS DISTINCT FROM \ + (d.expiry_date + c.close_time) AT TIME ZONE c.timezone", + ) + .execute(pool) + .await? + .rows_affected(); + + Ok(n) +} + +/// Retire every instrument whose expiry instant has passed. Returns rows retired. +/// +/// Only `ACTIVE` rows are touched, so a contract someone deliberately set to `HALTED` +/// or `INACTIVE` keeps that status — the sweep records that an expiry happened, it +/// does not overwrite a judgement about why something was already not trading. +/// +/// `expires_at IS NOT NULL` is doing real work: a NULL means the instant could not be +/// derived (no calendar), not that the contract is perpetual. Sweeping on a NULL would +/// be sweeping on an unknown. +pub async fn sweep_expired(pool: &PgPool) -> Result { + let n = sqlx::query( + "UPDATE instrument i \ + SET status = 'EXPIRED', updated_at = now() \ + FROM instrument_derivative d \ + WHERE d.instrument_id = i.id \ + AND i.status = 'ACTIVE' \ + AND d.expires_at IS NOT NULL \ + AND d.expires_at < now()", + ) + .execute(pool) + .await? + .rows_affected(); + + Ok(n) +} + +/// One pass: derive instants, then retire what has passed. Returns +/// `(recomputed, expired)`. +/// +/// Order matters — a contract seeded minutes ago has no instant yet, and sweeping +/// before deriving would skip it for a full interval. Shared with the admin trigger so +/// running it by hand and running it on the timer cannot drift apart. +pub async fn run_once(pool: &PgPool) -> Result<(u64, u64), sqlx::Error> { + let recomputed = recompute_expires_at(pool).await?; + let expired = sweep_expired(pool).await?; + Ok((recomputed, expired)) +} + +/// Background task: sweep at boot, then once an interval, forever. +/// +/// A failed pass is logged and the loop continues. The alternative — letting the task +/// die on a transient database error — would silently stop expiring instruments for +/// the lifetime of the process, and the symptom would surface much later as a feed +/// subscribing to a contract that no longer exists. +pub async fn run(pool: PgPool) { + loop { + match run_once(&pool).await { + // Quiet on a no-op tick, which is almost every tick. + Ok((0, 0)) => {} + Ok((recomputed, expired)) => { + info!(recomputed, expired, "expiry sweep"); + } + Err(e) => error!(error = %e, "expiry sweep failed; retrying next interval"), + } + tokio::time::sleep(SWEEP_INTERVAL).await; + } +} diff --git a/src/main.rs b/src/main.rs index 7953ae8..c8f60df 100644 --- a/src/main.rs +++ b/src/main.rs @@ -60,6 +60,7 @@ mod binance_feed; mod bybit_feed; mod feeds; mod preflight; +mod expiry; mod fix; #[derive(OpenApi)] @@ -109,6 +110,7 @@ mod fix; admin::list_feeds, admin::resolve_symbology, admin::backfill_symbology, + admin::expiry_sweep, ), components(schemas( SubmitOrder, SubmitOrderRequest, CancelOrder, OrderSide, OrderType, TimeInForce, OrderAggregateState, @@ -125,6 +127,7 @@ mod fix; admin::RiskLimit, admin::CreateRiskLimit, admin::UpdateRiskLimit, admin::InstrumentSummary, admin::FeedSummary, admin::ResolveRequest, admin::BackfillRequest, admin::BackfillResult, + admin::ExpirySweepResult, crate::symbology_resolver::ResolveOutcome, crate::symbology_resolver::ResolvedIdentity, )), modifiers(&SecurityAddon), @@ -455,6 +458,12 @@ async fn serve() { let (quote_tx, quote_rx) = tokio::sync::mpsc::channel::(1024); tokio::spawn(mark_router::run(quote_rx, state.marks().clone(), state.pool().clone())); + // Retire dated contracts once their expiry instant passes, so the feeds below + // stop resubscribing to them and the order path stops accepting them. Not + // supervised: `stream_supervisor` exists to reconnect streams, and treats a clean + // return as a disconnect to back off from — wrong shape for a periodic job. + tokio::spawn(expiry::run(state.pool().clone())); + if env::var("DATABENTO_API_KEY").map(|k| !k.is_empty()).unwrap_or(false) { let (opra_pos_tx, opra_pos_rx) = tokio::sync::mpsc::channel::<()>(1); marks_doorbells.push(opra_pos_tx); @@ -608,6 +617,7 @@ async fn serve() { .route("/admin/feeds", get(admin::list_feeds)) .route("/admin/symbology/resolve", post(admin::resolve_symbology)) .route("/admin/symbology/backfill", post(admin::backfill_symbology)) + .route("/admin/instruments/expiry-sweep", post(admin::expiry_sweep)) .layer(middleware::from_fn_with_state(state.clone(), auth::admin_middleware)); let scalar_html = { diff --git a/src/preflight.rs b/src/preflight.rs index aa430a3..6888267 100644 --- a/src/preflight.rs +++ b/src/preflight.rs @@ -38,10 +38,74 @@ impl std::fmt::Display for Fatal { /// from fatal to expected. pub async fn run(pool: &PgPool, auto_sync_pending: bool) -> Result<(), Fatal> { check_catalog(pool, auto_sync_pending).await?; + report_expiry(pool).await; report_held(pool).await; Ok(()) } +/// Degraded checks around instrument expiry. Logs; never fails the boot. +/// +/// Both conditions are silent by nature — one is a contract that will never be +/// retired, the other a position in a contract that already was — so a boot-time line +/// is the only thing that makes them visible. +async fn report_expiry(pool: &PgPool) { + // An option whose venue has no calendar row (or whose calendar has no close_time) + // gets no `expires_at`, so `crate::expiry` can never sweep it: it stays ACTIVE + // past expiry, keeps being subscribed, and keeps being orderable. Naming it here + // is the difference between a missing seed row and a mystery in the feed logs. + let undated = sqlx::query_scalar::<_, String>( + "SELECT i.symbol \ + FROM instrument i \ + JOIN instrument_derivative d ON d.instrument_id = i.id \ + WHERE i.status = 'ACTIVE' \ + AND d.expiry_date IS NOT NULL \ + AND d.expires_at IS NULL \ + ORDER BY 1", + ) + .fetch_all(pool) + .await; + match undated { + Ok(rows) if !rows.is_empty() => { + let names: Vec<&str> = rows.iter().map(String::as_str).collect(); + error!( + "preflight: {} dated instrument(s) have no expiry instant — their venue \ + has no calendar (run `make db-seed`), so they will never expire: {}", + names.len(), + sample(&names) + ); + } + Ok(_) => {} + Err(e) => error!("preflight: could not check instrument expiry: {e}"), + } + + // A position left in an expired contract. The OMS is not the book of record and + // must not guess at how it resolved — expired worthless, or exercised into the + // underlying — so this reports and stops. It is also unpriceable and unorderable + // by then, which the checks below would otherwise blame on a feed or a mapping. + let stranded = sqlx::query_scalar::<_, String>( + "SELECT DISTINCT i.symbol \ + FROM position p \ + JOIN instrument i ON i.id::text = p.instrument_id \ + WHERE p.net_qty <> 0 AND i.status = 'EXPIRED' \ + ORDER BY 1", + ) + .fetch_all(pool) + .await; + match stranded { + Ok(rows) if !rows.is_empty() => { + let names: Vec<&str> = rows.iter().map(String::as_str).collect(); + error!( + "preflight: {} held position(s) are in an expired instrument — the custodian \ + resolves these (expiry/assignment), the OMS does not: {}", + names.len(), + sample(&names) + ); + } + Ok(_) => {} + Err(e) => error!("preflight: could not check held expired positions: {e}"), + } +} + /// Fatal checks: the master catalog and the FK targets it depends on. /// /// An empty `venue` or `currency` table is the signature of a DB that never got diff --git a/src/setup/brokers.rs b/src/setup/brokers.rs index a3336c3..75e5620 100644 --- a/src/setup/brokers.rs +++ b/src/setup/brokers.rs @@ -148,8 +148,13 @@ pub async fn run(args: Args) -> Result<(), Box> { }; let (summary, ids) = catalog::upsert_catalog(&pool, &defs, &enrichers).await?; info!( - "master: upserted={} skipped_fk={} skipped_expired={} derivatives={} enriched={}", - summary.upserted, summary.skipped_fk(), summary.skipped_expired, summary.derivatives, summary.enriched + "master: upserted={} skipped_fk={} skipped_expired={} derivatives={} dated={} enriched={}", + summary.upserted, + summary.skipped_fk(), + summary.skipped_expired, + summary.derivatives, + summary.dated, + summary.enriched ); // 2) Attach the broker routing mapping for every instrument that landed. diff --git a/src/setup/catalog.rs b/src/setup/catalog.rs index b84f14b..55ae37b 100644 --- a/src/setup/catalog.rs +++ b/src/setup/catalog.rs @@ -26,6 +26,10 @@ pub struct CatalogSummary { pub skipped_currency: u64, pub skipped_expired: u64, pub derivatives: u64, + /// Contracts whose UTC `expires_at` was derived from the venue calendar. Zero on + /// a re-sync that changed nothing, or when the venue has no calendar row — + /// preflight reports the latter. + pub dated: u64, pub enriched: u64, /// The distinct venue/currency codes that failed the FK check. Distinct rather /// than per-row: a skip count alone says something is wrong but not what, and @@ -132,6 +136,12 @@ pub async fn upsert_catalog( } tx.commit().await?; + // Derive the UTC expiry instant for whatever was just written, rather than + // leaving a freshly seeded chain without one until the hourly sweep catches up. + // Same function the sweep runs, so a contract cannot end up with an instant that + // depends on which path wrote it. + summary.dated = crate::expiry::recompute_expires_at(pool).await?; + if !enrichers.is_empty() { summary.enriched = run_enrichers(pool, enrichers, defs).await?; } From d4c2d2269972c4b590bf35ce66a44f6fec6cdd3e Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 14 Aug 2026 15:21:48 +0200 Subject: [PATCH 5/7] feat(api): name instruments by Symbol@Venue, add grant-scoped order views MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A trading token could not actually trade. Submit takes instrument_id as a stringified BIGINT, and the only symbol->id lookups were /admin/instruments and /admin/symbology/resolve; listing orders meant /admin/orders. Closing either gap with the admin token was not an option — that one shared secret also mints trading tokens and edits risk limits, so a trading script would have been able to issue itself credentials and disable the guardrails meant to constrain it. Nothing new was needed. instrument is uniquely keyed (symbol, venue), and Symbol@Venue is already this codebase's name for instrument identity; this only makes it addressable on the wire. Submit now accepts instrument_id, symbol + venue, or symbol as SYMBOL@VENUE. An unqualified symbol resolves only when unique — 1,510 active tickers are listed under more than one MIC — and the 422 names the candidates rather than guessing a venue to trade on. principal_portfolio_grant already carried can_trade/can_view/can_allocate and get_portfolio_positions already enforced it, so GET /portfolios and GET /orders are that same check widened from one portfolio to the granted set. The blotter scopes by portfolio grant rather than by submitter: it shows a portfolio's activity, which is what a blotter is for, not a personal history. Admin and trading share one query builder behind a scope parameter so the two views cannot drift. Also closes a pre-existing hole: get_order and orders_cancel took no AuthContext, so any valid trading token could read or cancel any order in the system, including another principal's. Both now require a grant on the order's portfolio — can_view to read, can_trade to cancel. A missing order stays 404 and an unentitled one is 403; order ids are UUIDs, so confirming one exists tells an attacker who already guessed it nothing. Verified with three principals holding different grants: scoped listing (5 rows vs 8 vs 17 for admin), 403 on another portfolio's order and cancel, 404 preserved, and all three instrument reference forms resolving to the same order. --- src/handlers.rs | 401 +++++++++++++++++++++++++++++++++++++++++++++--- src/main.rs | 5 + 2 files changed, 381 insertions(+), 25 deletions(-) diff --git a/src/handlers.rs b/src/handlers.rs index 88afd69..3243f38 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -1,6 +1,6 @@ use uuid::Uuid; use chrono::{DateTime, Utc}; -use sqlx::{query_scalar, Postgres, QueryBuilder, Row}; +use sqlx::{query_scalar, PgPool, Postgres, QueryBuilder, Row}; use axum::{ extract::Extension, extract::Path, @@ -95,9 +95,19 @@ pub struct SubmitOrderRequest { /// Optional account UUID; omit to use the portfolio's default account. #[schema(example = json!(null))] pub account_id: Option, - /// Instrument surrogate key (BIGINT as string), not a UUID. - #[schema(example = "42")] - pub instrument_id: String, + /// Instrument surrogate key (BIGINT as string), not a UUID. Omit when naming the + /// instrument by `symbol` instead. + #[schema(example = json!(null))] + pub instrument_id: Option, + /// The instrument's venue-native symbol, as an alternative to `instrument_id`. + /// Accepts the `Symbol@Venue` shorthand (`SPY260918C00770000@OPRA`) or a bare + /// symbol qualified by the `venue` field. + #[schema(example = "SPY260918C00770000@OPRA")] + pub symbol: Option, + /// Venue (MIC) qualifying `symbol`. Omit when `symbol` already carries `@VENUE`, + /// or when the symbol is unique across venues. + #[schema(example = json!(null))] + pub venue: Option, pub side: OrderSide, pub order_type: OrderType, pub time_in_force: TimeInForce, @@ -109,13 +119,16 @@ pub struct SubmitOrderRequest { } impl SubmitOrderRequest { - fn into_command(self, account_id: String) -> SubmitOrder { + /// `instrument_id` is the already-resolved surrogate key: the request may have + /// named the instrument by symbol, so the caller resolves first and the command + /// only ever carries the id. + fn into_command(self, account_id: String, instrument_id: i64) -> SubmitOrder { SubmitOrder { order_id: self.order_id, client_order_id: self.client_order_id, portfolio_id: self.portfolio_id, account_id, - instrument_id: self.instrument_id, + instrument_id: instrument_id.to_string(), side: self.side, order_type: self.order_type, time_in_force: self.time_in_force, @@ -125,6 +138,169 @@ impl SubmitOrderRequest { } } +/// Which entitlement a route demands on the portfolio an order books against. +#[derive(Clone, Copy)] +enum OrderPermission { + View, + Trade, +} + +impl OrderPermission { + /// The grant column. A fixed `&'static str` per variant — never caller input — + /// so interpolating it into SQL cannot carry anything injectable. + fn column(self) -> &'static str { + match self { + Self::View => "can_view", + Self::Trade => "can_trade", + } + } +} + +/// Assert this principal may act on the given order, by way of its portfolio. +/// +/// Orders are reachable by UUID alone, so without this any valid trading token could +/// read or cancel any order in the system, including another principal's. The grant +/// lives on the portfolio, matching `get_portfolio_positions`. +/// +/// A missing order is 404 and an unentitled one is 403 — deliberately distinguished. +/// Collapsing both to 404 would hide typos behind "not found"; order ids are UUIDs, so +/// confirming one exists tells an attacker who already guessed it nothing new. +async fn require_order_grant( + pool: &PgPool, + principal_id: Uuid, + order_id: Uuid, + permission: OrderPermission, +) -> Result<(), ApiError> { + let granted: Option = query_scalar(&format!( + "SELECT EXISTS (SELECT 1 FROM principal_portfolio_grant g \ + WHERE g.portfolio_id = os.portfolio_id AND g.principal_id = $2 \ + AND g.{} = true) \ + FROM order_state os WHERE os.order_id = $1", + permission.column() + )) + .bind(order_id) + .bind(principal_id) + .fetch_optional(pool) + .await + .map_err(|err| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("failed to check grant: {:?}", err), + })?; + + match granted { + Some(true) => Ok(()), + Some(false) => Err(ApiError { + status: StatusCode::FORBIDDEN, + message: "unauthorized".to_string(), + }), + None => Err(ApiError { + status: StatusCode::NOT_FOUND, + message: "order not found".to_string(), + }), + } +} + +/// Split the `Symbol@Venue` shorthand into its parts. `None` when there is no `@`. +/// +/// `Symbol@Venue` is already this codebase's name for instrument identity (see +/// `adapters::BrokerInstrument` and migration 0017); this only makes it addressable +/// on the wire. Splits on the last `@` so a symbol containing one is still reachable +/// by qualifying it — no venue code contains `@`. +fn split_symbol_at_venue(symbol: &str) -> Option<(&str, &str)> { + let (sym, venue) = symbol.rsplit_once('@')?; + (!sym.is_empty() && !venue.is_empty()).then_some((sym, venue)) +} + +/// Resolve whichever instrument reference the request carried into the surrogate key. +/// +/// Three accepted forms — `instrument_id`, `symbol` + `venue`, or `symbol` as +/// `Symbol@Venue` — because `instrument` is uniquely keyed on `(symbol, venue)`, so a +/// symbol plus its venue is as precise as the id and far easier to write by hand. +/// +/// Only ACTIVE rows resolve, matching `symbology_resolver`. That means an expired +/// contract fails here with `instrument not found` rather than reaching the +/// `instrument not active` check below — the same refusal, one step earlier. +async fn resolve_instrument_ref( + pool: &PgPool, + req: &SubmitOrderRequest, +) -> Result { + let bad_request = |message: &str| ApiError { + status: StatusCode::BAD_REQUEST, + message: message.to_string(), + }; + + // The id wins when given: it is unambiguous, and honouring it keeps every existing + // caller working unchanged. + if let Some(id) = req.instrument_id.as_deref().filter(|s| !s.is_empty()) { + return id.parse().map_err(|_| bad_request("instrument_id must be a BIGINT")); + } + + let Some(symbol) = req.symbol.as_deref().filter(|s| !s.is_empty()) else { + return Err(bad_request("one of instrument_id or symbol is required")); + }; + + // A venue in both places is a contradiction, not something to silently pick from — + // resolving one over the other would trade on a venue the caller did not name. + let (symbol, venue) = match (split_symbol_at_venue(symbol), req.venue.as_deref()) { + (Some(_), Some(v)) if !v.is_empty() => { + return Err(bad_request( + "venue given both in symbol (SYMBOL@VENUE) and in the venue field", + )) + } + (Some((s, v)), _) => (s, Some(v)), + (None, v) => (symbol, v.filter(|v| !v.is_empty())), + }; + + if let Some(venue) = venue { + return query_scalar::<_, i64>( + "SELECT id FROM instrument WHERE symbol = $1 AND venue = $2 AND status = 'ACTIVE'", + ) + .bind(symbol) + .bind(venue) + .fetch_optional(pool) + .await + .map_err(|err| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("failed to resolve instrument: {:?}", err), + })? + .ok_or_else(|| ApiError { + status: StatusCode::UNPROCESSABLE_ENTITY, + message: "instrument not found".to_string(), + }); + } + + // Unqualified: usable only when the symbol names exactly one instrument. Options + // always do (one venue, OPRA); equities often do not, because the same ticker is + // listed under several exchange labels and each maps to a distinct MIC. + let rows = sqlx::query_as::<_, (i64, String)>( + "SELECT id, venue FROM instrument WHERE symbol = $1 AND status = 'ACTIVE' ORDER BY venue", + ) + .bind(symbol) + .fetch_all(pool) + .await + .map_err(|err| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("failed to resolve instrument: {:?}", err), + })?; + + match rows.as_slice() { + [(id, _)] => Ok(*id), + [] => Err(ApiError { + status: StatusCode::UNPROCESSABLE_ENTITY, + message: "instrument not found".to_string(), + }), + // Name the candidates: the caller cannot guess which venues exist, and the fix + // is mechanical once they can see them. + many => Err(ApiError { + status: StatusCode::UNPROCESSABLE_ENTITY, + message: format!( + "symbol {symbol} is ambiguous across venues: [{}] — qualify it as {symbol}@VENUE", + many.iter().map(|(_, v)| v.as_str()).collect::>().join(", ") + ), + }), + } +} + #[utoipa::path( post, path = "/orders/submit", tag = "orders", request_body = SubmitOrderRequest, @@ -179,13 +355,12 @@ pub async fn orders_submit( })?, }; // instrument.id is a BIGINT surrogate key (the mdm master instrument), not a UUID. - let instrument_id_bigint: i64 = req.instrument_id.parse().map_err(|_| ApiError { - status: StatusCode::BAD_REQUEST, - message: "instrument_id must be a BIGINT".to_string(), - })?; + // The request may have named the instrument by symbol instead; either way the + // command below carries only the id. + let instrument_id_bigint = resolve_instrument_ref(&pool, &req).await?; // Boundary → domain: fold the resolved account into the pure SubmitOrder command. - let cmd = req.into_command(account_id.to_string()); + let cmd = req.into_command(account_id.to_string(), instrument_id_bigint); // Pre-flight: resolve the account's routing coordinates from its broker_connection // (broker_code, environment) + the custodial ref, so we can validate the instrument @@ -641,9 +816,10 @@ pub async fn orders_submit( )] pub async fn orders_cancel( State(app_state): State, + Extension(auth): Extension, Json(req): Json ) -> Result{ - info!(?req, "cancel order received"); + info!(?req, principal_id = %auth.principal_id, "cancel order received"); let pool = app_state.pool().clone(); let event_store = OrderEventStore::new(pool.clone()); @@ -653,6 +829,9 @@ pub async fn orders_cancel( message: "order_id must be a UUID".to_string(), })?; + // Cancelling is acting on the position, so it takes can_trade rather than can_view. + require_order_grant(&pool, auth.principal_id, order_id, OrderPermission::Trade).await?; + // start transaction let mut tx = pool.begin().await.map_err(|err| ApiError { status: StatusCode::INTERNAL_SERVER_ERROR, @@ -905,6 +1084,7 @@ pub async fn orders_cancel( )] pub async fn get_order( State(state): State, + Extension(auth): Extension, Path(order_id_str): Path, ) -> Result, ApiError> { let order_id = Uuid::parse_str(&order_id_str).map_err(|_| ApiError { @@ -912,6 +1092,8 @@ pub async fn get_order( message: "id must be a UUID".to_string(), })?; + require_order_grant(state.pool(), auth.principal_id, order_id, OrderPermission::View).await?; + let row = sqlx::query( r#" SELECT @@ -964,6 +1146,69 @@ pub async fn get_order( Ok(Json(order)) } +/// One portfolio the caller is entitled to, with the entitlement itself. +/// +/// The flags are returned rather than filtered on, so a client can grey out what it +/// cannot do instead of discovering it from a 403 mid-order. +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct GrantedPortfolio { + pub portfolio_id: String, + pub code: String, + pub name: String, + pub status: String, + pub base_currency: Option, + pub can_trade: bool, + pub can_view: bool, + pub can_allocate: bool, +} + +/// The portfolios this principal may act on. +/// +/// Exists so a trading token is self-sufficient: every other trading route needs a +/// portfolio UUID, and without this the caller has to be handed one out of band (or +/// read the admin surface, which is exactly the authority a trading token must not +/// have). Scope is the principal's own grants — there is no filter to widen it. +#[utoipa::path( + get, path = "/portfolios", tag = "orders", + responses((status = 200, description = "OK", body = [GrantedPortfolio])), + security(("basic_auth" = []), ("bearer_token" = [])) +)] +pub async fn list_portfolios( + State(state): State, + Extension(auth): Extension, +) -> Result>, ApiError> { + let rows = sqlx::query( + "SELECT p.id, p.code, p.name, p.status, p.base_currency, \ + g.can_trade, g.can_view, g.can_allocate \ + FROM principal_portfolio_grant g \ + JOIN portfolio p ON p.id = g.portfolio_id \ + WHERE g.principal_id = $1 \ + ORDER BY p.code", + ) + .bind(auth.principal_id) + .fetch_all(state.pool()) + .await + .map_err(|err| ApiError { + status: StatusCode::INTERNAL_SERVER_ERROR, + message: format!("failed to list portfolios: {:?}", err), + })?; + + Ok(Json( + rows.into_iter() + .map(|r| GrantedPortfolio { + portfolio_id: r.get::("id").to_string(), + code: r.get("code"), + name: r.get("name"), + status: r.get("status"), + base_currency: r.get("base_currency"), + can_trade: r.get("can_trade"), + can_view: r.get("can_view"), + can_allocate: r.get("can_allocate"), + }) + .collect(), + )) +} + #[utoipa::path( get, path = "/portfolios/{id}/positions", tag = "orders", params(("id" = Uuid, Path, description = "Portfolio ID")), @@ -1294,16 +1539,24 @@ pub struct BlotterRow { pub updated_at: DateTime, } -#[utoipa::path( - get, path = "/admin/orders", tag = "admin", - params(BlotterFilter), - responses((status = 200, description = "OK", body = [BlotterRow])), - security(("bearer_token" = [])) -)] -pub async fn get_orders_blotter( - State(state): State, - Query(f): Query, -) -> Result>, ApiError> { +/// Who a blotter query is allowed to see. +/// +/// The admin blotter sees everything and may filter by principal; a trading token +/// sees only what its principal is granted. Making that a parameter of one query — +/// rather than two similar queries — is what keeps the two views from drifting into +/// disagreeing about the same order. +enum BlotterScope { + /// Oversight: every order, `principal_id` usable as a filter. + All, + /// This principal's entitled portfolios only. Not a filter — a ceiling. + GrantedTo(Uuid), +} + +async fn load_blotter( + state: &AppState, + f: &BlotterFilter, + scope: BlotterScope, +) -> Result, ApiError> { let mut qb = QueryBuilder::::new( "SELECT os.order_id, os.principal_id, p.code AS principal_code, \ os.portfolio_id, pf.code AS portfolio_code, os.account_id, \ @@ -1321,10 +1574,27 @@ pub async fn get_orders_blotter( LEFT JOIN instrument i ON i.id::text = os.instrument_id \ WHERE TRUE", ); + // Scope first, so it can never be widened by a filter appended after it. + match scope { + BlotterScope::All => { + if let Some(v) = f.principal_id { qb.push(" AND os.principal_id = ").push_bind(v); } + } + // EXISTS rather than a join: an order must sit in a portfolio this principal + // may view, and a join would duplicate rows if the grant model ever allows + // more than one matching grant per portfolio. + BlotterScope::GrantedTo(principal_id) => { + qb.push( + " AND EXISTS (SELECT 1 FROM principal_portfolio_grant g \ + WHERE g.portfolio_id = os.portfolio_id AND g.can_view = true \ + AND g.principal_id = ", + ) + .push_bind(principal_id) + .push(")"); + } + } if let Some(v) = &f.status { qb.push(" AND os.status = ").push_bind(v.clone()); } if let Some(v) = f.portfolio_id { qb.push(" AND os.portfolio_id = ").push_bind(v); } if let Some(v) = &f.instrument_id { qb.push(" AND os.instrument_id = ").push_bind(v.clone()); } - if let Some(v) = f.principal_id { qb.push(" AND os.principal_id = ").push_bind(v); } if let Some(v) = &f.broker_connection_code { qb.push(" AND os.broker_connection_code = ").push_bind(v.clone()); } @@ -1365,7 +1635,40 @@ pub async fn get_orders_blotter( updated_at: r.get("updated_at"), }) .collect(); - Ok(Json(out)) + Ok(out) +} + +#[utoipa::path( + get, path = "/admin/orders", tag = "admin", + params(BlotterFilter), + responses((status = 200, description = "OK", body = [BlotterRow])), + security(("bearer_token" = [])) +)] +pub async fn get_orders_blotter( + State(state): State, + Query(f): Query, +) -> Result>, ApiError> { + load_blotter(&state, &f, BlotterScope::All).await.map(Json) +} + +/// The caller's own blotter. +/// +/// Same rows and same filters as the admin blotter, bounded to the portfolios this +/// principal may view. `principal_id` in the query string is ignored rather than +/// honoured — the authenticated identity is the only thing that decides scope, so +/// passing someone else's id does nothing. +#[utoipa::path( + get, path = "/orders", tag = "orders", + params(BlotterFilter), + responses((status = 200, description = "OK", body = [BlotterRow])), + security(("basic_auth" = []), ("bearer_token" = [])) +)] +pub async fn list_orders( + State(state): State, + Extension(auth): Extension, + Query(f): Query, +) -> Result>, ApiError> { + load_blotter(&state, &f, BlotterScope::GrantedTo(auth.principal_id)).await.map(Json) } // Function to issue an api error @@ -1447,12 +1750,60 @@ fn domain_event_to_new_event(event: &OrderDomainEvent) -> Result why was command sent schema_version: 0, // indicate the schema version of the payload }) } +#[cfg(test)] +mod tests { + use super::*; + + /// The case the shorthand exists for: an OSI symbol qualified by its venue. + #[test] + fn splits_symbol_at_venue() { + assert_eq!( + split_symbol_at_venue("SPY260918C00770000@OPRA"), + Some(("SPY260918C00770000", "OPRA")) + ); + assert_eq!(split_symbol_at_venue("AAPL@XNAS"), Some(("AAPL", "XNAS"))); + } + + /// No `@` means the whole string is the symbol — the caller either qualified it + /// with the `venue` field or is relying on it being unique. + #[test] + fn unqualified_symbol_does_not_split() { + assert_eq!(split_symbol_at_venue("SPY260918C00770000"), None); + assert_eq!(split_symbol_at_venue("BTCUSDT"), None); + } + + /// Splitting on the *last* `@` keeps a symbol that itself contains one reachable: + /// no venue code contains `@`, so the trailing segment is always the venue. + #[test] + fn splits_on_the_last_at() { + assert_eq!(split_symbol_at_venue("WEIRD@SYM@XNAS"), Some(("WEIRD@SYM", "XNAS"))); + } + + /// A dangling `@` names neither a symbol nor a venue. Declining here means the + /// caller gets "one of instrument_id or symbol is required" or a not-found, rather + /// than a lookup on an empty string. + #[test] + fn declines_half_empty_forms() { + assert_eq!(split_symbol_at_venue("@OPRA"), None); + assert_eq!(split_symbol_at_venue("SPY@"), None); + assert_eq!(split_symbol_at_venue("@"), None); + } + + /// The grant column is chosen by the route, never by the caller — these two fixed + /// strings are the only things ever interpolated into the grant query. + #[test] + fn order_permission_maps_to_grant_column() { + assert_eq!(OrderPermission::View.column(), "can_view"); + assert_eq!(OrderPermission::Trade.column(), "can_trade"); + } +} + diff --git a/src/main.rs b/src/main.rs index c8f60df..472a6ca 100644 --- a/src/main.rs +++ b/src/main.rs @@ -71,6 +71,8 @@ mod fix; handlers::orders_submit, handlers::orders_cancel, handlers::get_order, + handlers::list_orders, + handlers::list_portfolios, handlers::get_portfolio_positions, handlers::create_allocations, handlers::list_allocations, @@ -116,6 +118,7 @@ mod fix; SubmitOrder, SubmitOrderRequest, CancelOrder, OrderSide, OrderType, TimeInForce, OrderAggregateState, crate::positions::Position, Allocation, CreateAllocations, AllocationSplit, BlotterRow, + handlers::GrantedPortfolio, Principal, Portfolio, Account, BrokerConnection, CreatePrincipal, UpdatePrincipal, CreatePortfolio, UpdatePortfolio, @@ -536,6 +539,8 @@ async fn serve() { .route("/orders/submit", post(handlers::orders_submit)) .route("/orders/cancel", post(handlers::orders_cancel)) .route("/orders/:id", get(handlers::get_order)) + .route("/orders", get(handlers::list_orders)) + .route("/portfolios", get(handlers::list_portfolios)) .route("/portfolios/:id/positions", get(handlers::get_portfolio_positions)) .route( "/orders/:id/allocations", From 68c06a68aa5b893cfabfed4fe7091d6e17ceda87 Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 14 Aug 2026 15:22:07 +0200 Subject: [PATCH 6/7] chore(catalog): script to drop options orphaned by the retired seeder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before broker-first seeding (0018), scripts/seed_instruments.py built the catalog from Databento definition files. Those rows carry Databento's strict 21-char OSI and — because that model had no notion of one — no broker_instrument routing handle, so they cannot be ordered or closed. They still match the OPRA feed's candidates() filter, so the quote feed kept subscribing to contracts the OMS could never trade. Not a migration, for two reasons: a catalog seeded fresh has none of these, so it would be a no-op; and the guard reads oms.position and oms.order_state, which the migration runner cannot see — it applies public-schema files as mdm_master, which has no access to the oms schema. Rows still referenced by a position or an order are kept. History has to stay resolvable, and a stranded position is something preflight reports rather than something a cleanup silently erases. Dry-run then applied here: 27,250 deleted, 2 kept. --- db/scripts/cleanup_orphaned_options.sql | 66 +++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 db/scripts/cleanup_orphaned_options.sql diff --git a/db/scripts/cleanup_orphaned_options.sql b/db/scripts/cleanup_orphaned_options.sql new file mode 100644 index 0000000..f1029a6 --- /dev/null +++ b/db/scripts/cleanup_orphaned_options.sql @@ -0,0 +1,66 @@ +-- One-off cleanup: remove option rows left behind by the retired Databento-dataset +-- seeding model. +-- +-- Before broker-first seeding (migration 0018), `scripts/seed_instruments.py` built +-- the catalog from Databento definition files. Those rows carry Databento's strict +-- 21-char OSI symbol (root space-padded to 6) and — because that model had no notion +-- of one — no `broker_instrument` routing handle at all. They cannot be ordered and +-- cannot be closed, but they still match the OPRA feed's `candidates()` filter, so +-- the quote feed keeps subscribing to contracts the OMS could never trade. +-- +-- Broker-first replaced this: an adapter's InstrumentProvider writes the master row +-- and the routing mapping in one pass (src/setup/brokers.rs), so "no broker mapping" +-- is now precisely the signature of a row from the old model. A catalog seeded fresh +-- has none of these, which is why this is a maintenance script rather than a +-- migration — on a clean database it is a no-op. +-- +-- Not a migration for a second reason: the guard below must read `oms.position` and +-- `oms.order_state`, and the migration runner executes public-schema files as +-- mdm_master, which has no access to the oms schema. Run this as the cluster admin +-- (ADMIN_USER), which can see both. +-- +-- Rows referenced by a position or an order are kept, not deleted. History must stay +-- resolvable — an order that names an instrument id has to be able to resolve it, and +-- a stranded position is something preflight reports rather than something a cleanup +-- silently erases. Those survivors keep whatever status they have; the expiry sweep +-- (src/expiry.rs) will already have marked the dated ones EXPIRED. +-- +-- instrument_derivative rows go with them via ON DELETE CASCADE. +-- +-- Idempotent: re-running deletes nothing once the orphans are gone. +-- +-- psql "$ADMIN_URL" -f db/scripts/cleanup_orphaned_options.sql + +\set ON_ERROR_STOP on + +BEGIN; + +CREATE TEMP TABLE orphaned_options ON COMMIT DROP AS +SELECT i.id, i.symbol +FROM public.instrument i +LEFT JOIN public.broker_instrument bi ON bi.instrument_id = i.id +WHERE i.venue = 'OPRA' + AND i.instrument_class = 'OPTION' + AND bi.instrument_id IS NULL + -- Keep anything the operational schema still points at. + AND NOT EXISTS (SELECT 1 FROM oms.position p WHERE p.instrument_id = i.id::text) + AND NOT EXISTS (SELECT 1 FROM oms.order_state s WHERE s.instrument_id = i.id::text) + AND NOT EXISTS (SELECT 1 FROM oms.allocation a WHERE a.instrument_id = i.id::text) + AND NOT EXISTS (SELECT 1 FROM oms.risk_limits r WHERE r.instrument_id = i.id::text) + -- And anything still serving as another instrument's underlying. + AND NOT EXISTS (SELECT 1 FROM public.instrument_derivative d WHERE d.underlying_id = i.id); + +SELECT count(*) AS to_delete FROM orphaned_options; + +DELETE FROM public.instrument i +USING orphaned_options o +WHERE i.id = o.id; + +-- What survived, and why it was kept. +SELECT i.symbol, i.status +FROM public.instrument i +LEFT JOIN public.broker_instrument bi ON bi.instrument_id = i.id +WHERE i.venue = 'OPRA' AND i.instrument_class = 'OPTION' AND bi.instrument_id IS NULL +ORDER BY 1; + +COMMIT; From 6ae6f4585bc807d49e621e89b01568042aee62db Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 14 Aug 2026 15:22:07 +0200 Subject: [PATCH 7/7] feat(sdk): python client and CLI for the trading API MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lets the OMS be driven from a script or a REPL with one trading token, so manual use does not have to wait for a UI. Deliberately has no admin surface: the admin token covers principal management, broker connections and risk limits, and none of that belongs in something that sends orders. Three properties of the wire protocol shape the client, and each is the kind of thing that is easy to get wrong and expensive when wrong: * Submit answers 204 with an empty body, so the client generates order_id and that id is the idempotency key. A 409 on retry therefore means the order was already accepted, and is swallowed rather than raised — raising would push callers toward resubmitting under a new id, which is how you end up with the position twice. A 502 is not swallowed: the OMS kept the order and only the broker leg failed. * Cancel answers 202 when the order is live at a broker and the cancel is confirmed asynchronously; only 204 means done. * Every error is text/plain, so nothing calls .json() on a failure. wait_for polls because the OMS has no push channel for order events. Includes an argparse CLI (oms orders list/get/cancel, positions, portfolios, submit) — the cheapest possible blotter — and a smoke-test example that sends a resting limit buy and cancels it. The order rests 20% below the market on purpose: a market order fills instantly and would leave nothing to cancel. 22 tests against a stubbed transport, plus a live run against Binance testnet: submitted, routed, cancel returned pending, polled to canceled. --- clients/python/README.md | 90 +++++++ clients/python/examples/smoke_test.py | 185 +++++++++++++ clients/python/oms_client.egg-info/PKG-INFO | 10 + .../python/oms_client.egg-info/SOURCES.txt | 14 + .../oms_client.egg-info/dependency_links.txt | 1 + .../oms_client.egg-info/entry_points.txt | 2 + .../python/oms_client.egg-info/requires.txt | 7 + .../python/oms_client.egg-info/top_level.txt | 1 + clients/python/oms_client/__init__.py | 46 ++++ clients/python/oms_client/cli.py | 226 ++++++++++++++++ clients/python/oms_client/client.py | 251 ++++++++++++++++++ clients/python/oms_client/errors.py | 92 +++++++ clients/python/oms_client/models.py | 178 +++++++++++++ clients/python/pyproject.toml | 22 ++ clients/python/tests/test_client.py | 246 +++++++++++++++++ 15 files changed, 1371 insertions(+) create mode 100644 clients/python/README.md create mode 100644 clients/python/examples/smoke_test.py create mode 100644 clients/python/oms_client.egg-info/PKG-INFO create mode 100644 clients/python/oms_client.egg-info/SOURCES.txt create mode 100644 clients/python/oms_client.egg-info/dependency_links.txt create mode 100644 clients/python/oms_client.egg-info/entry_points.txt create mode 100644 clients/python/oms_client.egg-info/requires.txt create mode 100644 clients/python/oms_client.egg-info/top_level.txt create mode 100644 clients/python/oms_client/__init__.py create mode 100644 clients/python/oms_client/cli.py create mode 100644 clients/python/oms_client/client.py create mode 100644 clients/python/oms_client/errors.py create mode 100644 clients/python/oms_client/models.py create mode 100644 clients/python/pyproject.toml create mode 100644 clients/python/tests/test_client.py diff --git a/clients/python/README.md b/clients/python/README.md new file mode 100644 index 0000000..2600485 --- /dev/null +++ b/clients/python/README.md @@ -0,0 +1,90 @@ +# oms-client + +Python client for the OMS trading API. Takes a trading token and nothing else — it +never touches the admin surface. + +```bash +pip install -e clients/python # add [pandas] for .to_pandas() +``` + +## Getting a token + +Trading tokens are minted by an admin, once, and shown once: + +```bash +curl -X POST "$OMS_URL/admin/trading-tokens" \ + -H "Authorization: Bearer $OMS_ADMIN_PASSWORD" \ + -H 'Content-Type: application/json' \ + -d '{"principal_id":"","portfolio_id":"","label":"my-laptop"}' +``` + +The `token` field of the response is what this client wants. It carries the +principal's grants (`can_trade` / `can_view` / `can_allocate`) on the portfolios it +was granted, and can do nothing else. + +## Library + +```python +from oms_client import OMS + +oms = OMS("http://localhost:3001", token=os.environ["OMS_TRADING_TOKEN"]) + +pf = oms.portfolios()[0] # what am I allowed to trade? +oid = oms.submit(portfolio=pf.portfolio_id, + symbol="SPY260918C00770000@OPRA", + side="buy", quantity=1) +order = oms.wait_for(oid) # polls until terminal +print(order.status, order.avg_px) + +for row in oms.orders(status="routed"): # the blotter + print(row.order_id, row.instrument_symbol, row.cum_qty) + +oms.orders().to_pandas() # needs the [pandas] extra +``` + +Naming an instrument works three ways: `symbol="SPY260918C00770000@OPRA"`, +`symbol=... , venue=...`, or `instrument_id="2972271"`. A bare symbol resolves only +when it is unique across venues — many equity tickers are listed under several +exchange MICs, and the server answers 422 naming the candidates. + +## CLI + +```bash +export OMS_URL=http://localhost:3001 +export OMS_TRADING_TOKEN=ak_....sk_.... + +oms portfolios +oms orders list --status routed --limit 20 +oms orders get +oms orders cancel +oms submit --portfolio --symbol SPY260918C00770000@OPRA --side buy --qty 1 --wait +oms positions +``` + +Add `--json` to any command to get raw JSON for piping into `jq`. + +## Behaviour worth knowing + +These follow from the server's contract, not from choices made here: + +- **`submit` returns only an order id.** The API answers 204 with an empty body, so + the client generates the id. Call `order()` or `wait_for()` to see the outcome. +- **Retrying a submit is safe.** `order_id` is the idempotency key, so a repeat is a + 409 — which this client treats as "already accepted" rather than an error. Never + resubmit under a *new* id after a failure; a 502 means the OMS kept the order and + only the broker leg failed. +- **`cancel` can be asynchronous.** It returns `"canceled"` when the OMS cancelled it + outright, or `"pending"` when the request went to the broker and will be confirmed + later. `"pending"` does not mean cancelled. +- **`wait_for` polls.** The OMS has no SSE or WebSocket channel for order events. +- **Errors are plain text**, mapped to typed exceptions (`Rejected`, `Forbidden`, + `BrokerRejected`, …), all subclasses of `OMSError`, each carrying `.status` and + `.message`. + +## Tests + +```bash +pip install -e "clients/python[dev]" && pytest clients/python +``` + +No server needed — the transport is stubbed. diff --git a/clients/python/examples/smoke_test.py b/clients/python/examples/smoke_test.py new file mode 100644 index 0000000..1d18cfe --- /dev/null +++ b/clients/python/examples/smoke_test.py @@ -0,0 +1,185 @@ +#!/usr/bin/env python3 +"""End-to-end walkthrough of the OMS Python client against a running OMS. + +Exercises every method on the client, then sends one real order and cancels it. + +The order is a **limit buy priced well below the market**, on purpose. A market +order fills immediately and cannot be cancelled, which would make the cancel half of +this script untestable; a resting bid sits in the book until we take it back. Crypto +is the default target because it trades 24/7 — the Alpaca options path only accepts +market orders during US market hours, so it cannot be exercised at 4am. + +Orders route to Binance **testnet** when the broker connection's environment is +PAPER, so nothing here touches real money. Note the OMS deliberately marks crypto +against *production* prices even though it routes to testnet, so a mark and a fill +price can legitimately disagree. + +Usage: + export OMS_URL=http://localhost:3001 + export OMS_TRADING_TOKEN=ak_....sk_.... + + python examples/smoke_test.py # BTCUSDT@BINANCE, ~12 USDT bid + python examples/smoke_test.py --dry-run # everything except submit/cancel + python examples/smoke_test.py --symbol ETHUSDT@BINANCE --notional 20 + python examples/smoke_test.py --portfolio # skip auto-selection +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import urllib.request + +from oms_client import OMS, OMSError, Rejected + +# Where to ask what a coin is worth. Testnet, not production: the order rests in +# testnet's book, so testnet's price is the one that decides whether it rests. +TESTNET_TICKER = "https://testnet.binance.vision/api/v3/ticker/price?symbol={}" + +# How far below the market to place the bid. Binance's PERCENT_PRICE_BY_SIDE filter +# rejects a bid below half the average price, so this stays comfortably inside that +# while being far enough out that nothing crosses it during the run. +DISCOUNT = 0.80 + + +def step(n: int, title: str) -> None: + print(f"\n\033[1m[{n}] {title}\033[0m") + + +def reference_price(symbol_at_venue: str) -> float: + """Current testnet price for the bare symbol part of SYMBOL@VENUE.""" + symbol = symbol_at_venue.split("@")[0] + with urllib.request.urlopen(TESTNET_TICKER.format(symbol), timeout=10) as resp: + return float(json.load(resp)["price"]) + + +def pick_portfolio(oms: OMS, explicit: str | None): + portfolios = oms.portfolios() + if not portfolios: + sys.exit("error: this token has no portfolio grants — ask an admin for one") + for p in portfolios: + print(f" {p.code:24} trade={p.can_trade} view={p.can_view} alloc={p.can_allocate}") + if explicit: + match = [p for p in portfolios if p.portfolio_id == explicit or p.code == explicit] + if not match: + sys.exit(f"error: {explicit} is not a portfolio this token can see") + return match[0] + tradeable = [p for p in portfolios if p.can_trade] + if not tradeable: + sys.exit("error: this token can view portfolios but not trade any (needs can_trade)") + # Prefer a crypto-looking portfolio, since the default symbol is crypto. + return next((p for p in tradeable if "crypto" in p.code.lower()), tradeable[0]) + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + ap.add_argument("--url", default=os.environ.get("OMS_URL", "http://localhost:3001")) + ap.add_argument("--symbol", default="BTCUSDT@BINANCE", help="SYMBOL@VENUE") + ap.add_argument("--portfolio", help="portfolio id or code (default: first tradeable)") + ap.add_argument("--notional", type=float, default=12.0, help="order size in quote ccy") + ap.add_argument("--price", type=float, help="explicit limit price (skips the lookup)") + ap.add_argument("--dry-run", action="store_true", help="read-only; no order is sent") + args = ap.parse_args() + + token = os.environ.get("OMS_TRADING_TOKEN") + if not token: + sys.exit("error: set OMS_TRADING_TOKEN (mint one via POST /admin/trading-tokens)") + + oms = OMS(args.url, token) + + step(1, "Portfolios this token may act on") + portfolio = pick_portfolio(oms, args.portfolio) + print(f" -> using {portfolio.code} ({portfolio.portfolio_id})") + + step(2, "Current positions") + positions = oms.positions(portfolio.portfolio_id) + if not positions: + print(" (none)") + for p in positions: + mark = "-" if p.mark is None else f"{p.mark:,.2f}" + print(f" instrument {p.instrument_id:>10} qty {p.net_qty:<12g} mark {mark}") + + step(3, "Blotter — most recent orders in these portfolios") + recent = oms.orders(limit=5) + if not recent: + print(" (none)") + for r in recent: + print(f" {r.order_id[:8]} {str(r.instrument_symbol):22} {r.side:4} " + f"{r.status:16} {r.cum_qty:g}/{r.original_qty:g}") + + step(4, "Error handling — an ambiguous symbol must be refused, legibly") + try: + oms.submit(portfolio=portfolio.portfolio_id, symbol="AAAU", side="buy", quantity=1) + print(" !! expected a rejection and did not get one") + except Rejected as err: + print(f" -> Rejected: {err.message}") + except OMSError as err: + # A token without an Alpaca-backed account cannot reach the ambiguity check; + # that is fine, the point is that it refused rather than guessed a venue. + print(f" -> refused ({err.status}): {err.message}") + + # ── the live half ──────────────────────────────────────────────────────── + + price = args.price or round(reference_price(args.symbol) * DISCOUNT, 2) + qty = round(args.notional / price, 5) + step(5, f"Submit a resting limit buy: {qty:g} {args.symbol} @ {price:,.2f}") + print(f" (market is ~{price / DISCOUNT:,.2f}; this bid sits {(1 - DISCOUNT) * 100:.0f}% " + f"below it so it rests instead of filling)") + if args.dry_run: + print(" --dry-run: stopping before anything is sent") + return 0 + + try: + order_id = oms.submit( + portfolio=portfolio.portfolio_id, + symbol=args.symbol, + side="buy", + quantity=qty, + order_type="limit", + limit_price=price, + time_in_force="gtc", + client_order_id="sdk-smoke-test", + ) + except OMSError as err: + sys.exit(f"error: submit refused ({err.status}): {err.message}") + print(f" -> order_id {order_id}") + + step(6, "Read it back") + order = oms.order(order_id) + print(f" status={order.status} leaves={order.leaves_qty:g} " + f"cum={order.cum_qty:g} version={order.version}") + + step(7, "Idempotency — resubmitting the same order_id is a no-op, not a duplicate") + again = oms.submit( + portfolio=portfolio.portfolio_id, symbol=args.symbol, side="buy", quantity=qty, + order_type="limit", limit_price=price, time_in_force="gtc", order_id=order_id, + ) + print(f" -> returned the same id, no exception: {again == order_id}") + + step(8, "Confirm it shows on the blotter") + mine = [r for r in oms.orders(limit=20) if r.order_id == order_id] + print(f" -> found: {bool(mine)}" + (f" status={mine[0].status}" if mine else "")) + + step(9, "Cancel it") + outcome = oms.cancel(order_id, reason="smoke test") + print(f" -> {outcome}" + (" (forwarded to broker; confirmed asynchronously)" + if outcome == "pending" else " (cancelled locally)")) + + step(10, "Wait for a terminal state") + try: + final = oms.wait_for(order_id, timeout=30, interval=1.0) + print(f" -> {final.status} (filled {final.cum_qty:g} of {final.original_qty:g})") + if final.status != "canceled": + print(f" !! expected 'canceled' — the resting bid may have been crossed") + except TimeoutError as err: + print(f" !! {err}") + print(" the cancel is still in flight; re-run `oms orders get` to follow it") + + print("\n\033[1mdone\033[0m — every client method exercised against a live OMS") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/clients/python/oms_client.egg-info/PKG-INFO b/clients/python/oms_client.egg-info/PKG-INFO new file mode 100644 index 0000000..a58427c --- /dev/null +++ b/clients/python/oms_client.egg-info/PKG-INFO @@ -0,0 +1,10 @@ +Metadata-Version: 2.4 +Name: oms-client +Version: 0.1.0 +Summary: Python client for the OMS trading API +Requires-Python: >=3.9 +Requires-Dist: requests>=2.31.0 +Provides-Extra: pandas +Requires-Dist: pandas>=2.0; extra == "pandas" +Provides-Extra: dev +Requires-Dist: pytest>=8.0; extra == "dev" diff --git a/clients/python/oms_client.egg-info/SOURCES.txt b/clients/python/oms_client.egg-info/SOURCES.txt new file mode 100644 index 0000000..eb0ab19 --- /dev/null +++ b/clients/python/oms_client.egg-info/SOURCES.txt @@ -0,0 +1,14 @@ +README.md +pyproject.toml +oms_client/__init__.py +oms_client/cli.py +oms_client/client.py +oms_client/errors.py +oms_client/models.py +oms_client.egg-info/PKG-INFO +oms_client.egg-info/SOURCES.txt +oms_client.egg-info/dependency_links.txt +oms_client.egg-info/entry_points.txt +oms_client.egg-info/requires.txt +oms_client.egg-info/top_level.txt +tests/test_client.py \ No newline at end of file diff --git a/clients/python/oms_client.egg-info/dependency_links.txt b/clients/python/oms_client.egg-info/dependency_links.txt new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/clients/python/oms_client.egg-info/dependency_links.txt @@ -0,0 +1 @@ + diff --git a/clients/python/oms_client.egg-info/entry_points.txt b/clients/python/oms_client.egg-info/entry_points.txt new file mode 100644 index 0000000..5fdac40 --- /dev/null +++ b/clients/python/oms_client.egg-info/entry_points.txt @@ -0,0 +1,2 @@ +[console_scripts] +oms = oms_client.cli:main diff --git a/clients/python/oms_client.egg-info/requires.txt b/clients/python/oms_client.egg-info/requires.txt new file mode 100644 index 0000000..5991695 --- /dev/null +++ b/clients/python/oms_client.egg-info/requires.txt @@ -0,0 +1,7 @@ +requests>=2.31.0 + +[dev] +pytest>=8.0 + +[pandas] +pandas>=2.0 diff --git a/clients/python/oms_client.egg-info/top_level.txt b/clients/python/oms_client.egg-info/top_level.txt new file mode 100644 index 0000000..c865946 --- /dev/null +++ b/clients/python/oms_client.egg-info/top_level.txt @@ -0,0 +1 @@ +oms_client diff --git a/clients/python/oms_client/__init__.py b/clients/python/oms_client/__init__.py new file mode 100644 index 0000000..b6aac45 --- /dev/null +++ b/clients/python/oms_client/__init__.py @@ -0,0 +1,46 @@ +"""Python client for the OMS trading API. + + from oms_client import OMS + + oms = OMS("http://localhost:3001", token=os.environ["OMS_TRADING_TOKEN"]) + oid = oms.submit(portfolio=pf, symbol="SPY260918C00770000@OPRA", + side="buy", quantity=1) + print(oms.wait_for(oid).status) + +Needs only a trading token (`key_id.secret`, minted by an admin via +`POST /admin/trading-tokens`). No admin credential is used anywhere in this package. +""" + +from .client import OMS +from .errors import ( + AlreadyExists, + AuthError, + BadRequest, + BrokerRejected, + Forbidden, + NotFound, + OMSError, + Rejected, + Unavailable, +) +from .models import Allocation, BlotterRow, Order, Portfolio, Position + +__version__ = "0.1.0" + +__all__ = [ + "OMS", + "OMSError", + "AuthError", + "Forbidden", + "NotFound", + "AlreadyExists", + "Rejected", + "BrokerRejected", + "Unavailable", + "BadRequest", + "Order", + "BlotterRow", + "Portfolio", + "Position", + "Allocation", +] diff --git a/clients/python/oms_client/cli.py b/clients/python/oms_client/cli.py new file mode 100644 index 0000000..ce46542 --- /dev/null +++ b/clients/python/oms_client/cli.py @@ -0,0 +1,226 @@ +#!/usr/bin/env python3 +"""Command-line front end for the OMS trading API — a blotter in a terminal. + +Credentials come from the environment so no token is ever typed into a shell history: + + export OMS_URL=http://localhost:3001 + export OMS_TRADING_TOKEN=ak_....sk_.... + +Usage: + oms portfolios + oms positions + oms orders list [--status routed] [--portfolio ID] [--limit 20] + oms orders get + oms orders cancel [--reason "..."] + oms submit --portfolio ID --symbol SPY260918C00770000@OPRA --side buy --qty 1 + oms submit --portfolio ID --symbol AAPL --venue XNAS --side buy --qty 10 \ + --type limit --limit-price 190.00 + +Every command takes `--json` to emit raw JSON instead of a table, for piping into jq. +""" + +from __future__ import annotations + +import argparse +import dataclasses +import json +import os +import sys +from typing import List, Sequence + +from .client import OMS +from .errors import OMSError + +ENV_URL = "OMS_URL" +ENV_TOKEN = "OMS_TRADING_TOKEN" + + +def _client(args: argparse.Namespace) -> OMS: + url = args.url or os.environ.get(ENV_URL) + token = os.environ.get(ENV_TOKEN) + if not url: + sys.exit(f"error: no server URL — set {ENV_URL} or pass --url") + if not token: + sys.exit(f"error: no trading token — set {ENV_TOKEN}") + return OMS(url, token) + + +def _print_table(rows: Sequence, columns: List[str]) -> None: + """Print dataclass rows as an aligned table, widening each column to fit.""" + if not rows: + print("(no rows)") + return + data = [[_cell(getattr(r, c, None)) for c in columns] for r in rows] + widths = [ + max(len(c), *(len(d[i]) for d in data)) for i, c in enumerate(columns) + ] + print(" ".join(c.upper().ljust(w) for c, w in zip(columns, widths))) + for row in data: + print(" ".join(v.ljust(w) for v, w in zip(row, widths))) + + +def _cell(value) -> str: + if value is None: + return "-" + if isinstance(value, float): + # Trim the trailing zeros a raw float repr leaves on whole quantities. + return f"{value:g}" + return str(value) + + +def _emit(args: argparse.Namespace, rows, columns: List[str]) -> None: + if args.json: + print(json.dumps([dataclasses.asdict(r) for r in rows], indent=2)) + else: + _print_table(rows, columns) + + +# ── commands ───────────────────────────────────────────────────────────────── + + +def cmd_portfolios(args: argparse.Namespace) -> None: + _emit( + args, + _client(args).portfolios(), + ["code", "name", "status", "can_trade", "can_view", "can_allocate", "portfolio_id"], + ) + + +def cmd_positions(args: argparse.Namespace) -> None: + _emit( + args, + _client(args).positions(args.portfolio_id), + ["instrument_id", "net_qty", "avg_cost", "mark", "market_value", "unrealized_pnl"], + ) + + +def cmd_orders_list(args: argparse.Namespace) -> None: + rows = _client(args).orders( + status=args.status, + portfolio_id=args.portfolio, + side=args.side, + limit=args.limit, + ) + _emit( + args, + rows, + [ + "order_id", + "instrument_symbol", + "side", + "order_type", + "status", + "original_qty", + "cum_qty", + "avg_px", + "portfolio_code", + "created_at", + ], + ) + + +def cmd_orders_get(args: argparse.Namespace) -> None: + order = _client(args).order(args.order_id) + print(json.dumps(dataclasses.asdict(order), indent=2)) + + +def cmd_orders_cancel(args: argparse.Namespace) -> None: + outcome = _client(args).cancel(args.order_id, reason=args.reason) + if outcome == "pending": + print("cancel sent to broker — still working; poll `oms orders get` to confirm") + else: + print("canceled") + + +def cmd_submit(args: argparse.Namespace) -> None: + oms = _client(args) + order_id = oms.submit( + portfolio=args.portfolio, + symbol=args.symbol, + venue=args.venue, + instrument_id=args.instrument_id, + side=args.side, + quantity=args.qty, + order_type=args.type, + time_in_force=args.tif, + limit_price=args.limit_price, + account_id=args.account, + ) + print(order_id) + if args.wait: + print(oms.wait_for(order_id).status) + + +# ── parser ─────────────────────────────────────────────────────────────────── + + +def build_parser() -> argparse.ArgumentParser: + # Global flags live on a parent every leaf command inherits, so they are accepted + # where they read naturally — `oms orders list --json`, not `oms --json orders + # list`. Defining them on the top-level parser instead would only allow the + # latter, and argparse would reject the form the help text advertises. + common = argparse.ArgumentParser(add_help=False) + common.add_argument("--url", help=f"OMS base URL (default: ${ENV_URL})") + common.add_argument("--json", action="store_true", help="emit raw JSON") + + p = argparse.ArgumentParser(prog="oms", description=__doc__.splitlines()[0]) + sub = p.add_subparsers(dest="command", required=True) + + sub.add_parser( + "portfolios", parents=[common], help="portfolios this token may act on" + ).set_defaults(func=cmd_portfolios) + + pos = sub.add_parser("positions", parents=[common], help="open positions in a portfolio") + pos.add_argument("portfolio_id") + pos.set_defaults(func=cmd_positions) + + orders = sub.add_parser("orders", help="list, inspect, and cancel orders") + osub = orders.add_subparsers(dest="orders_command", required=True) + + lst = osub.add_parser("list", parents=[common], help="the blotter") + lst.add_argument("--status", help="filter by status (e.g. routed, filled)") + lst.add_argument("--portfolio", help="filter by portfolio id") + lst.add_argument("--side", choices=["buy", "sell"]) + lst.add_argument("--limit", type=int, default=100) + lst.set_defaults(func=cmd_orders_list) + + get = osub.add_parser("get", parents=[common], help="one order's full state") + get.add_argument("order_id") + get.set_defaults(func=cmd_orders_get) + + can = osub.add_parser("cancel", parents=[common], help="cancel an order") + can.add_argument("order_id") + can.add_argument("--reason") + can.set_defaults(func=cmd_orders_cancel) + + sb = sub.add_parser("submit", parents=[common], help="submit an order") + sb.add_argument("--portfolio", required=True) + sb.add_argument("--symbol", help="SYMBOL or SYMBOL@VENUE") + sb.add_argument("--venue", help="venue MIC, when not embedded in --symbol") + sb.add_argument("--instrument-id", dest="instrument_id", help="instead of --symbol") + sb.add_argument("--side", required=True, choices=["buy", "sell"]) + sb.add_argument("--qty", required=True, type=float) + sb.add_argument("--type", default="market", choices=["market", "limit"]) + sb.add_argument("--tif", default="day", choices=["day", "gtc", "ioc", "fok"]) + sb.add_argument("--limit-price", dest="limit_price", type=float) + sb.add_argument("--account", help="override the portfolio's default account") + sb.add_argument("--wait", action="store_true", help="poll until terminal") + sb.set_defaults(func=cmd_submit) + + return p + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + args.func(args) + except OMSError as err: + # The server's message is the useful part; a traceback is not. + sys.exit(f"error: {err.message} (HTTP {err.status})") + except TimeoutError as err: + sys.exit(f"error: {err}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/clients/python/oms_client/client.py b/clients/python/oms_client/client.py new file mode 100644 index 0000000..190eec7 --- /dev/null +++ b/clients/python/oms_client/client.py @@ -0,0 +1,251 @@ +"""Synchronous client for the OMS trading API. + +One trading token is all this needs — it never touches the admin surface. That is +deliberate: the admin token is a single shared secret covering principal management, +broker connections, and risk limits, so a script holding it could mint itself +credentials and edit the limits meant to constrain it. + +Three properties of the wire protocol shape this whole module: + +* **Submit returns 204 with an empty body.** The client generates `order_id` and that + id *is* the idempotency key. Nothing comes back, so observing the order means + fetching it (`order`) or waiting on it (`wait_for`). +* **Cancel is asynchronous when the order is live at a broker.** 202 means the cancel + was forwarded and will be finalized by the execution stream; only 204 means done. +* **Errors are plain text.** See `errors.raise_for_status`. + +There is no push channel — no SSE, no WebSocket — so `wait_for` polls. That is a +property of the server, not a shortcut here. +""" + +from __future__ import annotations + +import time +import uuid +from typing import Any, Dict, List, Optional + +import requests + +from .errors import AlreadyExists, OMSError, raise_for_status +from .models import ( + Allocation, + BlotterRow, + Order, + Portfolio, + Position, + RowList, + parse_list, +) + +DEFAULT_TIMEOUT = 30.0 + + +class OMS: + """A trading-token client. + + >>> oms = OMS("http://localhost:3001", token="ak_....sk_....") + >>> oid = oms.submit(portfolio="3a8d…", symbol="SPY260918C00770000@OPRA", + ... side="buy", quantity=1) + >>> oms.wait_for(oid).status + 'filled' + """ + + def __init__( + self, + base_url: str, + token: str, + *, + timeout: float = DEFAULT_TIMEOUT, + session: Optional[requests.Session] = None, + ): + self.base_url = base_url.rstrip("/") + self.timeout = timeout + # `Bearer key_id.secret` is one of the two credential forms the server accepts + # (the other being Basic key_id:secret); this one survives a shell env var. + self._session = session or requests.Session() + self._session.headers.update({"Authorization": f"Bearer {token}"}) + + # ── plumbing ───────────────────────────────────────────────────────────── + + def _request(self, method: str, path: str, **kw) -> requests.Response: + resp = self._session.request( + method, f"{self.base_url}{path}", timeout=self.timeout, **kw + ) + if resp.status_code >= 400: + # Never .json() here: failures are text/plain. + raise_for_status(resp.status_code, resp.text) + return resp + + def _get_json(self, path: str, params: Optional[Dict[str, Any]] = None) -> Any: + clean = {k: v for k, v in (params or {}).items() if v is not None} + return self._request("GET", path, params=clean).json() + + # ── reference ──────────────────────────────────────────────────────────── + + def portfolios(self) -> "RowList[Portfolio]": + """Portfolios this token may act on, with its own permission flags.""" + return parse_list(Portfolio, self._get_json("/portfolios")) + + def positions(self, portfolio_id: str) -> "RowList[Position]": + """Open positions in a portfolio. Needs `can_view`.""" + return parse_list(Position, self._get_json(f"/portfolios/{portfolio_id}/positions")) + + # ── orders ─────────────────────────────────────────────────────────────── + + def submit( + self, + *, + portfolio: str, + side: str, + quantity: float, + symbol: Optional[str] = None, + venue: Optional[str] = None, + instrument_id: Optional[str] = None, + order_type: str = "market", + time_in_force: str = "day", + limit_price: Optional[float] = None, + account_id: Optional[str] = None, + client_order_id: Optional[str] = None, + order_id: Optional[str] = None, + ) -> str: + """Submit an order. Returns its `order_id`. + + Name the instrument any of three ways: `instrument_id`, `symbol` + `venue`, or + `symbol` alone as ``SYMBOL@VENUE``. A bare `symbol` works only when it is + unique across venues — many equity tickers are not, and the server answers 422 + naming the candidates. + + `order_id` is generated when not supplied, and is the idempotency key. Because + of that, a 409 on retry means the order was *already accepted* — so it is + swallowed here rather than raised, making `submit` safe to retry on a timeout. + Passing your own `order_id` opts into that same guarantee across processes. + + Note the server returns 204 with no body, so nothing but the id can be + returned; call `order()` or `wait_for()` to see what became of it. + """ + oid = order_id or str(uuid.uuid4()) + body: Dict[str, Any] = { + "order_id": oid, + "client_order_id": client_order_id or oid, + "portfolio_id": portfolio, + "side": side, + "order_type": order_type, + "time_in_force": time_in_force, + "quantity": quantity, + } + if instrument_id is not None: + body["instrument_id"] = instrument_id + if symbol is not None: + body["symbol"] = symbol + if venue is not None: + body["venue"] = venue + if limit_price is not None: + body["limit_price"] = limit_price + if account_id is not None: + body["account_id"] = account_id + + try: + self._request("POST", "/orders/submit", json=body) + except AlreadyExists: + # The id is ours and the OMS already has it: the earlier attempt landed. + pass + return oid + + def order(self, order_id: str) -> Order: + """Fetch one order's current state. Needs `can_view` on its portfolio.""" + return Order.from_dict(self._get_json(f"/orders/{order_id}")) + + def orders( + self, + *, + status: Optional[str] = None, + portfolio_id: Optional[str] = None, + instrument_id: Optional[str] = None, + side: Optional[str] = None, + since: Optional[str] = None, + until: Optional[str] = None, + limit: int = 100, + offset: int = 0, + ) -> "RowList[BlotterRow]": + """The blotter, bounded to portfolios this token may view. + + Scope comes from the authenticated principal's grants, so this returns every + order in those portfolios — including ones another principal submitted. It is + a portfolio's activity, not a personal history. + + `since`/`until` are RFC3339 timestamps; `limit` is clamped server-side to + 1..500. + """ + return parse_list( + BlotterRow, + self._get_json( + "/orders", + { + "status": status, + "portfolio_id": portfolio_id, + "instrument_id": instrument_id, + "side": side, + "since": since, + "until": until, + "limit": limit, + "offset": offset, + }, + ), + ) + + def cancel(self, order_id: str, reason: Optional[str] = None) -> str: + """Cancel an order. Needs `can_trade`. + + Returns ``"canceled"`` when the OMS cancelled it outright (it had not reached a + broker), or ``"pending"`` when the cancel was forwarded to the broker and will + be confirmed asynchronously by the execution stream. On ``"pending"`` the order + is not yet cancelled — poll `order()` to see it land. + """ + body: Dict[str, Any] = {"order_id": order_id} + if reason is not None: + body["reason"] = reason + resp = self._request("POST", "/orders/cancel", json=body) + return "pending" if resp.status_code == 202 else "canceled" + + def wait_for( + self, + order_id: str, + *, + timeout: float = 60.0, + interval: float = 0.5, + ) -> Order: + """Poll until the order reaches a terminal status, then return it. + + Polling because the OMS has no push channel for order events. Raises + `TimeoutError` if it is still working when `timeout` elapses — the order is + untouched by that, it just has not finished. + """ + deadline = time.monotonic() + timeout + while True: + current = self.order(order_id) + if current.is_terminal: + return current + if time.monotonic() >= deadline: + raise TimeoutError( + f"order {order_id} still {current.status} after {timeout:g}s" + ) + time.sleep(interval) + + # ── allocations ────────────────────────────────────────────────────────── + + def allocate(self, order_id: str, splits: List[Dict[str, Any]]) -> "RowList[Allocation]": + """Split a filled block order into sub-portfolios. Needs `can_allocate`. + + `splits` is a list of ``{"portfolio_id": ..., "qty": ...}``. + """ + payload = self._request( + "POST", f"/orders/{order_id}/allocations", json={"splits": splits} + ).json() + return parse_list(Allocation, payload) + + def allocations(self, order_id: str) -> "RowList[Allocation]": + """Existing allocations for an order.""" + return parse_list(Allocation, self._get_json(f"/orders/{order_id}/allocations")) + + +__all__ = ["OMS", "OMSError"] diff --git a/clients/python/oms_client/errors.py b/clients/python/oms_client/errors.py new file mode 100644 index 0000000..d179f14 --- /dev/null +++ b/clients/python/oms_client/errors.py @@ -0,0 +1,92 @@ +"""Exceptions mapped from the OMS's HTTP responses. + +The OMS returns errors as **plain text, not JSON** — `ApiError` on the Rust side is +`(StatusCode, String)`, so a failed request's body is a bare message like +``instrument not active`` or ``risk check failed [invalid_quantity]: ...``. Parsing a +failure as JSON would raise the wrong exception entirely, so nothing here touches +`.json()`. + +Every exception carries `status` and `message`, so a caller that does not want to +match on types can still branch on the status code. +""" + +from __future__ import annotations + + +class OMSError(Exception): + """Base for every error the OMS returns. Catch this to catch them all.""" + + def __init__(self, status: int, message: str): + super().__init__(f"{status}: {message}") + self.status = status + self.message = message + + +class AuthError(OMSError): + """401 — the token is missing, malformed, unknown, or revoked.""" + + +class Forbidden(OMSError): + """403 — authenticated, but lacking the grant this route requires. + + Trading needs `can_trade` on the portfolio, reading needs `can_view`, and + allocating needs `can_allocate`. + """ + + +class NotFound(OMSError): + """404 — no such order.""" + + +class AlreadyExists(OMSError): + """409 — an order with this `order_id` was already accepted. + + Not a failure for a retry. `order_id` is the idempotency key, so re-sending one + the OMS already has means the first attempt landed — see `OMS.submit`, which + treats this as success for an id it generated itself. + """ + + +class Rejected(OMSError): + """422 — the order was refused before reaching a broker. + + Covers an unknown or inactive instrument, an ambiguous symbol, an option with a + time-in-force other than `day`, a size below the broker minimum, and pre-trade + risk. Risk rejections carry their code inline: + ``risk check failed [invalid_quantity]: ...``. + """ + + +class BrokerRejected(OMSError): + """502 — the OMS accepted the order but the broker refused it. + + The order still exists in the OMS as `submitted`: the event was committed before + routing was attempted. Do not resubmit under a fresh id, or the position doubles + if the first one actually landed. + """ + + +class Unavailable(OMSError): + """503 — no adapter configured for the account's broker, or the DB is unreachable.""" + + +class BadRequest(OMSError): + """400 — malformed request: a bad UUID, or a missing/contradictory instrument reference.""" + + +_BY_STATUS = { + 400: BadRequest, + 401: AuthError, + 403: Forbidden, + 404: NotFound, + 409: AlreadyExists, + 422: Rejected, + 502: BrokerRejected, + 503: Unavailable, +} + + +def raise_for_status(status: int, body: str) -> None: + """Raise the exception matching `status`, or `OMSError` for anything unmapped.""" + cls = _BY_STATUS.get(status, OMSError) + raise cls(status, body.strip()) diff --git a/clients/python/oms_client/models.py b/clients/python/oms_client/models.py new file mode 100644 index 0000000..d3f85a5 --- /dev/null +++ b/clients/python/oms_client/models.py @@ -0,0 +1,178 @@ +"""Typed views over the OMS's JSON responses. + +Dataclasses rather than raw dicts, so a typo is an AttributeError at the call site +instead of a KeyError three frames later. Every one is built with `_of`, which drops +keys it does not know about — the server can add a field without breaking a client +that has not been updated. + +Ids stay strings. The OMS serializes them that way (order and portfolio ids are +UUIDs, `instrument_id` is a stringified BIGINT), and converting would mean converting +back on every request. +""" + +from __future__ import annotations + +from dataclasses import dataclass, fields +from typing import Any, Dict, List, Optional + +# The statuses no further event can move an order out of. Mirrors +# `OrderStatus::is_terminal` in src/domain/orders/state.rs — `OMS.wait_for` stops here. +TERMINAL_STATUSES = frozenset({"filled", "rejected", "canceled", "expired"}) + + +def _of(cls, data: Dict[str, Any]): + """Build a dataclass from a response dict, ignoring unknown keys.""" + known = {f.name for f in fields(cls)} + return cls(**{k: v for k, v in data.items() if k in known}) + + +@dataclass +class Order: + """An order's current state, from ``GET /orders/{id}``.""" + + order_id: str + client_order_id: str + portfolio_id: str + account_id: str + instrument_id: str + side: str + order_type: str + time_in_force: str + original_qty: float + leaves_qty: float + cum_qty: float + version: int + status: str + limit_price: Optional[float] = None + avg_px: Optional[float] = None + resume_to_status: Optional[str] = None + + @property + def is_terminal(self) -> bool: + return self.status in TERMINAL_STATUSES + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "Order": + return _of(cls, d) + + +@dataclass +class BlotterRow: + """One row of the blotter, from ``GET /orders``. + + Denormalized for display — it carries the portfolio and instrument *codes*, which + `Order` does not. Note it omits `client_order_id`, `limit_price`, and + `time_in_force`; fetch the order itself when those matter. + """ + + order_id: str + principal_id: str + principal_code: str + portfolio_id: str + portfolio_code: str + account_id: str + broker_connection_code: str + instrument_id: str + side: str + order_type: str + status: str + original_qty: float + leaves_qty: float + cum_qty: float + created_at: str + updated_at: str + instrument_symbol: Optional[str] = None + instrument_name: Optional[str] = None + avg_px: Optional[float] = None + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "BlotterRow": + return _of(cls, d) + + +@dataclass +class Portfolio: + """A portfolio this token may act on, from ``GET /portfolios``. + + The permission flags are the caller's own grant, so a client can tell in advance + what it may do rather than finding out from a 403. + """ + + portfolio_id: str + code: str + name: str + status: str + can_trade: bool + can_view: bool + can_allocate: bool + base_currency: Optional[str] = None + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "Portfolio": + return _of(cls, d) + + +@dataclass +class Position: + """A position, from ``GET /portfolios/{id}/positions``. + + The mark-derived fields are `None` when no live quote exists for the instrument — + an unpriceable or expired contract, or a feed that is down. + """ + + portfolio_id: str + instrument_id: str + net_qty: float + avg_cost: float + realized_pnl: float + updated_at: str + mark: Optional[float] = None + market_value: Optional[float] = None + unrealized_pnl: Optional[float] = None + mark_ts: Optional[str] = None + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "Position": + return _of(cls, d) + + +@dataclass +class Allocation: + """One split of a block order into a sub-portfolio.""" + + id: str + order_id: str + from_portfolio_id: str + to_portfolio_id: str + instrument_id: str + qty: float + price: float + created_at: str + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "Allocation": + return _of(cls, d) + + +class RowList(list): + """A plain `list` that can also turn itself into a DataFrame. + + Subclassing `list` keeps indexing, iteration, and `len()` working exactly as + expected; `to_pandas()` is the only addition, and it imports pandas lazily so the + dependency is genuinely optional. + """ + + def to_pandas(self): + try: + import pandas as pd + except ImportError as exc: # pragma: no cover - depends on the environment + raise ImportError( + "to_pandas() needs pandas — install with: pip install 'oms-client[pandas]'" + ) from exc + from dataclasses import asdict + + return pd.DataFrame([asdict(r) for r in self]) + + +def parse_list(cls, payload: List[Dict[str, Any]]) -> RowList: + return RowList(cls.from_dict(d) for d in payload) diff --git a/clients/python/pyproject.toml b/clients/python/pyproject.toml new file mode 100644 index 0000000..32e446b --- /dev/null +++ b/clients/python/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "oms-client" +version = "0.1.0" +description = "Python client for the OMS trading API" +requires-python = ">=3.9" +dependencies = ["requests>=2.31.0"] + +[project.optional-dependencies] +# pandas is deliberately optional: the client is usable in a plain script without +# it, and only `.to_pandas()` needs it. +pandas = ["pandas>=2.0"] +dev = ["pytest>=8.0"] + +[project.scripts] +oms = "oms_client.cli:main" + +[tool.setuptools.packages.find] +include = ["oms_client*"] diff --git a/clients/python/tests/test_client.py b/clients/python/tests/test_client.py new file mode 100644 index 0000000..1ecfc7e --- /dev/null +++ b/clients/python/tests/test_client.py @@ -0,0 +1,246 @@ +"""Unit tests for the OMS client — no server, no network. + +Everything here is about the rules that are easy to get wrong and expensive when +wrong: that a plain-text error becomes the right exception, that a 409 on a retry is +treated as success rather than failure, and that a 202 cancel is not mistaken for a +completed one. + +The transport is stubbed with a fake `requests.Session`, so these run anywhere. +""" + +from __future__ import annotations + +import pytest + +from oms_client import OMS +from oms_client.errors import ( + AlreadyExists, + AuthError, + BrokerRejected, + Forbidden, + NotFound, + OMSError, + Rejected, +) +from oms_client.models import Order + + +class FakeResponse: + def __init__(self, status_code: int, payload=None, text: str = ""): + self.status_code = status_code + self._payload = payload + self.text = text + + def json(self): + return self._payload + + +class FakeSession: + """Stands in for requests.Session, returning queued responses and recording calls.""" + + def __init__(self, *responses: FakeResponse): + self.headers = {} + self._responses = list(responses) + self.calls = [] + + def request(self, method, url, **kw): + self.calls.append((method, url, kw)) + return self._responses.pop(0) if self._responses else FakeResponse(204) + + +def client(*responses: FakeResponse) -> tuple: + session = FakeSession(*responses) + return OMS("http://oms.test", "ak_x.sk_y", session=session), session + + +# ── auth ───────────────────────────────────────────────────────────────────── + + +def test_token_is_sent_as_bearer(): + """The `Bearer key_id.secret` form — one of the two the server accepts.""" + _, session = client() + assert session.headers["Authorization"] == "Bearer ak_x.sk_y" + + +# ── error mapping ──────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "status,exc,body", + [ + (401, AuthError, "unauthorized"), + (403, Forbidden, "unauthorized"), + (404, NotFound, "order not found"), + (422, Rejected, "instrument not active"), + (502, BrokerRejected, "broker rejected order: closed"), + ], +) +def test_status_maps_to_exception(status, exc, body): + oms, _ = client(FakeResponse(status, text=body)) + with pytest.raises(exc) as caught: + oms.order("some-id") + assert caught.value.status == status + assert caught.value.message == body + + +def test_unmapped_status_still_raises_oms_error(): + """A status the client has never seen must not escape as a raw requests error.""" + oms, _ = client(FakeResponse(418, text="teapot")) + with pytest.raises(OMSError) as caught: + oms.order("some-id") + assert caught.value.status == 418 + + +def test_error_body_is_never_parsed_as_json(): + """Errors are text/plain; calling .json() on them would raise the wrong thing.""" + + class Exploding(FakeResponse): + def json(self): + raise AssertionError("json() must not be called on an error response") + + oms, _ = client(Exploding(422, text="risk check failed [invalid_quantity]: too big")) + with pytest.raises(Rejected) as caught: + oms.order("some-id") + assert "invalid_quantity" in caught.value.message + + +# ── submit ─────────────────────────────────────────────────────────────────── + + +def test_submit_generates_and_returns_order_id(): + """Submit answers 204 with no body, so the id the client made is the only handle.""" + oms, session = client(FakeResponse(204)) + order_id = oms.submit(portfolio="pf", symbol="SPY@ARCX", side="buy", quantity=1) + assert session.calls[0][2]["json"]["order_id"] == order_id + assert len(order_id) == 36 + + +def test_submit_swallows_409_because_the_id_is_ours(): + """409 means the OMS already accepted this exact order — a retry, not a failure. + + Raising here would push callers toward resubmitting under a new id, which is how + you end up with the position twice. + """ + oms, _ = client(FakeResponse(409, text="order already exists")) + assert oms.submit( + portfolio="pf", symbol="SPY@ARCX", side="buy", quantity=1, order_id="fixed-id" + ) == "fixed-id" + + +def test_submit_still_raises_on_broker_rejection(): + """502 is not idempotent-safe to ignore: the order exists but never routed.""" + oms, _ = client(FakeResponse(502, text="broker rejected order: market closed")) + with pytest.raises(BrokerRejected): + oms.submit(portfolio="pf", symbol="SPY@ARCX", side="buy", quantity=1) + + +def test_submit_omits_absent_instrument_fields(): + """Sending symbol AND venue when only one was given would trip the server's + 'venue given both in symbol and in the venue field' check.""" + oms, session = client(FakeResponse(204)) + oms.submit(portfolio="pf", symbol="SPY260918C00770000@OPRA", side="buy", quantity=1) + body = session.calls[0][2]["json"] + assert "venue" not in body and "instrument_id" not in body + assert "limit_price" not in body + + +def test_submit_passes_limit_price_and_venue_when_given(): + oms, session = client(FakeResponse(204)) + oms.submit( + portfolio="pf", symbol="AAPL", venue="XNAS", side="buy", quantity=10, + order_type="limit", limit_price=190.5, + ) + body = session.calls[0][2]["json"] + assert body["venue"] == "XNAS" + assert body["limit_price"] == 190.5 + assert body["order_type"] == "limit" + + +# ── cancel ─────────────────────────────────────────────────────────────────── + + +def test_cancel_202_is_pending_not_done(): + """202 means forwarded to the broker; the order is still live until confirmed.""" + oms, _ = client(FakeResponse(202)) + assert oms.cancel("oid") == "pending" + + +def test_cancel_204_is_final(): + oms, _ = client(FakeResponse(204)) + assert oms.cancel("oid") == "canceled" + + +# ── models ─────────────────────────────────────────────────────────────────── + + +ORDER_JSON = { + "order_id": "o1", "client_order_id": "c1", "portfolio_id": "p1", + "account_id": "a1", "instrument_id": "42", "side": "buy", + "order_type": "market", "time_in_force": "day", "limit_price": None, + "original_qty": 1.0, "leaves_qty": 0.0, "cum_qty": 1.0, "avg_px": 5.0, + "status": "filled", "resume_to_status": None, "version": 3, +} + + +def test_order_parses_and_reports_terminal(): + order = Order.from_dict(ORDER_JSON) + assert order.status == "filled" + assert order.is_terminal + + +def test_working_order_is_not_terminal(): + assert not Order.from_dict({**ORDER_JSON, "status": "partially_filled"}).is_terminal + + +def test_unknown_fields_are_ignored(): + """The server may add a field; a client that has not been updated must not break.""" + order = Order.from_dict({**ORDER_JSON, "brand_new_field": "surprise"}) + assert order.order_id == "o1" + + +# ── wait_for ───────────────────────────────────────────────────────────────── + + +def test_wait_for_polls_until_terminal(): + oms, _ = client( + FakeResponse(200, {**ORDER_JSON, "status": "routed"}), + FakeResponse(200, {**ORDER_JSON, "status": "partially_filled"}), + FakeResponse(200, ORDER_JSON), + ) + assert oms.wait_for("o1", interval=0).status == "filled" + + +def test_wait_for_times_out_while_still_working(): + oms, _ = client(*[FakeResponse(200, {**ORDER_JSON, "status": "routed"})] * 50) + with pytest.raises(TimeoutError): + oms.wait_for("o1", timeout=0, interval=0) + + +# ── blotter ────────────────────────────────────────────────────────────────── + + +BLOTTER_JSON = { + "order_id": "o1", "principal_id": "pr1", "principal_code": "trader", + "portfolio_id": "p1", "portfolio_code": "PF1", "account_id": "a1", + "broker_connection_code": "alpaca-paper", "instrument_id": "42", + "instrument_symbol": "SPY260918C00770000", "instrument_name": "SPY option", + "side": "buy", "order_type": "market", "status": "filled", + "original_qty": 1.0, "leaves_qty": 0.0, "cum_qty": 1.0, "avg_px": 5.0, + "created_at": "2026-08-14T07:00:00Z", "updated_at": "2026-08-14T07:00:01Z", +} + + +def test_orders_drops_none_filters_from_the_query(): + """A `status=None` sent as a literal would filter on the string 'None'.""" + oms, session = client(FakeResponse(200, [BLOTTER_JSON])) + rows = oms.orders(limit=10) + params = session.calls[0][2]["params"] + assert params == {"limit": 10, "offset": 0} + assert rows[0].instrument_symbol == "SPY260918C00770000" + + +def test_orders_keeps_supplied_filters(): + oms, session = client(FakeResponse(200, [])) + oms.orders(status="routed", side="buy") + params = session.calls[0][2]["params"] + assert params["status"] == "routed" and params["side"] == "buy"