from __future__ import annotations import json import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any from analysis.safe_btc5.types import parse_dt ACTIVITY_WS_URL = "wss://ws-live-data.polymarket.com" @dataclass(frozen=True) class TargetMarket: slug: str condition_id: str market_id: str question: str outcomes: list[str] token_ids: list[str] kind: str def json_list(raw: Any) -> list[Any]: if raw is None: return [] if isinstance(raw, list): return raw if isinstance(raw, str): try: value = json.loads(raw) except json.JSONDecodeError: return [] return value if isinstance(value, list) else [] return [] def market_kind(event_slug: str, market_slug: str) -> str | None: if market_slug == event_slug: return "moneyline" for game_no in range(1, 5): if market_slug == f"{event_slug}-game{game_no}": return f"game{game_no}" return None def infer_requested_markets(event_slug: str, markets: list[dict[str, Any]]) -> set[str]: """Select main winner markets only. BO3 usually has moneyline/game1/game2. BO5 should use moneyline plus game1-game4; the final decisive map is represented by the moneyline outcome. """ requested = {"moneyline"} for market in markets: kind = market_kind(event_slug, str(market.get("slug") or "")) if kind and kind != "moneyline": requested.add(kind) return requested def select_target_markets( event_slug: str, markets: list[dict[str, Any]], requested: set[str] | None, ) -> list[TargetMarket]: if requested is None: requested = infer_requested_markets(event_slug, markets) targets: list[TargetMarket] = [] for market in markets: slug = str(market.get("slug") or "") kind = market_kind(event_slug, slug) if kind is None or kind not in requested: continue targets.append( TargetMarket( slug=slug, condition_id=str(market.get("conditionId") or ""), market_id=str(market.get("id") or ""), question=str(market.get("question") or ""), outcomes=[str(item) for item in json_list(market.get("outcomes"))], token_ids=[str(item) for item in json_list(market.get("clobTokenIds"))], kind=kind, ) ) order = {"moneyline": 0, **{f"game{i}": i for i in range(1, 5)}} return sorted(targets, key=lambda item: order.get(item.kind, 99)) def target_indexes(targets: list[TargetMarket]) -> dict[str, dict[str, TargetMarket]]: by_slug = {target.slug: target for target in targets} by_condition = { target.condition_id.lower(): target for target in targets if target.condition_id } by_market_id = {target.market_id: target for target in targets if target.market_id} by_token = { token_id: target for target in targets for token_id in target.token_ids } return { "slug": by_slug, "condition": by_condition, "market_id": by_market_id, "token": by_token, } def match_activity_trade( item: dict[str, Any], targets: list[TargetMarket], ) -> TargetMarket | None: payload = item.get("payload") if isinstance(item.get("payload"), dict) else item if not isinstance(payload, dict): return None indexes = target_indexes(targets) slug = str(payload.get("slug") or payload.get("marketSlug") or payload.get("market_slug") or "") if slug in indexes["slug"]: return indexes["slug"][slug] condition_id = str(payload.get("conditionId") or payload.get("condition_id") or "").lower() if condition_id in indexes["condition"]: return indexes["condition"][condition_id] market_id = str(payload.get("market") or payload.get("marketId") or payload.get("market_id") or "") if market_id in indexes["market_id"]: return indexes["market_id"][market_id] asset = str( payload.get("asset") or payload.get("asset_id") or payload.get("token_id") or payload.get("outcomeTokenId") or "" ) if asset in indexes["token"]: return indexes["token"][asset] for maker_order in payload.get("maker_orders") or []: if not isinstance(maker_order, dict): continue maker_asset = str( maker_order.get("asset") or maker_order.get("asset_id") or maker_order.get("token_id") or "" ) if maker_asset in indexes["token"]: return indexes["token"][maker_asset] return None def parse_activity_ts(value: Any) -> datetime | None: if value in (None, ""): return None if isinstance(value, (int, float)): ts = float(value) if ts > 1e12: ts /= 1000.0 try: return datetime.fromtimestamp(ts, tz=timezone.utc) except (OSError, OverflowError, ValueError): return None if isinstance(value, str): try: ts = float(value) except ValueError: parsed = parse_dt(value) return parsed.astimezone(timezone.utc) if parsed is not None else None if ts > 1e12: ts /= 1000.0 try: return datetime.fromtimestamp(ts, tz=timezone.utc) except (OSError, OverflowError, ValueError): return None parsed = parse_dt(value) return parsed.astimezone(timezone.utc) if parsed is not None else None def normalized_activity_row( *, item: dict[str, Any], target: TargetMarket, event_slug: str, received_at_wall: datetime, received_at_monotonic_ns: int, message_index: int, ) -> dict[str, Any]: payload = item.get("payload") if isinstance(item.get("payload"), dict) else item ws_ts = parse_activity_ts(item.get("timestamp")) trade_ts = parse_activity_ts(payload.get("timestamp") if isinstance(payload, dict) else None) source_ts = ws_ts or trade_ts delay_ms = ( round((received_at_wall - source_ts).total_seconds() * 1000.0, 3) if source_ts is not None else None ) return { "event_slug": event_slug, "matched_market_slug": target.slug, "matched_market_kind": target.kind, "matched_condition_id": target.condition_id, "matched_market_id": target.market_id, "topic": item.get("topic", ""), "type": item.get("type", ""), "connection_id": item.get("connection_id", ""), "message_index": message_index, "ws_timestamp": ws_ts.isoformat() if ws_ts else "", "trade_timestamp": trade_ts.isoformat() if trade_ts else "", "received_at_wall": received_at_wall.isoformat(), "received_at_monotonic_ns": received_at_monotonic_ns, "delay_ms": delay_ms, "asset": payload.get("asset", "") if isinstance(payload, dict) else "", "outcome": payload.get("outcome", "") if isinstance(payload, dict) else "", "outcome_index": payload.get("outcomeIndex", "") if isinstance(payload, dict) else "", "side": payload.get("side", "") if isinstance(payload, dict) else "", "price": payload.get("price", "") if isinstance(payload, dict) else "", "size": payload.get("size", "") if isinstance(payload, dict) else "", "fee": payload.get("fee", "") if isinstance(payload, dict) else "", "proxy_wallet": payload.get("proxyWallet", "") if isinstance(payload, dict) else "", "name": payload.get("name", "") if isinstance(payload, dict) else "", "pseudonym": payload.get("pseudonym", "") if isinstance(payload, dict) else "", "transaction_hash": payload.get("transactionHash", "") if isinstance(payload, dict) else "", "raw_json": item, } class JsonlWriter: def __init__( self, path: Path, *, flush_every: int = 100, flush_interval_seconds: float = 0.5, ) -> None: self.path = path self.flush_every = max(1, flush_every) self.flush_interval_seconds = max(0.0, flush_interval_seconds) self._pending = 0 self._last_flush_monotonic = time.monotonic() self.path.parent.mkdir(parents=True, exist_ok=True) self._file = path.open("a", encoding="utf-8") def write(self, row: dict[str, Any]) -> None: self._file.write(json.dumps(row, ensure_ascii=False, sort_keys=True)) self._file.write("\n") self._pending += 1 now = time.monotonic() if ( self._pending >= self.flush_every or now - self._last_flush_monotonic >= self.flush_interval_seconds ): self.flush() def flush(self) -> None: self._file.flush() self._pending = 0 self._last_flush_monotonic = time.monotonic() def close(self) -> None: self.flush() self._file.close()