Hardening

.env.example

@@ -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

.gitignore

@@ -1,3 +1,4 @@
+markets.csv
 output.txt
 data/
 credentials.json

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"

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

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

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

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}")

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(?,?,?)",

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 <slug>`.")
     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."""

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)

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 {}

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
 
 

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},

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)

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

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

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