gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

启用概率校准和 WebSocket 报价配置

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-21 21:06:48 +08:00
parent 582ded8cfb
commit 82ec594277
22 changed files with 3703 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data_collection/polymarket_readonly.py
+143 -3
View File
@@ -23,6 +23,7 @@ import httpx
from loguru import logger
from src.data_collection.city_registry import ALIASES, CITY_REGISTRY
from src.data_collection.polymarket_ws_cache import PolymarketWsQuoteCache
try:
from py_clob_client.client import ClobClient # type: ignore
@@ -173,6 +174,16 @@ def _extract_price(value: Any) -> Optional[float]:
return None
def _clamp_probability(value: Optional[float]) -> Optional[float]:
if value is None:
return None
if value < 0.0:
return 0.0
if value > 1.0:
return 1.0
return value
def _extract_iso_date(value: Any) -> Optional[str]:
if not value:
return None
@@ -394,6 +405,7 @@ class PolymarketReadOnlyLayer:
self._lock = threading.Lock()
self._clob_client: Any = None
self._clob_unavailable_reason: Optional[str] = None
self._ws_quote_cache = PolymarketWsQuoteCache.from_env()
def build_market_scan(
self,
@@ -430,6 +442,7 @@ class PolymarketReadOnlyLayer:
"last_trade_price": None,
"liquidity": None,
"volume": None,
"price_analysis": None,
"sparkline": fallback_sparkline or [],
"top_buckets": [],
"recent_trades": [],
@@ -570,6 +583,8 @@ class PolymarketReadOnlyLayer:
"sell_price": _extract_price(yes_prices.get("sell")),
"midpoint": _extract_price(yes_prices.get("midpoint")),
"last_trade_price": _extract_price(yes_prices.get("last_trade_price")),
"quote_source": yes_prices.get("quote_source"),
"quote_age_ms": _safe_int(yes_prices.get("quote_age_ms"), 0),
"book": yes_prices.get("book"),
}
no_payload = {
@@ -580,8 +595,17 @@ class PolymarketReadOnlyLayer:
"sell_price": _extract_price(no_prices.get("sell")),
"midpoint": _extract_price(no_prices.get("midpoint")),
"last_trade_price": _extract_price(no_prices.get("last_trade_price")),
"quote_source": no_prices.get("quote_source"),
"quote_age_ms": _safe_int(no_prices.get("quote_age_ms"), 0),
"book": no_prices.get("book"),
}
price_analysis = self._build_price_analysis(
model_probability=model_probability,
yes_buy=_extract_price(yes_payload.get("buy_price")),
yes_sell=_extract_price(yes_payload.get("sell_price")),
no_buy=_extract_price(no_payload.get("buy_price")),
no_sell=_extract_price(no_payload.get("sell_price")),
)
sparkline_values: List[float] = []
for candidate in (
@@ -617,12 +641,15 @@ class PolymarketReadOnlyLayer:
"last_trade_price": last_trade_price,
"liquidity": liquidity,
"volume": volume,
"price_analysis": price_analysis,
"sparkline": sparkline_values,
"top_buckets": top_buckets,
"all_buckets": all_buckets,
"websocket": {
"market_url": market_url,
"asset_ids": [
"websocket": {
"enabled": self._ws_quote_cache.enabled,
"status": self._ws_quote_cache.status(),
"market_url": market_url,
"asset_ids": [
token
for token in [
yes_payload.get("token_id"),
@@ -636,6 +663,112 @@ class PolymarketReadOnlyLayer:
)
return scan
def _build_price_analysis(
self,
*,
model_probability: Optional[float],
yes_buy: Optional[float],
yes_sell: Optional[float],
no_buy: Optional[float],
no_sell: Optional[float],
) -> Dict[str, Any]:
"""Build read-only market price diagnostics.
Polymarket CLOB naming is from the user's perspective:
BUY is the executable ask to buy that outcome, SELL is the executable bid.
Kelly here is a sizing reference only; no order execution is performed.
"""
p_yes = _clamp_probability(_safe_float(model_probability))
p_no = _clamp_probability(1.0 - p_yes if p_yes is not None else None)
yes_ask = _clamp_probability(_safe_float(yes_buy))
no_ask = _clamp_probability(_safe_float(no_buy))
yes_bid = _clamp_probability(_safe_float(yes_sell))
no_bid = _clamp_probability(_safe_float(no_sell))
yes = self._build_side_price_analysis("yes", p_yes, yes_ask, yes_bid)
no = self._build_side_price_analysis("no", p_no, no_ask, no_bid)
ask_sum = None
lock_edge = None
lock_available = False
if yes_ask is not None and no_ask is not None:
ask_sum = yes_ask + no_ask
lock_edge = 1.0 - ask_sum
lock_available = lock_edge > 0
bid_sum = None
sell_side_edge = None
if yes_bid is not None and no_bid is not None:
bid_sum = yes_bid + no_bid
sell_side_edge = bid_sum - 1.0
best_side = None
side_rows = [
row
for row in [yes, no]
if isinstance(row.get("edge"), (int, float))
and isinstance(row.get("kelly_fraction"), (int, float))
and row.get("kelly_fraction") > 0
]
if side_rows:
best_side = max(
side_rows,
key=lambda row: (
float(row.get("edge") or 0.0),
float(row.get("kelly_fraction") or 0.0),
),
).get("side")
return {
"available": any(
value is not None
for value in (yes_ask, no_ask, yes_bid, no_bid, p_yes)
),
"source": "polymarket_clob_orderbook",
"model_probability": p_yes,
"yes": yes,
"no": no,
"best_side": best_side,
"lock": {
"available": lock_available,
"ask_sum": ask_sum,
"edge": lock_edge,
},
"sell_side": {
"bid_sum": bid_sum,
"edge": sell_side_edge,
},
}
def _build_side_price_analysis(
self,
side: str,
probability: Optional[float],
ask: Optional[float],
bid: Optional[float],
) -> Dict[str, Any]:
edge = None
kelly_fraction = None
if probability is not None and ask is not None:
edge = probability - ask
if 0.0 < ask < 1.0:
kelly_fraction = edge / (1.0 - ask)
return {
"side": side,
"model_probability": probability,
"ask": ask,
"bid": bid,
"edge": edge,
"edge_percent": edge * 100.0 if edge is not None else None,
"kelly_fraction": kelly_fraction,
"quarter_kelly": (
max(0.0, kelly_fraction) / 4.0
if kelly_fraction is not None
else None
),
}
def _market_trade_state(self, market: Dict[str, Any]) -> Dict[str, Any]:
active = _safe_bool(market.get("active"))
closed_raw = _safe_bool(market.get("closed"))
@@ -1193,6 +1326,11 @@ class PolymarketReadOnlyLayer:
if not token_id:
return {}
self._ws_quote_cache.subscribe([token_id])
ws_data = self._ws_quote_cache.get_market_data(token_id)
if ws_data is not None:
return ws_data
now = time.time()
with self._lock:
cached = self._price_cache.get(token_id)
@@ -1473,6 +1611,8 @@ class PolymarketReadOnlyLayer:
"yes_sell": yes_sell,
"no_buy": no_buy,
"no_sell": no_sell,
"quote_source": yes_prices.get("quote_source"),
"quote_age_ms": _safe_int(yes_prices.get("quote_age_ms"), 0),
"slug": market_slug or None,
"question": market.get("question") or market.get("title"),
"is_primary": bool(
src/data_collection/polymarket_ws_cache.py
+404
View File
@@ -0,0 +1,404 @@
"""
Read-only Polymarket market WebSocket quote cache.
The cache subscribes to public market-channel asset ids and stores executable
best bid / ask updates. It is deliberately optional: callers should keep REST
or CLOB polling as a fallback when the WebSocket client is unavailable.
"""
from __future__ import annotations
import asyncio
import json
import math
import os
import threading
import time
from typing import Any, Dict, Iterable, Optional, Set
from loguru import logger
def _safe_float(value: Any) -> Optional[float]:
if value is None:
return None
try:
if isinstance(value, str):
value = value.strip()
if not value:
return None
numeric = float(value)
if math.isnan(numeric) or math.isinf(numeric):
return None
return numeric
except Exception:
return None
def _env_bool(name: str, default: bool = False) -> bool:
raw = os.getenv(name)
if raw is None:
return default
return raw.strip().lower() in {"1", "true", "yes", "on"}
class PolymarketWsQuoteCache:
def __init__(
self,
*,
enabled: bool = False,
endpoint: Optional[str] = None,
quote_ttl_sec: int = 8,
max_assets: int = 256,
reconnect_delay_sec: float = 3.0,
) -> None:
self.enabled = enabled
self.endpoint = (
endpoint
or os.getenv(
"POLYMARKET_WS_MARKET_URL",
"wss://ws-subscriptions-clob.polymarket.com/ws/market",
)
or ""
).strip()
self.quote_ttl_sec = max(1, int(quote_ttl_sec or 8))
self.max_assets = max(1, int(max_assets or 256))
self.reconnect_delay_sec = max(0.5, float(reconnect_delay_sec or 3.0))
self._desired_assets: Set[str] = set()
self._quotes: Dict[str, Dict[str, Any]] = {}
self._lock = threading.Lock()
self._thread: Optional[threading.Thread] = None
self._stop_event = threading.Event()
self._started = False
self._last_error: Optional[str] = None
self._last_connected_at: Optional[float] = None
self._last_message_at: Optional[float] = None
@classmethod
def from_env(cls) -> "PolymarketWsQuoteCache":
return cls(
enabled=_env_bool("POLYMARKET_WS_PRICE_ENABLED", False),
endpoint=os.getenv("POLYMARKET_WS_MARKET_URL"),
quote_ttl_sec=int(os.getenv("POLYMARKET_WS_QUOTE_TTL_SEC", "8")),
max_assets=int(os.getenv("POLYMARKET_WS_MAX_ASSETS", "256")),
reconnect_delay_sec=float(
os.getenv("POLYMARKET_WS_RECONNECT_DELAY_SEC", "3")
),
)
def start(self) -> None:
if not self.enabled or not self.endpoint:
return
with self._lock:
if self._started:
return
self._started = True
self._thread = threading.Thread(
target=self._thread_main,
name="polymarket-ws-quotes",
daemon=True,
)
self._thread.start()
def stop(self) -> None:
self._stop_event.set()
def subscribe(self, asset_ids: Iterable[Any]) -> None:
if not self.enabled:
return
normalized = []
for asset_id in asset_ids:
text = str(asset_id or "").strip()
if text:
normalized.append(text)
if not normalized:
return
with self._lock:
remaining = self.max_assets - len(self._desired_assets)
for asset_id in normalized:
if asset_id in self._desired_assets:
continue
if remaining <= 0:
break
self._desired_assets.add(asset_id)
remaining -= 1
self.start()
def get_market_data(self, asset_id: Any) -> Optional[Dict[str, Any]]:
quote = self.get_quote(asset_id)
if not quote:
return None
best_bid = _safe_float(quote.get("best_bid"))
best_ask = _safe_float(quote.get("best_ask"))
if best_bid is None and best_ask is None:
return None
midpoint = None
if best_bid is not None and best_ask is not None:
midpoint = (best_bid + best_ask) / 2.0
age_ms = int((time.time() - float(quote.get("t") or time.time())) * 1000)
return {
"buy": best_ask,
"sell": best_bid,
"midpoint": midpoint,
"last_trade_price": _safe_float(quote.get("last_trade_price")),
"book": {
"best_bid": best_bid,
"best_ask": best_ask,
"bid_levels": [[best_bid, 0.0]] if best_bid is not None else [],
"ask_levels": [[best_ask, 0.0]] if best_ask is not None else [],
},
"book_liquidity": None,
"quote_source": "polymarket_ws",
"quote_age_ms": age_ms,
}
def get_quote(self, asset_id: Any) -> Optional[Dict[str, Any]]:
text = str(asset_id or "").strip()
if not text:
return None
now = time.time()
with self._lock:
quote = self._quotes.get(text)
if not quote:
return None
if now - float(quote.get("t") or 0.0) > self.quote_ttl_sec:
return None
return dict(quote)
def status(self) -> Dict[str, Any]:
with self._lock:
return {
"enabled": self.enabled,
"started": self._started,
"endpoint": self.endpoint,
"asset_count": len(self._desired_assets),
"quote_count": len(self._quotes),
"last_error": self._last_error,
"last_connected_at": self._last_connected_at,
"last_message_at": self._last_message_at,
}
def _thread_main(self) -> None:
try:
asyncio.run(self._run_forever())
except Exception as exc: # pragma: no cover - defensive thread guard
with self._lock:
self._last_error = str(exc)
logger.warning(f"Polymarket WS quote cache stopped: {exc}")
async def _run_forever(self) -> None:
try:
import websockets # type: ignore
except Exception as exc:
with self._lock:
self._last_error = f"websockets import failed: {exc}"
logger.warning(self._last_error)
return
while not self._stop_event.is_set():
try:
async with websockets.connect(
self.endpoint,
ping_interval=None,
close_timeout=2,
) as ws:
with self._lock:
self._last_connected_at = time.time()
self._last_error = None
subscribed: Set[str] = set()
last_ping = 0.0
while not self._stop_event.is_set():
desired = self._snapshot_assets()
missing = desired - subscribed
if missing:
await self._send_subscription(
ws,
missing,
initial=not subscribed,
)
subscribed.update(missing)
now = time.time()
if now - last_ping >= 10:
await ws.send(json.dumps({}))
last_ping = now
try:
raw = await asyncio.wait_for(ws.recv(), timeout=1.0)
except asyncio.TimeoutError:
continue
self._handle_message(raw)
except Exception as exc:
with self._lock:
self._last_error = str(exc)
logger.warning(f"Polymarket WS reconnecting after error: {exc}")
await asyncio.sleep(self.reconnect_delay_sec)
def _snapshot_assets(self) -> Set[str]:
with self._lock:
return set(self._desired_assets)
async def _send_subscription(
self,
ws: Any,
asset_ids: Iterable[str],
*,
initial: bool,
) -> None:
batch = [asset_id for asset_id in asset_ids if asset_id]
if not batch:
return
payload = {
"assets_ids": batch,
"custom_feature_enabled": True,
}
if initial:
payload["type"] = "market"
else:
payload["operation"] = "subscribe"
await ws.send(json.dumps(payload))
def _handle_message(self, raw: Any) -> None:
if raw in (None, "", "PONG"):
return
try:
payload = json.loads(raw) if isinstance(raw, str) else raw
except Exception:
return
if isinstance(payload, list):
for item in payload:
self._handle_event(item)
return
self._handle_event(payload)
def _handle_event(self, event: Any) -> None:
if not isinstance(event, dict):
return
if "event_type" in event:
event_type = str(event.get("event_type") or "").strip().lower()
else:
event_type = str(event.get("type") or "").strip().lower()
if event_type in {
"best_bid_ask",
"price_change",
"book",
"last_trade_price",
}:
self._handle_quote_event(event_type, event)
def _handle_quote_event(self, event_type: str, event: Dict[str, Any]) -> None:
candidates = (
event.get("price_changes")
or event.get("changes")
or event.get("assets")
or event.get("data")
)
if isinstance(candidates, list):
for item in candidates:
if isinstance(item, dict):
self._upsert_quote(event_type, item, parent=event)
return
self._upsert_quote(event_type, event, parent=event)
def _upsert_quote(
self,
event_type: str,
item: Dict[str, Any],
*,
parent: Dict[str, Any],
) -> None:
asset_id = str(
item.get("asset_id")
or item.get("assetId")
or item.get("token_id")
or item.get("tokenId")
or parent.get("asset_id")
or parent.get("assetId")
or ""
).strip()
if not asset_id:
return
best_bid = (
_safe_float(item.get("best_bid"))
or _safe_float(item.get("bid"))
or _safe_float(item.get("bestBid"))
)
best_ask = (
_safe_float(item.get("best_ask"))
or _safe_float(item.get("ask"))
or _safe_float(item.get("bestAsk"))
)
if event_type == "book":
parsed_bid, parsed_ask = self._extract_book_top(item)
best_bid = best_bid if best_bid is not None else parsed_bid
best_ask = best_ask if best_ask is not None else parsed_ask
price = _safe_float(item.get("price"))
side = str(item.get("side") or "").strip().upper()
if event_type == "price_change" and price is not None:
if side == "BUY":
best_ask = price
elif side == "SELL":
best_bid = price
last_trade = (
_safe_float(item.get("last_trade_price"))
or _safe_float(item.get("lastTradePrice"))
or (price if event_type == "last_trade_price" else None)
)
now = time.time()
with self._lock:
previous = dict(self._quotes.get(asset_id) or {})
if best_bid is not None:
previous["best_bid"] = best_bid
if best_ask is not None:
previous["best_ask"] = best_ask
if last_trade is not None:
previous["last_trade_price"] = last_trade
previous["asset_id"] = asset_id
previous["event_type"] = event_type
previous["t"] = now
self._quotes[asset_id] = previous
self._last_message_at = now
def _extract_book_top(
self,
payload: Dict[str, Any],
) -> tuple[Optional[float], Optional[float]]:
best_bid = None
best_ask = None
bids = payload.get("bids")
if isinstance(bids, list):
for item in bids:
price = self._extract_level_price(item)
if price is None:
continue
best_bid = price if best_bid is None else max(best_bid, price)
asks = payload.get("asks")
if isinstance(asks, list):
for item in asks:
price = self._extract_level_price(item)
if price is None:
continue
best_ask = price if best_ask is None else min(best_ask, price)
return best_bid, best_ask
@staticmethod
def _extract_level_price(level: Any) -> Optional[float]:
if isinstance(level, dict):
return _safe_float(level.get("price"))
if isinstance(level, (list, tuple)) and level:
return _safe_float(level[0])
return None