From 6b21210d9f0ae0d82a824217e035e6fd45a2e7fb Mon Sep 17 00:00:00 2001 From: Yuki Chu Date: Thu, 22 Oct 2026 20:35:06 +0100 Subject: [PATCH] Warm combo boards in serve and name the legs a venue lacks serve now builds every combo venue's board in the background at start-up and again every refresh_interval_s ([combo_board] warm, refresh_interval_s, pause_s), keeping the last good board, so the board route never waits on a build. The board is served in parts: GET /trading/combos/board/categories for the rail, and category / cursor / limit / q / sort pages on the board route. Polymarket's catalogue is about 214,000 markets; Gamma reads are cached between builds, and a first build publishes partial boards. The build stays off the event loop that answers health checks and orders: its CPU steps run in worker threads that hand back the GIL every 100 items, the GIL switch interval drops to 0.5 ms while boards are warmed, and the board's objects are frozen out of the cyclic collector as they are built. Requests stay under 0.8 s during a whole-catalogue build (before: 1-6 s). An RFQ group now matches every leg before skipping a venue, and each skip lists missing_legs (market_id, label, code, mapped_id, detail), with codes missing_leg, matching_error, ineligible_leg and outside_collections; labels come from the warm board, else a shortened catalogue question. Co-Authored-By: Claude Opus 5.5 --- docs/engine.md | 6 + docs/rfq-combo.md | 40 +- docs/server.md | 64 ++- synpath/engine/combo_board.py | 455 ++++++++++++++++-- synpath/engine/rfq_group.py | 168 +++++-- synpath/server/combo_warmer.py | 245 ++++++++++ synpath/server/rfq_routes.py | 67 ++- synpath/server/serve.py | 30 +- .../polymarket_combo_skip_legs_live.json | 160 ++++++ tests/test_combo_warmer.py | 453 +++++++++++++++++ tests/test_engine_rfq_group.py | 12 +- 11 files changed, 1603 insertions(+), 97 deletions(-) create mode 100644 synpath/server/combo_warmer.py create mode 100644 tests/samples/polymarket_combo_skip_legs_live.json create mode 100644 tests/test_combo_warmer.py diff --git a/docs/engine.md b/docs/engine.md index 431f2c5..d6806f9 100644 --- a/docs/engine.md +++ b/docs/engine.md @@ -506,6 +506,12 @@ price_collar = "0.10" daily_loss_limit = 500 max_open_orders = 50 self_trade_prevention = "firm" + +# synpath serve only: the combo boards it keeps warm (docs/server.md, "The combo board"). +[combo_board] +warm = true # false: build a board on request (5 pages, kept 30 s) +refresh_interval_s = 60 # rebuild each board this long after its last build ended +pause_s = 0.1 # between two reads of one venue during a build ``` Credentials never appear in the configuration: they come from the diff --git a/docs/rfq-combo.md b/docs/rfq-combo.md index 4c005b1..fa559e2 100644 --- a/docs/rfq-combo.md +++ b/docs/rfq-combo.md @@ -747,12 +747,19 @@ asked directly. `fetch_combo_eligible_markets`): every mapped leg must be on the list, read up to 20 pages and kept 60 s; Polymarket legs get their `info.position_id` from it. A venue with neither is `unchecked`. -- **Skipped venues**, each with `venue`, `code`, `reason` and the leg: +- **Skipped venues**, each with `venue`, `code`, `reason` and the first leg: `excluded` (`venues`, `exclude_venues`), `forbidden` (the HTTP key may not trade it), `not_configured`, `no_rfq`, `no_combos`, `no_match`, - `matching_error`, `missing_leg`, `ineligible_leg`, `duplicate_leg`, - `one_sided_size` (Polymarket global asks a YES buy by amount and a YES - sell by quantity), `member_disabled`. No venue left is `BadRequest` with the + `matching_error`, `missing_leg`, `ineligible_leg`, `outside_collections`, + `duplicate_leg`, `one_sided_size` (Polymarket global asks a YES buy by + amount and a YES sell by quantity), `member_disabled`. A combo venue names + every leg it lacks in `missing_legs` (`market_id`, `label`, `code`, + `mapped_id`, `detail`) and its `reason` names them by label + ([server](server.md#combo-skip-reasons)): no match (`missing_leg`), + matching unavailable (`matching_error`), matched but not combo-eligible + there (`ineligible_leg`), or outside the venue's combo collections + (`outside_collections`, Kalshi). Labels come from + `engine.rfq_groups.labeler` (serve sets its combo board warmer's). No venue left is `BadRequest` with the list on `exc.skipped` (`400 no_venue` over HTTP), and nothing is journaled. - **One-sided venues.** An adapter with `rfq_one_sided = True` (Polymarket global) is asked the YES side the group takes there; two-sided venues are @@ -947,9 +954,32 @@ live until the venue closes it: and Gamma's events by id; `TestPreview` in `tests/test_engine_rfq_group.py` and `tests/test_server_rfq_group.py`, `TestEventsByIds` in `tests/test_kalshi.py`. +### Warm boards and skip reasons (2026-10-22) + +- **Skip reasons name every leg.** Yuki's Polymarket combo "Will Al Fayha Saudi Club win on + 2026-10-22?" (`polymarket:5441701`: hosted matching lists it on Polymarket US as + `polymarket_us:atc-spl-haz-fay-2026-10-22-fay`, not on Kalshi) and "Will the U.S. invade Iran + before 2027?" (`polymarket:665374`: matched nowhere) was skipped correctly on both venues, but a + skip named only its first leg, by id, and the terminal showed the leg's outcome (`no "YES"`). Now + `_plan_combo` maps every leg before deciding, eligibility names every leg it refuses, and + `SkippedVenue.missing_legs` carries each with a label: Kalshi "no match for “Al Fayha to win”, + “U.S. invades Iran”", Polymarket US "no match for “U.S. invades Iran”". A Kalshi leg whose event + is in no open collection is its own code, `outside_collections` (it was `ineligible_leg`). +- **The board is warm** (`synpath/server/combo_warmer.py`, [server](server.md#the-combo-board)): + built at serve's start and every minute after, through the venue's whole list (the board read on + request stopped at 5 pages: 5,000 rows, `complete: false`; the whole catalogue is about 214,000 + rows, so a first build serves its board so far as it goes, and Gamma's reads are kept between + builds); requests answer from it at once + (`status: "building"` until the first build ends) and read it in parts: + `GET /trading/combos/board/categories` and `GET /trading/combos/board?category=&cursor=&limit=`. +- Tests: `tests/test_combo_warmer.py` (the warmer, the routes, the skip reasons on both real legs + with Gamma's real rows in `tests/samples/polymarket_combo_skip_legs_live.json` and matching + mocked from its real answers, and serve's wiring). + ### Over HTTP -`POST /trading/rfq-groups`, `POST /trading/rfq-groups/preview`, `GET /trading/combos/board`, `GET /trading/rfq-groups`, +`POST /trading/rfq-groups`, `POST /trading/rfq-groups/preview`, `GET /trading/combos/board`, +`GET /trading/combos/board/categories`, `GET /trading/rfq-groups`, `GET /trading/rfq-groups/{id}`, `POST /trading/rfq-groups/{id}/accept`, `DELETE /trading/rfq-groups/{id}` ([server](server.md#cross-venue-rfq-groups)). diff --git a/docs/server.md b/docs/server.md index 60fc2e1..4ab40cc 100644 --- a/docs/server.md +++ b/docs/server.md @@ -323,7 +323,8 @@ python -m synpath.server bootstrap --control control.db # the same, from a she | `POST /quotes`, `PUT /quotes/{id}`, `DELETE /quotes/{id}`, `POST /quotes/{id}/confirm` | `trade` | send, replace, withdraw a quote; confirm an accepted one (`202`/`200` as accept) | | `POST /combos`, `POST /combos/lookup`, `GET /combos/{market_id}` | `trade` / `view` / `view` | create or look up the combo for some legs (idempotent; `collection_id` on Kalshi); look one up without creating it; read its legs | | `GET /combos/markets` (`venue`, `event_id`, `tag`, `limit`, `cursor`), `GET /combos/collections` (`venue`, `status`, `event_id`, `limit`, `cursor`) | `view` | where combos can be built: Polymarket's and Polymarket US's combo-eligible markets; Kalshi's collections with their event tickers. Cached 15 s | -| `GET /combos/board` (`venue`, `pages`) | `view` | a combo venue's eligible markets as games (moneyline / spread / total picks per participant) and YES / NO questions, with categories and counts ([combo board](#the-combo-board)). Cached 30 s | +| `GET /combos/board/categories` (`venue`) | `view` | a combo board's rail: its categories with counts, `status` (`building` / `ready`), `built_at`, `complete` ([combo board](#the-combo-board)) | +| `GET /combos/board` (`venue`; `category`, `cursor`, `limit`, `q`, `sort`) | `view` | a combo venue's eligible markets as games (moneyline / spread / total picks per participant) and YES / NO questions, with categories and counts: the whole board, or one category a page at a time ([combo board](#the-combo-board)). serve keeps it warm; never waits on a build | | `GET`/`POST`/`DELETE /grants`, `POST /keys`, `DELETE /keys/{id}` | `manage_members` / `manage_credentials` | who may do what | | `GET /audit` | `manage_members` | the append-only history of those changes | | `WS /ws/events` | `view` | the engine's events, replayed from a cursor then live | @@ -496,7 +497,9 @@ list; accepting one quote cancels the rest - **`RfqGroup`**: `id`, `instrument`, the size, `outcome`, `action`, `members` (one per venue: `venue`, `instrument_id`, `flip`, `via`, mapped `legs`, `combo`, `collection_id`, `rfq_id`, `status`, `detail`), - `skipped` (`venue`, `code`, `reason`, `leg`), `quotes` best first + `skipped` (`venue`, `code`, `reason`, `leg`, and for a combo `missing_legs`: + every leg the venue lacks, each `market_id`, `label`, `code`, `mapped_id`, + `detail`; [skip reasons](#combo-skip-reasons)), `quotes` best first (`quote_id`, `venue`, `rfq_id`, `raw_bid`/`raw_ask`, `raw_price` (the member's YES price), `price` (the group's), `outcome_price`, `quantity`, `fee`, `fee_per_contract`, `effective_price`, `effective_outcome_price`, @@ -542,10 +545,60 @@ list; accepting one quote cancels the rest - **Halt**: `POST /halt` with the `cancel` policy cancels every open group's members first (`rfq.groups` in the result). +### Combo skip reasons + +A combo venue that would not be asked (`skipped` on a preview or a group) names **every** leg it +lacks, not the first: `missing_legs`, each with `market_id` (the leg as given), `label` (a short +readable name for its market, never its outcome), `code`, `mapped_id` (the venue's matched market, +where there is one) and `detail` (what matching or the venue said). The venue's own `code` is the +first of its legs' codes in this order, and `reason` says them all, one clause per code: + +| `code` | Meaning | `reason` | +|---|---|---| +| `missing_leg` | hosted matching lists no such market on the venue | `no match for “Al Fayha to win”, “U.S. invades Iran”` | +| `matching_error` | the matching service could not be asked | `matching unavailable for “Al Fayha to win”` | +| `ineligible_leg` | matched, but not on the venue's combo-eligible list | `not combo-eligible on Polymarket US: “U.S. invades Iran”` | +| `outside_collections` | matched, but its event is in none of the venue's open combo collections, or no one collection takes the legs together (Kalshi) | `Kalshi combos are sports and cross-category only: “U.S. invades Iran” is in none` | + +A label is the leg's pick on serve's warm board (a moneyline by its participant's short name and +`to win`, a spread by its line, a total by its game and side), else its market's question from the +catalogue, shortened (`Will the U.S. invade Iran before 2027?` -> `U.S. invades Iran`), else its +`info.label`, else its market id. Matching is checked for every leg before eligibility, so a venue +that lacks a match for some legs reports those, and eligibility is reported when every leg matched. + ### The combo board `GET /trading/combos/board?venue=` turns a combo venue's eligibility into what a combo builder draws -(`synpath/engine/combo_board.py`): `ComboBoard` `{venue, count, complete, as_of, categories, events}`. +(`synpath/engine/combo_board.py`): `ComboBoard` `{venue, count, complete, as_of, status, built_at, +error, categories, events}`, and on a page `category`, `total` and `next_cursor`. + +- **Kept warm** (`synpath/server/combo_warmer.py`). serve builds every combo venue's board at + start-up, through the venue's whole list (no page cap: Polymarket's catalogue is about 214,000 + rows in 214 pages, with their Gamma markets and events; every event of Kalshi's open + collections), sleeping `pause_s` between two venue reads, and builds it again + `refresh_interval_s` after the last build ended + (`[combo_board]` in the configuration, [engine](engine.md)). A request never waits on a build: + until the first build ends the board is `status: "building"` with no events, answered at once; + after that it is the last good board (`status: "ready"`, `built_at`), kept while a refresh runs + or fails (`error` says why). A first build serves its board so far (`complete: false`) after its + first 5 pages (the venue lists by volume) and every 40 after. Gamma's markets (without their + prices) and the games' events are kept between builds, so a refresh reads Gamma only for markets + it has not seen and takes the catalogue's prices. A bare `create_trading_app` (no warmer) or `warm = false` builds on + request instead, 5 pages, kept 30 s. +- **Beside trading.** A whole-catalogue build is minutes of CPU work in serve's own process, so + it is kept off the event loop that answers health checks and order routes: its steps run in + worker threads that hand back the GIL every 100 items, the GIL switch interval is 0.5 ms while + boards are warmed, and the board's objects are frozen out of the cyclic collector (`gc.freeze`) + as they are built. Measured on the whole Polymarket catalogue: requests stay under 0.8 s during + a build (before: 1-6 s, enough for a terminal to report the engine offline). +- **In parts.** `GET /trading/combos/board/categories?venue=` is the rail: `categories` with counts, + `count` (markets read), `events`, `status`, `built_at`, `complete`, a few hundred bytes. `GET + /trading/combos/board?venue=&category=&cursor=&limit=&q=&sort=` is one page of one category + (`all`: every one): `limit` events (default 50, at most 500) from `cursor` (the last page's + `next_cursor`), those whose title, category or participants hold every word of `q`, in board + order (`sort=board`, by start) or by `volume` or `start` within each category; `total` counts the + category's matching events, and `categories` are the whole board's. Without any of these the + answer is the whole board, as before. - **Events**: a `game` (`rows`: one per participant, `name`, `abbr` and `picks` by `moneyline`, `spread`, `total`, each a `BoardPick` or null where the venue lists no such market) or a @@ -557,14 +610,15 @@ list; accepting one quote cancels the rest - Everything comes from the venue's own fields, never a URL slug or title wording; a participant the venue gives no abbreviation shows its name. - **Kalshi**: the events of its open collections (`associated_event_tickers`), their markets read - from the server's catalogue (`fetch_events_by_ids`, at most 150 events). The type is the event's + from the server's catalogue (`fetch_events_by_ids`, in batches of 100; every event when warm, at + most 150 on request). The type is the event's `product_metadata.competition_scope` (`Game`, `Spread`, `Point Total`; any other scope, such as `1st Half Winner` or `Team Total`, is a question); a game joins the events that share the `sub_title` (`NE vs CHI (Oct 22)`) within one `competition`. Participants are the winner markets' `yes_sub_title` (`Tie` is the draw), identified by `custom_strike` (the team's id, which a spread market names too); codes are the `sub_title`'s, side by side with the title's names; a spread's line is its `floor_strike` (`CHI -2.5` for "Chicago wins by over 2.5"), a total's too. -- **Polymarket**: its catalogue rows (up to `pages` pages), with Gamma's market for each (from the +- **Polymarket**: its catalogue rows (every page when warm; `pages` on request), with Gamma's market for each (from the server's catalogue: `sportsMarketType` `moneyline` / `spreads` / `totals`, `line`, `groupItemTitle`, its event) and the games' Gamma events (`fetch_events_by_ids`: `teams` with `abbreviation`). A two-sided market backs its outcomes' teams (YES the first; a spread at `line` diff --git a/synpath/engine/combo_board.py b/synpath/engine/combo_board.py index deaf0fe..096fe06 100644 --- a/synpath/engine/combo_board.py +++ b/synpath/engine/combo_board.py @@ -31,9 +31,13 @@ """ from __future__ import annotations +import asyncio +import gc +import inspect import re +import time from decimal import Decimal -from typing import Any, Iterable, Literal +from typing import Any, Awaitable, Callable, Iterable, Literal from pydantic import Field @@ -94,6 +98,8 @@ class BoardRow(_Base): name: str abbr: str | None = None + alias: str | None = Field(default=None, description="The venue's short name for the participant, where it has " + "one (Gamma's team `alias`: `Al Fayha`)") picks: dict[str, BoardPick | None] = Field(default_factory=dict, description="By `moneyline`, `spread`, `total`") @@ -123,8 +129,12 @@ class BoardCategory(_Base): count: int = Field(description="Events on the board in this category") +BoardStatus = Literal["building", "ready"] + + class ComboBoard(_Base): - """A venue's combo-eligible markets, grouped for a builder.""" + """A venue's combo-eligible markets, grouped for a builder: the whole + board, or one page of one category of it (`category`, `next_cursor`).""" venue: str count: int = Field(description="Combo-eligible markets read") @@ -132,6 +142,27 @@ class ComboBoard(_Base): categories: list[BoardCategory] = Field(default_factory=list, description="By count, sports first") complete: bool = Field(default=True, description="Every page of the venue's list was read") as_of: int | None = None + status: BoardStatus = Field(default="ready", description="`building` while serve's first build of this venue's " + "board runs (no events yet), else `ready`") + built_at: int | None = Field(default=None, description="When the board served was built, ms") + category: str | None = Field(default=None, description="A page: the category it is of (`all` for every one)") + total: int | None = Field(default=None, description="A page: the events in its category (and search)") + next_cursor: str | None = Field(default=None, description="A page: the cursor of the next one; null at the end") + error: str | None = Field(default=None, description="Why serve's last build of this board failed, while the " + "board served is an older one (or none yet)") + + +class ComboBoardCategories(_Base): + """A board's rail: its categories with counts, without the events.""" + + venue: str + status: BoardStatus = "ready" + built_at: int | None = None + complete: bool = True + count: int = Field(default=0, description="Combo-eligible markets read") + events: int = Field(default=0, description="Events on the board") + categories: list[BoardCategory] = Field(default_factory=list, description="By count, sports first") + error: str | None = None # --------------------------------------------------------------------------- @@ -145,6 +176,7 @@ class Side(_Base): "one, else its name") name: str abbr: str | None = Field(default=None, description="The venue's own abbreviation, where it has one") + alias: str | None = Field(default=None, description="The venue's own short name, where it has one") draw: bool = False @@ -175,6 +207,20 @@ class BoardMarket(_Base): volume: Decimal | None = None +# The builders below run in worker threads beside serve's event loop +# (`board_for`). Under the GIL a thread building the whole catalogue starves +# the loop for seconds: health checks and order routes waited 1-6 s. Sleeping +# 1 ms every `LOCK_SHARE_EVERY` items hands the loop the lock on a fixed beat, +# at the cost of a slower background build. +LOCK_SHARE_EVERY = 100 +LOCK_SHARE_S = 0.001 + + +def share_the_lock(items_done: int) -> None: + if items_done % LOCK_SHARE_EVERY == 0: + time.sleep(LOCK_SHARE_S) + + def board_of(venue: str, markets: Iterable[BoardMarket], *, complete: bool = True, as_of: int | None = None) -> ComboBoard: """Games and questions from a venue's eligible markets (already @@ -185,7 +231,8 @@ def board_of(venue: str, markets: Iterable[BoardMarket], *, complete: bool = Tru for market in rows: by_event.setdefault(market.event_key, []).append(market) events: list[BoardEvent] = [] - for key, members in by_event.items(): + for events_done, (key, members) in enumerate(by_event.items(), 1): + share_the_lock(events_done) game = game_of(venue, key, members) if game is not None: events.append(game) @@ -220,7 +267,7 @@ def game_of(venue: str, key: str, members: list[BoardMarket]) -> BoardEvent | No return None first = members[0] category, label, group = category_of([t for m in members for t in m.tags], first.series) - rows = [BoardRow(name=s.name, abbr=s.abbr) for s in sides] + rows = [BoardRow(name=s.name, abbr=s.abbr, alias=s.alias) for s in sides] used: list[str] = [] for kind in KINDS: for row, pick in zip(rows, picks_for(kind, [m for m in typed if m.kind == kind], sides)): @@ -354,7 +401,8 @@ def polymarket_markets(rows: Iterable[Any], details: dict[str, dict[str, Any]] | Without Gamma's market a row is a question: the catalogue names no type.""" details, events = details or {}, events or {} out = [] - for row in rows: + for rows_done, row in enumerate(rows, 1): + share_the_lock(rows_done) info = dict(getattr(row, "info", None) or {}) native = str(row.market_id).partition(":")[2] or str(row.market_id) gamma = details.get(native) or {} @@ -388,7 +436,8 @@ def polymarket_us_markets(rows: Iterable[Any]) -> list[BoardMarket]: `sportsMarketTypeV2`, each outcome's participant from `marketSides[].team` (`name`, `displayAbbreviation`), the event's `teams` in order.""" out = [] - for row in rows: + for rows_done, row in enumerate(rows, 1): + share_the_lock(rows_done) info = dict(getattr(row, "info", None) or {}) event = info.get("event") if isinstance(info.get("event"), dict) else {} teams = teams_of(event.get("teams") or [], abbr_field="displayAbbreviation") @@ -437,7 +486,8 @@ def kalshi_markets(events: Iterable[Event]) -> list[BoardMarket]: `sub_title`'s, side by side with the title's names (`New England vs Chicago`); a spread's and a total's line is the market's `floor_strike`.""" out = [] - for event in events: + for events_done, event in enumerate(events, 1): + share_the_lock(events_done) raw = dict(event.info or {}) meta = raw.get("product_metadata") or {} kind = KALSHI_SCOPES.get(str(meta.get("competition_scope") or "").strip().lower()) @@ -520,8 +570,9 @@ def teams_of(raw: list[Any], *, abbr_field: str = "abbreviation") -> list[Side]: if not isinstance(team, dict) or not team.get("name"): continue abbr = team.get(abbr_field) or team.get("abbreviation") or None + alias = str(team.get("alias") or "").strip() or None out.append(Side(key=f"team:{team.get('id') or team['name']}", name=str(team["name"]), - abbr=str(abbr).upper() if abbr else None)) + abbr=str(abbr).upper() if abbr else None, alias=alias if alias != team["name"] else None)) return out @@ -661,78 +712,375 @@ def millis(value: Any) -> int | None: return None +# --------------------------------------------------------------------------- +# Leg labels: a short name for each market on a board +# --------------------------------------------------------------------------- + +LABEL_LENGTH = 60 +"""A label longer than this is cut, with an ellipsis.""" + + +def market_labels(board: ComboBoard) -> dict[str, str]: + """A short readable name for every market on the board, for a person to + recognise a leg by (`Al Fayha to win`, `Patriots -2.5`, `Patriots vs. + Bears over 41.5`, `U.S. invades Iran`): keyed by market id, and by + `market_id|outcome` where the two outcomes back different participants + (a two-sided moneyline or spread, a total). `label_of` reads it.""" + labels: dict[str, str] = {} + for events_done, event in enumerate(board.events, 1): + share_the_lock(events_done) + if event.kind == "game": + game_labels(event, labels) + else: + for pick in (event.yes, event.no): + if pick is not None: + labels.setdefault(pick.market_id, short_question(pick.title or event.title)) + return labels + + +def game_labels(event: BoardEvent, labels: dict[str, str]) -> None: + """A game's picks: the participant (by its short name) and what it backs.""" + for row in event.rows: + name = row.alias or row.name + for kind, pick in row.picks.items(): + if pick is None: + continue + if kind == "moneyline": + text = f"Draw in {event.title}" if name == "Draw" else f"{name} to win" + elif kind == "spread": + # The button says `CHI -2.5`: the signed line is its last word. + text = f"{name} {pick.label.rsplit(' ', 1)[-1]}" if " " in pick.label else name + else: + text = f"{event.title} {'over' if pick.outcome == 'yes' else 'under'} {pick.line or ''}".strip() + labels[f"{pick.market_id}|{pick.outcome}"] = text + two_sided = kind == "total" or any( + other is not row and (other.picks.get(kind) or None) is not None + and other.picks[kind].market_id == pick.market_id for other in event.rows) # type: ignore[union-attr] + labels.setdefault(pick.market_id, f"{event.title} {kind}" if two_sided else text) + + +def label_of(labels: dict[str, str], market_id: str, outcome: str | None = None) -> str | None: + """A leg's label from `market_labels`: its outcome's where the outcomes + differ, else its market's.""" + return (labels.get(f"{market_id}|{outcome}") if outcome else None) or labels.get(market_id) + + +_WHEN = re.compile( + r"\s+(?:on|by|before|after|in|during|at|through|until|ending)\s+(?:the\s+end\s+of\s+)?" + r"(?:\d{4}-\d{2}-\d{2}|(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\.?\s+\d{1,2}(?:st|nd|rd|th)?" + r"(?:,?\s+\d{4})?|(?:q[1-4]\s+)?\d{4}|today|tonight|tomorrow|this\s+(?:week|month|year))$", re.I) +_IRREGULAR = {"be": "is", "have": "has", "do": "does", "go": "goes"} + + +def short_question(title: str, *, length: int = LABEL_LENGTH) -> str: + """A market's question as a short statement: `Will the U.S. invade Iran + before 2027?` -> `U.S. invades Iran`; `Will Al Fayha Saudi Club win on + 2026-10-22?` -> `Al Fayha Saudi Club wins`. The question's own words, + without its `Will`, its question mark and a trailing date; anything else + is kept as it is, cut to `length`.""" + text = " ".join(str(title or "").split()).rstrip("?").strip() + text = _WHEN.sub("", text).strip() + if text.lower().startswith("will "): + words = text[5:].split(" ") + if len(words) > 1 and words[0].lower() == "the" and words[1][:1].isupper(): + words = words[1:] + for at, word in enumerate(words): + if at and word.isalpha() and word.islower(): + if word != "not": + words[at] = third_person(word) + break + text = " ".join(words) + if len(text) > length: + text = text[:length - 1].rstrip() + "…" + return text or str(title) + + +def third_person(verb: str) -> str: + """`win` -> `wins`, `reach` -> `reaches`, `carry` -> `carries`, `be` -> `is`.""" + if verb in _IRREGULAR: + return _IRREGULAR[verb] + if verb.endswith(("s", "sh", "ch", "x", "z", "o")): + return verb + "es" + if verb.endswith("y") and len(verb) > 1 and verb[-2] not in "aeiou": + return verb[:-1] + "ies" + return verb + "s" + + +# --------------------------------------------------------------------------- +# Serving a board in parts: the rail, and one category a page at a time +# --------------------------------------------------------------------------- + +PAGE_LIMIT = 50 +MAX_PAGE_LIMIT = 500 +BoardSort = Literal["board", "volume", "start"] + + +def categories_view(board: ComboBoard) -> ComboBoardCategories: + return ComboBoardCategories(venue=board.venue, status=board.status, built_at=board.built_at, + complete=board.complete, count=board.count, events=len(board.events), + categories=board.categories, error=board.error) + + +def board_page(board: ComboBoard, *, category: str | None = None, cursor: str | None = None, + limit: int = PAGE_LIMIT, query: str | None = None, sort: BoardSort = "board") -> ComboBoard: + """One page of the board: the events of `category` (`all` or none: every + one) whose title, category or participants hold every word of `query`, + in board order (sports first, the most events first, by start), or by + `volume` (largest first) or `start` within each category; from `cursor` + (an offset from the last page), `limit` events. The categories are the + whole board's, for the rail.""" + events = [e for e in board.events if category in (None, "", "all") or e.category == category] + words = [w for w in (query or "").lower().split() if w] + if words: + events = [e for e in events if all(w in searchable(e) for w in words)] + if sort != "board": + order = {c.key: index for index, c in enumerate(board.categories)} + + def key(e: BoardEvent) -> tuple: + within = -(e.volume or Decimal(0)) if sort == "volume" else Decimal(e.start_time or 2**62) + return (order.get(e.category, len(order)), within, e.title) + events = sorted(events, key=key) + start = page_offset(cursor) + limit = max(1, min(int(limit), MAX_PAGE_LIMIT)) + page = events[start:start + limit] + more = start + limit < len(events) + return board.model_copy(update={"events": page, "category": category or "all", "total": len(events), + "next_cursor": str(start + limit) if more else None}) + + +def searchable(event: BoardEvent) -> str: + return " ".join([event.title, event.category_label, + *(f"{r.name} {r.abbr or ''} {r.alias or ''}" for r in event.rows)]).lower() + + +def page_offset(cursor: str | None) -> int: + if not cursor: + return 0 + try: + return max(0, int(cursor)) + except ValueError: + from ..errors import BadRequest + raise BadRequest(f"not a cursor of this board: {cursor!r}") from None + + # --------------------------------------------------------------------------- # Reading a venue # --------------------------------------------------------------------------- BOARD_PAGES = 5 -"""Pages of a venue's eligible list (or of Kalshi's collections) a board reads.""" +"""Pages of a venue's eligible list (or of Kalshi's collections) a board read +on request reads. serve's warmer (`synpath.server.combo_warmer`) reads every +page (`pages=None`).""" BOARD_EVENTS = 150 -"""Kalshi: the most collection events whose markets a board reads.""" -PRICE_BATCH = 50 +"""Kalshi: the most collection events whose markets a board read on request +reads (`max_events=None`: every one).""" +PRICE_BATCH = 100 +"""Gamma's largest batch.""" +KALSHI_EVENT_BATCH = 100 +EVENT_BATCH = 20 + + +FIRST_PUBLISH_PAGES = 5 +"""A whole-list build with `progress` publishes its board after this many +pages (the venue lists by volume, so the busiest markets come first)...""" +PUBLISH_EVERY_PAGES = 40 +"""...and again every this many pages, until the list ends.""" + +class BoardCache: + """What a board keeps between builds of one venue: Polymarket's Gamma + markets (their type, line, label and event: fields that do not change; + prices are left out, so each build takes the catalogue's own) and the + games' events (their teams). A rebuild reads Gamma only for markets it + has not seen; what a build no longer lists is dropped.""" -async def board_for(venue: str, adapter: Any, reader: Any | None = None, *, pages: int = BOARD_PAGES, - max_events: int = BOARD_EVENTS) -> ComboBoard: + def __init__(self) -> None: + self.markets: dict[str, dict[str, Any]] = {} + self.events: dict[str, dict[str, Any]] = {} + + def prune(self, market_natives: set[str], event_ids: set[str]) -> None: + self.markets = {k: v for k, v in self.markets.items() if k in market_natives} + self.events = {k: v for k, v in self.events.items() if k in event_ids} + + +GAMMA_FIELDS = ("question", "sportsMarketType", "line", "groupItemTitle", "gameStartTime", "endDate", "outcomes") +GAMMA_EVENT_FIELDS = ("id", "title", "seriesSlug", "startTime", "teams") + + +def static_gamma(info: dict[str, Any]) -> dict[str, Any]: + """A Gamma market's fields a board reads that do not move with the market.""" + out = {k: info[k] for k in GAMMA_FIELDS if info.get(k) is not None} + events = info.get("events") + if isinstance(events, list) and events and isinstance(events[0], dict): + out["events"] = [{k: events[0][k] for k in GAMMA_EVENT_FIELDS if events[0].get(k) is not None}] + return out + + +async def board_for(venue: str, adapter: Any, reader: Any | None = None, *, pages: int | None = BOARD_PAGES, + max_events: int | None = BOARD_EVENTS, pause_s: float = 0.0, cache: BoardCache | None = None, + progress: Callable[[ComboBoard], None | Awaitable[None]] | None = None) -> ComboBoard: """The board of one combo venue. `adapter` is the engine's trading adapter (its eligibility: Kalshi's collections, or the venue's eligible list); `reader` is the server's read adapter for the venue, where there is one: Kalshi's event markets come from it (`fetch_events_by_ids`); Polymarket's rows get its Gamma markets (types, lines, labels, quotes) and - the games' events (teams); Polymarket US's rows carry their own.""" - import asyncio + the games' events (teams); Polymarket US's rows carry their own. + + `pages` (None: until the venue's list ends) and `max_events` (None: no + limit) bound the read; `pause_s` is slept between two venue reads, so a + whole-catalogue build stays polite to the venue. `cache` keeps + Polymarket's Gamma reads between builds (`BoardCache`). `progress`, where + given, is handed the board so far (`complete: false`) after the first + `FIRST_PUBLISH_PAGES` pages of a listed venue and every + `PUBLISH_EVERY_PAGES` after: Polymarket lists about 214,000 markets. + + The CPU-heavy steps (rows into board markets, markets into the board) run + in worker threads: on the whole catalogue they take seconds, and serve's + event loop must keep answering the terminal's health checks meanwhile.""" import time from .rfq_group import lists_combo_markets now = int(time.time() * 1000) if callable(getattr(adapter, "fetch_combo_collections", None)): - tickers, complete = await collection_events(adapter, pages) + tickers, complete = await collection_events(adapter, pages, pause_s) fetch = getattr(reader, "fetch_events_by_ids", None) if fetch is None or not tickers: return ComboBoard(venue=venue, count=0, complete=complete and not tickers, as_of=now) - events = await asyncio.to_thread(fetch, [f"{venue}:{t}" for t in tickers[:max_events]]) - return board_of(venue, kalshi_markets(events), complete=complete and len(tickers) <= max_events, as_of=now) + wanted = tickers if max_events is None else tickers[:max_events] + events = await kalshi_events_of(venue, wanted, fetch, pause_s) + markets = await asyncio.to_thread(kalshi_markets, events) + return await asyncio.to_thread(board_of, venue, markets, complete=complete and len(wanted) == len(tickers), + as_of=now) if not lists_combo_markets(adapter): return ComboBoard(venue=venue, count=0, complete=True, as_of=now) - rows: list[Any] = [] + return await listed_board(venue, adapter, reader, pages=pages, pause_s=pause_s, cache=cache, progress=progress, + now=now) + + +async def listed_board(venue: str, adapter: Any, reader: Any | None, *, pages: int | None, pause_s: float, + cache: BoardCache | None, progress: Callable[[ComboBoard], None | Awaitable[None]] | None, + now: int) -> ComboBoard: + """A listed venue's board (Polymarket, Polymarket US): its eligible list + page by page, each chunk of rows turned into board markets as it comes, + the board so far handed to `progress` along the way.""" + cache = cache if cache is not None else BoardCache() + markets: list[BoardMarket] = [] + seen: set[str] = set() + pending: list[Any] = [] cursor: str | None = None + read = 0 complete = False - for _ in range(pages): + while pages is None or read < pages: + if read: + await pause(pause_s) page = await adapter.fetch_combo_eligible_markets(cursor=cursor) - rows.extend(page) + read += 1 + for row in page: + if row.market_id not in seen: + seen.add(row.market_id) + pending.append(row) cursor = page.next_cursor if not cursor: complete = True break - if venue == "polymarket_us" or all(getattr(r, "venue", "") == "polymarket_us" for r in rows if rows): - markets = polymarket_us_markets(rows) - if reader is not None and callable(getattr(reader, "fetch_markets_by_ids", None)): - await with_catalogue_quotes(markets, reader) - return board_of(venue, markets, complete=complete, as_of=now) - details: dict[str, dict[str, Any]] = {} + if progress is not None and (read == FIRST_PUBLISH_PAGES or read % PUBLISH_EVERY_PAGES == 0): + markets += await listed_markets_of(venue, pending, reader, pause_s, cache) + pending = [] + # The board so far is long-lived: keep the cyclic collector from + # rescanning it on every full collection (`share_the_lock`). + gc.freeze() + published = progress(await asyncio.to_thread(board_of, venue, markets, complete=False, as_of=now)) + if inspect.isawaitable(published): + await published + markets += await listed_markets_of(venue, pending, reader, pause_s, cache) + if complete: + cache.prune({m.market_id.partition(":")[2] for m in markets}, + {m.event_key.partition(":")[2] for m in markets if m.event_key.startswith("polymarket-event:")}) + return await asyncio.to_thread(board_of, venue, markets, complete=complete, as_of=now) + + +async def listed_markets_of(venue: str, rows: list[Any], reader: Any | None, pause_s: float, + cache: BoardCache) -> list[BoardMarket]: + """Board markets for a chunk of a listed venue's rows: Polymarket US's + carry their own fields; Polymarket's take Gamma's market (from `cache`, + else read now, with its prices) and their game's event (teams).""" + if not rows: + return [] + can_read = reader is not None and callable(getattr(reader, "fetch_markets_by_ids", None)) + if venue == "polymarket_us" or all(getattr(r, "venue", "") == "polymarket_us" for r in rows): + markets = await asyncio.to_thread(polymarket_us_markets, rows) + if can_read: + await with_catalogue_quotes(markets, reader, pause_s) + return markets + natives, unseen = await asyncio.to_thread(natives_not_cached, rows, cache) quotes: dict[str, Market] = {} - if reader is not None and callable(getattr(reader, "fetch_markets_by_ids", None)): - quotes = await catalogue_markets([r.market_id for r in rows], reader) - details = {m.venue_market_id: dict(m.info or {}) for m in quotes.values()} - events: dict[str, dict[str, Any]] = {} + if can_read and unseen: + quotes = await catalogue_markets(unseen, reader, pause_s) + details = await asyncio.to_thread(remember_gamma_markets, natives, unseen if can_read else [], quotes, cache) fetch_events = getattr(reader, "fetch_events_by_ids", None) if callable(fetch_events): - wanted = list(dict.fromkeys( - str((d.get("events") or [{}])[0].get("id")) for d in details.values() - if POLYMARKET_KINDS.get(str(d.get("sportsMarketType") or "")) and d.get("events"))) - events = await game_events(wanted, fetch_events) - markets = polymarket_markets(rows, details, events) - apply_quotes(markets, quotes) - return board_of(venue, markets, complete=complete, as_of=now) + missing = await asyncio.to_thread(game_events_not_cached, details, cache) + for event_id, info in (await game_events(missing, fetch_events, pause_s)).items(): + cache.events[event_id] = {"teams": info.get("teams") or []} + markets = await asyncio.to_thread(polymarket_markets, rows, details, cache.events) + await asyncio.to_thread(apply_quotes, markets, quotes) + return markets + + +# The bookkeeping of one chunk of Polymarket rows (about 40,000 markets), kept +# off serve's event loop like the builders above (`share_the_lock`). + +def natives_not_cached(rows: list[Any], cache: BoardCache) -> tuple[dict[str, str], list[str]]: + """Each row's native Gamma id, by market id, and the market ids whose + Gamma market the cache does not hold yet.""" + natives: dict[str, str] = {} + for rows_done, row in enumerate(rows, 1): + share_the_lock(rows_done) + natives[row.market_id] = str(row.market_id).partition(":")[2] or str(row.market_id) + return natives, [m for m, n in natives.items() if n not in cache.markets] -async def collection_events(adapter: Any, pages: int) -> tuple[list[str], bool]: +def remember_gamma_markets(natives: dict[str, str], unseen: list[str], quotes: dict[str, Market], + cache: BoardCache) -> dict[str, dict[str, Any]]: + """Gamma's market for each row, by native id: the one just read (with + its prices), else the cache's. What was just read goes into the cache; + a market Gamma did not serve is cached empty, so it is not asked for again.""" + for market_id in unseen: + cache.markets.setdefault(natives[market_id], {}) + fresh: dict[str, dict[str, Any]] = {} + for quotes_done, market in enumerate(quotes.values(), 1): + share_the_lock(quotes_done) + info = dict(market.info or {}) + fresh[market.venue_market_id] = info + cache.markets[market.venue_market_id] = static_gamma(info) + return {n: fresh.get(n) or cache.markets[n] for n in natives.values() if n in fresh or n in cache.markets} + + +def game_events_not_cached(details: dict[str, dict[str, Any]], cache: BoardCache) -> list[str]: + """The Gamma event ids of the chunk's games whose teams the cache lacks.""" + wanted = list(dict.fromkeys( + str((d.get("events") or [{}])[0].get("id")) for d in details.values() + if POLYMARKET_KINDS.get(str(d.get("sportsMarketType") or "")) and d.get("events"))) + return [e for e in wanted if e not in cache.events] + + +async def pause(seconds: float) -> None: + if seconds > 0: + import asyncio + await asyncio.sleep(seconds) + + +async def collection_events(adapter: Any, pages: int | None, pause_s: float = 0.0) -> tuple[list[str], bool]: """Every event ticker an open collection takes, in the venue's order.""" tickers: list[str] = [] cursor: str | None = None - for _ in range(pages): + read = 0 + while pages is None or read < pages: + if read: + await pause(pause_s) page = await adapter.fetch_combo_collections(status="open", limit=200, cursor=cursor) + read += 1 for row in page: tickers.extend(str(t) for t in row.get("associated_event_tickers") or []) cursor = page.next_cursor @@ -741,12 +1089,26 @@ async def collection_events(adapter: Any, pages: int) -> tuple[list[str], bool]: return list(dict.fromkeys(tickers)), False -async def catalogue_markets(wanted: list[str], reader: Any) -> dict[str, Market]: +async def kalshi_events_of(venue: str, tickers: list[str], fetch: Any, pause_s: float) -> list[Event]: + """Kalshi's collection events with their markets, in batches.""" + import asyncio + + out: list[Event] = [] + for start in range(0, len(tickers), KALSHI_EVENT_BATCH): + if start: + await pause(pause_s) + out.extend(await asyncio.to_thread(fetch, [f"{venue}:{t}" for t in tickers[start:start + KALSHI_EVENT_BATCH]])) + return out + + +async def catalogue_markets(wanted: list[str], reader: Any, pause_s: float = 0.0) -> dict[str, Market]: """The catalogue's markets for these ids, best effort, by id.""" import asyncio found: dict[str, Market] = {} for start in range(0, len(wanted), PRICE_BATCH): + if start: + await pause(pause_s) try: batch = await asyncio.to_thread(reader.fetch_markets_by_ids, wanted[start:start + PRICE_BATCH]) except Exception: @@ -755,12 +1117,14 @@ async def catalogue_markets(wanted: list[str], reader: Any) -> dict[str, Market] return found -async def game_events(event_ids: list[str], fetch: Any) -> dict[str, dict[str, Any]]: +async def game_events(event_ids: list[str], fetch: Any, pause_s: float = 0.0) -> dict[str, dict[str, Any]]: """Polymarket's game events (their `teams`), by event id, best effort.""" import asyncio out: dict[str, dict[str, Any]] = {} for start in range(0, len(event_ids), EVENT_BATCH): + if start: + await pause(pause_s) try: batch = await asyncio.to_thread(fetch, [f"polymarket:{e}" for e in event_ids[start:start + EVENT_BATCH]]) except Exception: @@ -770,9 +1134,6 @@ async def game_events(event_ids: list[str], fetch: Any) -> dict[str, dict[str, A return out -EVENT_BATCH = 20 - - def apply_quotes(markets: list[BoardMarket], quotes: dict[str, Market]) -> None: for market in markets: held = quotes.get(market.market_id) @@ -783,6 +1144,6 @@ def apply_quotes(markets: list[BoardMarket], quotes: dict[str, Market]) -> None: market.yes_bid, market.yes_ask = to_decimal(quote.bid), to_decimal(quote.ask) -async def with_catalogue_quotes(markets: list[BoardMarket], reader: Any) -> None: +async def with_catalogue_quotes(markets: list[BoardMarket], reader: Any, pause_s: float = 0.0) -> None: """Replace listed prices with the catalogue's quotes, best effort.""" - apply_quotes(markets, await catalogue_markets([m.market_id for m in markets], reader)) + apply_quotes(markets, await catalogue_markets([m.market_id for m in markets], reader, pause_s)) diff --git a/synpath/engine/rfq_group.py b/synpath/engine/rfq_group.py index a6367bf..09200b0 100644 --- a/synpath/engine/rfq_group.py +++ b/synpath/engine/rfq_group.py @@ -96,6 +96,20 @@ """How long a matching answer (a listing, or none) is reused: a preview and the group it leads to ask the same legs within seconds.""" +Labeler = Callable[[list[ComboLeg]], Awaitable[dict[str, str]]] +"""`labeler(legs)` -> a short readable name for each leg's market, by market +id (`Al Fayha to win`, `U.S. invades Iran`). serve's combo board warmer is +one (`synpath.server.combo_warmer`); without one a leg is named by its +`info["label"]`, else its market id.""" + +VENUE_NAMES = {"kalshi": "Kalshi", "polymarket": "Polymarket", "polymarket_us": "Polymarket US", + "limitless": "Limitless", "opinion": "Opinion", "predict_fun": "predict.fun", "sx_bet": "SX Bet"} +COLLECTION_SCOPES = {"kalshi": "Kalshi combos are sports and cross-category only"} +"""What a collection venue's combos take, said when a leg's event is in none +of its open collections.""" +LEG_CODES = ("missing_leg", "matching_error", "ineligible_leg", "outside_collections") +"""Why a venue lacks a leg, in the order a skip reason names them.""" + Matcher = Callable[[str, str], Awaitable[MarketLink | None]] """`matcher(anchor, venue)` -> the other venue's listing of the same proposition, or `None` when it has none. Raises `MarketNotFound` when the @@ -179,15 +193,34 @@ def key(self) -> str: return f"{self.venue}:{self.rfq_id}" +class SkippedLeg(_Base): + """One leg a venue lacks, and why.""" + + market_id: str = Field(description="The leg, as given") + label: str = Field(description="A short readable name for the leg's market (`Al Fayha to win`, `U.S. invades " + "Iran`), never just its outcome") + code: str = Field(description="`missing_leg`: no matching market there; `matching_error`: the matching service " + "could not be asked; `ineligible_leg`: matched, but the venue's combos do not take " + "that market; `outside_collections`: matched, but its event is in none of the " + "venue's open combo collections (Kalshi)") + mapped_id: str | None = Field(default=None, description="The venue's matched market, where there is one") + detail: str | None = Field(default=None, description="What matching or the venue said") + + class SkippedVenue(_Base): """A venue the group does not ask, and why.""" venue: str code: str = Field(description="`excluded`, `forbidden`, `not_configured`, `no_rfq`, `no_combos`, `no_match`, " - "`matching_error`, `missing_leg`, `ineligible_leg`, `duplicate_leg`, " - "`one_sided_size`, `member_disabled`") - reason: str - leg: str | None = Field(default=None, description="The leg (as given) that could not be placed here") + "`matching_error`, `missing_leg`, `ineligible_leg`, `outside_collections`, " + "`duplicate_leg`, `one_sided_size`, `member_disabled`. A combo venue lacking legs " + "has the code of the first reason in `missing_legs`, in that order") + reason: str = Field(description="For a person: a combo venue's names every leg it lacks (`no match for “Al " + "Fayha to win”, “U.S. invades Iran”`)") + leg: str | None = Field(default=None, description="The first leg (as given) that could not be placed here") + missing_legs: list[SkippedLeg] = Field(default_factory=list, description="A combo: every leg this venue lacks, " + "each with its label and reason") + detail: str | None = Field(default=None, description="What matching or the venue said, for a log") class GroupQuote(_Base): @@ -333,6 +366,8 @@ class RfqGroupDesk: def __init__(self, engine: "Engine", *, matcher: Matcher | None = None): self.engine = engine self.matcher: Matcher = matcher or hosted_matcher() + self.labeler: Labeler | None = None + """Names each leg's market in a skip reason (`Labeler`); set by serve.""" self.groups: dict[str, RfqGroup] = {} self._by_rfq: dict[str, str] = {} """`venue:rfq_id` -> the group it is a member of.""" @@ -663,7 +698,11 @@ async def _map_leg(self, leg: ComboLeg, venue: str) -> tuple[ComboLeg | None, st async def _plan_combo(self, group: RfqGroup, candidates: list[str], skipped: list[SkippedVenue]) -> list[GroupMember]: + """Each combo venue's legs (mapped, then checked eligible), or every + leg it lacks and why: no match, matching unavailable, matched but not + combo-eligible there, or outside its combo collections.""" legs = group.instrument.legs or [] + labels: dict[str, str] | None = None members = [] for venue in candidates: adapter = self.engine.adapters[venue] @@ -671,29 +710,45 @@ async def _plan_combo(self, group: RfqGroup, candidates: list[str], skipped.append(SkippedVenue(venue=venue, code="no_combos", reason=f"{venue} has no combos")) continue mapped: list[ComboLeg] = [] - missing: SkippedVenue | None = None + lacking: list[tuple[ComboLeg, str, str, str | None]] = [] for leg in legs: found, reason, code = await self._map_leg(leg, venue) if found is None: - missing = SkippedVenue(venue=venue, code=code, reason=reason, leg=leg.market_id) - break - mapped.append(found) - if missing is not None: - skipped.append(missing) - continue - if len({leg.market_id for leg in mapped}) != len(mapped): + lacking.append((leg, code, reason, None)) + else: + mapped.append(found) + if not lacking and len({leg.market_id for leg in mapped}) != len(mapped): skipped.append(SkippedVenue(venue=venue, code="duplicate_leg", reason=f"two legs are the same market on {venue}")) continue - try: - mapped, collection, how = await self._eligible(venue, adapter, mapped, legs) - except _Ineligible as exc: - skipped.append(SkippedVenue(venue=venue, code="ineligible_leg", reason=str(exc), leg=exc.leg)) + collection = how = None + together = False + if not lacking: + try: + mapped, collection, how = await self._eligible(venue, adapter, mapped, legs) + except _Ineligible as exc: + together = exc.together + lacking = exc.legs or [(next(g for g in legs if g.market_id == exc.leg) if exc.leg else legs[-1], + exc.code, str(exc), None)] + if lacking: + labels = labels if labels is not None else await self._labels(legs) + skipped.append(skipped_combo_venue(venue, lacking, labels, together=together)) continue members.append(GroupMember(venue=venue, via="combo", legs=mapped, collection_id=collection, eligibility=how)) return members + async def _labels(self, legs: list[ComboLeg]) -> dict[str, str]: + """A short name for each leg's market (`Labeler`), best effort.""" + found: dict[str, str] = {} + if self.labeler is not None: + try: + found = dict(await self.labeler(legs)) + except Exception as exc: # a label is for reading; a skip must still be reported + log.info("synpath.engine: labelling combo legs failed: %s", exc) + return {leg.market_id: found.get(leg.market_id) or str(leg.info.get("label") or "") or leg.market_id + for leg in legs} + # -- eligibility ----------------------------------------------------------- async def _eligible(self, venue: str, adapter: Any, legs: list[ComboLeg], @@ -741,13 +796,19 @@ async def _eligible_collections(self, venue: str, adapter: Any, legs: list[Combo fits.append(str(row.get("collection_ticker"))) if not fits: if not rows: - raise _Ineligible(f"no open collection on {venue} takes event {wanted[0]}", leg=given[0].market_id) - for leg, mine, event in zip(given, legs, events): - if event not in seen: - raise _Ineligible(f"event {event} ({mine.market_id}) is in no open collection on {venue} with " - f"event {wanted[0]}", leg=leg.market_id) - raise _Ineligible(f"no open collection on {venue} takes these {len(legs)} legs together " - f"(events {', '.join(wanted)})", leg=given[-1].market_id) + raise _Ineligible(f"no open collection on {venue} takes event {wanted[0]}", leg=given[0].market_id, + code="outside_collections", + legs=[(given[0], "outside_collections", + f"event {wanted[0]} is in no open collection on {venue}", legs[0].market_id)]) + outside = [(leg, "outside_collections", f"event {event} is in no open collection on {venue} with event " + f"{wanted[0]}", mine.market_id) + for leg, mine, event in zip(given, legs, events) if event not in seen] + if outside: + raise _Ineligible(outside[0][2], leg=outside[0][0].market_id, code="outside_collections", legs=outside) + together = f"no open collection on {venue} takes these {len(legs)} legs together (events {', '.join(wanted)})" + raise _Ineligible(together, leg=given[-1].market_id, code="outside_collections", + legs=[(leg, "outside_collections", together, mine.market_id) + for leg, mine in zip(given, legs)], together=True) fits = sorted(set(fits)) how = f"collection {fits[0]}" + (f" (of {len(fits)}: {', '.join(fits)})" if len(fits) > 1 else "") return legs, fits[0], how @@ -775,11 +836,13 @@ async def _eligible_listed(self, venue: str, adapter: Any, legs: list[ComboLeg], break self._eligible_lists[venue] = (now, known, complete) out = [] + where = "is not on" if complete else f"is not in the first {ELIGIBLE_PAGES} pages of" + absent = [(original, "ineligible_leg", f"{leg.market_id} {where} {venue}'s combo-eligible list", leg.market_id) + for leg, original in zip(legs, given) if leg.market_id not in known] + if absent: + raise _Ineligible(absent[0][2], leg=absent[0][0].market_id, code="ineligible_leg", legs=absent) for leg, original in zip(legs, given): - row = known.get(leg.market_id) - if row is None: - where = "is not on" if complete else f"is not in the first {ELIGIBLE_PAGES} pages of" - raise _Ineligible(f"{leg.market_id} {where} {venue}'s combo-eligible list", leg=original.market_id) + row = known[leg.market_id] position = (row.position_ids or {}).get(leg.outcome) if position and not leg.info.get("position_id"): leg = leg.model_copy(update={"info": {**leg.info, "position_id": position}}) @@ -1387,9 +1450,56 @@ def find_quote(group: RfqGroup, venue: str, quote_id: str) -> GroupQuote | None: class _Ineligible(Exception): - def __init__(self, message: str, *, leg: str | None = None): + """The venue's combos do not take these legs: each `(given leg, code, + detail, the venue's market)`, the code `ineligible_leg` or + `outside_collections`; `together` when each is in a collection, but no + one collection takes them all.""" + + def __init__(self, message: str, *, leg: str | None = None, code: str = "ineligible_leg", + legs: list[tuple[ComboLeg, str, str, str | None]] | None = None, together: bool = False): super().__init__(message) - self.leg = leg + self.leg, self.code, self.legs, self.together = leg, code, list(legs or []), together + + +def skipped_combo_venue(venue: str, lacking: list[tuple[ComboLeg, str, str, str | None]], + labels: Mapping[str, str], *, together: bool = False) -> SkippedVenue: + """A combo venue skipped for the legs it lacks: each leg by its label, + with its reason, and one sentence naming them all.""" + missing = [SkippedLeg(market_id=leg.market_id, label=labels.get(leg.market_id) or leg.market_id, code=code, + detail=detail or None, mapped_id=mapped) + for leg, code, detail, mapped in lacking] + codes = [c for c in LEG_CODES if any(m.code == c for m in missing)] or [missing[0].code] + return SkippedVenue(venue=venue, code=codes[0], reason=skip_reason(venue, missing, together=together), + leg=missing[0].market_id, missing_legs=missing, + detail="; ".join(dict.fromkeys(m.detail for m in missing if m.detail)) or None) + + +def skip_reason(venue: str, missing: list[SkippedLeg], *, together: bool = False) -> str: + """`no match for “Al Fayha to win”, “U.S. invades Iran”`, one clause per + reason, in `LEG_CODES` order.""" + name = VENUE_NAMES.get(venue, venue) + + def quoted(code: str) -> str: + return ", ".join(f"“{m.label}”" for m in missing if m.code == code) + + clauses = [] + for code in [*LEG_CODES, *sorted({m.code for m in missing} - set(LEG_CODES))]: + if not any(m.code == code for m in missing): + continue + if code == "missing_leg": + clauses.append(f"no match for {quoted(code)}") + elif code == "matching_error": + clauses.append(f"matching unavailable for {quoted(code)}") + elif code == "ineligible_leg": + clauses.append(f"not combo-eligible on {name}: {quoted(code)}") + elif code == "outside_collections" and together: + clauses.append(f"no open {name} combo collection takes {quoted(code)} together") + elif code == "outside_collections": + scope = COLLECTION_SCOPES.get(venue, f"{name} combos take only the events of its open collections") + clauses.append(f"{scope}: {quoted(code)} is in none") + else: + clauses.append(f"{code.replace('_', ' ')}: {quoted(code)}") + return "; ".join(clauses) def _params(intent: Intent) -> dict[str, Any]: diff --git a/synpath/server/combo_warmer.py b/synpath/server/combo_warmer.py new file mode 100644 index 0000000..5942933 --- /dev/null +++ b/synpath/server/combo_warmer.py @@ -0,0 +1,245 @@ +"""serve's combo boards, kept warm: built at start-up, refreshed on a timer. + +A combo venue's board (`synpath.engine.combo_board.board_for`) reads the +venue's whole combo-eligible list (Polymarket: about 5,000 markets, plus +Gamma's markets and events for them; Kalshi: every open collection's events) +and takes seconds to minutes to build. A request must never wait for that, +so serve builds each combo venue's board in the background +(`ComboBoardWarmer.loop`, one of serve's background tasks) and the board +routes (`GET /trading/combos/board`, `/combos/board/categories`) answer from +the last build: + +* **start-up**: every combo venue's first build starts at once; until it ends + the routes answer `status: "building"` with no events, at once; +* **refresh**: every `refresh_interval_s` (60 s) after the last build ended, + each board is built again, through the venue's whole list (no page cap), + sleeping `pause_s` between two venue reads; +* **the first build fills in**: Polymarket lists about 214,000 combo-eligible + markets (214 pages), so its first build takes minutes; the board so far is + served (`complete: false`) after its first 5 pages (the busiest markets) and + every 40 after. Gamma's markets and events are kept between builds + (`BoardCache`), so a refresh reads Gamma only for new markets; +* **the last good board stays**: a refresh that fails, or is still running, + leaves the last board served; `refreshing` and `error` say so. + +The warmer also names markets for the RFQ groups' skip reasons +(`labels_for`, the engine's `Labeler`): a leg on a warm board by its pick +(`Al Fayha to win`), else by its market's question from the catalogue, +shortened (`U.S. invades Iran`). + +Configured in `engine.toml`: + +```toml +[combo_board] +warm = true # false: build on request, as a bare trading app does +refresh_interval_s = 60 +pause_s = 0.1 # between two reads of one venue during a build +``` +""" +from __future__ import annotations + +import asyncio +import gc +import logging +import sys +import time +from dataclasses import dataclass +from typing import Any, Callable + +from ..engine.combo_board import BoardCache, ComboBoard, board_for, label_of, market_labels, short_question +from ..engine.rfq_group import lists_combo_markets +from ..trading.types import ComboLeg + +log = logging.getLogger("synpath.serve.combo_boards") + +REFRESH_INTERVAL_S = 60.0 +PAUSE_S = 0.1 +SWITCH_INTERVAL_S = 0.0005 +"""The GIL switch interval while boards are warmed (`loop`).""" + + +@dataclass +class WarmBoard: + """One venue's board as the warmer holds it.""" + + board: ComboBoard | None = None + built_at: int | None = None + """When the board held was built, ms.""" + build_ms: int | None = None + """How long that build took.""" + refreshing: bool = False + error: str | None = None + """Why the last build failed, while the board held is an older one (or none).""" + + @property + def status(self) -> str: + return "ready" if self.board is not None else "building" + + +class ComboBoardWarmer: + """Every combo venue's board, built in the background (module doc).""" + + def __init__(self, engine: Any, registry: Any | None = None, *, refresh_interval_s: float = REFRESH_INTERVAL_S, + pause_s: float = PAUSE_S, clock: Callable[[], float] = time.time): + self.engine = engine + self.registry = registry + self.refresh_interval_s = float(refresh_interval_s) + self.pause_s = float(pause_s) + self.clock = clock + self.boards: dict[str, WarmBoard] = {} + self.labels: dict[str, dict[str, str]] = {} + """Per venue, `market_labels` of its last board.""" + self._builds: dict[str, asyncio.Task] = {} + self.caches: dict[str, BoardCache] = {} + """Per venue, what a build keeps for the next (Gamma's markets and events).""" + + # -- which venues ----------------------------------------------------------- + + def venues(self) -> list[str]: + """The engine's venues with a combo board: Kalshi's collections, or a + venue's own eligible list.""" + return [name for name, adapter in self.engine.adapters.items() + if adapter.has.get("combos") and (callable(getattr(adapter, "fetch_combo_collections", None)) + or lists_combo_markets(adapter))] + + def warms(self, venue: str) -> bool: + return venue in self.venues() + + # -- what the routes read ------------------------------------------------- + + def board(self, venue: str) -> ComboBoard: + """The venue's last board, with its `status` and `built_at`; while its + first build runs, an empty board with `status: "building"`. Never + waits: a venue not yet being built starts building here.""" + held = self.boards.get(venue) + if held is None or held.board is None: + self.ensure(venue) + return ComboBoard(venue=venue, count=0, complete=False, status="building", + error=held.error if held is not None else None) + return held.board.model_copy(update={"status": "ready", "built_at": held.built_at, "error": held.error}) + + def state(self, venue: str) -> dict[str, Any]: + held = self.boards.get(venue) or WarmBoard() + board = held.board + return {"venue": venue, "status": held.status, "built_at": held.built_at, "build_ms": held.build_ms, + "complete": board.complete if board is not None else False, "refreshing": held.refreshing, + "events": len(board.events) if board is not None else 0, "error": held.error} + + # -- building --------------------------------------------------------------- + + def ensure(self, venue: str) -> asyncio.Task: + """The venue's build: the one running, else a new one.""" + task = self._builds.get(venue) + if task is None or task.done(): + task = asyncio.get_running_loop().create_task(self.build(venue), name=f"synpath-combo-board-{venue}") + self._builds[venue] = task + return task + + async def build(self, venue: str) -> None: + """Build one venue's board through its whole list. A failure keeps + the last good board and is logged; it never raises.""" + held = self.boards.setdefault(venue, WarmBoard()) + held.refreshing = True + started = self.clock() + try: + board = await board_for(venue, self.engine.adapters[venue], self.reader(venue), pages=None, + max_events=None, pause_s=self.pause_s, + cache=self.caches.setdefault(venue, BoardCache()), + progress=lambda partial: self.partial(venue, partial)) + except Exception as exc: + held.error = f"{type(exc).__name__}: {exc}" + log.warning("synpath serve: building the %s combo board failed (%s); %s", venue, held.error, + "keeping the last board" if held.board is not None else "no board yet") + return + finally: + held.refreshing = False + held.board, held.error = board, None + held.built_at = int(self.clock() * 1000) + held.build_ms = int((self.clock() - started) * 1000) + self.labels[venue] = await asyncio.to_thread(market_labels, board) + keep_out_of_gc_scans() + log.info("synpath serve: %s combo board built: %d events from %d markets in %.1f s%s", venue, + len(board.events), board.count, held.build_ms / 1000, "" if board.complete else " (incomplete)") + + async def partial(self, venue: str, board: ComboBoard) -> None: + """The first build's board so far (`complete: false`), served until it + ends; a refresh keeps serving the last whole board instead.""" + held = self.boards.setdefault(venue, WarmBoard()) + if held.board is not None and held.board.complete: + return + held.board, held.built_at = board, int(self.clock() * 1000) + self.labels[venue] = await asyncio.to_thread(market_labels, board) + keep_out_of_gc_scans() + + async def refresh_all(self) -> None: + venues = self.venues() + await asyncio.gather(*(self.ensure(venue) for venue in venues), return_exceptions=True) + + async def loop(self) -> None: + """serve's background task: build every board now, then again each + `refresh_interval_s` after the last build ended.""" + # Builds run in worker threads (`share_the_lock`). Under the GIL, + # serve's event loop also waits up to one switch interval (5 ms by + # default) each time it takes the lock back, and one request takes it + # many times; a 0.5 ms interval keeps requests fast while a board builds. + sys.setswitchinterval(min(sys.getswitchinterval(), SWITCH_INTERVAL_S)) + while True: + try: + await self.refresh_all() + except Exception: # a board is never a reason to stop serve + log.exception("synpath serve: refreshing the combo boards failed") + await asyncio.sleep(self.refresh_interval_s) + + async def close(self) -> None: + for task in self._builds.values(): + task.cancel() + await asyncio.gather(*self._builds.values(), return_exceptions=True) + + def reader(self, venue: str) -> Any | None: + if self.registry is None: + return None + try: + return self.registry.get(venue) + except Exception: + return None + + # -- naming legs (the RFQ groups' `Labeler`) ----------------------------------- + + async def labels_for(self, legs: list[ComboLeg]) -> dict[str, str]: + """A short name for each leg's market: its pick on a warm board, else + its question from the catalogue, shortened. Legs neither knows are + left out (the group names them by id).""" + out: dict[str, str] = {} + unknown: dict[str, list[str]] = {} + for leg in legs: + venue = leg.market_id.partition(":")[0] + found = label_of(self.labels.get(venue) or {}, leg.market_id, leg.outcome) + if found: + out[leg.market_id] = found + else: + unknown.setdefault(venue, []).append(leg.market_id) + for venue, wanted in unknown.items(): + reader = self.reader(venue) + if reader is None or not callable(getattr(reader, "fetch_markets_by_ids", None)): + continue + try: + markets = await asyncio.to_thread(reader.fetch_markets_by_ids, wanted) + except Exception as exc: + log.info("synpath serve: reading %s legs' titles failed: %s", venue, exc) + continue + for market in markets: + if market.title: + out.setdefault(market.id, short_question(market.title)) + return out + + +def keep_out_of_gc_scans() -> None: + """Move every object alive now out of the cyclic collector's reach. + + A whole Polymarket board is millions of objects, and each full collection + rescans them all while every thread waits: with a board held, collections + took 0.3-1.4 s each and froze serve's requests. The board and its cache + are long-lived and acyclic, so `gc.freeze` loses nothing: reference + counting still frees them when the next build replaces them.""" + gc.freeze() + diff --git a/synpath/server/rfq_routes.py b/synpath/server/rfq_routes.py index 0af5526..7227297 100644 --- a/synpath/server/rfq_routes.py +++ b/synpath/server/rfq_routes.py @@ -545,6 +545,34 @@ async def combo_markets( boards = TtlCache(ttl_s=COMBO_BOARD_TTL_S, size=16) + async def whole_board(request: Request, engine: Engine, name: str, pages: int) -> ComboBoard: + """serve's warm board (never waiting on a build: `status: building` + until the first ends); without a warmer (a bare trading app, or + `[combo_board] warm = false`), built on request and kept 30 s.""" + warmer = getattr(request.app.state, "combo_boards", None) + if warmer is not None and warmer.warms(name): + return warmer.board(name) + reader = catalogue_reader(request, name) + return await boards.get(("board", name, pages), + lambda: tagged(name, board_for(name, engine.adapters[name], reader, pages=pages))) + + @router.get("/combos/board/categories", tags=["rfq"], summary="A combo board's categories, with counts", + responses=RFQ_RESPONSES) + async def combo_board_categories( + principal: Caller, + engine: EngineDep, + request: Request, + venue: Annotated[str, Query(description="A venue with `combos`: `kalshi`, `polymarket`, `polymarket_us`")], + ) -> ComboBoardCategories: + """The rail of a combo builder: the board's categories (leagues, + kinds) with their event counts, the markets read and the events on + the board, its `status` (`building` until serve's first build of it + ends), `built_at` and `complete`. A few hundred bytes; read the + events a category at a time from `GET /combos/board?category=`.""" + name = rfq_venue(engine, venue, capability="combos") + require_view(principal, engine, name) + return categories_view(await whole_board(request, engine, name, BOARD_PAGES)) + @router.get("/combos/board", tags=["rfq"], summary="A venue's combo-eligible markets as games and questions", responses=RFQ_RESPONSES) async def combo_board( @@ -552,7 +580,18 @@ async def combo_board( engine: EngineDep, request: Request, venue: Annotated[str, Query(description="A venue with `combos`: `kalshi`, `polymarket`, `polymarket_us`")], - pages: Annotated[int, Query(ge=1, le=20, description="Pages of the venue's list to read")] = 5, + pages: Annotated[int, Query(ge=1, le=20, description="Without serve's warmer only: pages of the venue's " + "list to read")] = BOARD_PAGES, + category: Annotated[str | None, Query(max_length=40, description="One category's events (`nfl`, " + "`politics`; `all`: every one), a page " + "at a time")] = None, + cursor: Annotated[str | None, Query(max_length=20, description="The page after the one that gave it")] = None, + limit: Annotated[int | None, Query(ge=1, le=MAX_PAGE_LIMIT, description="Events per page (default " + f"{PAGE_LIMIT})")] = None, + q: Annotated[str | None, Query(max_length=100, description="Only events whose title, category or " + "participants hold every word")] = None, + sort: Annotated[BoardSort, Query(description="`board` (start time), `volume` (largest first) or `start`, " + "within each category")] = "board", ) -> ComboBoard: """What a combo builder draws (`synpath/engine/combo_board.py`): games (participants, and moneyline / spread / total picks where the @@ -561,12 +600,24 @@ async def combo_board( `market_id` and `outcome`, with the catalogue's prices. Kalshi's board is its open collections' events (their markets read from the server's catalogue); Polymarket's and Polymarket US's, their own - eligible lists. Served from memory for 30 s.""" + eligible lists. + + serve keeps every combo venue's board warm (built at start-up from + the venue's whole list, rebuilt each minute; docs/server.md), so + this never waits on a venue: until the first build ends it answers + `status: "building"` with no events. With `category`, `cursor`, + `limit`, `q` or `sort` it answers one page (`category`, `total`, + `next_cursor`) with the whole board's categories; without any, the + whole board, as before.""" name = rfq_venue(engine, venue, capability="combos") require_view(principal, engine, name) - reader = catalogue_reader(request, name) - return await boards.get(("board", name, pages), - lambda: tagged(name, board_for(name, engine.adapters[name], reader, pages=pages))) + board = await whole_board(request, engine, name, pages) + if category is None and cursor is None and limit is None and not q and sort == "board": + return board + try: + return board_page(board, category=category, cursor=cursor, limit=limit or PAGE_LIMIT, query=q, sort=sort) + except BadRequest as exc: + raise HTTPException(status_code=400, detail=str(exc)) from None @router.get("/combos/{market_id}", tags=["rfq"], summary="One combo and its legs", responses=RFQ_RESPONSES) async def get_combo(principal: Caller, engine: EngineDep, market_id: Annotated[str, Path()], @@ -586,7 +637,11 @@ async def get_combo(principal: Caller, engine: EngineDep, market_id: Annotated[s # Cross-venue RFQ groups (docs/rfq-combo.md, "Cross-venue RFQ") # --------------------------------------------------------------------------- -from ..engine.combo_board import ComboBoard, board_for # noqa: E402 +from ..engine.combo_board import ( # noqa: E402 + BOARD_PAGES, MAX_PAGE_LIMIT, PAGE_LIMIT, BoardSort, ComboBoard, ComboBoardCategories, board_for, board_page, + categories_view, +) +from ..errors import BadRequest # noqa: E402 from ..engine.rfq_group import GroupPreview, RfqGroup # noqa: E402 diff --git a/synpath/server/serve.py b/synpath/server/serve.py index b24a011..738ccca 100644 --- a/synpath/server/serve.py +++ b/synpath/server/serve.py @@ -13,6 +13,9 @@ streams, so stops see prices and parents hear their children fill; * **reconciliation**, the **end-of-day** close, **halt/resume** requests written by `synpath halt`, and a periodic status line; +* the **combo boards** (`synpath.server.combo_warmer`): every combo venue's + board built at start-up and rebuilt each minute, so `/trading/combos/board` + never waits on a venue; * the **HTTP API**: the read app at `/` (with batch reads and search), matching and tick history from Synpath's hosted services at `/matching` and `/history` (serve holds the Synpath API key), live market data at @@ -82,6 +85,8 @@ class Stack: extra_tasks: dict[str, Any] = field(default_factory=dict) market_streams: Any | None = None """The `/ws/market` hub: venue market connections shared with the feeds.""" + combo_boards: Any | None = None + """Every combo venue's board, kept warm (`synpath.server.combo_warmer`); None with `[combo_board] warm = false`.""" async def build( @@ -192,10 +197,29 @@ async def build( # The combo board and group previews read leg markets and quotes from the # same catalogue as the read app. trading_app.state.registry = app.state.registry + combo_boards = combo_board_warmer(engine, app.state.registry, config) + if combo_boards is not None: + # The board routes answer from it, never waiting on a build; the RFQ + # groups name the legs a venue lacks by its boards' picks. + trading_app.state.combo_boards = combo_boards + engine.rfq_groups.labeler = combo_boards.labels_for app.mount("/trading", trading_app) app.state.trading_path = "/trading" return Stack(engine=engine, feeds=feeds, store=store, app=app, reconciler=reconciler, eod=eod, - config=config, owner_key=owner_key, local_key=stack_key, market_streams=market_streams) + config=config, owner_key=owner_key, local_key=stack_key, market_streams=market_streams, + combo_boards=combo_boards) + + +def combo_board_warmer(engine: Any, registry: Any, config: dict[str, Any]) -> Any | None: + """The combo board warmer `[combo_board]` configures (on by default).""" + from .combo_warmer import PAUSE_S, REFRESH_INTERVAL_S, ComboBoardWarmer + + section = dict(config.get("combo_board") or {}) + if not section.get("warm", True): + return None + return ComboBoardWarmer(engine, registry, + refresh_interval_s=float(section.get("refresh_interval_s", REFRESH_INTERVAL_S)), + pause_s=float(section.get("pause_s", PAUSE_S))) def background(stack: Stack) -> dict[str, Any]: @@ -210,6 +234,8 @@ def background(stack: Stack) -> dict[str, Any]: loops["control"] = _control_loop(engine, asyncio.Event()) loops["eod"] = stack.eod.loop() loops["status"] = _status_loop(engine, float(stack.config.get("status_interval_s", 60))) + if stack.combo_boards is not None and stack.combo_boards.venues(): + loops["combo_boards"] = stack.combo_boards.loop() return loops @@ -220,6 +246,8 @@ async def shutdown(stack: Stack) -> None: await engine.halt("the server is shutting down", policy=engine.config.halt_policy) except Exception: log.exception("synpath serve: halting on exit failed") + if stack.combo_boards is not None: + await stack.combo_boards.close() if stack.market_streams is not None: await stack.market_streams.close() if stack.feeds is not None: diff --git a/tests/samples/polymarket_combo_skip_legs_live.json b/tests/samples/polymarket_combo_skip_legs_live.json new file mode 100644 index 0000000..1e4ed58 --- /dev/null +++ b/tests/samples/polymarket_combo_skip_legs_live.json @@ -0,0 +1,160 @@ +{ + "_source": "gamma-api.polymarket.com, no auth, 2026-10-22: GET /markets?id=... and GET /events?slug=spl-haz-fay-2026-10-22 (trimmed)", + "gamma_markets": [ + { + "id": "665374", + "question": "Will the U.S. invade Iran before 2027?", + "conditionId": "0x5db999fad322cea2914535aae5517060c3f80ad6d8c0231cde2124a434d16846", + "slug": "will-the-us-invade-iran-before-2027", + "outcomes": "[\"Yes\", \"No\"]", + "outcomePrices": "[\"0.165\", \"0.835\"]", + "groupItemTitle": "", + "endDate": "2027-01-01T04:59:00Z", + "bestBid": 0.16, + "bestAsk": 0.17, + "volume": "74992514.50771101", + "active": true, + "closed": false, + "clobTokenIds": "[\"55115078421062885512539156303747803058407616201213034911037320915726138659123\", \"1910830010387565971650098373488592514702818137344973088263643820608151819241\"]", + "events": [ + { + "id": "73130", + "title": "Will the U.S. invade Iran before 2027?", + "slug": "will-the-us-invade-iran-before-2027", + "seriesSlug": "us-to-invade-iran" + } + ] + }, + { + "id": "5441699", + "question": "Will Al Hazem SC win on 2026-10-22?", + "conditionId": "0x018ec4ff84aa4be8dbc516f219330875b24d62f6de5d01322d28980f32ab9270", + "slug": "spl-haz-fay-2026-10-22-haz", + "outcomes": "[\"Yes\", \"No\"]", + "outcomePrices": "[\"0.055\", \"0.945\"]", + "groupItemTitle": "Al Hazem SC", + "sportsMarketType": "moneyline", + "gameStartTime": "2026-10-22 13:40:00+00", + "endDate": "2026-10-22T13:40:00Z", + "bestBid": 0.05, + "bestAsk": 0.06, + "volume": "34255.961589000006", + "active": true, + "closed": false, + "clobTokenIds": "[\"55820637656080051236779553744171718550430743317298849483643993923931369327674\", \"49887698749361420283253703773210173901670246597450902289338216286026577039943\"]", + "events": [ + { + "id": "1155505", + "title": "Al Hazem SC vs. Al Fayha Saudi Club", + "slug": "spl-haz-fay-2026-10-22", + "seriesSlug": "saudi-professional-league", + "startTime": "2026-10-22T13:40:00Z" + } + ] + }, + { + "id": "5441700", + "question": "Will Al Hazem SC vs. Al Fayha Saudi Club end in a draw?", + "conditionId": "0xcb917d0199f13258315a4d139a984b864cb9d69b958de0dff8e5e9260bc5182a", + "slug": "spl-haz-fay-2026-10-22-draw", + "outcomes": "[\"Yes\", \"No\"]", + "outcomePrices": "[\"0.245\", \"0.755\"]", + "groupItemTitle": "Draw (Al Hazem SC vs. Al Fayha Saudi Club)", + "sportsMarketType": "moneyline", + "gameStartTime": "2026-10-22 13:40:00+00", + "endDate": "2026-10-22T13:40:00Z", + "bestBid": 0.24, + "bestAsk": 0.25, + "volume": "4995.146306", + "active": true, + "closed": false, + "clobTokenIds": "[\"45724549326613598249462400116700482919332896353917976499269550515147661134216\", \"59241911495708626709881233002614477825022443457172355568157022255230177659246\"]", + "events": [ + { + "id": "1155505", + "title": "Al Hazem SC vs. Al Fayha Saudi Club", + "slug": "spl-haz-fay-2026-10-22", + "seriesSlug": "saudi-professional-league", + "startTime": "2026-10-22T13:40:00Z" + } + ] + }, + { + "id": "5441701", + "question": "Will Al Fayha Saudi Club win on 2026-10-22?", + "conditionId": "0x16532e62e59a43814d88904862e72f499e732564a12538eaff19bbe72ced22ee", + "slug": "spl-haz-fay-2026-10-22-fay", + "outcomes": "[\"Yes\", \"No\"]", + "outcomePrices": "[\"0.705\", \"0.295\"]", + "groupItemTitle": "Al Fayha Saudi Club", + "sportsMarketType": "moneyline", + "gameStartTime": "2026-10-22 13:40:00+00", + "endDate": "2026-10-22T13:40:00Z", + "bestBid": 0.7, + "bestAsk": 0.71, + "volume": "44744.349081", + "active": true, + "closed": false, + "clobTokenIds": "[\"89021557173236511193207323280147232960197796554895843150971900545567946374202\", \"72594579263945692457243696883608040759678572474107456550546424327901331244223\"]", + "events": [ + { + "id": "1155505", + "title": "Al Hazem SC vs. Al Fayha Saudi Club", + "slug": "spl-haz-fay-2026-10-22", + "seriesSlug": "saudi-professional-league", + "startTime": "2026-10-22T13:40:00Z" + } + ] + } + ], + "gamma_events": [ + { + "id": "1155505", + "title": "Al Hazem SC vs. Al Fayha Saudi Club", + "slug": "spl-haz-fay-2026-10-22", + "seriesSlug": "saudi-professional-league", + "startTime": "2026-10-22T13:40:00Z", + "teams": [ + { + "id": 119066, + "name": "Al Hazem SC", + "abbreviation": "haz", + "alias": "Al Hazem", + "league": "spl", + "ordering": "home" + }, + { + "id": 119054, + "name": "Al Fayha Saudi Club", + "abbreviation": "fay", + "alias": "Al Fayha", + "league": "spl", + "ordering": "away" + } + ], + "tags": [ + { + "id": "1", + "label": "Sports", + "slug": "sports" + }, + { + "id": "100639", + "label": "Games", + "slug": "games" + }, + { + "id": "100350", + "label": "Soccer", + "slug": "soccer" + }, + { + "id": "102650", + "label": "Saudi Professional League", + "slug": "saudi-professional-league" + } + ], + "markets": [] + } + ] +} \ No newline at end of file diff --git a/tests/test_combo_warmer.py b/tests/test_combo_warmer.py new file mode 100644 index 0000000..e573cd1 --- /dev/null +++ b/tests/test_combo_warmer.py @@ -0,0 +1,453 @@ +"""serve's warm combo boards and the combo skip reasons, offline. + +* The warmer (`synpath/server/combo_warmer.py`): every combo venue's board + built at start-up, through the venue's whole list (no page cap), rebuilt on + its interval, the last good board kept when a rebuild fails, and its state. +* The board routes on it: `GET /combos/board/categories`, one category a page + at a time on `GET /combos/board?category=&cursor=&limit=&q=&sort=`, the + whole board as before, and `status: "building"` at once while the first + build runs. +* The skip reasons of a cross-venue combo, on Yuki's real Polymarket combo of + 2026-10-22: "Will Al Fayha Saudi Club win on 2026-10-22?" + (`polymarket:5441701`, matched on Polymarket US only) and "Will the U.S. + invade Iran before 2027?" (`polymarket:665374`, matched nowhere), with + hosted matching mocked from its real answers and Gamma's real rows for both + (`samples/polymarket_combo_skip_legs_live.json`, no auth). +""" +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import Any + +import httpx +import pytest + +from synpath.engine import Engine, EngineConfig, RiskConfig +from synpath.engine.combo_board import board_of, market_labels, polymarket_markets, short_question +from synpath.engine.paper import PaperRfqVenue +from synpath.engine.rfq_group import hosted_matcher +from synpath.polymarket import normalize_market as gamma_market +from synpath.server.combo_warmer import ComboBoardWarmer +from synpath.server.store import ControlStore +from synpath.server.trading import create_trading_app +from synpath.trading.types import Account, ComboLeg, ComboMarket +from synpath.types import Page + +from test_engine_rfq_group import RISK, CollectionVenue, ListedVenue + +pytestmark = pytest.mark.anyio + +SAMPLE = json.loads((Path(__file__).parent / "samples" / "polymarket_combo_skip_legs_live.json").read_text()) +AL_FAYHA, IRAN = "polymarket:5441701", "polymarket:665374" +AL_FAYHA_US = "polymarket_us:atc-spl-haz-fay-2026-10-22-fay" + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.fixture(autouse=True) +def no_saved_key(monkeypatch, tmp_path): + monkeypatch.delenv("SYNPATH_API_KEY", raising=False) + monkeypatch.setenv("SYNPATH_CREDENTIALS_FILE", str(tmp_path / "no-credentials.json")) + + +# -- a venue with a long eligible list ---------------------------------------------- + +def question(n: int, tag: str) -> ComboMarket: + return ComboMarket(market_id=f"polymarket:{n}", venue="polymarket", title=f"Will thing {n} happen?", tags=[tag], + info={"volume": str(n), "outcome_prices": ["0.5", "0.5"]}) + + +class PagedVenue(PaperRfqVenue): + """A Polymarket-like list of `pages` pages; each build may be held or fail.""" + + def __init__(self, pages: int, per_page: int = 10, **kw: Any): + super().__init__(venue="polymarket", **kw) + self.has = {**self.has, "combos": True} + tags = ("nfl", "politics", "crypto") + self.rows = [question(n, tags[n % 3]) for n in range(pages * per_page)] + self.per_page = per_page + self.reads = 0 + self.fail = False + self.hold: asyncio.Event | None = None + self.block_at: int | None = None + """Hold the read of this page (1-based) until `hold` is set.""" + + async def fetch_combo_eligible_markets(self, *, event_id=None, tag=None, limit=None, cursor=None) -> Page: + if self.hold is not None and (self.block_at is None or self.reads + 1 == self.block_at): + await self.hold.wait() + if self.fail: + raise ConnectionError("the catalogue is down") + self.reads += 1 + start = int(cursor or 0) + page = self.rows[start:start + self.per_page] + more = start + self.per_page < len(self.rows) + return Page(page, next_cursor=str(start + self.per_page) if more else None) + + +class FakeEngine: + def __init__(self, adapters): + self.adapters = adapters + + +class TestWarmer: + async def test_builds_the_whole_list_at_start_up_with_no_page_cap(self): + venue = PagedVenue(pages=12) # more than the 5 pages a board read on request takes + warmer = ComboBoardWarmer(FakeEngine({"polymarket": venue}), pause_s=0) + assert warmer.venues() == ["polymarket"] + first = warmer.board("polymarket") # starts the build, does not wait for it + assert first.status == "building" and first.events == [] + await warmer.ensure("polymarket") + board = warmer.board("polymarket") + assert board.status == "ready" and board.complete and board.count == 120 and venue.reads == 12 + assert board.built_at and warmer.state("polymarket")["status"] == "ready" + + async def test_refreshes_and_keeps_the_last_good_board_when_a_refresh_fails(self): + venue = PagedVenue(pages=2) + warmer = ComboBoardWarmer(FakeEngine({"polymarket": venue}), pause_s=0) + await warmer.build("polymarket") + built = warmer.board("polymarket").built_at + venue.rows = venue.rows[:5] + await warmer.build("polymarket") + assert warmer.board("polymarket").count == 5, "a refresh replaces the board" + venue.fail = True + await warmer.build("polymarket") + kept = warmer.board("polymarket") + assert kept.count == 5 and kept.status == "ready" and "catalogue is down" in kept.error + assert kept.built_at >= built and not warmer.state("polymarket")["refreshing"] + + async def test_the_loop_builds_at_once_then_on_its_interval_and_never_raises(self): + venue = PagedVenue(pages=1) + warmer = ComboBoardWarmer(FakeEngine({"polymarket": venue}), refresh_interval_s=0.01, pause_s=0) + task = asyncio.create_task(warmer.loop()) + for _ in range(200): + await asyncio.sleep(0.005) + if venue.reads >= 3: + break + venue.fail = True + await asyncio.sleep(0.05) + assert not task.done(), "a failing build never stops serve's loop" + task.cancel() + await warmer.close() + assert venue.reads >= 3 and warmer.board("polymarket").count == 10 + + async def test_a_long_first_build_serves_its_busiest_pages_first(self): + venue = PagedVenue(pages=12) + venue.hold, venue.block_at = asyncio.Event(), 7 + warmer = ComboBoardWarmer(FakeEngine({"polymarket": venue}), pause_s=0) + task = warmer.ensure("polymarket") + for _ in range(200): + await asyncio.sleep(0.005) + if warmer.board("polymarket").status == "ready": + break + partial = warmer.board("polymarket") + assert partial.status == "ready" and not partial.complete and partial.count == 50, "after the first 5 pages" + venue.hold.set() + await task + whole = warmer.board("polymarket") + assert whole.complete and whole.count == 120 + # A refresh never swaps the whole board for a partial one. + venue.hold, venue.reads = asyncio.Event(), 0 + refresh = warmer.ensure("polymarket") + for _ in range(50): + await asyncio.sleep(0.005) + assert warmer.board("polymarket").complete and warmer.state("polymarket")["refreshing"] + venue.hold.set() + await refresh + + async def test_a_venue_without_combo_lists_is_not_warmed(self): + warmer = ComboBoardWarmer(FakeEngine({"limitless": PaperRfqVenue(venue="limitless")})) + assert warmer.venues() == [] and not warmer.warms("limitless") + + +# -- the routes on a warm board ---------------------------------------------------------- + +@pytest.fixture +async def server(tmp_path: Path): + from fastapi.testclient import TestClient + + venue = PagedVenue(pages=3) + venue.account = Account(venue="polymarket", name="desk-a") + engine = Engine({"polymarket": venue}, EngineConfig(journal_path=str(tmp_path / "engine.db"), in_doubt_timeout_s=0, + rfq_incoming_interval_s=None), + risk=RiskConfig(**RISK), accounts={"polymarket": venue.account}) + await engine.start() + store = await ControlStore(str(tmp_path / "control.db")).open() + owner, issued = await store.bootstrap("owner") + app = create_trading_app(engine, store) + warmer = ComboBoardWarmer(engine, pause_s=0) + app.state.combo_boards = warmer + with TestClient(app) as client: + yield {"client": client, "venue": venue, "warmer": warmer, "headers": {"Authorization": f"Bearer {issued.secret}"}} + await store.close() + await engine.stop() + + +def wait_ready(server) -> None: + client = server["client"] + for _ in range(200): + if client.get("/combos/board/categories?venue=polymarket", headers=server["headers"]).json()["status"] == "ready": + return + import time + time.sleep(0.01) + raise AssertionError("the board never became ready") + + +class TestRoutes: + async def test_building_answers_at_once_then_the_board_fills_in(self, server): + client, venue = server["client"], server["venue"] + venue.hold = asyncio.Event() + categories = client.get("/combos/board/categories?venue=polymarket", headers=server["headers"]) + assert categories.status_code == 200 and categories.json()["status"] == "building" + whole = client.get("/combos/board?venue=polymarket", headers=server["headers"]).json() + assert whole["status"] == "building" and whole["events"] == [] + page = client.get("/combos/board?venue=polymarket&category=nfl", headers=server["headers"]).json() + assert page["status"] == "building" and page["total"] == 0 and page["next_cursor"] is None + client.portal.call(venue.hold.set) + wait_ready(server) + + async def test_categories_then_one_category_a_page_at_a_time(self, server): + client = server["client"] + wait_ready(server) + rail = client.get("/combos/board/categories?venue=polymarket", headers=server["headers"]).json() + assert rail["events"] == 30 and rail["complete"] and rail["built_at"] + assert {c["key"]: c["count"] for c in rail["categories"]} == {"nfl": 10, "politics": 10, "crypto": 10} + assert "events" not in rail["categories"][0] + first = client.get("/combos/board?venue=polymarket&category=politics&limit=4&sort=volume", + headers=server["headers"]).json() + assert first["category"] == "politics" and first["total"] == 10 and len(first["events"]) == 4 + assert first["next_cursor"] == "4" and {e["category"] for e in first["events"]} == {"politics"} + volumes = [float(e["volume"]) for e in first["events"]] + assert volumes == sorted(volumes, reverse=True) + assert len(first["categories"]) == 3, "a page carries the rail's categories" + seen = [e["id"] for e in first["events"]] + cursor = first["next_cursor"] + while cursor: + page = client.get(f"/combos/board?venue=polymarket&category=politics&limit=4&sort=volume&cursor={cursor}", + headers=server["headers"]).json() + seen += [e["id"] for e in page["events"]] + cursor = page["next_cursor"] + assert len(seen) == len(set(seen)) == 10 + found = client.get("/combos/board?venue=polymarket&category=all&q=thing%2012", headers=server["headers"]).json() + assert [e["id"] for e in found["events"]] == ["polymarket:12"] and found["total"] == 1 + bad = client.get("/combos/board?venue=polymarket&category=nfl&cursor=x", headers=server["headers"]) + assert bad.status_code == 400 + + async def test_the_whole_board_call_still_works(self, server): + wait_ready(server) + whole = server["client"].get("/combos/board?venue=polymarket", headers=server["headers"]).json() + assert whole["status"] == "ready" and len(whole["events"]) == 30 and whole["next_cursor"] is None + assert whole["category"] is None and whole["count"] == 30 + + async def test_openapi_lists_the_new_routes(self, server): + paths = server["client"].get("/openapi.json").json()["paths"] + assert "get" in paths["/combos/board/categories"] + params = {p["name"] for p in paths["/combos/board"]["get"]["parameters"]} + assert {"venue", "category", "cursor", "limit", "q", "sort", "pages"} <= params + + +# -- skip reasons on the two real legs ------------------------------------------------------- + +class GammaReader: + """serve's Polymarket read adapter, on Gamma's real rows for both legs.""" + + def fetch_markets_by_ids(self, market_ids): + wanted = {m.partition(":")[2] for m in market_ids} + return [gamma_market(raw) for raw in SAMPLE["gamma_markets"] if raw["id"] in wanted] + + def fetch_events_by_ids(self, event_ids): + from synpath.polymarket import normalize_event + return [normalize_event(e) for e in SAMPLE["gamma_events"] if f"polymarket:{e['id']}" in event_ids] + + +class Registry: + def __init__(self, readers): + self.readers = readers + + def get(self, venue): + return self.readers[venue] + + +def catalogue_row(raw: dict) -> ComboMarket: + tags = ["sports", "games", "soccer", "saudi-professional-league"] if raw.get("sportsMarketType") else ["politics"] + return ComboMarket(market_id=f"polymarket:{raw['id']}", venue="polymarket", title=raw["question"], tags=tags, + position_ids={"yes": f"{raw['id']}-Y", "no": f"{raw['id']}-N"}, info={"volume": raw.get("volume")}) + + +# Hosted matching's answers for Yuki's legs on 2026-10-22: Al Fayha is listed on +# Polymarket US and not on Kalshi; the Iran question is listed nowhere else. +MATCHES = { + (AL_FAYHA, "polymarket_us"): {"id": AL_FAYHA_US, "venue": "polymarket_us", "side_map": {"yes": "yes", "no": "no"}}, + (AL_FAYHA, "kalshi"): None, + (IRAN, "polymarket_us"): None, + (IRAN, "kalshi"): None, +} + + +def matching(down: bool = False): + def handler(request: httpx.Request) -> httpx.Response: + if down: + return httpx.Response(503, json={"detail": "matching is down"}) + key = (request.url.params["id"], request.url.params.get("venue")) + if key not in MATCHES: + return httpx.Response(404, json={"detail": "market not found"}) + return httpx.Response(200, json={"anchor": key[0], "event_id": "evt", "matched": MATCHES[key]}) + + return hosted_matcher(client=httpx.Client(transport=httpx.MockTransport(handler)), api_key="spk_test") + + +def combo_venues() -> dict[str, PaperRfqVenue]: + rows = [catalogue_row(raw) for raw in SAMPLE["gamma_markets"]] + poly = ListedVenue(venue="polymarket", eligible=[r.market_id for r in rows]) + poly.eligible = rows + us = ListedVenue(venue="polymarket_us", eligible=[]) # the US match is not combo-eligible there + kalshi = CollectionVenue(venue="kalshi", collections=[{"collection_ticker": "KXMVESPORTS", "associated_event_tickers": ["KXNFLGAME-26OCT22NECHI"]}]) + return {"kalshi": kalshi, "polymarket": poly, "polymarket_us": us} + + +LEGS = [ComboLeg(market_id=AL_FAYHA, outcome="yes"), ComboLeg(market_id=IRAN, outcome="yes")] + + +async def engine_with(tmp_path, adapters, *, warm: bool = True, down: bool = False) -> Engine: + engine = Engine(adapters, EngineConfig(journal_path=str(tmp_path / "engine.db"), in_doubt_timeout_s=0, + rfq_incoming_interval_s=None), + risk=RiskConfig(**RISK), matcher=matching(down)) + await engine.start() + warmer = ComboBoardWarmer(engine, Registry({"polymarket": GammaReader()}), pause_s=0) + if warm: + await warmer.build("polymarket") + engine.rfq_groups.labeler = warmer.labels_for + return engine + + +class CountingGamma(GammaReader): + def __init__(self): + self.asked: list[list[str]] = [] + + def fetch_markets_by_ids(self, market_ids): + self.asked.append(list(market_ids)) + return super().fetch_markets_by_ids(market_ids) + + +class TestGammaCache: + async def test_a_rebuild_reads_gamma_only_for_new_markets_and_keeps_prices_fresh(self): + rows = [catalogue_row(raw) for raw in SAMPLE["gamma_markets"]] + venue = ListedVenue(venue="polymarket", eligible=[]) + venue.eligible = rows + gamma = CountingGamma() + warmer = ComboBoardWarmer(FakeEngine({"polymarket": venue}), Registry({"polymarket": gamma}), pause_s=0) + await warmer.build("polymarket") + assert sum(len(a) for a in gamma.asked) == 4 + first = warmer.board("polymarket") + assert any(e.kind == "game" for e in first.events), "Gamma's types and the event's teams" + venue.eligible = rows + [ComboMarket(market_id="polymarket:1", venue="polymarket", title="Will it rain?", + tags=["weather"], info={"outcome_prices": ["0.4", "0.6"]})] + gamma.asked.clear() + await warmer.build("polymarket") + assert gamma.asked == [["polymarket:1"]], "only the new market" + again = warmer.board("polymarket") + assert [e.kind for e in again.events if e.kind == "game"] == [e.kind for e in first.events if e.kind == "game"] + cache = warmer.caches["polymarket"] + assert all("bestBid" not in m and "outcomePrices" not in m for m in cache.markets.values()), \ + "prices are never kept: each build takes the catalogue's" + + +class TestSkipReasons: + def test_the_board_names_al_fayha_by_its_short_name_and_questions_shortened(self): + rows = [catalogue_row(raw) for raw in SAMPLE["gamma_markets"]] + board = board_of("polymarket", polymarket_markets( + rows, {m["id"]: m for m in SAMPLE["gamma_markets"]}, {e["id"]: e for e in SAMPLE["gamma_events"]})) + labels = market_labels(board) + assert labels[AL_FAYHA] == "Al Fayha to win" and labels["polymarket:5441699"] == "Al Hazem to win" + assert labels[IRAN] == "U.S. invades Iran" + assert short_question("Will Al Fayha Saudi Club win on 2026-10-22?") == "Al Fayha Saudi Club wins" + assert short_question("Bitcoin above $100k on December 31?") == "Bitcoin above $100k" + + async def test_each_venue_names_every_leg_it_lacks(self, tmp_path): + engine = await engine_with(tmp_path, combo_venues()) + preview = await engine.preview_rfq_group(legs=LEGS) + skipped = {s.venue: s for s in preview.skipped} + assert [v.venue for v in preview.offered] == ["polymarket"] + kalshi, us = skipped["kalshi"], skipped["polymarket_us"] + assert kalshi.code == "missing_leg" + assert kalshi.reason == "no match for “Al Fayha to win”, “U.S. invades Iran”" + assert [(m.market_id, m.label, m.code) for m in kalshi.missing_legs] == [ + (AL_FAYHA, "Al Fayha to win", "missing_leg"), (IRAN, "U.S. invades Iran", "missing_leg")] + assert us.code == "missing_leg" and us.reason == "no match for “U.S. invades Iran”" + assert [m.market_id for m in us.missing_legs] == [IRAN] and us.leg == IRAN + await engine.stop() + + async def test_without_a_warm_board_the_catalogue_question_is_shortened(self, tmp_path): + engine = await engine_with(tmp_path, combo_venues(), warm=False) + preview = await engine.preview_rfq_group(legs=LEGS) + kalshi = next(s for s in preview.skipped if s.venue == "kalshi") + assert kalshi.reason == "no match for “Al Fayha Saudi Club wins”, “U.S. invades Iran”" + await engine.stop() + + async def test_a_matched_leg_the_venue_will_not_combo_is_not_a_missing_match(self, tmp_path): + global MATCHES + saved = dict(MATCHES) + MATCHES[(IRAN, "polymarket_us")] = {"id": "polymarket_us:iran", "venue": "polymarket_us", + "side_map": {"yes": "yes", "no": "no"}} + try: + engine = await engine_with(tmp_path, combo_venues()) + us = next(s for s in (await engine.preview_rfq_group(legs=LEGS)).skipped if s.venue == "polymarket_us") + finally: + MATCHES.clear() + MATCHES.update(saved) + assert us.code == "ineligible_leg" + assert us.reason == "not combo-eligible on Polymarket US: “Al Fayha to win”, “U.S. invades Iran”" + assert [m.mapped_id for m in us.missing_legs] == [AL_FAYHA_US, "polymarket_us:iran"] + await engine.stop() + + async def test_outside_kalshis_collections_and_matching_unavailable_are_their_own_reasons(self, tmp_path): + global MATCHES + saved = dict(MATCHES) + MATCHES[(AL_FAYHA, "kalshi")] = {"id": "kalshi:KXSPL-26OCT22HAZFAY-FAY", "venue": "kalshi", + "side_map": {"yes": "yes", "no": "no"}} + MATCHES[(IRAN, "kalshi")] = {"id": "kalshi:KXUSAIRAN-27-YES", "venue": "kalshi", + "side_map": {"yes": "yes", "no": "no"}} + try: + engine = await engine_with(tmp_path, combo_venues()) + kalshi = next(s for s in (await engine.preview_rfq_group(legs=LEGS)).skipped if s.venue == "kalshi") + finally: + MATCHES.clear() + MATCHES.update(saved) + assert kalshi.code == "outside_collections" + assert kalshi.reason.startswith("Kalshi combos are sports and cross-category only: “Al Fayha to win”") + await engine.stop() + (tmp_path / "down").mkdir() + down = await engine_with(tmp_path / "down", combo_venues(), down=True) + lost = {s.venue: s for s in (await down.preview_rfq_group(legs=LEGS)).skipped} + assert lost["kalshi"].code == "matching_error" + assert lost["kalshi"].reason == "matching unavailable for “Al Fayha to win”, “U.S. invades Iran”" + await down.stop() + + +class TestServe: + async def test_serve_warms_every_combo_venue_unless_told_not_to(self, tmp_path): + from synpath.server import serve + + for warm, home in ((True, "on"), (False, "off")): + venue = PagedVenue(pages=1) + venue.account = Account(venue="polymarket", name="paper") + (tmp_path / home).mkdir() + stack = await serve.build( + config={"risk": RISK, "combo_board": {"warm": warm, "refresh_interval_s": 5, "pause_s": 0}}, + journal=str(tmp_path / home / "serve.db"), control=str(tmp_path / home / "control.db"), + adapters={"polymarket": venue}, streams=False, home_dir=str(tmp_path / home / "home")) + loops = serve.background(stack) + try: + if warm: + assert stack.combo_boards.refresh_interval_s == 5 and "combo_boards" in loops + assert stack.engine.rfq_groups.labeler == stack.combo_boards.labels_for + else: + assert stack.combo_boards is None and "combo_boards" not in loops + finally: + for coro in loops.values(): + coro.close() + await serve.shutdown(stack) diff --git a/tests/test_engine_rfq_group.py b/tests/test_engine_rfq_group.py index 3101936..e7cd92c 100644 --- a/tests/test_engine_rfq_group.py +++ b/tests/test_engine_rfq_group.py @@ -215,12 +215,15 @@ async def test_maps_legs_by_matching_and_skips_a_missing_and_an_ineligible_leg(s skipped = {s.venue: s for s in group.skipped} assert skipped["polymarket_us"].code == "missing_leg" and skipped["polymarket_us"].leg == K_B assert skipped["limitless"].code == "ineligible_leg" and skipped["limitless"].leg == K_B - assert "combo-eligible list" in skipped["limitless"].reason + assert skipped["limitless"].reason == f"not combo-eligible on Limitless: “{K_B}”" + assert "combo-eligible list" in skipped["limitless"].detail + assert [(m.market_id, m.code, m.mapped_id) for m in skipped["limitless"].missing_legs] == [ + (K_B, "ineligible_leg", L_B)] assert all(auth == "Bearer spk_test" for *_, auth in seen) assert len(group.quotes) == 2 await engine.stop() - async def test_a_kalshi_leg_in_no_collection_is_an_ineligible_leg(self, tmp_path): + async def test_a_kalshi_leg_in_no_collection_is_outside_its_collections(self, tmp_path): adapters = venues() adapters["kalshi"].collections = [{"collection_ticker": "KXMVE-1", "associated_event_tickers": ["KXA-26"]}] engine = await started(tmp_path, adapters) @@ -228,8 +231,9 @@ async def test_a_kalshi_leg_in_no_collection_is_an_ineligible_leg(self, tmp_path legs=[ComboLeg(market_id=K_A, outcome="yes"), ComboLeg(market_id=K_B, outcome="no")], quantity=D("10"), venues=["kalshi", "polymarket"]) skipped = {s.venue: s for s in group.skipped} - assert skipped["kalshi"].code == "ineligible_leg" and skipped["kalshi"].leg == K_B - assert "KXB-26" in skipped["kalshi"].reason + assert skipped["kalshi"].code == "outside_collections" and skipped["kalshi"].leg == K_B + assert skipped["kalshi"].reason == f"Kalshi combos are sports and cross-category only: “{K_B}” is in none" + assert "KXB-26" in skipped["kalshi"].detail assert skipped["polymarket_us"].code == "excluded" assert [m.venue for m in group.members] == ["polymarket"] await engine.stop()