Mastering the Orderbook

from dataclasses import dataclass
from typing import Optional

@dataclass
class EdgeResult:
    fair_prob: float
    fill_price: float
    raw_edge: float          # fair_prob - fill_price
    net_edge: float          # after fee and latency buffers
    tradeable: bool

FEE_BUFFER    = 0.002   # ~0.2 c round-trip fee estimate
LATENCY_BUFFER = 0.003  # conservative latency adverse-selection buffer
TICK_SIZE     = 0.01    # Polymarket minimum price increment

def compute_taker_buy_edge(
    fair_prob: float,
    effective_ask: float,   # depth-walked fill price, NOT midpoint
    fee_buffer: float = FEE_BUFFER,
    latency_buffer: float = LATENCY_BUFFER,
) -> EdgeResult:
    raw   = fair_prob - effective_ask
    net   = raw - fee_buffer - latency_buffer
    # Edge must exceed one tick to be meaningful
    return EdgeResult(
        fair_prob=fair_prob,
        fill_price=effective_ask,
        raw_edge=round(raw, 6),
        net_edge=round(net, 6),
        tradeable=(net > TICK_SIZE),
    )

# Example: looks bullish on midpoint, negative on ask
result = compute_taker_buy_edge(fair_prob=0.53, effective_ask=0.54)
print(result)
# EdgeResult(fair_prob=0.53, fill_price=0.54, raw_edge=-0.01,
#            net_edge=-0.015, tradeable=False)

import httpx
from datetime import datetime, timezone
from typing import TypedDict

GAMMA_API = "https://gamma-api.polymarket.com"
CLOB_API  = "https://clob.polymarket.com"

class TokenMap(TypedDict):
    UP:   str
    DOWN: str

class MarketMeta(TypedDict):
    condition_id:       str
    question:           str
    tokens:             TokenMap
    end_time_utc:       str
    tick_size:          float
    min_order_size:     float
    neg_risk:           bool
    enable_order_book:  bool

def discover_btc_updown_market(slug: str) -> MarketMeta:
    """Fetch metadata for a BTC Up/Down market by event slug."""
    resp = httpx.get(f"{GAMMA_API}/markets", params={"slug": slug}, timeout=10)
    resp.raise_for_status()
    markets = resp.json()
    if not markets:
        raise ValueError(f"No market found for slug: {slug}")

    m = markets[0]  # take the first matching market

    # Build explicit token map — never assume index order
    outcomes   = m["outcomes"]          # e.g. ["Up", "Down"] or ["Down", "Up"]
    token_ids  = m["clobTokenIds"]      # same length, same order
    token_map: TokenMap = {}
    for outcome, tid in zip(outcomes, token_ids):
        key = outcome.strip().upper()   # normalise to "UP" or "DOWN"
        if key in ("UP", "DOWN"):
            token_map[key] = tid

    if set(token_map.keys()) != {"UP", "DOWN"}:
        raise ValueError(f"Unexpected outcomes: {outcomes}")

    return MarketMeta(
        condition_id      = m["conditionId"],
        question          = m["question"],
        tokens            = token_map,
        end_time_utc      = m["endDateIso"],
        tick_size         = float(m.get("tickSize", 0.01)),
        min_order_size    = float(m.get("minOrderSize", 1)),
        neg_risk          = bool(m.get("negRisk", False)),
        enable_order_book = bool(m.get("enableOrderBook", True)),
    )

if __name__ == "__main__":
    meta = discover_btc_updown_market("btc-up-down-5-minute")
    print(meta)

from datetime import datetime, timezone

def seconds_remaining(end_time_utc: str) -> float:
    """Return tau: seconds until market close. Returns 0.0 if already closed."""
    end = datetime.fromisoformat(end_time_utc.replace("Z", "+00:00"))
    now = datetime.now(timezone.utc)
    return max(0.0, (end - now).total_seconds())

# Usage
tau = seconds_remaining(meta["end_time_utc"])
print(f"tau = {tau:.1f}s")

import time
from collections import defaultdict
from typing import Literal, Optional

Side = Literal["bids", "asks"]

class LocalBook:
    """Level-2 orderbook for a single token, maintained via WS deltas."""

    STALE_THRESHOLD_S = 5.0  # resync if no update for this many seconds

    def __init__(self, token_id: str):
        self.token_id   = token_id
        self.bids: dict[float, float] = {}  # price -> size
        self.asks: dict[float, float] = {}
        self._last_update = 0.0
        self._snapshot_ts = 0.0

    def load_snapshot(self, bids: list[dict], asks: list[dict]) -> None:
        """Initialise from REST /book response."""
        self.bids = {float(b["price"]): float(b["size"]) for b in bids}
        self.asks = {float(a["price"]): float(a["size"]) for a in asks}
        now = time.monotonic()
        self._last_update = now
        self._snapshot_ts = now

    def apply_delta(self, side: str, price: float, size: float) -> None:
        """Apply a single price_change event from the WebSocket."""
        book = self.bids if side.upper() == "BUY" else self.asks
        if size == 0.0:
            book.pop(price, None)   # remove zero-size level
        else:
            book[price] = size      # upsert
        self._last_update = time.monotonic()

    @property
    def is_stale(self) -> bool:
        return (time.monotonic() - self._last_update) > self.STALE_THRESHOLD_S

    def best_bid(self) -> Optional[float]:
        return max(self.bids) if self.bids else None

    def best_ask(self) -> Optional[float]:
        return min(self.asks) if self.asks else None

    def spread(self) -> Optional[float]:
        bb, ba = self.best_bid(), self.best_ask()
        return round(ba - bb, 6) if bb and ba else None

import asyncio
import json
import time
import httpx
import websockets
from book import LocalBook

CLOB_REST = "https://clob.polymarket.com"
CLOB_WS   = "wss://ws-subscriptions-clob.polymarket.com/ws/market"

async def fetch_snapshot(token_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{CLOB_REST}/book", params={"token_id": token_id})
        r.raise_for_status()
        return r.json()

async def run_book_feed(
    up_token_id: str,
    down_token_id: str,
    on_update,          # async callback(up_book, down_book, recv_ts)
) -> None:
    books = {
        up_token_id:   LocalBook(up_token_id),
        down_token_id: LocalBook(down_token_id),
    }

    # 1. Seed both books from REST before opening WebSocket
    for tid, book in books.items():
        snap = await fetch_snapshot(tid)
        book.load_snapshot(snap.get("bids", []), snap.get("asks", []))

    subscribe_msg = json.dumps({
        "auth": {},
        "type": "Market",
        "assets_ids": [up_token_id, down_token_id],
    })

    async for ws in websockets.connect(CLOB_WS, ping_interval=20):
        try:
            await ws.send(subscribe_msg)
            async for raw in ws:
                recv_ts = time.time()
                events = json.loads(raw)
                if not isinstance(events, list):
                    events = [events]
                for event in events:
                    if event.get("event_type") != "price_change":
                        continue
                    tid   = event["asset_id"]
                    price = float(event["price"])
                    size  = float(event["size"])
                    side  = event["side"]   # "BUY" or "SELL"
                    if tid in books:
                        books[tid].apply_delta(side, price, size)

                # Resync stale books
                for tid, book in books.items():
                    if book.is_stale:
                        snap = await fetch_snapshot(tid)
                        book.load_snapshot(snap.get("bids", []), snap.get("asks", []))

                await on_update(
                    books[up_token_id],
                    books[down_token_id],
                    recv_ts,
                )
        except websockets.ConnectionClosed:
            continue  # reconnect via the async-for loop

from typing import Optional
from book import LocalBook

def depth_within(book: LocalBook, side: str, band: float) -> float:
    """Total size available within `band` cents of best price."""
    if side == "ask":
        best = book.best_ask()
        if best is None:
            return 0.0
        return sum(sz for px, sz in book.asks.items() if px <= best + band)
    else:
        best = book.best_bid()
        if best is None:
            return 0.0
        return sum(sz for px, sz in book.bids.items() if px >= best - band)

def microprice(book: LocalBook) -> Optional[float]:
    """Size-weighted mid: leans toward the heavier side."""
    bb, ba = book.best_bid(), book.best_ask()
    if bb is None or ba is None:
        return None
    bid_sz = book.bids.get(bb, 0.0)
    ask_sz = book.asks.get(ba, 0.0)
    total  = bid_sz + ask_sz
    if total == 0:
        return (bb + ba) / 2
    return (bb * ask_sz + ba * bid_sz) / total  # weighted toward thinner side

def effective_fill_price(book: LocalBook, side: str, quantity: float) -> Optional[float]:
    """Depth-walk the book for `quantity` shares. Returns None if not enough depth."""
    if side == "buy":
        levels = sorted(book.asks.items())       # ascending price
    else:
        levels = sorted(book.bids.items(), reverse=True)  # descending price

    remaining = quantity
    cost      = 0.0
    for price, size in levels:
        take       = min(remaining, size)
        cost      += take * price
        remaining -= take
        if remaining <= 1e-9:
            break

    if remaining > 1e-9:
        return None  # insufficient depth
    return cost / quantity

def book_summary(book: LocalBook, qty: float = 10.0) -> dict:
    return {
        "best_bid":      book.best_bid(),
        "best_ask":      book.best_ask(),
        "spread":        book.spread(),
        "depth_1c_bid":  depth_within(book, "bid", 0.01),
        "depth_1c_ask":  depth_within(book, "ask", 0.01),
        "depth_3c_ask":  depth_within(book, "ask", 0.03),
        "microprice":    microprice(book),
        "fill_price_10": effective_fill_price(book, "buy", 10.0),
        "fill_price_50": effective_fill_price(book, "buy", 50.0),
    }

import math
from scipy.stats import norm
from typing import Optional

def diffusion_fair_prob(
    spot: float,          # current BTC price (e.g. from Binance/Coinbase)
    strike: float,        # cycle reference/comparison price
    sigma_per_sqrt_s: float,  # realized vol per sqrt-second (e.g. 0.00012)
    tau_seconds: float,   # seconds remaining in cycle
    epsilon: float = 0.5, # floor to avoid division by zero near expiry
) -> float:
    """
    Binary diffusion probability that BTC closes above strike.
    Uses the same math as a digital call option.
    Returns a float in (0, 1).
    """
    tau = max(tau_seconds, epsilon)
    log_moneyness = math.log(spot / strike)
    z = log_moneyness / (sigma_per_sqrt_s * math.sqrt(tau))
    return float(norm.cdf(z))

def realized_vol_per_sqrt_second(
    returns: list[float],  # list of log-returns, one per second
) -> float:
    """Estimate sigma_per_sqrt_s from recent second-by-second log-returns."""
    if len(returns) < 2:
        return 1e-4  # fallback
    mean = sum(returns) / len(returns)
    var  = sum((r - mean) ** 2 for r in returns) / (len(returns) - 1)
    return math.sqrt(var)  # already per sqrt-second since returns are per second

# Example: BTC at 67,450, strike 67,400, 45s remaining, sigma 0.00015/sqrt(s)
p_up = diffusion_fair_prob(spot=67450, strike=67400,
                           sigma_per_sqrt_s=0.00015, tau_seconds=45)
print(f"P(UP) = {p_up:.4f}")  # e.g. 0.5832

import math
from dataclasses import dataclass
from typing import Literal
from book import LocalBook

RegimeLabel = Literal["calm", "high_vol_boundary", "stale_book", "close_gamma", "thin_liquidity"]

@dataclass
class RegimeState:
    label:         RegimeLabel
    trade_allowed: bool
    reason:        str

# Thresholds — tune these against your paper-trade log
MIN_DEPTH_3C       = 15.0   # minimum shares within 3c of ask
MAX_SPREAD         = 0.08   # suppress if spread > 8c
CLOSE_GAMMA_TAU    = 30.0   # seconds: final window, spreads blow out
HIGH_VOL_Z_THRESH  = 0.25   # abs(z) < this AND high vol = boundary risk
HIGH_VOL_SIGMA_MUL = 2.0    # sigma > 2x rolling median = high-vol regime
STALE_BOOK_DELTA   = 0.005  # book mid unchanged by < 0.5c despite BTC move
STALE_BTC_MOVE_BPS = 3.0    # BTC moved > 3 bps but book did not reprice

def classify_regime(
    up_book:       LocalBook,
    tau:           float,
    z:             float,       # log(S/K) / (sigma * sqrt(tau))
    sigma:         float,
    sigma_median:  float,       # rolling median sigma for comparison
    btc_move_bps:  float,       # abs BTC move in bps over last 2s
    book_mid_move: float,       # abs change in UP microprice over last 2s
) -> RegimeState:

    spread = up_book.spread()
    depth  = sum(sz for px, sz in up_book.asks.items()
                 if up_book.best_ask() and px <= up_book.best_ask() + 0.03)

    if up_book.is_stale:
        return RegimeState("stale_book", False, "WebSocket book is stale")

    if tau <= CLOSE_GAMMA_TAU:
        return RegimeState("close_gamma", False,
                           f"tau={tau:.0f}s: final window, spreads unreliable")

    if depth < MIN_DEPTH_3C:
        return RegimeState("thin_liquidity", False,
                           f"depth_3c={depth:.1f} < {MIN_DEPTH_3C}")

    if spread and spread > MAX_SPREAD:
        return RegimeState("thin_liquidity", False,
                           f"spread={spread:.3f} > {MAX_SPREAD}")

    high_vol = sigma > HIGH_VOL_SIGMA_MUL * sigma_median
    near_boundary = abs(z) < HIGH_VOL_Z_THRESH
    if high_vol and near_boundary:
        return RegimeState("high_vol_boundary", False,
                           f"sigma={sigma:.5f} high and abs(z)={abs(z):.2f} near boundary")

    stale_book = (btc_move_bps > STALE_BTC_MOVE_BPS
                  and book_mid_move < STALE_BOOK_DELTA)
    if stale_book:
        return RegimeState("stale_book", False,
                           f"BTC moved {btc_move_bps:.1f} bps but book mid unchanged")

    return RegimeState("calm", True, "all filters pass")

import time
from collections import deque

class ChurnGuard:
    """
    Track how often the best-ask price and size change.
    High churn rate = market makers are not committed to their quotes.
    """

    def __init__(self, window_s: float = 5.0, max_churn_rate: float = 4.0):
        self.window_s      = window_s
        self.max_churn_rate = max_churn_rate  # changes per second
        self._events: deque[float] = deque()  # timestamps of best-ask changes
        self._last_best_ask: float | None = None
        self._last_best_size: float | None = None

    def update(self, best_ask: float | None, best_ask_size: float | None) -> None:
        now = time.monotonic()
        # Prune old events outside the window
        while self._events and now - self._events[0] > self.window_s:
            self._events.popleft()

        changed = (
            best_ask  != self._last_best_ask or
            best_ask_size != self._last_best_size
        )
        if changed:
            self._events.append(now)
            self._last_best_ask  = best_ask
            self._last_best_size = best_ask_size

    @property
    def churn_rate(self) -> float:
        """Changes per second over the rolling window."""
        return len(self._events) / self.window_s

    @property
    def is_churning(self) -> bool:
        return self.churn_rate > self.max_churn_rate

    @property
    def quote_age_s(self) -> float:
        """Seconds since the last best-ask change."""
        if not self._events:
            return float("inf")
        return time.monotonic() - self._events[-1]

import logging
import os

log = logging.getLogger(__name__)

# Jurisdictions blocked for order placement as of Polymarket docs reviewed 2026-05-18.
# Verify current restrictions at https://polymarket.com/geographic-restrictions
# before enabling live execution.
BLOCKED_JURISDICTIONS = {"US", "USA", "UNITED STATES"}

def check_compliance(jurisdiction: str, paper_trade: bool = True) -> bool:
    """
    Returns True if order placement is allowed.
    In paper_trade mode, logs a warning but does not raise.
    In live mode, raises RuntimeError for blocked jurisdictions.
    """
    j = jurisdiction.strip().upper()
    if j in BLOCKED_JURISDICTIONS:
        msg = (
            f"Jurisdiction '{jurisdiction}' is listed as blocked for Polymarket "
            f"order placement. Verify current restrictions before enabling execution."
        )
        if paper_trade:
            log.warning("[COMPLIANCE] %s (paper-trade mode: logging only)", msg)
            return False
        else:
            raise RuntimeError(f"[COMPLIANCE BLOCK] {msg}")
    return True

# Usage: call this as the FIRST line of any order-submission function
# allowed = check_compliance(os.environ.get("JURISDICTION", "US"), paper_trade=True)
# if not allowed:
#     return  # skip order, log suppression

import asyncio
import time
import uuid
from dataclasses import dataclass, field
from typing import Optional
from book import LocalBook
from metrics import effective_fill_price
from edge import compute_taker_buy_edge
from compliance import check_compliance
import os

@dataclass
class PaperFill:
    fill_id:        str
    token_side:     str          # "UP" or "DOWN"
    signal_time:    float
    fill_time:      float
    intended_ask:   float
    fill_price:     Optional[float]  # None = no fill (ask moved)
    fill_size:      float
    fair_prob_at_signal: float
    raw_edge_at_signal:  float
    regime:         str
    markout_5s:     Optional[float] = None
    markout_10s:    Optional[float] = None
    markout_30s:    Optional[float] = None
    markout_final:  Optional[float] = None

LATENCY_DELAY_S = 1.5   # simulate round-trip latency before checking fill
MAX_FILL_SIZE   = 50.0  # cap per-entry size

async def attempt_paper_fill(
    book_at_signal:  LocalBook,
    book_after_delay: LocalBook,   # same book object, updated by WS
    token_side:      str,
    quantity:        float,
    fair_prob:       float,
    raw_edge:        float,
    regime:          str,
    jurisdiction:    str = "US",
) -> PaperFill:

    fill_id      = str(uuid.uuid4())[:8]
    signal_time  = time.time()
    intended_ask = book_at_signal.best_ask()

    # Compliance check first — always
    allowed = check_compliance(jurisdiction, paper_trade=True)

    # Simulate latency
    await asyncio.sleep(LATENCY_DELAY_S)
    fill_time = time.time()

    # Check if ask is still available after delay
    delayed_ask = book_after_delay.best_ask()
    if delayed_ask is None or (intended_ask and delayed_ask > intended_ask + 0.01):
        return PaperFill(
            fill_id=fill_id, token_side=token_side,
            signal_time=signal_time, fill_time=fill_time,
            intended_ask=intended_ask, fill_price=None,
            fill_size=0.0, fair_prob_at_signal=fair_prob,
            raw_edge_at_signal=raw_edge, regime=regime,
        )

    # Cap size to available depth within 3c
    available = sum(
        sz for px, sz in book_after_delay.asks.items()
        if px <= delayed_ask + 0.03
    )
    capped_qty = min(quantity, MAX_FILL_SIZE, available)
    fill_price = effective_fill_price(book_after_delay, "buy", capped_qty)

    return PaperFill(
        fill_id=fill_id, token_side=token_side,
        signal_time=signal_time, fill_time=fill_time,
        intended_ask=intended_ask, fill_price=fill_price,
        fill_size=capped_qty, fair_prob_at_signal=fair_prob,
        raw_edge_at_signal=raw_edge, regime=regime,
    )

import time
from typing import Callable
from paper_trader import PaperFill
from fair_value import diffusion_fair_prob

async def record_markouts(
    fill:          PaperFill,
    get_spot:      Callable[[], float],   # live BTC spot price getter
    strike:        float,
    sigma:         float,
    get_tau:       Callable[[], float],   # live tau getter
    final_outcome: Callable[[], float | None],  # returns 1.0/0.0 or None
) -> PaperFill:
    """Wait for markout horizons and record fair value vs fill price."""
    if fill.fill_price is None:
        return fill  # no fill, nothing to mark

    horizons = {"markout_5s": 5, "markout_10s": 10, "markout_30s": 30}
    for attr, delay in horizons.items():
        await asyncio.sleep(delay)
        spot = get_spot()
        tau  = get_tau()
        fv   = diffusion_fair_prob(spot, strike, sigma, tau)
        # markout = fair_value_now - fill_price (positive = good for buyer)
        setattr(fill, attr, round(fv - fill.fill_price, 6))

    # Wait for settlement
    while True:
        outcome = final_outcome()
        if outcome is not None:
            fill.markout_final = round(outcome - fill.fill_price, 6)
            break
        await asyncio.sleep(1)

    return fill

import asyncio  # ensure asyncio is available for asyncio.sleep calls above