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()