03 ライブデータをストリーミングする

復元力のある collector を実行し、WebSocket、REST での突合、Cloud への書き込みを検証します。

開始条件

polymarket の 6 つのオブジェクトが存在し、.env.polymarket を source 済みです。

何が起動するか

ステートレスな Python コンテナが 1 つ:

Gamma を通じてアクティブな市場を 5 件見つける;
公開 CLOB WebSocket で両アウトカムのトークンを購読する;
10 秒ごとに公開取引を突合する;
WebSocket が停止しているときは CLOB の板をポーリングする; および
確認応答付きの非同期挿入を ClickHouse Cloud に書き込む。

ローカルのデータベース、ブローカー、ダッシュボードサーバー、Polymarket の資格情報はいずれもありません。

ステップ 1 — collector をビルドして起動する

docker compose --env-file .env.polymarket up -d --build collector
docker compose --env-file .env.polymarket ps

市場の探索と最初の Cloud への書き込みが成功すると、ステータスが healthy になります。アプリケーションステータスが degraded でも、REST が最新で WebSocket が再接続中であれば Docker 上は healthy です。

ステップ 2 — ヘルスの契約を読む

curl --fail --silent http://localhost:8090/health \
  | python3 -m json.tool

期待されるフィールド:

{
  "status": "live",
  "websocket": "connected",
  "queue_depth": 0,
  "queue_capacity": 10000,
  "watched_markets": 5,
  "watched_tokens": 10,
  "fresh_tokens": 10
}

reason: websocket_stale_rest_active を伴う status: degraded は、last_trade_reconcile_at と last_book_fallback_at が進み続けているなら許容されます。unhealthy は許容されません。トラブルシューティングを参照してください。

ステップ 3 — ソースと書き込みのイベントを観察する

docker compose --env-file .env.polymarket logs --tail=30 collector

ログは JSON です。collector_ready を探してください。ソースまたは ClickHouse の障害には、長さを制限したエラーのプレビューとリトライ待ち時間が含まれます。パスワードはログに出力されません。

ステップ 4 — Cloud に行があることを確認する

clickhouse client \
  --host "$CLICKHOUSE_HOST" \
  --port "$CLICKHOUSE_PORT" \
  --user "$CLICKHOUSE_USER" \
  --password "$CLICKHOUSE_PASSWORD" \
  --secure \
  --query "
    SELECT 'markets' AS table, count() AS rows FROM polymarket.markets
    UNION ALL
    SELECT 'quote_midpoints', countIf(midpoint > 0) FROM polymarket.price_ticks
    UNION ALL
    SELECT 'trades', count() FROM polymarket.trades_clean
    UNION ALL
    SELECT 'one_minute_states', count() FROM polymarket.market_midpoints_1m
  "

モジュール 04 に進む前に、markets、quote_midpoints、one_minute_states が 0 より大きくなっている必要があります。trades は通常 1 分以内に増えますが、動きの少ない市場では遅れることがあります。

ステップ 5 — 決定的な fixture モードは必要なときだけ使う

会場のネットワークが Polymarket をブロックしている場合、または選ばれた市場が 60 秒経っても動かない場合:

sed -i.bak 's/^POLYMARKET_MODE=.*/POLYMARKET_MODE=fixture/' .env.polymarket
set -a; source ./.env.polymarket; set +a
docker compose --env-file .env.polymarket up -d --build --force-recreate collector

ヘルスと行数のチェックをもう一度実行します。期待されるステータス: fixture。tick と trade の件数が 5 秒ごとに増えます。バックアップファイルはモジュールが終わるまで残しておいてください。

完了条件

ヘルスが live、REST のタイムスタンプが新しい degraded、または fixture である;
watched_markets が 5 である; および
Cloud の 4 つの行数がすべて返り、クォートと 1 分足の行が 0 より大きい。

次: インクリメンタルな集計をクエリする。