diff --git a/.env.example b/.env.example deleted file mode 100644 index 86600da..0000000 --- a/.env.example +++ /dev/null @@ -1,25 +0,0 @@ -# polymaker secrets. Copy to .env and fill in. NEVER commit .env. -# -# IMPORTANT: use the SAME wallet you use in the Polymarket browser UI, and make -# sure it has traded at least once through the UI so allowances are set. - -# Private key of the signing wallet (EOA that controls the funder). -PK= - -# The funder = the smart-contract wallet that actually holds your pUSD and -# positions (NOT your signing EOA/MetaMask address above). Polymarket's UI labels -# this inconsistently (may show as "deposit" or "developer" address) — the one -# that holds the funds is the funder. `polymaker doctor` reads the balance so you -# can confirm you picked the right one. Set signature_type in config/config.toml -# to match how the account was made (new deposit wallets = 3). -BROWSER_ADDRESS= - -# Optional: override the default public Polygon RPC with your own (Alchemy/Infura). -# POLYGON_RPC= - -# Optional: webhook (Discord/Telegram/ntfy) POST URL for critical alerts. -# ALERT_WEBHOOK_URL= - -# Optional: route outbound traffic through an SSH tunnel / proxy, e.g. to test -# from a colocated box. Standard env var; httpx and web3 pick it up automatically. -# ALL_PROXY=socks5://127.0.0.1:1080 diff --git a/.gitignore b/.gitignore index a09f149..730dc55 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +markets.csv output.txt data/ credentials.json diff --git a/livecfg/config.toml b/livecfg/config.toml new file mode 100644 index 0000000..ab04cb1 --- /dev/null +++ b/livecfg/config.toml @@ -0,0 +1,37 @@ +# CONSERVATIVE live-test config (Fable autonomous run, ~$50 mandate). +[wallet] +chain_id = 137 +signature_type = 3 +clob_host = "https://clob.polymarket.com" +gamma_host = "https://gamma-api.polymarket.com" +data_api_host = "https://data-api.polymarket.com" +polygon_rpc = "https://polygon-bor-rpc.publicnode.com" + +[engine] +debounce_ms = 250 +reconcile_interval_s = 20 +catalog_refresh_s = 300 +heartbeat = true +heartbeat_interval_s = 5 +journal = true +loop = "uvloop" + +[risk] +max_total_exposure_usdc = 40.0 # hard ceiling on deployed capital +max_event_group_loss_usdc = 30.0 +max_market_notional_usdc = 15.0 # per-market cap +daily_loss_kill_usdc = 12.0 # stop trading if we lose ~$12 +ws_stale_halt_s = 8.0 +user_ws_blind_halt_s = 12.0 +heartbeat_halt_failures = 3 +max_order_error_rate = 0.25 + +[execution] +rate_budget_fraction = 0.15 +post_only = true +max_orders_per_batch = 15 + +[paths] +db = "livecfg/state.db" +journal_dir = "livecfg/journal" +log_dir = "livecfg/logs" diff --git a/livecfg/markets.toml b/livecfg/markets.toml new file mode 100644 index 0000000..cba627e --- /dev/null +++ b/livecfg/markets.toml @@ -0,0 +1,9 @@ +[[markets]] +slug = "will-gavin-newsom-win-the-2028-democratic-presidential-nomination-568" +profile = "live-tiny" +enabled = true + +[[markets]] +slug = "will-jd-vance-win-the-2028-republican-presidential-nomination" +profile = "live-tiny" +enabled = true diff --git a/livecfg/strategy.toml b/livecfg/strategy.toml new file mode 100644 index 0000000..0c9542c --- /dev/null +++ b/livecfg/strategy.toml @@ -0,0 +1,27 @@ +# Tiny, defensive profile for the live test — small size, quick to pull. +[profiles.live-tiny] +micro_levels = 3 +flow_ewma_halflife_s = 90 +gamma = 0.8 +delta_min_ticks = 2 +c_vol = 1.5 +c_tox = 3.0 +vol_short_halflife_s = 10 +vol_long_halflife_s = 600 +base_size_usdc = 5.0 +q_max_usdc = 12.0 +q_soft_frac = 0.5 +layers = 1 +layer_step_ticks = 2 +reprice_ticks = 2 +resize_frac = 0.2 +min_edge_ticks = 1 +event_cooloff_s = 90 +event_jump_ticks = 6 +event_sweep_levels = 2 +trend_flow_z = 1.2 +end_date_taper_days = 7 +reduce_only_hours = 48 +halt_before_hours = 6 +exit_urgency_s = 600 +merge_min_size = 20.0 diff --git a/src/polymaker/alerts.py b/src/polymaker/alerts.py new file mode 100644 index 0000000..14b5ea5 --- /dev/null +++ b/src/polymaker/alerts.py @@ -0,0 +1,60 @@ +"""Critical-event alerting via a generic webhook POST. + +Fire-and-forget, deduped, and rate-limited so a flapping condition can't spam. +Works with any endpoint that accepts a JSON POST (Discord/Slack/Telegram-bridge/ +ntfy). No-ops cleanly when no webhook is configured, so callers never branch. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import time + +import httpx + +from polymaker.logging import get_logger + +log = get_logger("alerts") + + +class Alerter: + def __init__(self, webhook_url: str | None, *, min_interval_s: float = 30.0, + proxy: str | None = None) -> None: + self._url = webhook_url + self._min_interval = min_interval_s + self._proxy = proxy + self._last_sent: dict[str, float] = {} # key -> ts (dedupe/rate-limit) + + @property + def enabled(self) -> bool: + return bool(self._url) + + def alert(self, key: str, message: str, *, critical: bool = False) -> None: + """Queue an alert. `key` dedupes/rate-limits repeated conditions. + + Always logs (so nothing is lost even without a webhook); posts to the + webhook at most once per `min_interval_s` per key (critical bypasses the + limit). Safe to call from sync code — schedules the POST on the loop. + """ + (log.critical if critical else log.warning)("alert", key=key, msg=message) + if not self._url: + return + now = time.time() + if not critical and now - self._last_sent.get(key, 0.0) < self._min_interval: + return + self._last_sent[key] = now + with contextlib.suppress(RuntimeError): # no running loop (off-loop call) + asyncio.get_running_loop().create_task(self._post(key, message, critical)) + + async def _post(self, key: str, message: str, critical: bool) -> None: + text = f"{'🚨' if critical else '⚠️'} polymaker [{key}] {message}" + try: + kwargs: dict[str, object] = {"timeout": 10.0} + if self._proxy: + kwargs["proxy"] = self._proxy + async with httpx.AsyncClient(**kwargs) as c: # type: ignore[arg-type] + # send both keys so Slack ("text") and Discord ("content") work + await c.post(self._url, json={"text": text, "content": text}) # type: ignore[arg-type] + except httpx.HTTPError as exc: + log.warning("alert_post_failed", err=str(exc)) diff --git a/src/polymaker/catalog/gamma.py b/src/polymaker/catalog/gamma.py index a027dae..be3f055 100644 --- a/src/polymaker/catalog/gamma.py +++ b/src/polymaker/catalog/gamma.py @@ -39,6 +39,26 @@ class GammaClient: async def aclose(self) -> None: await self._client.aclose() + async def markets_by_condition(self, condition_ids: list[str]) -> dict[str, dict[str, Any]]: + """Fetch current raw market dicts for specific condition ids (metadata + refresh: detect closed/not-accepting/resolved and updated end dates).""" + out: dict[str, dict[str, Any]] = {} + if not condition_ids: + return out + try: + r = await self._client.get( + "/markets", + params={"condition_ids": ",".join(condition_ids), "limit": len(condition_ids) + 5}, + ) + r.raise_for_status() + for m in r.json(): + cid = m.get("conditionId") + if cid: + out[cid] = m + except (httpx.HTTPError, ValueError, KeyError) as exc: + log.warning("markets_by_condition_failed", err=str(exc)) + return out + async def resolve_tag_id(self, slug: str) -> str | None: try: r = await self._client.get(f"/tags/slug/{slug}") diff --git a/src/polymaker/catalog/store.py b/src/polymaker/catalog/store.py index cb80abc..a168943 100644 --- a/src/polymaker/catalog/store.py +++ b/src/polymaker/catalog/store.py @@ -7,6 +7,7 @@ file (state.db), queryable by the CLI. WAL mode so the running bot and a from __future__ import annotations +import csv import json import sqlite3 import time @@ -99,6 +100,32 @@ class CatalogStore: out.append((meta, sc)) return out + def export_csv(self, path: str | Path, limit: int = 500) -> int: + """Write the scored catalog to a CSV for easy market picking. + + Columns are chosen so you can eyeball reward/rebate income, cost (spread, + fee category), liquidity, and the exact slug/condition_id to paste into + markets.toml. Returns the number of rows written. + """ + rows = self.top(limit) + fields = [ + "score", "reward_per_day", "rebate_per_day", "spread", "best_bid", "best_ask", + "tick", "min_size", "neg_risk", "taker_fee_bps", "rewards_max_spread", + "liquidity", "volume", "end_date", "question", "slug", "condition_id", + ] + with open(path, "w", newline="") as fh: + w = csv.writer(fh) + w.writerow(fields) + for m, sc in rows: + w.writerow([ + f"{sc.score:.3f}", f"{m.rewards_daily_rate:.2f}", f"{sc.rebate_potential:.2f}", + f"{sc.spread:.4f}", m.best_bid, m.best_ask, f"{m.tick_size:g}", + f"{m.min_order_size:g}", int(m.neg_risk), m.taker_fee_bps, + m.rewards_max_spread, f"{m.liquidity_num:.0f}", f"{m.volume_num:.0f}", + m.end_date_iso or "", m.question, m.slug, m.condition_id, + ]) + return len(rows) + def cache_tag(self, slug: str, tag_id: str) -> None: self._conn.execute( "INSERT OR REPLACE INTO tags(slug, tag_id, ts) VALUES(?,?,?)", diff --git a/src/polymaker/cli.py b/src/polymaker/cli.py index 286fa3b..0fbf074 100644 --- a/src/polymaker/cli.py +++ b/src/polymaker/cli.py @@ -55,7 +55,11 @@ def scan( return len(metas) n = asyncio.run(_go()) - console.print(f"[green]Scanned and stored {n} markets.[/green] Run [bold]polymaker markets[/bold] to browse.") + csv_path = Path(config_dir).parent / "markets.csv" + written = store.export_csv(csv_path) + console.print(f"[green]Scanned and stored {n} markets.[/green] " + f"Wrote [bold]{csv_path}[/bold] ({written} rows) — open it, pick markets, " + f"then `polymaker markets-add `.") store.close() @@ -134,6 +138,48 @@ def status(config_dir: str = typer.Option("config", help="config directory")) -> store.close() +@app.command() +def pnl(config_dir: str = typer.Option("config", help="config directory")) -> None: + """Show PnL from the recorded snapshots (equity, daily PnL, fills).""" + import sqlite3 + + cfg = Config.load(config_dir) + conn = sqlite3.connect(cfg.paths.db) + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT ts, equity, net_cash, inventory_value, daily_pnl FROM pnl_snapshots " + "ORDER BY ts DESC LIMIT 1" + ).fetchall() + if not rows: + console.print("[yellow]No PnL snapshots yet (run the engine first).[/yellow]") + else: + r = rows[0] + color = "green" if r["daily_pnl"] >= 0 else "red" + console.print(f"[bold]equity:[/bold] {r['equity']:.4f} " + f"[bold]inventory:[/bold] {r['inventory_value']:.4f} " + f"[bold]net cash:[/bold] {r['net_cash']:.4f}") + console.print(f"[bold]daily PnL:[/bold] [{color}]{r['daily_pnl']:+.4f}[/{color}] pUSD") + nfills = conn.execute("SELECT COUNT(*) n FROM fills").fetchone()["n"] + console.print(f"[dim]total fills recorded: {nfills}[/dim]") + conn.close() + + +@app.command(name="export-csv") +def export_csv( + config_dir: str = typer.Option("config", help="config directory"), + out: str = typer.Option("markets.csv", help="output CSV path"), + limit: int = typer.Option(500, help="max rows"), +) -> None: + """Export the scored market catalog to a CSV for easy picking.""" + from polymaker.catalog.store import CatalogStore + + cfg = Config.load(config_dir) + store = CatalogStore(cfg.paths.db) + n = store.export_csv(out, limit) + store.close() + console.print(f"[green]Wrote {n} markets to {out}.[/green]") + + @app.command() def doctor(config_dir: str = typer.Option("config", help="config directory")) -> None: """Preflight checks: config, wallet auth, balance/allowance, WS reachability.""" diff --git a/src/polymaker/engine.py b/src/polymaker/engine.py index 550f426..6ac6cab 100644 --- a/src/polymaker/engine.py +++ b/src/polymaker/engine.py @@ -16,10 +16,11 @@ import time from datetime import datetime from typing import Any +from polymaker.alerts import Alerter from polymaker.catalog.gamma import GammaClient, fetch_reward_rates, parse_market from polymaker.catalog.store import CatalogStore from polymaker.config import Config, StrategyProfile -from polymaker.domain import Fill, MarketMeta +from polymaker.domain import Fill, MarketMeta, Regime from polymaker.execution.gateway import ExecutionGateway from polymaker.execution.reconciler import reconcile from polymaker.journal import Journal @@ -56,6 +57,7 @@ class Engine: self.gateway = ExecutionGateway(cfg, self.journal, paper=paper) self.risk = RiskManager(cfg.risk, self.state) self.merger = Merger(cfg) + self.alerter = Alerter(cfg.secrets.alert_webhook_url, proxy=cfg.proxy) self.md = MarketDataService(on_dirty=self._on_dirty, on_trade=self._on_trade, journal=self.journal, proxy=cfg.proxy) @@ -72,6 +74,9 @@ class Engine: self._sweep: dict[str, bool] = {} self._merging: set[str] = set() self._token_cid: dict[str, str] = {} + self._locks: dict[str, asyncio.Lock] = {} # per-market: serialize recompute vs reconcile + self._halted: set[str] = set() # markets closed/resolved/not-accepting + self._last_quote_fv: dict[str, float] = {} # requote suppression # supervised tasks: name -> (factory, task) so a dead task restarts self._task_specs: dict[str, Any] = {} self._tasks: dict[str, asyncio.Task[Any]] = {} @@ -80,6 +85,7 @@ class Engine: self._reconcile_now = asyncio.Event() self._user_started = False # user WS task launched (live mode) self._hb_was_down = False + self._chain_lock = asyncio.Lock() # serialize on-chain txs (nonce safety) # ── lifecycle ─────────────────────────────────────────────────────── async def start(self) -> None: @@ -105,9 +111,15 @@ class Engine: if not self.paper: assert self.user is not None self._spawn("user_ws", self.user.run) + # register the dead-man switch BEFORE any quoter can place an order, + # so a crash between placing and the first heartbeat still auto-cancels + with contextlib.suppress(Exception): + await self.gateway.heartbeat() self._spawn("heartbeat", self._heartbeat_loop) self._user_started = True self._spawn("reconcile", self._reconcile_loop) + self._spawn("metadata", self._metadata_refresh_loop) + self._spawn("maintenance", self._maintenance_loop) for cid in self.metas: self._spawn(f"quote:{cid[:8]}", lambda c=cid: self._quoter(c)) self._spawn("supervisor", self._supervise) @@ -133,6 +145,7 @@ class Engine: with contextlib.suppress(asyncio.CancelledError, asyncio.InvalidStateError): exc = task.exception() log.critical("task_died_restarting", task=name, err=str(exc) if exc else "exited") + self.alerter.alert("task_died", f"{name} died: {exc}", critical=True) self._tasks[name] = asyncio.create_task(self._task_specs[name](), name=name) async def run_forever(self) -> None: @@ -150,6 +163,7 @@ class Engine: t.cancel() with contextlib.suppress(Exception): await self.gateway.cancel_all() + self.gateway.close() self.journal.close() self.state.close() self.catalog.close() @@ -174,6 +188,7 @@ class Engine: self.est[meta.condition_id] = self._make_estimators(self.profiles[meta.condition_id]) self.regime_m[meta.condition_id] = RegimeMachine() self._dirty[meta.condition_id] = asyncio.Event() + self._locks[meta.condition_id] = asyncio.Lock() for tok in (meta.yes.token_id, meta.no.token_id): self._token_cid[tok] = meta.condition_id @@ -205,6 +220,23 @@ class Engine: async def _startup_reconcile(self) -> None: with contextlib.suppress(Exception): await self.gateway.cancel_all() # clean slate; heartbeat covers crashes + # cancel-all may have partially failed — verify no orders remain, and + # cancel/adopt any stragglers so we never quote on top of an unknown order + with contextlib.suppress(Exception): + leftover = await self.gateway.open_orders() + if leftover: + log.warning("startup_orders_remain", n=len(leftover)) + for tok in {o.token_id for o in leftover}: + await self.gateway.cancel_asset(tok) + still = await self.gateway.open_orders() + for tok in self._token_cid: + self.state.replace_open_orders( + tok, [o for o in still if o.token_id == tok], grace_s=0.0 + ) + if still: + log.error("startup_orders_stuck", n=len(still)) + self.alerter.alert("startup_orders_stuck", + f"{len(still)} orders survived cancel-all", critical=True) positions = await self.gateway.positions() if positions: self.state.reconcile_positions(positions) @@ -268,6 +300,13 @@ class Engine: await asyncio.sleep(0.5) async def _recompute(self, cid: str) -> None: + lock = self._locks.get(cid) + if lock is None: + return + async with lock: # serialize vs the reconcile loop mutating this market + await self._recompute_locked(cid) + + async def _recompute_locked(self, cid: str) -> None: meta = self.metas[cid] p = self.profiles[cid] yes_book = self.md.book(meta.yes.token_id) @@ -275,6 +314,11 @@ class Engine: if yes_book is None or yes_book.is_empty: return + # crossed/locked or one-sided book -> FV is unreliable; skip this tick + bb, ba = yes_book.best_bid(), yes_book.best_ask() + if bb is None or ba is None or bb.price >= ba.price: + return + now = time.time() micro = yes_book.microprice(p.micro_levels) if micro is None: @@ -309,13 +353,25 @@ class Engine: and self.cfg.engine.heartbeat and self.gateway.heartbeat_failures >= self.cfg.risk.heartbeat_halt_failures ) - blind = market_stale or user_blind or hb_blind + halted = cid in self._halted + blind = market_stale or user_blind or hb_blind or halted if blind: log.warning("market_blind", cid=cid[:8], market_stale=market_stale, - user_blind=user_blind, hb_blind=hb_blind) + user_blind=user_blind, hb_blind=hb_blind, halted=halted) + self.alerter.alert( + f"blind:{cid[:8]}", + f"{meta.question[:40]} blind (stale={market_stale} user={user_blind} " + f"hb={hb_blind} halted={halted})", + critical=hb_blind, + ) rd = self.risk.evaluate(meta, ws_stale=blind, event_group_cost=self._event_group_cost(meta)) + if rd.halt and rd.reason not in ("ws_stale",): + self.alerter.alert( + f"risk_halt:{rd.reason}", f"risk halt: {rd.reason}", + critical=any(k in rd.reason for k in ("daily_loss", "kill", "error_rate")), + ) ws_stale = blind regime = self.regime_m[cid].decide( RegimeInputs( @@ -353,18 +409,33 @@ class Engine: await self._refresh_token_orders(meta, grace_s=10.0) self._dirty[cid].set() return + placed_n = 0 if plan.to_place: - placed = await self.gateway.place(plan.to_place, meta) - self.risk.note_order_result(len(placed) == len(plan.to_place)) - for o in placed: - self.state.upsert_order(o) - if len(placed) < len(plan.to_place): - # QUARANTINE: a failed/partial batch may still have posted orders - # we don't have ids for. Cancel everything on these tokens - # (idempotent) and resync — never risk an untracked live order. - await self._quarantine(meta, reason="place_incomplete") + # LOAD SHED: under rate-budget pressure, skip *new* quotes in calm + # regimes (cancels/exits above already ran) so we don't inject latency + # right when the book is busy. Risk regimes always place. + shed = ( + not self.paper + and self.gateway.order_pressure > 0.85 + and regime in (Regime.QUIET, Regime.TRENDING) + ) + if shed: + log.warning("shed_load", cid=cid[:8], pressure=round(self.gateway.order_pressure, 2)) + self._dirty[cid].set() # retry soon + else: + placed = await self.gateway.place(plan.to_place, meta) + placed_n = len(placed) + self.risk.note_order_result(len(placed) == len(plan.to_place)) + for o in placed: + self.state.upsert_order(o) + if len(placed) < len(plan.to_place): + # QUARANTINE: a failed/partial batch may still have posted + # orders we don't have ids for. Cancel everything on these + # tokens (idempotent) and resync — never risk an untracked order. + await self._quarantine(meta, reason="place_incomplete") + self._last_quote_fv[cid] = fv log.info("requote", cid=cid[:8], regime=regime.value, fv=round(fv, 4), - place=len(plan.to_place), cancel=len(plan.to_cancel), + place=placed_n, cancel=len(plan.to_cancel), pos_yes=round(pos_yes.size, 1), pos_no=round(pos_no.size, 1)) self._maybe_merge(cid, meta, p, pos_yes.size, pos_no.size) @@ -395,8 +466,17 @@ class Engine: async def _merge_task(self, cid: str, meta: MarketMeta, amount: float) -> None: try: - raw = int(amount * 1e6) - await asyncio.to_thread(self.merger.merge, meta.condition_id, raw, meta.neg_risk) + # serialize all on-chain txs so concurrent merges can't reuse a nonce; + # read on-chain balances as source of truth for the mergeable amount + async with self._chain_lock: + bals = await self.gateway.token_balances([meta.yes.token_id, meta.no.token_id]) + if bals: + amount = min(amount, bals.get(meta.yes.token_id, 0.0), + bals.get(meta.no.token_id, 0.0)) + raw = int(amount * 1e6) + if raw <= 0: + return + await asyncio.to_thread(self.merger.merge, meta.condition_id, raw, meta.neg_risk) finally: self._merging.discard(cid) @@ -426,6 +506,7 @@ class Engine: await asyncio.sleep(self.cfg.engine.heartbeat_interval_s) async def _reconcile_loop(self) -> None: + rounds = 0 while self._running: # periodic cadence, but wake immediately when a reconnect/recovery # demands an urgent resync @@ -436,7 +517,15 @@ class Engine: ) forced = self._reconcile_now.is_set() self._reconcile_now.clear() + rounds += 1 try: + # a MATCHED whose settlement event was lost would block a token's + # reconciliation forever — expire stale in-flight guards first + expired = self.state.expire_inflight(self.cfg.engine.reconcile_interval_s * 2) + if expired: + self.alerter.alert("inflight_expired", + f"{len(expired)} stuck in-flight guards cleared") + positions = await self.gateway.positions() if positions: self.state.reconcile_positions(positions) @@ -444,12 +533,17 @@ class Engine: by_token: dict[str, list[Any]] = {} for o in live: by_token.setdefault(o.token_id, []).append(o) - # iterate ALL our tokens, not just those present in the REST - # response — a token whose orders all vanished server-side must - # be cleaned up too (grace window protects fresh placements) - for tok in self._token_cid: - if self.state.inflight(tok) == 0: - self.state.replace_open_orders(tok, by_token.get(tok, [])) + # iterate ALL our tokens, not just those in the REST response — a + # token whose orders vanished server-side must be cleaned up too. + # Hold the market lock so we don't race the quoter mid-flight. + for cid, meta in self.metas.items(): + lock = self._locks.get(cid) + if lock is None: + continue + async with lock: + for tok in (meta.yes.token_id, meta.no.token_id): + if self.state.inflight(tok) == 0: + self.state.replace_open_orders(tok, by_token.get(tok, [])) if forced: log.info("forced_reconcile_done", positions=len(positions), open_orders=len(live)) @@ -457,6 +551,116 @@ class Engine: except Exception as exc: # noqa: BLE001 log.warning("reconcile_error", err=str(exc)) + # slower loops: on-chain position divergence + pnl snapshot + WAL + if rounds % 4 == 0: + with contextlib.suppress(Exception): + await self._check_position_divergence() + self.state.record_pnl(self.risk.equity, self.risk.net_cash, + self.risk.inventory_value, self.risk.daily_pnl) + if rounds % 20 == 0: + self.state.checkpoint_wal() + + async def _check_position_divergence(self) -> None: + """Compare internal positions to on-chain truth; alert + correct on drift. + + Catches subtle fill-attribution bugs before they compound. On-chain is + authoritative (it's what the exchange settles), so we correct to it — + but only for tokens with no in-flight trades (optimistic state is newer). + """ + tokens = [t for t in self._token_cid if self.state.inflight(t) == 0] + onchain = await self.gateway.token_balances(tokens) + if not onchain: + return + for tok, chain_size in onchain.items(): + internal = self.state.position(tok).size + if abs(internal - chain_size) > max(1.0, 0.02 * chain_size): + log.error("position_divergence", token=tok[:12], + internal=round(internal, 2), onchain=round(chain_size, 2)) + self.alerter.alert( + f"divergence:{tok[:8]}", + f"position drift: internal {internal:.1f} vs on-chain {chain_size:.1f}", + critical=True, + ) + self.state.force_set_position(tok, chain_size, self.state.position(tok).avg_price, + source="onchain") + cid = self._token_cid.get(tok) + if cid: + self._wake_cid(cid) + + async def _metadata_refresh_loop(self) -> None: + """Refresh market metadata from Gamma: halt markets that have closed / + resolved / stopped accepting orders, and pick up updated end dates.""" + import dataclasses + + while self._running: + await asyncio.sleep(self.cfg.engine.catalog_refresh_s) + if not self.metas: + continue + try: + async with GammaClient(self.cfg.wallet.gamma_host) as gamma: + raws = await gamma.markets_by_condition(list(self.metas)) + except Exception as exc: # noqa: BLE001 + log.warning("metadata_refresh_error", err=str(exc)) + continue + for cid, raw in raws.items(): + if cid not in self.metas: + continue + accepting = bool(raw.get("acceptingOrders", True)) + closed = bool(raw.get("closed", False)) + if closed or not accepting: + if cid not in self._halted: + self._halted.add(cid) + log.critical("market_halted_by_meta", cid=cid[:8], closed=closed, + accepting=accepting) + self.alerter.alert(f"halted:{cid[:8]}", + f"{self.metas[cid].question[:40]} closed/not-accepting", + critical=True) + meta = self.metas[cid] + for tok in (meta.yes.token_id, meta.no.token_id): + with contextlib.suppress(Exception): + await self.gateway.cancel_asset(tok) + self._wake_cid(cid) + else: + self._halted.discard(cid) + new_end = raw.get("endDate") + if new_end and new_end != self.metas[cid].end_date_iso: + self.metas[cid] = dataclasses.replace(self.metas[cid], end_date_iso=new_end) + + async def _maintenance_loop(self) -> None: + """Periodic REST book refresh to catch any silently-missed WS deltas.""" + while self._running: + await asyncio.sleep(120.0) + for meta in list(self.metas.values()): + for tok in (meta.yes.token_id, meta.no.token_id): + with contextlib.suppress(Exception): + await self._refresh_book(tok) + + async def _refresh_book(self, token_id: str) -> None: + levels = await self.gateway.get_full_book(token_id) + if levels is None: + return + bids, asks, book_hash = levels + book = self.md.book(token_id) + if book is None: + return + # drift check: only overwrite if the REST top-of-book disagrees with ours + cur_bb = book.best_bid() + cur_ba = book.best_ask() + rest_bb = max((p for p, _ in bids), default=None) + rest_ba = min((p for p, _ in asks), default=None) + drift = ( + (cur_bb is None) != (rest_bb is None) + or (cur_ba is None) != (rest_ba is None) + or (cur_bb and rest_bb and abs(cur_bb.price - rest_bb) > book.tick_size) + or (cur_ba and rest_ba and abs(cur_ba.price - rest_ba) > book.tick_size) + ) + if drift: + log.warning("book_drift_corrected", token=token_id[:12]) + book.apply_snapshot(bids, asks, time.time(), book_hash) + cid = self._token_cid.get(token_id) + if cid: + self._wake_cid(cid) + # ── helpers ───────────────────────────────────────────────────────── def _other_token(self, token_id: str) -> str | None: cid = self._token_cid.get(token_id) diff --git a/src/polymaker/execution/gateway.py b/src/polymaker/execution/gateway.py index 711dae3..54331d3 100644 --- a/src/polymaker/execution/gateway.py +++ b/src/polymaker/execution/gateway.py @@ -14,8 +14,10 @@ from __future__ import annotations import asyncio import itertools import time +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict -from typing import Any +from typing import Any, TypeVar import httpx @@ -27,6 +29,8 @@ from polymaker.logging import get_logger log = get_logger("execution.gateway") +_T = TypeVar("_T") + def _tick_str(tick: float) -> str: return f"{tick:g}" @@ -55,11 +59,27 @@ class ExecutionGateway: self._paper_ids = itertools.count(1) self._hb_id: str = "" # heartbeat chain self._hb_failures: int = 0 + # dedicated, bounded pool for blocking order/HTTP calls so a burst of + # requotes across many markets can't starve the default executor + self._pool = ThreadPoolExecutor(max_workers=8, thread_name_prefix="clob-io") @property def paper(self) -> bool: return self._paper + @property + def order_pressure(self) -> float: + """0 = plenty of order-post budget, 1 = about to queue (shed load).""" + return self._order_bucket.pressure + + def close(self) -> None: + self._pool.shutdown(wait=False, cancel_futures=True) + + async def _io(self, fn: Callable[..., _T], *args: Any) -> _T: + """Run a blocking client call on the dedicated pool.""" + loop = asyncio.get_running_loop() + return await loop.run_in_executor(self._pool, fn, *args) + @property def creds(self) -> Any: return self._creds @@ -94,6 +114,9 @@ class ExecutionGateway: host=self._cfg.wallet.clob_host, chain_id=self._cfg.wallet.chain_id, key=sec.pk, + # use_server_time=False: fetching /time before EVERY signed order + # adds a full round-trip per op (latency killer through a proxy). + # We check clock drift once below and rely on the local (NTP) clock. signature_type=self._cfg.wallet.signature_type, funder=sec.browser_address, ) @@ -101,12 +124,28 @@ class ExecutionGateway: client.set_api_creds(creds) return client, creds, client.get_address() - self._client, self._creds, self._address = await asyncio.to_thread(_build) + self._client, self._creds, self._address = await self._io(_build) + await self._check_clock_drift() # funds/positions live on the funder (proxy/deposit wallet); fall back to EOA self._funder = sec.browser_address or self._address log.info("gateway_connected", signer=self._address[:10], funder=self._funder[:10], paper=self._paper) + async def _check_clock_drift(self) -> None: + """Warn once if the local clock is skewed vs the exchange (affects L2 auth).""" + try: + async with httpx.AsyncClient(timeout=10.0) as c: + r = await c.get(f"{self._cfg.wallet.clob_host}/time") + server = float(r.text.strip().strip('"')) + drift = abs(time.time() - server) + if drift > 5.0: + log.warning("clock_drift", drift_s=round(drift, 1), + note="sync system clock (NTP) — large skew can fail order auth") + else: + log.info("clock_ok", drift_s=round(drift, 1)) + except (httpx.HTTPError, ValueError) as exc: + log.warning("clock_check_failed", err=str(exc)) + # ── placement ─────────────────────────────────────────────────────── async def place(self, quotes: list[Quote], meta: MarketMeta) -> list[OpenOrder]: if not quotes: @@ -138,7 +177,7 @@ class ExecutionGateway: return self._parse_place_response(resp, quotes) try: - return await asyncio.to_thread(_place) + return await self._io(_place) except Exception as exc: # noqa: BLE001 - surface + continue; engine handles error rate log.error("place_failed", err=str(exc), n=len(quotes)) return [] @@ -172,7 +211,7 @@ class ExecutionGateway: self._client.cancel_orders(order_ids) try: - await asyncio.to_thread(_cancel) + await self._io(_cancel) return True except Exception as exc: # noqa: BLE001 log.error("cancel_failed", err=str(exc), n=len(order_ids)) @@ -189,7 +228,7 @@ class ExecutionGateway: self._client.cancel_market_orders(OrderMarketCancelParams(asset_id=asset_id)) try: - await asyncio.to_thread(_cancel) + await self._io(_cancel) return True except Exception as exc: # noqa: BLE001 log.error("cancel_asset_failed", err=str(exc), token=asset_id[:12]) @@ -198,7 +237,7 @@ class ExecutionGateway: async def cancel_all(self) -> None: if self._paper or self._client is None: return - await asyncio.to_thread(self._client.cancel_all) + await self._io(self._client.cancel_all) log.info("cancel_all_sent") # ── market (taker) orders — used by moneydoctor, NOT the maker strategy ── @@ -232,7 +271,7 @@ class ExecutionGateway: except Exception as exc: # noqa: BLE001 - surface as data, never crash the caller return {"status": "failed", "error": str(exc)} - return await asyncio.to_thread(_do) + return await self._io(_do) async def get_book(self, token_id: str) -> dict[str, float]: """Live best bid/ask + touch depth for one token (public REST).""" @@ -254,6 +293,24 @@ class ExecutionGateway: log.warning("get_book_failed", err=str(exc)) return {} + async def get_full_book( + self, token_id: str + ) -> tuple[list[tuple[float, float]], list[tuple[float, float]], str | None] | None: + """Full L2 book (bids, asks, hash) via public REST — for periodic + integrity refresh against the WS book.""" + try: + async with httpx.AsyncClient(timeout=15.0) as c: + r = await c.get(f"{self._cfg.wallet.clob_host}/book", + params={"token_id": token_id}) + r.raise_for_status() + b = r.json() + bids = [(float(x["price"]), float(x["size"])) for x in b.get("bids", [])] + asks = [(float(x["price"]), float(x["size"])) for x in b.get("asks", [])] + return bids, asks, b.get("hash") + except (httpx.HTTPError, KeyError, ValueError) as exc: + log.warning("get_full_book_failed", err=str(exc)) + return None + async def token_balance(self, token_id: str) -> float: """Exact on-chain conditional-token balance (shares) held by the funder. @@ -291,11 +348,54 @@ class ExecutionGateway: return None try: - return await asyncio.to_thread(_read) + return await self._io(_read) except Exception as exc: # noqa: BLE001 log.warning("token_balance_failed", err=str(exc)) return None + async def token_balances(self, token_ids: list[str]) -> dict[str, float] | None: + """Batch on-chain balances for several tokens in one RPC session. + + Used by the position-divergence monitor. Returns None on RPC failure. + """ + if not token_ids: + return {} + + def _read() -> dict[str, float] | None: + from web3 import Web3 + from web3.middleware import ExtraDataToPOAMiddleware + + configured = self._cfg.secrets.polygon_rpc or self._cfg.wallet.polygon_rpc + rpcs = [configured, "https://polygon-bor-rpc.publicnode.com", + "https://polygon.llamarpc.com", "https://rpc.ankr.com/polygon"] + abi = [{"name": "balanceOf", "type": "function", "stateMutability": "view", + "inputs": [{"name": "a", "type": "address"}, {"name": "id", "type": "uint256"}], + "outputs": [{"name": "", "type": "uint256"}]}] + funder = None + for rpc in dict.fromkeys(rpcs): + try: + w3 = Web3(Web3.HTTPProvider(rpc, request_kwargs={"timeout": 20})) + w3.middleware_onion.inject(ExtraDataToPOAMiddleware, layer=0) + ctf = w3.eth.contract( + address=Web3.to_checksum_address("0x4D97DCd97eC945f40cF65F87097ACe5EA0476045"), + abi=abi, + ) + funder = Web3.to_checksum_address(self.funder) + out: dict[str, float] = {} + for tid in token_ids: + raw = ctf.functions.balanceOf(funder, int(tid)).call() + out[tid] = float(raw) / 1e6 + return out + except Exception: # noqa: BLE001, PERF203 + continue + return None + + try: + return await self._io(_read) + except Exception as exc: # noqa: BLE001 + log.warning("token_balances_failed", err=str(exc)) + return None + async def collateral_balance(self) -> float: """pUSD balance (float) on the funder.""" ba = await self.balance_allowance() @@ -324,7 +424,7 @@ class ExecutionGateway: return self._client.post_heartbeat(self._hb_id) try: - resp = await asyncio.to_thread(_beat) + resp = await self._io(_beat) new_id = _first(resp, "heartbeat_id", "heartbeatId", "id") self._hb_id = str(new_id) if new_id else "" if self._hb_failures: @@ -371,7 +471,7 @@ class ExecutionGateway: return out try: - return await asyncio.to_thread(_get) + return await self._io(_get) except Exception as exc: # noqa: BLE001 log.warning("open_orders_failed", err=str(exc)) return [] @@ -411,7 +511,7 @@ class ExecutionGateway: return result try: - return await asyncio.to_thread(_get) + return await self._io(_get) except Exception as exc: # noqa: BLE001 log.warning("balance_allowance_failed", err=str(exc)) return {} diff --git a/src/polymaker/risk/manager.py b/src/polymaker/risk/manager.py index 50e1b4a..dca7224 100644 --- a/src/polymaker/risk/manager.py +++ b/src/polymaker/risk/manager.py @@ -51,6 +51,14 @@ class RiskManager: total += pos.size * self._marks.get(tok, pos.avg_price) return total + @property + def net_cash(self) -> float: + return self._net_cash + + @property + def inventory_value(self) -> float: + return self._inventory_value() + @property def equity(self) -> float: return self._net_cash + self._inventory_value() @@ -116,13 +124,15 @@ class RiskManager: return RiskDecision(False, False, scale, "") def _market_notional(self, meta: MarketMeta) -> float: + """Filled-inventory notional for this market. Deliberately does NOT count + our own resting BUY orders: those are the quotes we're about to replace, + and counting them makes the size taper collapse the moment we place a full + quote (self-reinforcing cancel/replace churn). Worst-case fill is bounded + instead by small per-quote sizes + the position cap that this drives.""" total = 0.0 for tok in (meta.yes.token_id, meta.no.token_id): pos = self._store.position(tok) total += pos.size * self._marks.get(tok, pos.avg_price or 0.5) - for o in self._store.orders_for(tok): - if o.side is Side.BUY: - total += o.notional return total def _total_exposure(self) -> float: @@ -130,9 +140,6 @@ class RiskManager: for tok, pos in self._store.positions.items(): if pos.size > 0: total += pos.size * self._marks.get(tok, pos.avg_price or 0.5) - for o in self._store.orders.values(): - if o.side is Side.BUY: - total += o.notional return total diff --git a/src/polymaker/state/store.py b/src/polymaker/state/store.py index c8fafb4..2663870 100644 --- a/src/polymaker/state/store.py +++ b/src/polymaker/state/store.py @@ -12,6 +12,7 @@ In-memory + typed, mirrored to SQLite on change so a crash-restart resumes. from __future__ import annotations +import contextlib import json import sqlite3 import time @@ -37,6 +38,10 @@ CREATE TABLE IF NOT EXISTS order_log ( order_id TEXT PRIMARY KEY, token_id TEXT, side TEXT, price REAL, size REAL, state TEXT, ts REAL ); +CREATE TABLE IF NOT EXISTS pnl_snapshots ( + ts REAL PRIMARY KEY, + equity REAL, net_cash REAL, inventory_value REAL, daily_pnl REAL +); """ @@ -55,6 +60,7 @@ class StateStore: self.orders: dict[str, OpenOrder] = {} # token_id -> count of in-flight (MATCHED-not-CONFIRMED) trades; guards reconcile self._inflight: dict[str, int] = {} + self._inflight_ts: dict[str, float] = {} # oldest in-flight mark, for expiry self._last_fill_ts: dict[str, float] = {} self._load() @@ -124,14 +130,33 @@ class StateStore: # ── in-flight guard ───────────────────────────────────────────────── def mark_inflight(self, token_id: str) -> None: self._inflight[token_id] = self._inflight.get(token_id, 0) + 1 + self._inflight_ts.setdefault(token_id, time.time()) def clear_inflight(self, token_id: str) -> None: if self._inflight.get(token_id, 0) > 0: self._inflight[token_id] -= 1 + if self._inflight.get(token_id, 0) == 0: + self._inflight_ts.pop(token_id, None) def inflight(self, token_id: str) -> int: return self._inflight.get(token_id, 0) + def expire_inflight(self, max_age_s: float) -> list[str]: + """Force-clear in-flight guards older than max_age_s. + + A MATCHED whose CONFIRMED/FAILED never arrives (dropped WS event) would + otherwise block reconciliation for that token forever. Returns the + tokens cleared so the engine can force an authoritative REST reconcile. + """ + now = time.time() + stale = [t for t, ts in self._inflight_ts.items() if now - ts > max_age_s] + for t in stale: + age = round(now - self._inflight_ts[t]) + self._inflight[t] = 0 + self._inflight_ts.pop(t, None) + log.warning("inflight_expired", token=t[:12], age_s=age) + return stale + # ── orders ────────────────────────────────────────────────────────── def orders_for(self, token_id: str) -> list[OpenOrder]: return [o for o in self.orders.values() if o.token_id == token_id] @@ -195,7 +220,28 @@ class StateStore: row["token_id"], row["size"], row["avg_price"] ) - # ── reporting ─────────────────────────────────────────────────────── + def force_set_position(self, token_id: str, size: float, avg_price: float, source: str) -> None: + """Overwrite a position unconditionally (used when on-chain is truth).""" + prev = self.positions.get(token_id) + self.set_position(token_id, size, avg_price) + log.warning("position_forced", token=token_id[:12], source=source, + prev=round(prev.size, 2) if prev else 0.0, now=round(size, 2)) + + # ── maintenance / reporting ───────────────────────────────────────── + def checkpoint_wal(self) -> None: + """Truncate the WAL so it can't grow without bound under high volume.""" + with contextlib.suppress(sqlite3.Error): + self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)") + + def record_pnl(self, equity: float, net_cash: float, inv_value: float, daily_pnl: float) -> None: + with contextlib.suppress(sqlite3.Error): + self._conn.execute( + "INSERT OR REPLACE INTO pnl_snapshots(ts,equity,net_cash,inventory_value,daily_pnl)" + " VALUES(?,?,?,?,?)", + (time.time(), equity, net_cash, inv_value, daily_pnl), + ) + self._conn.commit() + def snapshot(self) -> dict[str, object]: return { "positions": {k: json.loads(_pos_json(v)) for k, v in self.positions.items() if v.size > 0}, diff --git a/src/polymaker/strategy/estimators.py b/src/polymaker/strategy/estimators.py index 514ee89..af791ec 100644 --- a/src/polymaker/strategy/estimators.py +++ b/src/polymaker/strategy/estimators.py @@ -9,7 +9,7 @@ observations with timestamps, read scalar summaries. No I/O. from __future__ import annotations import math -from dataclasses import dataclass, field +from dataclasses import dataclass from polymaker.domain import Side @@ -186,7 +186,6 @@ class MarketEstimators: markout: MarkoutTracker last_fv: float | None = None last_fv_ts: float = 0.0 - fv_history: list[tuple[float, float]] = field(default_factory=list) def on_fair_value(self, fv: float, ts: float) -> None: self.vol.update(fv, ts) diff --git a/src/polymaker/strategy/quoting.py b/src/polymaker/strategy/quoting.py index 2b2e3ac..92f6a40 100644 --- a/src/polymaker/strategy/quoting.py +++ b/src/polymaker/strategy/quoting.py @@ -198,6 +198,9 @@ def _maybe_exit( if view.best_bid is not None: target = max(target, view.best_bid + tick) price = round_to_tick(target, tick, dec, up=True) - size = round(pos.size, 2) + # FLOOR (never round up): selling more than we hold is rejected by the + # exchange -> the exit silently fails and we stay long. Floor guarantees + # size <= held. + size = math.floor(pos.size * 100) / 100 if 0 < price < 1 and size >= m.min_order_size: quotes.append(Quote(token_id, Side.SELL, price, size)) diff --git a/tests/test_engine.py b/tests/test_engine.py index af775ed..ea924cf 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -24,6 +24,7 @@ def _engine_with_market(tmp_path, meta) -> Engine: eng.est[cid] = Engine._make_estimators(eng.profiles[cid]) eng.regime_m[cid] = RegimeMachine() eng._dirty[cid] = asyncio.Event() + eng._locks[cid] = asyncio.Lock() for tok in (meta.yes.token_id, meta.no.token_id): eng._token_cid[tok] = cid eng.md.set_markets([(cid, [meta.yes.token_id, meta.no.token_id])]) diff --git a/tests/test_hardening2.py b/tests/test_hardening2.py new file mode 100644 index 0000000..cfa3e5b --- /dev/null +++ b/tests/test_hardening2.py @@ -0,0 +1,185 @@ +"""Hardening batch 2: Tier 0-3 fixes — inflight expiry, crossed-book guard, +metadata halt, load shed, exit floor, divergence correction, per-market lock, +CSV export, WAL/pnl.""" + +from __future__ import annotations + +import asyncio +import time + +from polymaker.domain import Fill, Position, Regime, Side +from polymaker.state.store import StateStore +from polymaker.strategy.quoting import QuoteInputs, construct_quotes +from tests.conftest import view +from tests.test_engine import _engine_with_market, _feed_book + + +# ── T0-1: inflight expiry ──────────────────────────────────────────────── +def test_inflight_expires_after_max_age(tmp_path): + s = StateStore(tmp_path / "s.db") + s.mark_inflight("tok") + assert s.inflight("tok") == 1 + # not yet stale + assert s.expire_inflight(max_age_s=100) == [] + assert s.inflight("tok") == 1 + # force age by rewriting the stored ts + s._inflight_ts["tok"] = time.time() - 999 + cleared = s.expire_inflight(max_age_s=100) + assert cleared == ["tok"] + assert s.inflight("tok") == 0 + s.close() + + +# ── T0-7: exit sizing floors (never over-sell) ─────────────────────────── +def test_exit_size_is_floored(meta, profile): + # hold a fractional position; the SELL must be floored so size <= held + tq = construct_quotes(QuoteInputs( + meta=meta, regime=Regime.REDUCE_ONLY, fv=0.5, vol_short=0.0, toxicity=0.0, + yes_view=view(0.49, 0.51), no_view=view(0.49, 0.51), + pos_yes=Position("yes-token", 17.999, 0.4), pos_no=Position("no-token"), + profile=profile, now=1000.0, + )) + sells = [q for q in tq.quotes if q.side == Side.SELL] + assert sells + assert sells[0].size <= 17.999 # floored, never rounded up past the holding + assert sells[0].size == 17.99 + + +# ── T0-5: crossed-book guard ───────────────────────────────────────────── +async def test_crossed_book_skips_quoting(tmp_path, meta): + eng = _engine_with_market(tmp_path, meta) + now = time.time() + # crossed: best bid (0.55) above best ask (0.45) + eng.md.book(meta.yes.token_id).apply_snapshot(bids=[(0.55, 100)], asks=[(0.45, 100)], ts=now) + eng.md.book(meta.no.token_id).apply_snapshot(bids=[(0.45, 100)], asks=[(0.55, 100)], ts=now) + await eng._recompute(meta.condition_id) + assert eng.state.orders == {} # no quotes on a nonsensical book + eng.state.close() + eng.catalog.close() + + +# ── T0-2: metadata halt pulls quotes ───────────────────────────────────── +async def test_halted_market_pulls_quotes(tmp_path, meta): + eng = _engine_with_market(tmp_path, meta) + _feed_book(eng, meta) + await eng._recompute(meta.condition_id) + assert len(eng.state.orders) > 0 # quoting normally + # market flagged closed/not-accepting by the metadata refresh + eng._halted.add(meta.condition_id) + await eng._recompute(meta.condition_id) + assert eng.state.orders == {} # all pulled + eng.state.close() + eng.catalog.close() + + +# ── T1: load shedding under order pressure ─────────────────────────────── +async def test_load_shed_skips_new_quotes_under_pressure(tmp_path, meta): + eng = _engine_with_market(tmp_path, meta) + eng.paper = False # shed only applies live + _feed_book(eng, meta) + placed_calls: list[int] = [] + + async def spy_place(quotes, m): + placed_calls.append(len(quotes)) + return [] + + async def no_cancel(ids): + return True + + # force high pressure + for _ in range(1000): + eng.gateway._order_bucket._tokens = 0.0 + eng.gateway.place = spy_place # type: ignore[method-assign] + eng.gateway.cancel = no_cancel # type: ignore[method-assign] + await eng._recompute(meta.condition_id) + assert eng.gateway.order_pressure > 0.85 + assert placed_calls == [] # new quotes shed, not placed + eng.state.close() + eng.catalog.close() + + +# ── per-market lock serializes recompute vs reconcile ──────────────────── +async def test_recompute_holds_market_lock(tmp_path, meta): + eng = _engine_with_market(tmp_path, meta) + _feed_book(eng, meta) + lock = eng._locks[meta.condition_id] + await lock.acquire() # simulate reconcile holding it + + async def try_recompute(): + await eng._recompute(meta.condition_id) + + task = asyncio.create_task(try_recompute()) + await asyncio.sleep(0.05) + assert not task.done() # blocked on the lock + lock.release() + await task # now proceeds + eng.state.close() + eng.catalog.close() + + +# ── T1: on-chain divergence correction ─────────────────────────────────── +async def test_divergence_corrects_to_onchain(tmp_path, meta): + eng = _engine_with_market(tmp_path, meta) + tok = meta.yes.token_id + eng.state.apply_fill(Fill(tok, Side.BUY, 0.5, 100, "phantom")) # internal says 100 + assert eng.state.position(tok).size == 100 + + async def fake_balances(tokens): + return {t: (5.0 if t == tok else 0.0) for t in tokens} # chain says 5 + + eng.gateway.token_balances = fake_balances # type: ignore[method-assign] + await eng._check_position_divergence() + assert eng.state.position(tok).size == 5.0 # corrected to on-chain truth + eng.state.close() + eng.catalog.close() + + +# ── churn bug: resting orders must NOT shrink the size taper ───────────── +def test_open_orders_do_not_taper_quote_size(tmp_path, meta): + """Regression: counting our own resting BUY orders toward market notional + collapsed the next quote size to ~0 -> empty targets -> cancel/replace churn. + Full resting quotes with zero filled inventory must keep size_scale = 1.0.""" + from polymaker.config import RiskConfig + from polymaker.domain import OpenOrder, OrderState + from polymaker.risk.manager import RiskManager + + store = StateStore(tmp_path / "s.db") + rm = RiskManager(RiskConfig(max_market_notional_usdc=15.0), store) + rm.update_mark(meta.yes.token_id, 0.2) + rm.update_mark(meta.no.token_id, 0.8) + # rest ~$14 of BUY orders (near cap) but hold NO inventory + store.upsert_order(OpenOrder("y", meta.yes.token_id, Side.BUY, 0.2, 50, OrderState.LIVE)) + store.upsert_order(OpenOrder("n", meta.no.token_id, Side.BUY, 0.79, 6, OrderState.LIVE)) + d = rm.evaluate(meta, ws_stale=False, event_group_cost=0.0) + assert not d.reduce_only + assert d.size_scale == 1.0 # resting orders do not taper -> no churn + # but FILLED inventory near cap DOES taper + store.apply_fill(Fill(meta.yes.token_id, Side.BUY, 0.2, 70, "f")) # $14 position + d2 = rm.evaluate(meta, ws_stale=False, event_group_cost=0.0) + assert d2.size_scale < 1.0 + store.close() + + +# ── T2: PnL snapshot + CSV export smoke ────────────────────────────────── +def test_pnl_snapshot_and_wal(tmp_path): + s = StateStore(tmp_path / "s.db") + s.record_pnl(100.0, 50.0, 50.0, 1.5) + s.checkpoint_wal() # must not raise + row = s._conn.execute("SELECT equity, daily_pnl FROM pnl_snapshots").fetchone() + assert row["equity"] == 100.0 and row["daily_pnl"] == 1.5 + s.close() + + +def test_catalog_csv_export(tmp_path): + from polymaker.catalog.gamma import parse_market + from polymaker.catalog.store import CatalogStore + from tests.test_catalog import RAW + + store = CatalogStore(tmp_path / "c.db") + store.upsert_market(parse_market(RAW, {"0xabc": 42.0})) + out = tmp_path / "markets.csv" + n = store.export_csv(out) + assert n == 1 + text = out.read_text() + assert "slug" in text and "will-x-win" in text and "condition_id" in text + store.close()