Publish Render-ready Polymarket collector

.dockerignore

@@ -0,0 +1,7 @@
+.git
+.venv*
+__pycache__
+*.py[cod]
+.env
+data
+*.zip

.env.example

@@ -0,0 +1,20 @@
+# Public market-data collector only. No wallet or Polymarket private key is used.
+DATA_DIR=/var/lib/polymarket-collector
+MAX_MARKETS=75
+MIN_DAILY_REWARD=0
+MIN_VOLUME_24H=0
+DISCOVERY_INTERVAL_SECONDS=900
+ROTATION_SECONDS=900
+MAX_FILE_MIB=64
+FLUSH_INTERVAL_SECONDS=2
+HEALTH_HOST=127.0.0.1
+HEALTH_PORT=8080
+STALE_AFTER_SECONDS=180
+LOG_LEVEL=INFO
+
+# Optional Google Drive archive through rclone.
+# Example after running `rclone config`: gdrive:polymarket-data
+RCLONE_REMOTE=
+LOCAL_RETENTION_DAYS=7
+REMOVE_AFTER_UPLOAD=false
+BACKUP_MIN_AGE=2m

.gitignore

@@ -0,0 +1,10 @@
+__pycache__/
+*.py[cod]
+.venv/
+.env
+.coverage
+data/
+dist/
+build/
+*.egg-info/
+

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    DATA_DIR=/data \
+    HEALTH_HOST=0.0.0.0 \
+    PORT=8080
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates tini \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+COPY pyproject.toml README.md ./
+COPY polymarket_collector ./polymarket_collector
+COPY scripts ./scripts
+COPY deploy/container-entrypoint.sh ./deploy/container-entrypoint.sh
+RUN pip install --no-cache-dir . \
+    && chmod +x /app/scripts/polymarket-backup /app/deploy/container-entrypoint.sh
+
+EXPOSE 8080
+ENTRYPOINT ["/usr/bin/tini", "--", "/app/deploy/container-entrypoint.sh"]

README.md

@@ -0,0 +1,161 @@
+---
+title: Polymarket LP Collector
+emoji: 📈
+colorFrom: green
+colorTo: blue
+sdk: docker
+app_port: 8080
+---
+
+# Collecteur de données LP Polymarket
+
+Ce projet enregistre en continu les données publiques de niveau 2 des marchés
+Polymarket qui distribuent des récompenses de liquidité. Il est conçu pour une
+petite VM Google Cloud `e2-micro` : mémoire limitée, fichiers rotatifs et aucune
+base de données résidente.
+
+Il ne signe aucun ordre, ne contient aucune clé de portefeuille et ne peut pas
+trader. Son unique rôle est de créer un historique propre pour les futurs
+backtests et modèles d'apprentissage automatique.
+
+## Données collectées
+
+- carnets complets (`book`) ;
+- changements de niveaux (`price_change`) ;
+- meilleurs bid/ask ;
+- dernières transactions ;
+- changements de tick et résolutions ;
+- métadonnées, volumes et paramètres de récompenses LP ;
+- heure de réception locale en nanosecondes pour mesurer les délais.
+
+Les fichiers terminés sont écrits sous la forme :
+
+```text
+data/
+  market_ws/YYYY/MM/DD/market_ws-....jsonl.gz
+  discovery/YYYY/MM/DD/discovery-....jsonl.gz
+```
+
+Chaque ligne est un objet JSON autonome. Les fichiers `.part` sont récupérés et
+compressés automatiquement au prochain démarrage après une coupure brutale.
+
+## Exécution locale
+
+```bash
+python3 -m venv .venv
+. .venv/bin/activate
+pip install -e .
+cp .env.example .env
+DATA_DIR="$PWD/data" polymarket-collector
+```
+
+Tester uniquement l'accès aux API publiques avant de démarrer :
+
+```bash
+polymarket-collector --check
+```
+
+État du service :
+
+```bash
+curl http://127.0.0.1:8080/healthz
+curl http://127.0.0.1:8080/metrics
+```
+
+Arrêter avec `Ctrl+C`. Le service finalise alors proprement les fichiers en
+cours.
+
+## Déploiement Google Cloud
+
+Depuis un ordinateur ou Google Cloud Shell ayant `gcloud` authentifié :
+
+```bash
+./scripts/deploy_gcp.sh ID_DU_PROJET_GCP
+```
+
+Le script crée par défaut une VM `e2-micro` Ubuntu 24.04 dans `us-east1-b`, un
+disque standard de 30 Go, copie le projet, puis installe et démarre le service.
+
+Commandes utiles après installation :
+
+```bash
+gcloud compute ssh polymarket-collector --zone us-east1-b
+sudo systemctl status polymarket-collector
+sudo journalctl -u polymarket-collector -f
+curl http://127.0.0.1:8080/healthz
+```
+
+## Sauvegarde vers Google Drive
+
+Sur la VM :
+
+```bash
+sudo rclone config
+sudoedit /etc/polymarket-collector.env
+```
+
+Définir ensuite, par exemple :
+
+```text
+RCLONE_REMOTE=gdrive:polymarket-data
+LOCAL_RETENTION_DAYS=7
+```
+
+Puis activer la sauvegarde :
+
+```bash
+sudo systemctl enable --now polymarket-backup.timer
+sudo systemctl start polymarket-backup.service
+sudo journalctl -u polymarket-backup.service -n 100
+```
+
+Les fichiers locaux âgés de plus de `LOCAL_RETENTION_DAYS` ne sont supprimés
+qu'après une copie `rclone` réussie. Mettre la valeur à `0` pour désactiver le
+nettoyage local.
+
+## Hébergement gratuit sans dépassement : Render + Hugging Face
+
+L'architecture recommandée pour une collecte continue sans VM personnelle est :
+
+1. un Web Service public **Render Free**, sans moyen de paiement enregistré ;
+2. UptimeRobot Free sur `/healthz`, toutes les cinq minutes, pour maintenir une
+   visite régulière et surveiller le service ;
+3. un dataset Hugging Face public pour les archives. Les fichiers locaux sont
+   supprimés uniquement après la réussite du commit distant.
+
+Le conteneur est prêt dans `Dockerfile` et le profil Render dans `render.yaml`.
+Créer un jeton Hugging Face avec droit d'écriture, puis publier le dépôt source :
+
+```bash
+python scripts/publish_hf_source.py
+```
+
+Créer ensuite un Web Service Render depuis l'URL Git publique affichée, choisir
+le plan `Free`, et fournir `HF_TOKEN` comme secret. Le profil utilise 40 marchés,
+une rotation à 256 Mio et un envoi groupé toutes les heures. Render dispose de
+750 heures gratuites par mois et suspend le service au lieu de facturer lorsque
+le compte ne contient aucun moyen de paiement. Son disque est temporaire : les
+archives ne sont supprimées qu'après confirmation du commit Hugging Face.
+
+Hugging Face Spaces n'est plus recommandé : les Spaces Docker sur `cpu-basic`
+exigent désormais un abonnement Hugging Face PRO.
+
+Koyeb n'est plus recommandé : depuis son rapprochement avec Mistral, les
+nouveaux utilisateurs doivent fournir un moyen de paiement et souscrire un
+forfait payant.
+
+## Tests
+
+```bash
+python -m unittest discover -s tests -v
+```
+
+Les tests ne contactent pas Polymarket et ne nécessitent aucune clé.
+
+## Limites importantes
+
+- Avec 1 Go de RAM et 30 Go de disque, commencer avec 75 marchés maximum.
+- Une VM gratuite aux États-Unis convient à la collecte, pas au futur bot de
+  trading sensible à la latence.
+- Le service conserve les messages bruts : les transformations en Parquet et
+  l'entraînement se feront sur l'ordinateur équipé de la RTX 3060.

deploy/container-entrypoint.sh

@@ -0,0 +1,70 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+DATA_DIR="${DATA_DIR:-/data}"
+BACKUP_INTERVAL_SECONDS="${BACKUP_INTERVAL_SECONDS:-120}"
+RCLONE_CONFIG_FILE="${RCLONE_CONFIG_FILE:-/tmp/rclone.conf}"
+collector_pid=""
+backup_pid=""
+
+if ! [[ "$BACKUP_INTERVAL_SECONDS" =~ ^[1-9][0-9]*$ ]]; then
+  echo "BACKUP_INTERVAL_SECONDS must be a positive integer." >&2
+  exit 1
+fi
+
+mkdir -p "$DATA_DIR"
+
+if [[ -n "${RCLONE_CONFIG_B64:-}" ]]; then
+  printf '%s' "$RCLONE_CONFIG_B64" | base64 --decode >"$RCLONE_CONFIG_FILE"
+  chmod 600 "$RCLONE_CONFIG_FILE"
+  export RCLONE_CONFIG="$RCLONE_CONFIG_FILE"
+fi
+
+if [[ "${REQUIRE_REMOTE_BACKUP:-false}" == "true" ]]; then
+  if [[ -n "${HF_TOKEN:-}" ]]; then
+    :
+  elif [[ -n "${RCLONE_REMOTE:-}" && -s "$RCLONE_CONFIG_FILE" ]]; then
+    :
+  else
+    echo "Remote backup is required but neither Hugging Face nor rclone is configured." >&2
+    exit 1
+  fi
+fi
+
+stop_children() {
+  trap - EXIT INT TERM
+  if [[ -n "$collector_pid" ]]; then
+    kill -TERM "$collector_pid" 2>/dev/null || true
+  fi
+  if [[ -n "$backup_pid" ]]; then
+    kill -TERM "$backup_pid" 2>/dev/null || true
+  fi
+  wait "$collector_pid" 2>/dev/null || true
+  wait "$backup_pid" 2>/dev/null || true
+}
+trap stop_children EXIT INT TERM
+
+if [[ -n "${HF_TOKEN:-}" ]]; then
+  python -m polymarket_collector.hf_backup --loop &
+  backup_pid=$!
+elif [[ -n "${RCLONE_REMOTE:-}" ]]; then
+  (
+    while true; do
+      if ! ENV_FILE=/dev/null DATA_DIR="$DATA_DIR" \
+        /app/scripts/polymarket-backup; then
+        echo "Backup failed; files are kept locally and will be retried." >&2
+      fi
+      sleep "$BACKUP_INTERVAL_SECONDS"
+    done
+  ) &
+  backup_pid=$!
+fi
+
+polymarket-collector &
+collector_pid=$!
+set +e
+wait "$collector_pid"
+status=$?
+set -e
+collector_pid=""
+exit "$status"

deploy/hf-dataset-card.md

@@ -0,0 +1,22 @@
+---
+license: other
+task_categories:
+- time-series-forecasting
+- reinforcement-learning
+tags:
+- polymarket
+- prediction-markets
+- order-book
+- market-microstructure
+---
+
+# Polymarket rewarded-market L2 history
+
+Append-only archives of public Polymarket CLOB WebSocket messages for markets
+eligible for liquidity rewards. The dataset is intended for market
+microstructure research, backtests and machine-learning experiments.
+
+No wallet data, private keys or authenticated trading information is collected.
+Files are gzip-compressed JSON Lines and grouped by source and UTC date.
+The source data comes from Polymarket's public CLOB interfaces and remains
+subject to the applicable source-platform terms.

deploy/koyeb.env.example

@@ -0,0 +1,15 @@
+# Keep the Koyeb service type strictly on "free".
+MAX_MARKETS=40
+DATA_DIR=/data
+HEALTH_HOST=0.0.0.0
+PORT=8080
+ROTATION_SECONDS=300
+MAX_FILE_MIB=32
+BACKUP_INTERVAL_SECONDS=120
+BACKUP_MIN_AGE=30s
+REMOVE_AFTER_UPLOAD=true
+REQUIRE_REMOTE_BACKUP=true
+RCLONE_REMOTE=gdrive:polymarket-data
+
+# Secret only: base64 of an rclone.conf containing the Google Drive token.
+RCLONE_CONFIG_B64=

deploy/polymarket-backup.service

@@ -0,0 +1,12 @@
+[Unit]
+Description=Archive Polymarket collector files with rclone
+After=network-online.target polymarket-collector.service
+Wants=network-online.target
+
+[Service]
+Type=oneshot
+EnvironmentFile=-/etc/polymarket-collector.env
+ExecStart=/usr/local/sbin/polymarket-backup
+Nice=10
+IOSchedulingClass=idle
+

deploy/polymarket-backup.timer

@@ -0,0 +1,12 @@
+[Unit]
+Description=Hourly Polymarket data archive
+
+[Timer]
+OnBootSec=20min
+OnUnitActiveSec=1h
+RandomizedDelaySec=5min
+Persistent=true
+
+[Install]
+WantedBy=timers.target
+

deploy/polymarket-collector.service

@@ -0,0 +1,33 @@
+[Unit]
+Description=Polymarket LP public market-data collector
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=polymarket
+Group=polymarket
+WorkingDirectory=/opt/polymarket-collector
+EnvironmentFile=-/etc/polymarket-collector.env
+ExecStart=/opt/polymarket-collector/.venv/bin/polymarket-collector
+Restart=always
+RestartSec=5
+TimeoutStopSec=30
+
+# Keep the small VPS usable even if a malformed feed causes a memory spike.
+MemoryMax=700M
+TasksMax=64
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictAddressFamilies=AF_INET AF_INET6 AF_UNIX
+ReadWritePaths=/var/lib/polymarket-collector
+
+[Install]
+WantedBy=multi-user.target
+

polymarket_collector/__init__.py

@@ -0,0 +1,4 @@
+"""Polymarket LP market-data collector."""
+
+__version__ = "0.1.0"
+

polymarket_collector/__main__.py

@@ -0,0 +1,6 @@
+from .main import cli
+
+
+if __name__ == "__main__":
+    cli()
+

polymarket_collector/config.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+
+def _load_env_file(path: Path) -> None:
+    """Load a small dotenv-compatible file without adding a dependency."""
+    if not path.is_file():
+        return
+    for raw_line in path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        key = key.strip()
+        value = value.strip().strip("\"").strip("'")
+        if key:
+            os.environ.setdefault(key, value)
+
+
+def _int(name: str, default: int, minimum: int = 0) -> int:
+    raw = os.getenv(name, str(default))
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise ValueError(f"{name} must be an integer, got {raw!r}") from exc
+    if value < minimum:
+        raise ValueError(f"{name} must be >= {minimum}, got {value}")
+    return value
+
+
+def _float(name: str, default: float, minimum: float = 0) -> float:
+    raw = os.getenv(name, str(default))
+    try:
+        value = float(raw)
+    except ValueError as exc:
+        raise ValueError(f"{name} must be a number, got {raw!r}") from exc
+    if value < minimum:
+        raise ValueError(f"{name} must be >= {minimum}, got {value}")
+    return value
+
+
+@dataclass(frozen=True, slots=True)
+class Config:
+    data_dir: Path
+    max_markets: int
+    min_daily_reward: float
+    min_volume_24h: float
+    discovery_interval_seconds: int
+    rotation_seconds: int
+    max_file_bytes: int
+    flush_interval_seconds: int
+    health_host: str
+    health_port: int
+    stale_after_seconds: int
+    log_level: str
+    clob_base_url: str = "https://clob.polymarket.com"
+    websocket_url: str = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
+
+    @classmethod
+    def from_env(cls) -> "Config":
+        env_file = Path(os.getenv("ENV_FILE", ".env"))
+        _load_env_file(env_file)
+        return cls(
+            data_dir=Path(os.getenv("DATA_DIR", "data")).expanduser().resolve(),
+            max_markets=_int("MAX_MARKETS", 75, 1),
+            min_daily_reward=_float("MIN_DAILY_REWARD", 0),
+            min_volume_24h=_float("MIN_VOLUME_24H", 0),
+            discovery_interval_seconds=_int(
+                "DISCOVERY_INTERVAL_SECONDS", 900, 30
+            ),
+            rotation_seconds=_int("ROTATION_SECONDS", 900, 60),
+            max_file_bytes=_int("MAX_FILE_MIB", 64, 1) * 1024 * 1024,
+            flush_interval_seconds=_int("FLUSH_INTERVAL_SECONDS", 2, 1),
+            health_host=os.getenv("HEALTH_HOST", "127.0.0.1"),
+            # PaaS providers such as Koyeb inject PORT. HEALTH_PORT keeps
+            # precedence so local/systemd installations remain configurable.
+            health_port=_int(
+                "HEALTH_PORT", int(os.getenv("PORT", "8080")), 1
+            ),
+            stale_after_seconds=_int("STALE_AFTER_SECONDS", 180, 30),
+            log_level=os.getenv("LOG_LEVEL", "INFO").upper(),
+            clob_base_url=os.getenv(
+                "CLOB_BASE_URL", "https://clob.polymarket.com"
+            ).rstrip("/"),
+            websocket_url=os.getenv(
+                "WEBSOCKET_URL",
+                "wss://ws-subscriptions-clob.polymarket.com/ws/market",
+            ),
+        )

polymarket_collector/discovery.py

@@ -0,0 +1,171 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any
+
+
+USER_AGENT = "polymarket-lp-collector/0.1 (+public-market-data-only)"
+PAGE_SIZE = 500
+CURRENT_REWARDS_PAGE_LIMIT = 4
+CANDIDATE_MULTIPLIER = 4
+
+
+@dataclass(frozen=True, slots=True)
+class MarketSelection:
+    markets: tuple[dict[str, Any], ...]
+    asset_ids: tuple[str, ...]
+
+
+def daily_reward(market: dict[str, Any]) -> float:
+    explicit = market.get("total_daily_rate")
+    if explicit is not None:
+        try:
+            return float(explicit)
+        except (TypeError, ValueError):
+            pass
+    total = 0.0
+    for reward in market.get("rewards_config") or []:
+        try:
+            total += float(reward.get("rate_per_day") or 0)
+        except (TypeError, ValueError):
+            continue
+    return total
+
+
+def select_markets(
+    markets: list[dict[str, Any]],
+    *,
+    max_markets: int,
+    min_daily_reward: float,
+    min_volume_24h: float,
+) -> MarketSelection:
+    eligible: list[dict[str, Any]] = []
+    for market in markets:
+        tokens = market.get("tokens") or []
+        token_ids = [str(t.get("token_id", "")) for t in tokens if t.get("token_id")]
+        if len(token_ids) < 2:
+            continue
+        try:
+            volume = float(market.get("volume_24hr") or 0)
+        except (TypeError, ValueError):
+            volume = 0.0
+        reward = daily_reward(market)
+        if reward < min_daily_reward or volume < min_volume_24h:
+            continue
+        normalized = dict(market)
+        normalized["collector_daily_reward"] = reward
+        normalized["collector_volume_24hr"] = volume
+        eligible.append(normalized)
+
+    eligible.sort(
+        key=lambda m: (
+            float(m.get("collector_daily_reward") or 0),
+            float(m.get("collector_volume_24hr") or 0),
+        ),
+        reverse=True,
+    )
+    selected = eligible[:max_markets]
+
+    seen: set[str] = set()
+    asset_ids: list[str] = []
+    for market in selected:
+        for token in market.get("tokens") or []:
+            token_id = str(token.get("token_id", ""))
+            if token_id and token_id not in seen:
+                seen.add(token_id)
+                asset_ids.append(token_id)
+    return MarketSelection(tuple(selected), tuple(asset_ids))
+
+
+def _get_json(url: str, timeout: int = 30) -> dict[str, Any]:
+    request = urllib.request.Request(
+        url,
+        headers={"Accept": "application/json", "User-Agent": USER_AGENT},
+    )
+    with urllib.request.urlopen(request, timeout=timeout) as response:
+        if response.status != 200:
+            raise RuntimeError(f"HTTP {response.status} from {url}")
+        payload = json.loads(response.read())
+    if not isinstance(payload, dict):
+        raise RuntimeError(f"Unexpected response type from {url}")
+    return payload
+
+
+async def discover_rewarded_markets(
+    base_url: str,
+    *,
+    max_markets: int,
+    min_daily_reward: float,
+    min_volume_24h: float,
+) -> MarketSelection:
+    endpoint = f"{base_url}/rewards/markets/multi"
+    cursor: str | None = None
+    all_markets: list[dict[str, Any]] = []
+    # The endpoint can contain many thousands of short-lived sports markets.
+    # It is already sorted by reward rate, so fetching a bounded candidate set
+    # avoids delaying the initial websocket subscription by several minutes.
+    candidate_limit = max(PAGE_SIZE, max_markets * CANDIDATE_MULTIPLIER)
+
+    for _ in range(100):
+        params: list[tuple[str, str]] = [
+            ("page_size", str(PAGE_SIZE)),
+            ("order_by", "rate_per_day"),
+            ("position", "DESC"),
+        ]
+        if min_volume_24h > 0:
+            params.append(("min_volume_24hr", str(min_volume_24h)))
+        if cursor:
+            params.append(("next_cursor", cursor))
+        url = f"{endpoint}?{urllib.parse.urlencode(params)}"
+        payload = await asyncio.to_thread(_get_json, url)
+        page = payload.get("data") or []
+        if not isinstance(page, list):
+            raise RuntimeError("Rewards API returned a non-list data field")
+        all_markets.extend(item for item in page if isinstance(item, dict))
+        cursor = str(payload.get("next_cursor") or "LTE=")
+        if cursor == "LTE=" or not page or len(all_markets) >= candidate_limit:
+            break
+    else:
+        raise RuntimeError("Rewards pagination exceeded 100 pages")
+
+    # The current endpoint includes native + sponsored daily totals. Preserve
+    # both raw responses so historic selection decisions remain reproducible.
+    current_by_condition: dict[str, dict[str, Any]] = {}
+    cursor = None
+    current_endpoint = f"{base_url}/rewards/markets/current"
+    # Current rewards are not sorted and can also span many thousands of
+    # entries. A bounded enrichment pass keeps discovery timely; the raw
+    # candidate reward configuration remains available as the fallback.
+    for _ in range(CURRENT_REWARDS_PAGE_LIMIT):
+        params = []
+        if cursor:
+            params.append(("next_cursor", cursor))
+        suffix = f"?{urllib.parse.urlencode(params)}" if params else ""
+        payload = await asyncio.to_thread(_get_json, current_endpoint + suffix)
+        page = payload.get("data") or []
+        if not isinstance(page, list):
+            raise RuntimeError("Current rewards API returned a non-list data field")
+        for item in page:
+            if isinstance(item, dict) and item.get("condition_id"):
+                current_by_condition[str(item["condition_id"])] = item
+        cursor = str(payload.get("next_cursor") or "LTE=")
+        if cursor == "LTE=" or not page:
+            break
+
+    for market in all_markets:
+        current = current_by_condition.get(str(market.get("condition_id", "")))
+        if current:
+            market["current_rewards"] = current
+            if current.get("total_daily_rate") is not None:
+                market["total_daily_rate"] = current["total_daily_rate"]
+
+    return select_markets(
+        all_markets,
+        max_markets=max_markets,
+        min_daily_reward=min_daily_reward,
+        min_volume_24h=min_volume_24h,
+    )

polymarket_collector/health.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Final
+
+from .state import CollectorState
+
+
+MAX_REQUEST_BYTES: Final = 8192
+
+
+class HealthServer:
+    def __init__(
+        self,
+        state: CollectorState,
+        host: str,
+        port: int,
+        stale_after_seconds: int,
+    ) -> None:
+        self.state = state
+        self.host = host
+        self.port = port
+        self.stale_after_seconds = stale_after_seconds
+        self._server: asyncio.Server | None = None
+
+    async def start(self) -> None:
+        self._server = await asyncio.start_server(
+            self._handle, host=self.host, port=self.port
+        )
+
+    async def close(self) -> None:
+        if self._server is not None:
+            self._server.close()
+            await self._server.wait_closed()
+
+    async def _handle(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        try:
+            raw = await asyncio.wait_for(
+                reader.readuntil(b"\r\n\r\n"), timeout=2
+            )
+            if len(raw) > MAX_REQUEST_BYTES:
+                raise ValueError("request too large")
+            first_line = raw.split(b"\r\n", 1)[0].decode("ascii", "replace")
+            _, path, _ = first_line.split(" ", 2)
+            if path == "/healthz":
+                payload = self.state.as_dict(self.stale_after_seconds)
+                status = 200 if payload["healthy"] else 503
+                await self._respond_json(writer, status, payload)
+            elif path == "/metrics":
+                await self._respond_metrics(writer)
+            else:
+                await self._respond_json(writer, 404, {"error": "not_found"})
+        except Exception:
+            await self._respond_json(writer, 400, {"error": "bad_request"})
+        finally:
+            writer.close()
+            await writer.wait_closed()
+
+    async def _respond_json(self, writer: asyncio.StreamWriter, status: int, payload: object) -> None:
+        body = json.dumps(payload, separators=(",", ":")).encode()
+        await self._respond(writer, status, "application/json", body)
+
+    async def _respond_metrics(self, writer: asyncio.StreamWriter) -> None:
+        healthy, _ = self.state.health(self.stale_after_seconds)
+        lines = [
+            "# TYPE polymarket_collector_up gauge",
+            f"polymarket_collector_up {1 if healthy else 0}",
+            "# TYPE polymarket_collector_markets gauge",
+            f"polymarket_collector_markets {self.state.markets}",
+            "# TYPE polymarket_collector_assets gauge",
+            f"polymarket_collector_assets {self.state.assets}",
+            "# TYPE polymarket_collector_messages_total counter",
+            f"polymarket_collector_messages_total {self.state.messages_total}",
+            "# TYPE polymarket_collector_reconnects_total counter",
+            f"polymarket_collector_reconnects_total {self.state.reconnects_total}",
+            "",
+        ]
+        await self._respond(writer, 200, "text/plain; version=0.0.4", "\n".join(lines).encode())
+
+    async def _respond(
+        self,
+        writer: asyncio.StreamWriter,
+        status: int,
+        content_type: str,
+        body: bytes,
+    ) -> None:
+        reason = {200: "OK", 400: "Bad Request", 404: "Not Found", 503: "Service Unavailable"}.get(status, "Error")
+        headers = (
+            f"HTTP/1.1 {status} {reason}\r\n"
+            f"Content-Type: {content_type}\r\n"
+            f"Content-Length: {len(body)}\r\n"
+            "Connection: close\r\n\r\n"
+        ).encode("ascii")
+        writer.write(headers + body)
+        await writer.drain()
+

polymarket_collector/hf_backup.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+import argparse
+import logging
+import os
+import time
+from pathlib import Path
+
+from huggingface_hub import CommitOperationAdd, HfApi
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+def completed_archives(
+    data_dir: Path,
+    *,
+    min_age_seconds: int,
+    max_files: int = 80,
+) -> list[Path]:
+    cutoff = time.time() - min_age_seconds
+    paths = [
+        path
+        for path in data_dir.rglob("*.jsonl.gz")
+        if path.is_file() and path.stat().st_mtime <= cutoff
+    ]
+    return sorted(paths)[:max_files]
+
+
+def upload_once(
+    *,
+    data_dir: Path,
+    repo_id: str,
+    token: str,
+    min_age_seconds: int,
+    api: HfApi | None = None,
+) -> int:
+    paths = completed_archives(
+        data_dir,
+        min_age_seconds=min_age_seconds,
+    )
+    if not paths:
+        return 0
+
+    client = api or HfApi(token=token)
+    operations = [
+        CommitOperationAdd(
+            path_in_repo=path.relative_to(data_dir).as_posix(),
+            path_or_fileobj=path,
+        )
+        for path in paths
+    ]
+    client.create_commit(
+        repo_id=repo_id,
+        repo_type="dataset",
+        operations=operations,
+        commit_message=f"Add {len(paths)} Polymarket archive(s)",
+    )
+    for path in paths:
+        path.unlink()
+    LOGGER.info("Uploaded and removed %d local archive(s)", len(paths))
+    return len(paths)
+
+
+def run_loop() -> None:
+    data_dir = Path(os.getenv("DATA_DIR", "/data")).resolve()
+    token = os.environ["HF_TOKEN"]
+    api = HfApi(token=token)
+    repo_id = os.getenv("HF_DATASET_REPO")
+    if not repo_id:
+        namespace = api.whoami()["name"]
+        repo_id = f"{namespace}/polymarket-l2-history"
+        LOGGER.info("Using inferred Hugging Face dataset: %s", repo_id)
+    interval = int(os.getenv("BACKUP_INTERVAL_SECONDS", "3600"))
+    min_age = int(os.getenv("BACKUP_MIN_AGE_SECONDS", "120"))
+    if interval < 60 or min_age < 0:
+        raise ValueError("Invalid backup interval or minimum age")
+
+    while True:
+        try:
+            upload_once(
+                data_dir=data_dir,
+                repo_id=repo_id,
+                token=token,
+                min_age_seconds=min_age,
+                api=api,
+            )
+        except Exception:
+            LOGGER.exception("Hugging Face backup failed; retrying later")
+        time.sleep(interval)
+
+
+def cli() -> None:
+    parser = argparse.ArgumentParser(
+        description="Upload completed collector archives to a HF dataset."
+    )
+    parser.add_argument("--loop", action="store_true")
+    args = parser.parse_args()
+    logging.basicConfig(
+        level=os.getenv("LOG_LEVEL", "INFO"),
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    if args.loop:
+        run_loop()
+        return
+
+    upload_once(
+        data_dir=Path(os.getenv("DATA_DIR", "/data")).resolve(),
+        repo_id=os.environ["HF_DATASET_REPO"],
+        token=os.environ["HF_TOKEN"],
+        min_age_seconds=int(os.getenv("BACKUP_MIN_AGE_SECONDS", "120")),
+    )
+
+
+if __name__ == "__main__":
+    cli()

polymarket_collector/main.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+import asyncio
+import argparse
+import json
+import logging
+import signal
+import sys
+import time
+from typing import Any
+
+from .config import Config
+from .discovery import MarketSelection, discover_rewarded_markets
+from .health import HealthServer
+from .state import CollectorState
+from .storage import RotatingJsonlWriter
+from .stream import MarketStream
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+async def _wait_or_stop(stop_event: asyncio.Event, timeout: float) -> None:
+    try:
+        await asyncio.wait_for(stop_event.wait(), timeout)
+    except TimeoutError:
+        pass
+
+
+async def _discover(config: Config) -> MarketSelection:
+    return await discover_rewarded_markets(
+        config.clob_base_url,
+        max_markets=config.max_markets,
+        min_daily_reward=config.min_daily_reward,
+        min_volume_24h=config.min_volume_24h,
+    )
+
+
+async def run(config: Config) -> None:
+    stop_event = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        try:
+            loop.add_signal_handler(sig, stop_event.set)
+        except NotImplementedError:
+            pass
+
+    state = CollectorState()
+    market_writer = RotatingJsonlWriter(
+        config.data_dir / "market_ws",
+        "market_ws",
+        rotation_seconds=config.rotation_seconds,
+        max_file_bytes=config.max_file_bytes,
+        flush_interval_seconds=config.flush_interval_seconds,
+    )
+    discovery_writer = RotatingJsonlWriter(
+        config.data_dir / "discovery",
+        "discovery",
+        rotation_seconds=config.rotation_seconds,
+        max_file_bytes=config.max_file_bytes,
+        flush_interval_seconds=config.flush_interval_seconds,
+    )
+    await market_writer.start()
+    await discovery_writer.start()
+
+    health = HealthServer(
+        state,
+        config.health_host,
+        config.health_port,
+        config.stale_after_seconds,
+    )
+    await health.start()
+    LOGGER.info(
+        "Health endpoint listening on http://%s:%d/healthz",
+        config.health_host,
+        config.health_port,
+    )
+
+    stream = MarketStream(config.websocket_url, market_writer, state)
+    stream_task = asyncio.create_task(stream.run(stop_event), name="market-stream")
+
+    try:
+        failure_delay = 10.0
+        while not stop_event.is_set():
+            try:
+                selection = await _discover(config)
+                discovered_at_ns = time.time_ns()
+                state.discovered_at = discovered_at_ns / 1_000_000_000
+                state.markets = len(selection.markets)
+                state.assets = len(selection.asset_ids)
+                state.last_error = None
+                await discovery_writer.write(
+                    {
+                        "record_type": "rewarded_market_discovery",
+                        "received_at_ns": discovered_at_ns,
+                        "market_count": len(selection.markets),
+                        "asset_count": len(selection.asset_ids),
+                        "markets": list(selection.markets),
+                    }
+                )
+                stream.set_assets(selection.asset_ids)
+                LOGGER.info(
+                    "Selected %d rewarded markets (%d assets)",
+                    len(selection.markets),
+                    len(selection.asset_ids),
+                )
+                failure_delay = 10.0
+                await _wait_or_stop(
+                    stop_event, config.discovery_interval_seconds
+                )
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                state.discovery_failures_total += 1
+                state.last_error = f"discovery: {type(exc).__name__}: {exc}"
+                LOGGER.exception("Market discovery failed")
+                await discovery_writer.write(
+                    {
+                        "record_type": "discovery_error",
+                        "received_at_ns": time.time_ns(),
+                        "error_type": type(exc).__name__,
+                        "error": str(exc),
+                    }
+                )
+                await _wait_or_stop(stop_event, failure_delay)
+                failure_delay = min(failure_delay * 2, 300)
+    finally:
+        stop_event.set()
+        try:
+            await asyncio.wait_for(stream_task, timeout=10)
+        except TimeoutError:
+            stream_task.cancel()
+            await asyncio.gather(stream_task, return_exceptions=True)
+        await health.close()
+        await asyncio.gather(
+            market_writer.close(), discovery_writer.close()
+        )
+        LOGGER.info("Collector stopped cleanly")
+
+
+def cli() -> None:
+    try:
+        parser = argparse.ArgumentParser(
+            description="Collect public L2 data for rewarded Polymarket markets."
+        )
+        parser.add_argument(
+            "--check",
+            action="store_true",
+            help="test public API discovery and exit without opening the collector",
+        )
+        args = parser.parse_args()
+        config = Config.from_env()
+        logging.basicConfig(
+            level=getattr(logging, config.log_level, logging.INFO),
+            format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+        )
+        if args.check:
+            selection = asyncio.run(_discover(config))
+            print(
+                json.dumps(
+                    {
+                        "ok": True,
+                        "markets": len(selection.markets),
+                        "assets": len(selection.asset_ids),
+                        "top_markets": [
+                            {
+                                "question": market.get("question"),
+                                "daily_reward": market.get("collector_daily_reward"),
+                                "volume_24hr": market.get("collector_volume_24hr"),
+                            }
+                            for market in selection.markets[:5]
+                        ],
+                    },
+                    ensure_ascii=False,
+                    indent=2,
+                )
+            )
+        else:
+            asyncio.run(run(config))
+    except KeyboardInterrupt:
+        pass
+    except Exception as exc:
+        print(f"fatal: {type(exc).__name__}: {exc}", file=sys.stderr)
+        raise SystemExit(1) from exc
+
+
+if __name__ == "__main__":
+    cli()

polymarket_collector/state.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from typing import Any
+
+
+def _iso(timestamp: float | None) -> str | None:
+    if timestamp is None:
+        return None
+    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat()
+
+
+@dataclass(slots=True)
+class CollectorState:
+    started_at: float = field(default_factory=time.time)
+    discovered_at: float | None = None
+    connected_at: float | None = None
+    disconnected_at: float | None = None
+    last_message_at: float | None = None
+    last_pong_at: float | None = None
+    connected: bool = False
+    markets: int = 0
+    assets: int = 0
+    messages_total: int = 0
+    invalid_messages_total: int = 0
+    reconnects_total: int = 0
+    discovery_failures_total: int = 0
+    last_error: str | None = None
+
+    def health(self, stale_after_seconds: int) -> tuple[bool, str]:
+        if self.discovered_at is None:
+            return False, "waiting_for_discovery"
+        if self.assets == 0:
+            return False, "no_rewarded_markets"
+        if not self.connected:
+            return False, "websocket_disconnected"
+        if self.last_message_at is None:
+            return False, "waiting_for_first_message"
+        if time.time() - self.last_message_at > stale_after_seconds:
+            return False, "market_data_stale"
+        return True, "ok"
+
+    def as_dict(self, stale_after_seconds: int) -> dict[str, Any]:
+        healthy, reason = self.health(stale_after_seconds)
+        return {
+            "healthy": healthy,
+            "reason": reason,
+            "started_at": _iso(self.started_at),
+            "uptime_seconds": round(time.time() - self.started_at, 3),
+            "discovered_at": _iso(self.discovered_at),
+            "connected_at": _iso(self.connected_at),
+            "disconnected_at": _iso(self.disconnected_at),
+            "last_message_at": _iso(self.last_message_at),
+            "last_pong_at": _iso(self.last_pong_at),
+            "connected": self.connected,
+            "markets": self.markets,
+            "assets": self.assets,
+            "messages_total": self.messages_total,
+            "invalid_messages_total": self.invalid_messages_total,
+            "reconnects_total": self.reconnects_total,
+            "discovery_failures_total": self.discovery_failures_total,
+            "last_error": self.last_error,
+        }
+

polymarket_collector/storage.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import asyncio
+import gzip
+import json
+import os
+import shutil
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, BinaryIO
+
+
+def _compress_part(path: Path) -> Path:
+    destination = path.with_suffix("")
+    if destination.suffix != ".jsonl":
+        destination = Path(str(path).removesuffix(".part"))
+    destination = destination.with_suffix(destination.suffix + ".gz")
+    temporary = Path(str(destination) + ".tmp")
+    try:
+        with path.open("rb") as source, gzip.open(
+            temporary, "wb", compresslevel=5
+        ) as target:
+            shutil.copyfileobj(source, target, length=1024 * 1024)
+        os.replace(temporary, destination)
+        path.unlink(missing_ok=True)
+        return destination
+    except Exception:
+        temporary.unlink(missing_ok=True)
+        raise
+
+
+class RotatingJsonlWriter:
+    def __init__(
+        self,
+        root: Path,
+        source: str,
+        *,
+        rotation_seconds: int,
+        max_file_bytes: int,
+        flush_interval_seconds: int,
+    ) -> None:
+        self.root = root
+        self.source = source
+        self.rotation_seconds = rotation_seconds
+        self.max_file_bytes = max_file_bytes
+        self.flush_interval_seconds = flush_interval_seconds
+        self._file: BinaryIO | None = None
+        self._path: Path | None = None
+        self._opened_monotonic = 0.0
+        self._last_flush_monotonic = 0.0
+        self._size = 0
+        self._sequence = 0
+        self._compression_tasks: set[asyncio.Task[Path]] = set()
+        self._compression_errors: list[BaseException] = []
+
+    @property
+    def current_path(self) -> str | None:
+        return str(self._path) if self._path else None
+
+    async def start(self) -> None:
+        self.root.mkdir(parents=True, exist_ok=True)
+        await asyncio.to_thread(self._recover_parts)
+
+    def _recover_parts(self) -> None:
+        for temporary in self.root.rglob("*.jsonl.gz.tmp"):
+            temporary.unlink(missing_ok=True)
+        for part in sorted(self.root.rglob("*.jsonl.part")):
+            if part.stat().st_size == 0:
+                part.unlink(missing_ok=True)
+            else:
+                _compress_part(part)
+
+    def _open(self) -> None:
+        now = datetime.now(timezone.utc)
+        directory = self.root / now.strftime("%Y/%m/%d")
+        directory.mkdir(parents=True, exist_ok=True)
+        self._sequence += 1
+        stamp = now.strftime("%Y%m%dT%H%M%S.%fZ")
+        self._path = directory / (
+            f"{self.source}-{stamp}-{os.getpid()}-{self._sequence:04d}.jsonl.part"
+        )
+        self._file = self._path.open("ab", buffering=64 * 1024)
+        self._size = self._path.stat().st_size
+        self._opened_monotonic = time.monotonic()
+        self._last_flush_monotonic = self._opened_monotonic
+
+    async def write(self, record: dict[str, Any]) -> None:
+        if self._file is None:
+            self._open()
+        encoded = json.dumps(
+            record, ensure_ascii=False, separators=(",", ":")
+        ).encode("utf-8") + b"\n"
+        assert self._file is not None
+        self._file.write(encoded)
+        self._size += len(encoded)
+
+        now = time.monotonic()
+        if now - self._last_flush_monotonic >= self.flush_interval_seconds:
+            self._file.flush()
+            self._last_flush_monotonic = now
+        if (
+            self._size >= self.max_file_bytes
+            or now - self._opened_monotonic >= self.rotation_seconds
+        ):
+            self._rotate()
+
+    def _rotate(self) -> None:
+        if self._file is None or self._path is None:
+            return
+        self._file.flush()
+        os.fsync(self._file.fileno())
+        self._file.close()
+        part = self._path
+        self._file = None
+        self._path = None
+        task = asyncio.create_task(asyncio.to_thread(_compress_part, part))
+        self._compression_tasks.add(task)
+        task.add_done_callback(self._compression_done)
+
+    def _compression_done(self, task: asyncio.Task[Path]) -> None:
+        self._compression_tasks.discard(task)
+        try:
+            task.result()
+        except BaseException as exc:
+            self._compression_errors.append(exc)
+
+    async def close(self) -> None:
+        self._rotate()
+        if self._compression_tasks:
+            await asyncio.gather(
+                *tuple(self._compression_tasks), return_exceptions=True
+            )
+        if self._compression_errors:
+            raise RuntimeError(f"Compression failed: {self._compression_errors[0]}")

polymarket_collector/stream.py

@@ -0,0 +1,200 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import random
+import time
+import uuid
+from collections.abc import Iterable
+from typing import Any
+
+from websockets.asyncio.client import connect
+from websockets.exceptions import ConnectionClosed
+
+from .state import CollectorState
+from .storage import RotatingJsonlWriter
+
+
+LOGGER = logging.getLogger(__name__)
+
+
+class MarketStream:
+    def __init__(
+        self,
+        url: str,
+        writer: RotatingJsonlWriter,
+        state: CollectorState,
+    ) -> None:
+        self.url = url
+        self.writer = writer
+        self.state = state
+        self._desired_assets: set[str] = set()
+        self._assets_changed = asyncio.Event()
+
+    def set_assets(self, asset_ids: Iterable[str]) -> None:
+        desired = {str(asset_id) for asset_id in asset_ids if asset_id}
+        if desired != self._desired_assets:
+            self._desired_assets = desired
+            self._assets_changed.set()
+
+    async def run(self, stop_event: asyncio.Event) -> None:
+        delay = 1.0
+        while not stop_event.is_set():
+            if not self._desired_assets:
+                await self._wait_for_assets_or_stop(stop_event)
+                continue
+            try:
+                connected_for = await self._connected_loop(stop_event)
+                delay = 1.0 if connected_for >= 30 else min(delay * 2, 60)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                self.state.last_error = f"websocket: {type(exc).__name__}: {exc}"
+                LOGGER.warning("WebSocket disconnected: %s", exc)
+                delay = min(delay * 2, 60)
+
+            self.state.connected = False
+            self.state.disconnected_at = time.time()
+            if not stop_event.is_set():
+                self.state.reconnects_total += 1
+                try:
+                    await asyncio.wait_for(stop_event.wait(), delay + random.random())
+                except TimeoutError:
+                    pass
+
+    async def _wait_for_assets_or_stop(self, stop_event: asyncio.Event) -> None:
+        assets_task = asyncio.create_task(self._assets_changed.wait())
+        stop_task = asyncio.create_task(stop_event.wait())
+        done, pending = await asyncio.wait(
+            {assets_task, stop_task}, return_when=asyncio.FIRST_COMPLETED
+        )
+        for task in pending:
+            task.cancel()
+        await asyncio.gather(*pending, return_exceptions=True)
+        if assets_task in done:
+            self._assets_changed.clear()
+
+    async def _connected_loop(self, stop_event: asyncio.Event) -> float:
+        connection_id = uuid.uuid4().hex
+        connected_monotonic = time.monotonic()
+        async with connect(
+            self.url,
+            open_timeout=20,
+            close_timeout=5,
+            ping_interval=None,
+            max_size=None,
+        ) as websocket:
+            subscribed = set(self._desired_assets)
+            await websocket.send(
+                json.dumps(
+                    {
+                        "assets_ids": sorted(subscribed),
+                        "type": "market",
+                        "custom_feature_enabled": True,
+                    },
+                    separators=(",", ":"),
+                )
+            )
+            if subscribed == self._desired_assets:
+                self._assets_changed.clear()
+            else:
+                self._assets_changed.set()
+            self.state.connected = True
+            self.state.connected_at = time.time()
+            self.state.last_error = None
+            LOGGER.info("WebSocket connected for %d assets", len(subscribed))
+            last_ping = 0.0
+
+            while not stop_event.is_set():
+                if self._assets_changed.is_set():
+                    self._assets_changed.clear()
+                    subscribed = await self._sync_subscriptions(websocket, subscribed)
+                    if subscribed != self._desired_assets:
+                        self._assets_changed.set()
+                    if not subscribed:
+                        break
+
+                now = time.monotonic()
+                if now - last_ping >= 10:
+                    await websocket.send("PING")
+                    last_ping = now
+
+                try:
+                    message = await asyncio.wait_for(websocket.recv(), timeout=1)
+                except TimeoutError:
+                    continue
+                except ConnectionClosed:
+                    break
+
+                if isinstance(message, bytes):
+                    message = message.decode("utf-8", "replace")
+                if message == "PONG":
+                    self.state.last_pong_at = time.time()
+                    continue
+                await self._record_message(message, connection_id)
+
+        return time.monotonic() - connected_monotonic
+
+    async def _sync_subscriptions(self, websocket: Any, subscribed: set[str]) -> set[str]:
+        desired = set(self._desired_assets)
+        removed = sorted(subscribed - desired)
+        added = sorted(desired - subscribed)
+        if removed:
+            await websocket.send(
+                json.dumps(
+                    {"assets_ids": removed, "operation": "unsubscribe"},
+                    separators=(",", ":"),
+                )
+            )
+        if added:
+            await websocket.send(
+                json.dumps(
+                    {
+                        "assets_ids": added,
+                        "operation": "subscribe",
+                        "custom_feature_enabled": True,
+                    },
+                    separators=(",", ":"),
+                )
+            )
+        if added or removed:
+            LOGGER.info(
+                "WebSocket subscriptions updated: +%d -%d (%d total)",
+                len(added),
+                len(removed),
+                len(desired),
+            )
+        return desired
+
+    async def _record_message(self, raw_message: str, connection_id: str) -> None:
+        received_at_ns = time.time_ns()
+        try:
+            payload = json.loads(raw_message)
+        except json.JSONDecodeError:
+            self.state.invalid_messages_total += 1
+            await self.writer.write(
+                {
+                    "record_type": "invalid_market_ws",
+                    "received_at_ns": received_at_ns,
+                    "connection_id": connection_id,
+                    "raw": raw_message,
+                }
+            )
+            return
+
+        messages = payload if isinstance(payload, list) else [payload]
+        for item in messages:
+            if not isinstance(item, dict):
+                self.state.invalid_messages_total += 1
+                continue
+            await self.writer.write(
+                {
+                    "record_type": "market_ws",
+                    "received_at_ns": received_at_ns,
+                    "connection_id": connection_id,
+                    "payload": item,
+                }
+            )
+            self.state.messages_total += 1
+        self.state.last_message_at = time.time()

polymarket_lp_collector.egg-info/PKG-INFO

@@ -0,0 +1,166 @@
+Metadata-Version: 2.4
+Name: polymarket-lp-collector
+Version: 0.1.0
+Summary: Low-resource Polymarket LP market-data collector
+Requires-Python: >=3.11
+Description-Content-Type: text/markdown
+Requires-Dist: huggingface-hub<2,>=1.0
+Requires-Dist: websockets<17,>=15
+
+---
+title: Polymarket LP Collector
+emoji: 📈
+colorFrom: green
+colorTo: blue
+sdk: docker
+app_port: 8080
+---
+
+# Collecteur de données LP Polymarket
+
+Ce projet enregistre en continu les données publiques de niveau 2 des marchés
+Polymarket qui distribuent des récompenses de liquidité. Il est conçu pour une
+petite VM Google Cloud `e2-micro` : mémoire limitée, fichiers rotatifs et aucune
+base de données résidente.
+
+Il ne signe aucun ordre, ne contient aucune clé de portefeuille et ne peut pas
+trader. Son unique rôle est de créer un historique propre pour les futurs
+backtests et modèles d'apprentissage automatique.
+
+## Données collectées
+
+- carnets complets (`book`) ;
+- changements de niveaux (`price_change`) ;
+- meilleurs bid/ask ;
+- dernières transactions ;
+- changements de tick et résolutions ;
+- métadonnées, volumes et paramètres de récompenses LP ;
+- heure de réception locale en nanosecondes pour mesurer les délais.
+
+Les fichiers terminés sont écrits sous la forme :
+
+```text
+data/
+  market_ws/YYYY/MM/DD/market_ws-....jsonl.gz
+  discovery/YYYY/MM/DD/discovery-....jsonl.gz
+```
+
+Chaque ligne est un objet JSON autonome. Les fichiers `.part` sont récupérés et
+compressés automatiquement au prochain démarrage après une coupure brutale.
+
+## Exécution locale
+
+```bash
+python3 -m venv .venv
+. .venv/bin/activate
+pip install -e .
+cp .env.example .env
+DATA_DIR="$PWD/data" polymarket-collector
+```
+
+Tester uniquement l'accès aux API publiques avant de démarrer :
+
+```bash
+polymarket-collector --check
+```
+
+État du service :
+
+```bash
+curl http://127.0.0.1:8080/healthz
+curl http://127.0.0.1:8080/metrics
+```
+
+Arrêter avec `Ctrl+C`. Le service finalise alors proprement les fichiers en
+cours.
+
+## Déploiement Google Cloud
+
+Depuis un ordinateur ou Google Cloud Shell ayant `gcloud` authentifié :
+
+```bash
+./scripts/deploy_gcp.sh ID_DU_PROJET_GCP
+```
+
+Le script crée par défaut une VM `e2-micro` Ubuntu 24.04 dans `us-east1-b`, un
+disque standard de 30 Go, copie le projet, puis installe et démarre le service.
+
+Commandes utiles après installation :
+
+```bash
+gcloud compute ssh polymarket-collector --zone us-east1-b
+sudo systemctl status polymarket-collector
+sudo journalctl -u polymarket-collector -f
+curl http://127.0.0.1:8080/healthz
+```
+
+## Sauvegarde vers Google Drive
+
+Sur la VM :
+
+```bash
+sudo rclone config
+sudoedit /etc/polymarket-collector.env
+```
+
+Définir ensuite, par exemple :
+
+```text
+RCLONE_REMOTE=gdrive:polymarket-data
+LOCAL_RETENTION_DAYS=7
+```
+
+Puis activer la sauvegarde :
+
+```bash
+sudo systemctl enable --now polymarket-backup.timer
+sudo systemctl start polymarket-backup.service
+sudo journalctl -u polymarket-backup.service -n 100
+```
+
+Les fichiers locaux âgés de plus de `LOCAL_RETENTION_DAYS` ne sont supprimés
+qu'après une copie `rclone` réussie. Mettre la valeur à `0` pour désactiver le
+nettoyage local.
+
+## Hébergement gratuit sans dépassement : Hugging Face Spaces
+
+L'architecture recommandée pour une collecte continue sans VM personnelle est :
+
+1. un Space public sur le matériel **CPU Basic gratuit** ;
+2. UptimeRobot Free sur `/healthz`, toutes les cinq minutes, pour maintenir une
+   visite régulière et surveiller le service ;
+3. un dataset Hugging Face public pour les archives. Les fichiers locaux sont
+   supprimés uniquement après la réussite du commit distant.
+
+Le conteneur est prêt dans `Dockerfile`. Créer un jeton Hugging Face avec droit
+d'écriture, puis lancer :
+
+```bash
+HF_TOKEN=hf_... python scripts/deploy_hf.py
+```
+
+Le script crée le Space et le dataset, force le matériel `cpu-basic`, ajoute le
+jeton comme secret du Space et applique le profil suivant : 40 marchés,
+rotation à 256 Mio et envoi groupé toutes les heures. Cela évite une quantité
+excessive de petits commits tout en gardant plusieurs heures de marge sur le
+disque temporaire.
+
+Koyeb n'est plus recommandé : depuis son rapprochement avec Mistral, les
+nouveaux utilisateurs doivent fournir un moyen de paiement et souscrire un
+forfait payant.
+
+## Tests
+
+```bash
+python -m unittest discover -s tests -v
+```
+
+Les tests ne contactent pas Polymarket et ne nécessitent aucune clé.
+
+## Limites importantes
+
+- Avec 1 Go de RAM et 30 Go de disque, commencer avec 75 marchés maximum.
+- Une VM gratuite aux États-Unis convient à la collecte, pas au futur bot de
+  trading sensible à la latence.
+- Le service conserve les messages bruts : les transformations en Parquet et
+  l'entraînement se feront sur l'ordinateur équipé de la RTX 3060.

polymarket_lp_collector.egg-info/SOURCES.txt

@@ -0,0 +1,24 @@
+README.md
+pyproject.toml
+polymarket_collector/__init__.py
+polymarket_collector/__main__.py
+polymarket_collector/config.py
+polymarket_collector/discovery.py
+polymarket_collector/health.py
+polymarket_collector/hf_backup.py
+polymarket_collector/main.py
+polymarket_collector/state.py
+polymarket_collector/storage.py
+polymarket_collector/stream.py
+polymarket_lp_collector.egg-info/PKG-INFO
+polymarket_lp_collector.egg-info/SOURCES.txt
+polymarket_lp_collector.egg-info/dependency_links.txt
+polymarket_lp_collector.egg-info/entry_points.txt
+polymarket_lp_collector.egg-info/requires.txt
+polymarket_lp_collector.egg-info/top_level.txt
+tests/test_config.py
+tests/test_discovery.py
+tests/test_health.py
+tests/test_hf_backup.py
+tests/test_storage.py
+tests/test_stream.py

polymarket_lp_collector.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

polymarket_lp_collector.egg-info/entry_points.txt

@@ -0,0 +1,2 @@
+[console_scripts]
+polymarket-collector = polymarket_collector.main:cli

polymarket_lp_collector.egg-info/requires.txt

@@ -0,0 +1,2 @@
+huggingface-hub<2,>=1.0
+websockets<17,>=15

polymarket_lp_collector.egg-info/top_level.txt

@@ -0,0 +1 @@
+polymarket_collector

pyproject.toml

@@ -0,0 +1,20 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "polymarket-lp-collector"
+version = "0.1.0"
+description = "Low-resource Polymarket LP market-data collector"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+  "huggingface-hub>=1.0,<2",
+  "websockets>=15,<17",
+]
+
+[project.scripts]
+polymarket-collector = "polymarket_collector.main:cli"
+
+[tool.setuptools.packages.find]
+include = ["polymarket_collector*"]

render.yaml

@@ -0,0 +1,27 @@
+services:
+  - type: web
+    name: polymarket-lp-collector
+    runtime: docker
+    plan: free
+    region: frankfurt
+    autoDeployTrigger: off
+    healthCheckPath: /healthz
+    envVars:
+      - key: HF_TOKEN
+        sync: false
+      - key: MAX_MARKETS
+        value: "40"
+      - key: DATA_DIR
+        value: /data
+      - key: HEALTH_HOST
+        value: 0.0.0.0
+      - key: ROTATION_SECONDS
+        value: "3600"
+      - key: MAX_FILE_MIB
+        value: "256"
+      - key: BACKUP_INTERVAL_SECONDS
+        value: "3600"
+      - key: BACKUP_MIN_AGE_SECONDS
+        value: "120"
+      - key: REQUIRE_REMOTE_BACKUP
+        value: "true"

scripts/deploy_gcp.sh

@@ -0,0 +1,65 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+PROJECT_ID="${1:-}"
+INSTANCE="${INSTANCE_NAME:-polymarket-collector}"
+ZONE="${GCP_ZONE:-us-east1-b}"
+MACHINE_TYPE="${GCP_MACHINE_TYPE:-e2-micro}"
+ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+
+if [[ -z "$PROJECT_ID" ]]; then
+  echo "Usage: $0 GOOGLE_CLOUD_PROJECT_ID" >&2
+  exit 2
+fi
+if ! command -v gcloud >/dev/null 2>&1; then
+  echo "gcloud is required. Run this script from Google Cloud Shell or install the Google Cloud CLI." >&2
+  exit 2
+fi
+if [[ ! -f "$ROOT_DIR/pyproject.toml" ]]; then
+  echo "Project files not found next to this script." >&2
+  exit 2
+fi
+
+ACTIVE_ACCOUNT="$(gcloud auth list --filter=status:ACTIVE --format='value(account)' | head -1)"
+if [[ -z "$ACTIVE_ACCOUNT" ]]; then
+  echo "No active Google Cloud account. Run: gcloud auth login" >&2
+  exit 2
+fi
+
+gcloud config set project "$PROJECT_ID" >/dev/null
+gcloud services enable compute.googleapis.com --project "$PROJECT_ID"
+
+if ! gcloud compute instances describe "$INSTANCE" --zone "$ZONE" \
+  --project "$PROJECT_ID" >/dev/null 2>&1; then
+  gcloud compute instances create "$INSTANCE" \
+    --project "$PROJECT_ID" \
+    --zone "$ZONE" \
+    --machine-type "$MACHINE_TYPE" \
+    --provisioning-model STANDARD \
+    --network-tier STANDARD \
+    --image-family ubuntu-2404-lts-amd64 \
+    --image-project ubuntu-os-cloud \
+    --boot-disk-type pd-standard \
+    --boot-disk-size 30GB \
+    --metadata enable-oslogin=TRUE \
+    --labels app=polymarket-collector
+else
+  echo "Instance $INSTANCE already exists; updating the collector."
+fi
+
+ARCHIVE="$(mktemp --suffix=.tar.gz)"
+trap 'rm -f "$ARCHIVE"' EXIT
+tar -czf "$ARCHIVE" \
+  --exclude=.git --exclude=.venv --exclude=data --exclude='*.egg-info' \
+  -C "$ROOT_DIR" .
+
+gcloud compute scp "$ARCHIVE" "$INSTANCE:/tmp/polymarket-collector.tar.gz" \
+  --zone "$ZONE" --project "$PROJECT_ID"
+gcloud compute ssh "$INSTANCE" --zone "$ZONE" --project "$PROJECT_ID" \
+  --command 'workdir=$(mktemp -d); tar -xzf /tmp/polymarket-collector.tar.gz -C "$workdir"; sudo bash "$workdir/scripts/install_vps.sh" "$workdir"'
+
+echo
+echo "Deployment complete."
+echo "Connect: gcloud compute ssh $INSTANCE --zone $ZONE --project $PROJECT_ID"
+echo "The collector uses public data only and has no Polymarket trading key."
+

scripts/deploy_hf.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import getpass
+import os
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+
+PROJECT_DIR = Path(__file__).resolve().parents[1]
+
+
+def main() -> None:
+    token = os.getenv("HF_TOKEN") or getpass.getpass("Hugging Face write token: ")
+    if not token.startswith("hf_"):
+        raise SystemExit("The token does not look like a Hugging Face token.")
+
+    api = HfApi(token=token)
+    account = api.whoami()
+    namespace = account["name"]
+    space_id = os.getenv(
+        "HF_SPACE_REPO", f"{namespace}/polymarket-lp-collector"
+    )
+    dataset_id = os.getenv(
+        "HF_DATASET_REPO", f"{namespace}/polymarket-l2-history"
+    )
+
+    api.create_repo(
+        repo_id=dataset_id,
+        repo_type="dataset",
+        private=False,
+        exist_ok=True,
+    )
+    api.upload_file(
+        path_or_fileobj=PROJECT_DIR / "deploy/hf-dataset-card.md",
+        path_in_repo="README.md",
+        repo_id=dataset_id,
+        repo_type="dataset",
+        commit_message="Add dataset card",
+    )
+
+    api.create_repo(
+        repo_id=space_id,
+        repo_type="space",
+        space_sdk="docker",
+        private=False,
+        exist_ok=True,
+    )
+    api.add_space_secret(
+        repo_id=space_id,
+        key="HF_TOKEN",
+        value=token,
+        description="Uploads collector archives to the public dataset.",
+    )
+    variables = {
+        "HF_DATASET_REPO": dataset_id,
+        "MAX_MARKETS": "40",
+        "DATA_DIR": "/data",
+        "HEALTH_HOST": "0.0.0.0",
+        "PORT": "8080",
+        "ROTATION_SECONDS": "3600",
+        "MAX_FILE_MIB": "256",
+        "BACKUP_INTERVAL_SECONDS": "3600",
+        "BACKUP_MIN_AGE_SECONDS": "120",
+        "REQUIRE_REMOTE_BACKUP": "true",
+    }
+    for key, value in variables.items():
+        api.add_space_variable(repo_id=space_id, key=key, value=value)
+
+    api.upload_folder(
+        folder_path=PROJECT_DIR,
+        repo_id=space_id,
+        repo_type="space",
+        ignore_patterns=[
+            ".git/**",
+            ".venv/**",
+            ".venv-hf/**",
+            ".tools/**",
+            "data/**",
+            "upload/**",
+            "**/__pycache__/**",
+            "*.zip",
+            ".env",
+            ".koyeb.yaml",
+        ],
+        commit_message="Deploy Polymarket LP collector",
+    )
+    api.request_space_hardware(repo_id=space_id, hardware="cpu-basic")
+    runtime = api.get_space_runtime(repo_id=space_id)
+
+    print(f"Space: https://huggingface.co/spaces/{space_id}")
+    print(f"Dataset: https://huggingface.co/datasets/{dataset_id}")
+    print(f"Hardware requested: {runtime.requested_hardware or runtime.hardware}")
+
+
+if __name__ == "__main__":
+    main()

scripts/deploy_koyeb.sh

@@ -0,0 +1,54 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+
+PROJECT_DIR="${1:-$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)}"
+RCLONE_CONFIG_FILE="${RCLONE_CONFIG_FILE:-$HOME/.config/rclone/rclone.conf}"
+APP_SERVICE="${KOYEB_APP_SERVICE:-polymarket-collector/collector}"
+SECRET_NAME="${KOYEB_RCLONE_SECRET:-polymarket-rclone-config}"
+
+if ! command -v koyeb >/dev/null 2>&1; then
+  echo "Koyeb CLI is missing: https://www.koyeb.com/docs/build-and-deploy/cli/installation" >&2
+  exit 1
+fi
+if [[ ! -f "$PROJECT_DIR/Dockerfile" ]]; then
+  echo "Dockerfile not found in $PROJECT_DIR" >&2
+  exit 1
+fi
+if [[ ! -s "$RCLONE_CONFIG_FILE" ]]; then
+  echo "rclone config not found: $RCLONE_CONFIG_FILE" >&2
+  exit 1
+fi
+
+if koyeb secrets describe "$SECRET_NAME" >/dev/null 2>&1; then
+  base64 -w0 "$RCLONE_CONFIG_FILE" \
+    | koyeb secrets update "$SECRET_NAME" --value-from-stdin >/dev/null
+else
+  base64 -w0 "$RCLONE_CONFIG_FILE" \
+    | koyeb secrets create "$SECRET_NAME" --value-from-stdin >/dev/null
+fi
+
+koyeb deploy "$PROJECT_DIR" "$APP_SERVICE" \
+  --archive-builder docker \
+  --instance-type free \
+  --regions fra \
+  --ports 8080:http \
+  --routes /:8080 \
+  --checks 8080:http:/healthz \
+  --checks-grace-period 8080=180 \
+  --scale 1 \
+  --env MAX_MARKETS=40 \
+  --env DATA_DIR=/data \
+  --env HEALTH_HOST=0.0.0.0 \
+  --env PORT=8080 \
+  --env ROTATION_SECONDS=300 \
+  --env MAX_FILE_MIB=32 \
+  --env BACKUP_INTERVAL_SECONDS=120 \
+  --env BACKUP_MIN_AGE=30s \
+  --env REMOVE_AFTER_UPLOAD=true \
+  --env REQUIRE_REMOTE_BACKUP=true \
+  --env RCLONE_REMOTE=gdrive:polymarket-data \
+  --env "RCLONE_CONFIG_B64={{secret.$SECRET_NAME}}" \
+  --wait \
+  --wait-timeout 10m
+
+koyeb services get "$APP_SERVICE"

scripts/install_vps.sh

@@ -0,0 +1,60 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+SOURCE_DIR="${1:-}"
+INSTALL_DIR=/opt/polymarket-collector
+DATA_DIR=/var/lib/polymarket-collector
+ENV_FILE=/etc/polymarket-collector.env
+
+if [[ -z "$SOURCE_DIR" || ! -f "$SOURCE_DIR/pyproject.toml" ]]; then
+  echo "Usage: sudo $0 /path/to/polymarket-collector" >&2
+  exit 2
+fi
+if [[ "${EUID}" -ne 0 ]]; then
+  echo "This installer must run as root." >&2
+  exit 2
+fi
+
+export DEBIAN_FRONTEND=noninteractive
+apt-get update
+apt-get install -y --no-install-recommends \
+  ca-certificates python3 python3-pip python3-venv rclone
+
+if ! id polymarket >/dev/null 2>&1; then
+  useradd --system --home-dir "$DATA_DIR" --create-home \
+    --shell /usr/sbin/nologin polymarket
+fi
+
+mkdir -p "$INSTALL_DIR" "$DATA_DIR"
+cp -a "$SOURCE_DIR/." "$INSTALL_DIR/"
+python3 -m venv "$INSTALL_DIR/.venv"
+"$INSTALL_DIR/.venv/bin/pip" install --no-cache-dir --upgrade pip
+"$INSTALL_DIR/.venv/bin/pip" install --no-cache-dir "$INSTALL_DIR"
+
+if [[ ! -f "$ENV_FILE" ]]; then
+  cp "$INSTALL_DIR/.env.example" "$ENV_FILE"
+fi
+chown root:polymarket "$ENV_FILE"
+chmod 0640 "$ENV_FILE"
+chown -R polymarket:polymarket "$DATA_DIR"
+
+install -m 0644 "$INSTALL_DIR/deploy/polymarket-collector.service" \
+  /etc/systemd/system/polymarket-collector.service
+install -m 0644 "$INSTALL_DIR/deploy/polymarket-backup.service" \
+  /etc/systemd/system/polymarket-backup.service
+install -m 0644 "$INSTALL_DIR/deploy/polymarket-backup.timer" \
+  /etc/systemd/system/polymarket-backup.timer
+install -m 0755 "$INSTALL_DIR/scripts/polymarket-backup" \
+  /usr/local/sbin/polymarket-backup
+
+systemctl daemon-reload
+systemctl enable --now polymarket-collector.service
+
+echo
+echo "Collector installed."
+echo "Status:  systemctl status polymarket-collector"
+echo "Logs:    journalctl -u polymarket-collector -f"
+echo "Health:  curl http://127.0.0.1:8080/healthz"
+echo "Drive:   run 'sudo rclone config', set RCLONE_REMOTE in $ENV_FILE,"
+echo "         then enable polymarket-backup.timer."
+

scripts/polymarket-backup

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ENV_FILE="${ENV_FILE:-/etc/polymarket-collector.env}"
+if [[ -f "$ENV_FILE" ]]; then
+  set -a
+  # shellcheck disable=SC1090
+  source "$ENV_FILE"
+  set +a
+fi
+
+DATA_DIR="${DATA_DIR:-/var/lib/polymarket-collector}"
+RCLONE_REMOTE="${RCLONE_REMOTE:-}"
+LOCAL_RETENTION_DAYS="${LOCAL_RETENTION_DAYS:-0}"
+REMOVE_AFTER_UPLOAD="${REMOVE_AFTER_UPLOAD:-false}"
+BACKUP_MIN_AGE="${BACKUP_MIN_AGE:-2m}"
+
+if [[ -z "$RCLONE_REMOTE" ]]; then
+  echo "RCLONE_REMOTE is empty; backup is not configured."
+  exit 0
+fi
+if [[ ! -d "$DATA_DIR" ]]; then
+  echo "Data directory does not exist: $DATA_DIR" >&2
+  exit 1
+fi
+if ! [[ "$LOCAL_RETENTION_DAYS" =~ ^[0-9]+$ ]]; then
+    echo "LOCAL_RETENTION_DAYS must be a non-negative integer." >&2
+    exit 1
+fi
+
+case "${REMOVE_AFTER_UPLOAD,,}" in
+  1|true|yes) RCLONE_ACTION="move" ;;
+  0|false|no) RCLONE_ACTION="copy" ;;
+  *)
+    echo "REMOVE_AFTER_UPLOAD must be true or false." >&2
+    exit 1
+    ;;
+esac
+
+LOCK_FILE="${BACKUP_LOCK_FILE:-/tmp/polymarket-backup.lock}"
+exec 9>"$LOCK_FILE"
+flock -n 9 || exit 0
+
+rclone "$RCLONE_ACTION" "$DATA_DIR" "$RCLONE_REMOTE" \
+  --include "*.jsonl.gz" \
+  --min-age "$BACKUP_MIN_AGE" \
+  --checksum \
+  --transfers 2 \
+  --checkers 4 \
+  --retries 3 \
+  --low-level-retries 10 \
+  --create-empty-src-dirs
+
+if (( LOCAL_RETENTION_DAYS > 0 )); then
+  find "$DATA_DIR" -type f -name "*.jsonl.gz" \
+    -mtime "+$LOCAL_RETENTION_DAYS" -delete
+fi

scripts/publish_hf_source.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import getpass
+import os
+from pathlib import Path
+
+from huggingface_hub import HfApi
+
+
+PROJECT_DIR = Path(__file__).resolve().parents[1]
+
+
+def main() -> None:
+    token = os.getenv("HF_TOKEN") or getpass.getpass("Hugging Face write token: ")
+    if not token.startswith("hf_"):
+        raise SystemExit("The token does not look like a Hugging Face token.")
+
+    api = HfApi(token=token)
+    namespace = api.whoami()["name"]
+    repo_id = f"{namespace}/polymarket-lp-collector-source"
+    api.create_repo(repo_id=repo_id, private=False, exist_ok=True)
+    api.upload_folder(
+        folder_path=PROJECT_DIR,
+        repo_id=repo_id,
+        ignore_patterns=[
+            ".git/**",
+            ".venv/**",
+            ".venv-hf/**",
+            ".tools/**",
+            "data/**",
+            "upload/**",
+            "**/__pycache__/**",
+            "*.zip",
+            ".env",
+            ".koyeb.yaml",
+        ],
+        commit_message="Publish Render-ready Polymarket collector",
+    )
+    print(f"Source: https://huggingface.co/{repo_id}")
+    print(f"Git: https://huggingface.co/{repo_id}")
+    print(f"Dataset: https://huggingface.co/datasets/{namespace}/polymarket-l2-history")
+
+
+if __name__ == "__main__":
+    main()

tests/test_config.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from polymarket_collector.config import Config
+
+
+class ConfigTests(unittest.TestCase):
+    def test_loads_dotenv_and_respects_process_environment(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            env_file = Path(temporary) / ".env"
+            env_file.write_text(
+                "MAX_MARKETS=12\nHEALTH_PORT=9090\nMIN_DAILY_REWARD=2.5\n",
+                encoding="utf-8",
+            )
+            clean = {
+                key: value
+                for key, value in os.environ.items()
+                if key
+                not in {"MAX_MARKETS", "HEALTH_PORT", "MIN_DAILY_REWARD"}
+            }
+            clean["ENV_FILE"] = str(env_file)
+            clean["MAX_MARKETS"] = "15"
+            with patch.dict(os.environ, clean, clear=True):
+                config = Config.from_env()
+            self.assertEqual(config.max_markets, 15)
+            self.assertEqual(config.health_port, 9090)
+            self.assertEqual(config.min_daily_reward, 2.5)
+
+    def test_port_is_used_when_health_port_is_absent(self) -> None:
+        clean = {
+            key: value
+            for key, value in os.environ.items()
+            if key not in {"HEALTH_PORT", "PORT", "ENV_FILE"}
+        }
+        clean["ENV_FILE"] = "/does/not/exist"
+        clean["PORT"] = "8765"
+        with patch.dict(os.environ, clean, clear=True):
+            config = Config.from_env()
+        self.assertEqual(config.health_port, 8765)
+
+    def test_rejects_invalid_limits(self) -> None:
+        with patch.dict(
+            os.environ,
+            {"ENV_FILE": "/does/not/exist", "MAX_MARKETS": "0"},
+            clear=True,
+        ):
+            with self.assertRaises(ValueError):
+                Config.from_env()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_discovery.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import unittest
+
+from polymarket_collector.discovery import daily_reward, select_markets
+
+
+def market(
+    identifier: str,
+    reward: float,
+    volume: float,
+    token_suffix: str,
+) -> dict:
+    return {
+        "condition_id": identifier,
+        "volume_24hr": volume,
+        "rewards_config": [{"rate_per_day": reward}],
+        "tokens": [
+            {"token_id": f"yes-{token_suffix}", "outcome": "YES"},
+            {"token_id": f"no-{token_suffix}", "outcome": "NO"},
+        ],
+    }
+
+
+class DiscoveryTests(unittest.TestCase):
+    def test_daily_reward_sums_configs(self) -> None:
+        self.assertEqual(
+            daily_reward(
+                {"rewards_config": [{"rate_per_day": 2}, {"rate_per_day": "1.5"}]}
+            ),
+            3.5,
+        )
+
+    def test_selects_by_reward_then_volume(self) -> None:
+        candidates = [
+            market("a", 2, 1_000, "a"),
+            market("b", 5, 10, "b"),
+            market("c", 2, 2_000, "c"),
+            market("d", 0.5, 10_000, "d"),
+        ]
+        selection = select_markets(
+            candidates,
+            max_markets=2,
+            min_daily_reward=1,
+            min_volume_24h=0,
+        )
+        self.assertEqual(
+            [item["condition_id"] for item in selection.markets], ["b", "c"]
+        )
+        self.assertEqual(
+            selection.asset_ids, ("yes-b", "no-b", "yes-c", "no-c")
+        )
+
+    def test_skips_market_without_two_tokens(self) -> None:
+        invalid = market("a", 10, 1_000, "a")
+        invalid["tokens"] = invalid["tokens"][:1]
+        selection = select_markets(
+            [invalid],
+            max_markets=10,
+            min_daily_reward=0,
+            min_volume_24h=0,
+        )
+        self.assertEqual(selection.markets, ())
+
+
+if __name__ == "__main__":
+    unittest.main()
+

tests/test_health.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+import unittest
+
+from polymarket_collector.health import HealthServer
+from polymarket_collector.state import CollectorState
+
+
+class HealthTests(unittest.IsolatedAsyncioTestCase):
+    async def test_health_endpoint(self) -> None:
+        state = CollectorState(
+            discovered_at=time.time(),
+            connected_at=time.time(),
+            last_message_at=time.time(),
+            connected=True,
+            markets=4,
+            assets=8,
+        )
+        server = HealthServer(state, "127.0.0.1", 0, stale_after_seconds=60)
+        await server.start()
+        assert server._server is not None
+        port = server._server.sockets[0].getsockname()[1]
+
+        reader, writer = await asyncio.open_connection("127.0.0.1", port)
+        writer.write(b"GET /healthz HTTP/1.1\r\nHost: localhost\r\n\r\n")
+        await writer.drain()
+        response = await reader.read()
+        writer.close()
+        await writer.wait_closed()
+        await server.close()
+
+        _, body = response.split(b"\r\n\r\n", 1)
+        payload = json.loads(body)
+        self.assertTrue(payload["healthy"])
+        self.assertEqual(payload["markets"], 4)
+
+
+if __name__ == "__main__":
+    unittest.main()
+

tests/test_hf_backup.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from polymarket_collector.hf_backup import completed_archives, upload_once
+
+
+class FakeApi:
+    def __init__(self, fail: bool = False) -> None:
+        self.fail = fail
+        self.calls: list[dict[str, object]] = []
+
+    def create_commit(self, **kwargs: object) -> None:
+        self.calls.append(kwargs)
+        if self.fail:
+            raise RuntimeError("upload failed")
+
+
+class HuggingFaceBackupTests(unittest.TestCase):
+    def test_lists_only_completed_archives(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            archive = root / "a.jsonl.gz"
+            archive.write_bytes(b"archive")
+            os.utime(archive, (1, 1))
+            (root / "active.jsonl.part").write_bytes(b"active")
+            self.assertEqual(
+                completed_archives(root, min_age_seconds=0), [archive]
+            )
+
+    def test_deletes_only_after_successful_commit(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            archive = root / "market_ws" / "a.jsonl.gz"
+            archive.parent.mkdir()
+            archive.write_bytes(b"archive")
+            os.utime(archive, (1, 1))
+
+            api = FakeApi()
+            count = upload_once(
+                data_dir=root,
+                repo_id="user/data",
+                token="token",
+                min_age_seconds=0,
+                api=api,  # type: ignore[arg-type]
+            )
+            self.assertEqual(count, 1)
+            self.assertFalse(archive.exists())
+            operations = api.calls[0]["operations"]
+            self.assertEqual(operations[0].path_in_repo, "market_ws/a.jsonl.gz")
+
+    def test_keeps_files_when_commit_fails(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            archive = root / "a.jsonl.gz"
+            archive.write_bytes(b"archive")
+            os.utime(archive, (1, 1))
+            with self.assertRaises(RuntimeError):
+                upload_once(
+                    data_dir=root,
+                    repo_id="user/data",
+                    token="token",
+                    min_age_seconds=0,
+                    api=FakeApi(fail=True),  # type: ignore[arg-type]
+                )
+            self.assertTrue(archive.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_storage.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import gzip
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from polymarket_collector.storage import RotatingJsonlWriter
+
+
+class StorageTests(unittest.IsolatedAsyncioTestCase):
+    async def test_rotates_and_compresses_jsonl(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            writer = RotatingJsonlWriter(
+                root,
+                "test",
+                rotation_seconds=3600,
+                max_file_bytes=1,
+                flush_interval_seconds=1,
+            )
+            await writer.start()
+            await writer.write({"hello": "world"})
+            await writer.close()
+
+            files = list(root.rglob("*.jsonl.gz"))
+            self.assertEqual(len(files), 1)
+            with gzip.open(files[0], "rt", encoding="utf-8") as handle:
+                self.assertEqual(json.loads(handle.readline()), {"hello": "world"})
+            self.assertEqual(list(root.rglob("*.part")), [])
+
+    async def test_recovers_part_file_on_start(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            part = root / "2026/07/22/crashed.jsonl.part"
+            part.parent.mkdir(parents=True)
+            part.write_text('{"recovered":true}\n', encoding="utf-8")
+
+            writer = RotatingJsonlWriter(
+                root,
+                "test",
+                rotation_seconds=3600,
+                max_file_bytes=1024,
+                flush_interval_seconds=1,
+            )
+            await writer.start()
+            await writer.close()
+
+            recovered = list(root.rglob("*.jsonl.gz"))
+            self.assertEqual(len(recovered), 1)
+            with gzip.open(recovered[0], "rt", encoding="utf-8") as handle:
+                self.assertEqual(json.loads(handle.readline()), {"recovered": True})
+
+
+if __name__ == "__main__":
+    unittest.main()
+

tests/test_stream.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import asyncio
+import gzip
+import json
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+from websockets.asyncio.server import serve
+
+from polymarket_collector.state import CollectorState
+from polymarket_collector.storage import RotatingJsonlWriter
+from polymarket_collector.stream import MarketStream
+
+
+class StreamTests(unittest.IsolatedAsyncioTestCase):
+    async def test_records_public_market_message(self) -> None:
+        subscription_received = asyncio.Event()
+
+        async def handler(websocket) -> None:
+            subscription = json.loads(await websocket.recv())
+            self.assertEqual(subscription["assets_ids"], ["asset-1", "asset-2"])
+            subscription_received.set()
+            await websocket.send(
+                json.dumps(
+                    {
+                        "event_type": "book",
+                        "asset_id": "asset-1",
+                        "market": "condition-1",
+                        "bids": [],
+                        "asks": [],
+                        "timestamp": "1",
+                    }
+                )
+            )
+            while True:
+                try:
+                    message = await websocket.recv()
+                except Exception:
+                    return
+                if message == "PING":
+                    await websocket.send("PONG")
+
+        with tempfile.TemporaryDirectory() as temporary:
+            writer = RotatingJsonlWriter(
+                Path(temporary),
+                "stream",
+                rotation_seconds=3600,
+                max_file_bytes=1024 * 1024,
+                flush_interval_seconds=1,
+            )
+            await writer.start()
+            state = CollectorState()
+            stop = asyncio.Event()
+            async with serve(handler, "127.0.0.1", 0) as websocket_server:
+                port = websocket_server.sockets[0].getsockname()[1]
+                stream = MarketStream(
+                    f"ws://127.0.0.1:{port}", writer, state
+                )
+                stream.set_assets(["asset-1", "asset-2"])
+                task = asyncio.create_task(stream.run(stop))
+                await asyncio.wait_for(subscription_received.wait(), 2)
+                deadline = time.monotonic() + 2
+                while state.messages_total < 1 and time.monotonic() < deadline:
+                    await asyncio.sleep(0.01)
+                stop.set()
+                await asyncio.wait_for(task, 3)
+            await writer.close()
+
+            self.assertEqual(state.messages_total, 1)
+            archive = next(Path(temporary).rglob("*.jsonl.gz"))
+            with gzip.open(archive, "rt", encoding="utf-8") as handle:
+                record = json.loads(handle.readline())
+            self.assertEqual(record["payload"]["event_type"], "book")
+            self.assertIn("received_at_ns", record)
+
+
+if __name__ == "__main__":
+    unittest.main()
+