ssmd-dq-checks

Reference catalog for all 13 DQ checks in data/dq.py.

Check Catalog

Check 1: data_completeness

What: Verifies parquet files exist and the archiver manifest is present.

Common failures: Archiver not running, GCS permissions, wrong date.

Check 2: record_counts

What: Reports row counts per message type from parquet files. Uses archiver manifest records_by_type or falls back to parquet-gen manifest records_written.

Note: Polymarket archiver manifest has files: [] — the check falls back to pq_manifest for type breakdown.

Check 3: schema

What: Validates parquet column names match expected schema per feed/type.

Check 4: null_rates

What: Checks null percentage for required (non-nullable) columns.

Check 5: duplicates

What: Checks for duplicate rows by primary key (feed-specific).

Primary keys: Kalshi trade: trade_id. Kalshi ticker: (market_ticker, ts, _nats_seq).

Check 6: nats_continuity

What: Checks for gaps in _nats_seq (NATS JetStream sequence numbers).

Common failures: NATS consumer restart, archiver reconnection.

Check 7: ts_continuity

What: Checks for time gaps in exchange timestamps. Divides the day into 15-minute slots and checks for empty slots.

Check 8: per_ticker_stats

What: Per-ticker row counts and time coverage. Reports top/bottom tickers by volume.

Check 9: parquet_vs_jsonl (accountability)

What: Verifies every JSONL line is accounted for in parquet. The zero-tolerance accountability check.

Sources (in priority order):

1. parquet-gen manifest (parquet-manifest.json) — preferred, scoped to what was actually processed
2. Archiver manifest (manifest.json) — fallback for pre-v2.0.0 data

Comparison logic:

• Data types: parquet_rows == jsonl_lines (exact match), or parquet_rows >= jsonl_lines for fanout types
• Control types: counted and reported, NOT converted to parquet
• Unknown types: any unrecognized type = FAIL (needs new schema)

Pipeline stats surfaced: lines_json_error, lines_type_unknown, lines_no_schema, parse_batch_dropped

u64 underflow guard: Values > 2^63 in parse_batch_dropped are treated as 0 (known Rust underflow from fanout).

Check 10: exchange_seq_gaps

What: Gaps in exchange-provided sequence numbers per file.

Grouping:

• Kalshi: group by _shard_id (v1.3.0+), falls back to sid (v1.2.0), then ticker_col
• Kraken Futures: group by product_id

Common failures: Kalshi pre-1.3.0 data groups by sid which collides across shards — expect false gaps on old data. Post-1.3.0 groups by _shard_id for accurate per-shard analysis.

Check 11: data_coverage

What: Percentage of 15-minute time slots (96/day) that have exchange-timestamped data.

Feed-specific timestamp columns: Kalshi ts, Kraken time, Polymarket timestamp_ms.

Note: Polymarket coverage may be lower due to activity concentrated in US hours.

Check 12: connection_uptime

What: WebSocket connection uptime from Cloud Monitoring websocket_connected gauge.

Known issue: Ghost shards from before connector restarts have low data point counts, dragging down the min. The check should consider only the latest shard generation.

Requires: google-cloud-monitoring package, Monitoring Viewer IAM role on the SA.

Check 13: schema_version

What: Verifies parquet files report their schema version via metadata.

Fanout Rules

Some message types produce multiple parquet rows per JSONL line (1:N fan-out):

For fanout types: parquet_rows >= jsonl_lines (not exact match). For non-fanout types: parquet_rows == jsonl_lines (exact match).

Per-Feed Expected Message Types

Trade Message Schemas

Kalshi NATS message (post-connector injection)

{
  "type": "trade",
  "sid": 2,
  "seq": 5724,
  "msg": {
    "trade_id": "uuid",
    "market_ticker": "KXBTCD-26FEB0317-T76999.99",
    "yes_price": 17,
    "count": 130,
    "taker_side": "no",
    "ts": 1770153448
  },
  "_shard_id": 3
}

Kalshi REST API trade

{
  "trade_id": "uuid",
  "ticker": "KXBTCD-26FEB0317-T76999.99",
  "yes_price": 17,
  "count": 130,
  "taker_side": "no",
  "created_time": "2026-02-03T21:17:28.18002Z"
}

Adding a New Check

1. Add def check_<name>(con, base, ...) -> dict to data/dq.py
2. Return dict with at minimum: {"check": "<name>", "status": "pass|warn|fail|skip", "detail": "..."}
3. Add to check_fns list in DQRunner.run()
4. Update this skill with the check's criteria and common failures
5. Deploy: tag dq-v*, update dq-daily.yaml manifest (see ssmd-deploy skill)

GCS Data Layout

gs://ssmd-data/{prefix}/{feed}/{stream}/{date}/
  manifest.json           # archiver manifest (records_by_type, files list)
  parquet-manifest.json   # parquet-gen manifest (parse_batch_input, records_written)
  ticker_000000.parquet   # parquet files per type
  trade_000000.parquet
  ...