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" 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/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; 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/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()); 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(",") ); 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/handlers.rs b/src/handlers.rs index fc44ecf..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, @@ -83,26 +83,52 @@ 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, - 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, + /// Required for `limit` orders; omit for `market`. + #[schema(example = json!(null))] pub limit_price: Option, + #[schema(example = 1.0)] pub quantity: f64, } 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, @@ -112,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, @@ -124,7 +313,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, @@ -166,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 @@ -624,13 +812,14 @@ 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, + 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()); @@ -640,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, @@ -888,10 +1080,11 @@ 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, + Extension(auth): Extension, Path(order_id_str): Path, ) -> Result, ApiError> { let order_id = Uuid::parse_str(&order_id_str).map_err(|_| ApiError { @@ -899,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 @@ -951,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")), @@ -958,7 +1216,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 +1334,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 +1464,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, @@ -1281,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, \ @@ -1308,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()); } @@ -1352,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 @@ -1434,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 8acba7e..472a6ca 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; @@ -60,6 +60,7 @@ mod binance_feed; mod bybit_feed; mod feeds; mod preflight; +mod expiry; mod fix; #[derive(OpenApi)] @@ -70,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, @@ -109,11 +112,13 @@ mod fix; admin::list_feeds, admin::resolve_symbology, admin::backfill_symbology, + admin::expiry_sweep, ), components(schemas( SubmitOrder, SubmitOrderRequest, CancelOrder, OrderSide, OrderType, TimeInForce, OrderAggregateState, crate::positions::Position, Allocation, CreateAllocations, AllocationSplit, BlotterRow, + handlers::GrantedPortfolio, Principal, Portfolio, Account, BrokerConnection, CreatePrincipal, UpdatePrincipal, CreatePortfolio, UpdatePortfolio, @@ -125,6 +130,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), @@ -141,11 +147,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(), + ), ); } } @@ -442,6 +461,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); @@ -514,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", @@ -595,6 +622,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?; }