03 สตรีมข้อมูลสด

รัน collector ที่ทนทาน และตรวจสอบ WebSocket การกระทบยอดผ่าน REST และการเขียนลง Cloud

จุดเริ่มต้น

ออบเจ็กต์ polymarket ทั้งหกมีอยู่แล้ว และ .env.polymarket ถูก source แล้ว

สิ่งที่จะเริ่มทำงาน

คอนเทนเนอร์ Python ไร้สถานะหนึ่งตัว:

ค้นหาตลาดที่ยังเปิดใช้งานห้าตลาดผ่าน Gamma
subscribe token ของทั้งสองผลลัพธ์บน CLOB WebSocket สาธารณะ
กระทบยอดการเทรดสาธารณะทุก 10 วินาที
poll CLOB book เมื่อ WebSocket ค้าง และ
เขียนข้อมูลด้วย async insert ที่ได้รับการยืนยันไปยัง ClickHouse Cloud

ไม่มีฐานข้อมูลบนเครื่อง broker เซิร์ฟเวอร์แดชบอร์ด หรือข้อมูลรับรองของ Polymarket

ขั้นที่ 1 — Build และเริ่ม collector

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

สถานะจะกลายเป็น healthy หลังการค้นหาตลาดและการเขียนลง Cloud สำเร็จครั้งแรก สถานะ degraded ของแอปพลิเคชันยังนับว่า Docker-healthy เมื่อ REST ยังทันเวลาและ WebSocket กำลังเชื่อมต่อใหม่

ขั้นที่ 2 — อ่านสัญญาของ health

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
}

status: degraded พร้อม reason: websocket_stale_rest_active ยอมรับได้หาก 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
  "

markets, quote_midpoints และ one_minute_states ต้องมากกว่าศูนย์ก่อนเข้า โมดูล 04 ส่วน trades ปกติจะเพิ่มขึ้นภายในหนึ่งนาที ตลาดที่เงียบอาจทำให้ช้ากว่านั้น

ขั้นที่ 5 — ใช้โหมด fixture แบบ deterministic เฉพาะเมื่อจำเป็น

หากเครือข่ายในห้องบล็อก 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

รันการตรวจสอบ health และจำนวนแถวอีกครั้ง สถานะที่ควรได้: fixture จำนวน tick และ trade เพิ่มขึ้นทุกห้าวินาที เก็บไฟล์สำรองไว้จนจบโมดูล

ถือว่าเสร็จเมื่อ

health เป็น live, degraded โดยมี timestamp ของ REST ที่สดใหม่ หรือ fixture
watched_markets เป็น 5 และ
จำนวนแถวใน Cloud ทั้งสี่ค่าคืนออกมา โดยแถวของ quote และรายหนึ่งนาทีมากกว่าศูนย์

ต่อไป: คิวรี aggregate แบบเพิ่มทีละส่วน