WebSocket stream

Record a wss venue’s frames first, then build a table from the recorded text relation.

The question

What price did the venue publish for each product, and when did each message arrive?

The source

Venue	Coinbase Exchange market-data feed, wss://ws-feed.exchange.coinbase.com
Members	BTC-USD, ETH-USD
Format	JSON frames, written by a recorder into a fixed text relation
Cadence	continuous while a subscription is open
Rights	read under the venue’s published market-data terms for its open feed, recorded as terms_of_service_asserted with the terms page digest as evidence

The feed is open, so it needs no credential. Its ticker messages carry a product identifier and a price, the smallest useful shape for this example.

Record before you build

A build never reaches the venue. A run reads the batches a recorder wrote before the run began. Four steps come in front of the ordinary build loop:

mr-data stream document register --file connector.json --name coinbase --json
mr-data stream registry create --digest DOCUMENT_DIGEST --json
mr-data stream probe --document-digest DOCUMENT_DIGEST --registry REGISTRY_ID \
  --member BTC-USD --json
mr-data stream capture --document-digest DOCUMENT_DIGEST --registry REGISTRY_ID \
  --member BTC-USD --member ETH-USD --seconds 300 --watch --json

1. Register the connector document. It names the venue, the message type, the event time and the subscription frame. Registration prints the fingerprint the recipe names. mr-data stream document show ID reads one back.
2. Admit it into a registry, the reviewed set a recording may name. Pass every fingerprint the recording will use.
3. A probe connects briefly, reports which message types the document classified and which it did not, and records nothing. Read it with mr-data stream status ID --kind probe, and fix a document that classifies nothing before spending a capture window on it.
4. Capture a bounded window with capture --seconds N, where --seconds is at most 600, and --max-events and --max-bytes bound it further. subscribe instead keeps recording until mr-data stream subscription stop ID, checkpointing as it goes.

mr-data stream recordings --document-digest D reports every batch a build could read for that venue, the bookmark each one carries, and when each expires.

The recipe

recipe.json

{
  "dataset": {
    "name": "Coinbase Exchange ticker frames, BTC-USD and ETH-USD",
    "description": "Each row is one ticker message the Coinbase Exchange market-data feed pushed while a recorder was subscribed, with the venue's own price beside the moment it arrived."
  },
  "question": {
    "text": "What price did the venue publish for each product, and when did the message arrive?"
  },
  "table": {
    "name": "ticker_frames",
    "description": "One row per recorded ticker message.",
    "grain": [
      "event_id"
    ],
    "columns": [
      {
        "name": "event_id",
        "type": "string",
        "nullable": false,
        "description": "The identity the recorder assigned this frame.",
        "presentation": {
          "chart": "none"
        }
      },
      {
        "name": "received_at",
        "type": "timestamp",
        "nullable": false,
        "description": "When the recorder received the frame, in UTC.",
        "presentation": {
          "chart": "timeline",
          "bucket": "hour"
        }
      },
      {
        "name": "product_id",
        "type": "string",
        "nullable": true,
        "description": "The market the message is about, as the venue names it.",
        "presentation": {
          "chart": "top_values"
        }
      },
      {
        "name": "price",
        "type": "decimal",
        "nullable": true,
        "description": "The last trade price the venue published in this message.",
        "presentation": {
          "chart": "histogram",
          "bins": 10
        }
      }
    ]
  },
  "sources": [
    {
      "name": "coinbase_ticker",
      "description": "The public Coinbase Exchange market-data feed, recorded frame by frame.",
      "source_class": "stream",
      "data_classification": "public",
      "locator": {
        "kind": "stream",
        "display_locator": "wss://ws-feed.exchange.coinbase.com"
      },
      "rights_claim": {
        "claimed_basis": "terms_of_service_asserted",
        "claim_evidence_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
        "claim_note": "Read under the venue's published market-data terms for its open feed; the digest is of those terms as read on 12 September 2026."
      },
      "connector": {
        "adapter_id": "stream.wss@1.0.0",
        "credential_mode": "none",
        "stream_connector_document_digest": "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
        "parameters": [
          {
            "name": "stream.members",
            "value": "BTC-USD,ETH-USD"
          }
        ]
      },
      "limits": {
        "max_source_bytes": 268435456,
        "max_rows": 5000000
      }
    }
  ],
  "transform": {
    "engine": "duckdb_sql",
    "steps": [
      {
        "step_id": "ticker_messages",
        "sql": "select event_id, received_at, cast(json_extract(message, '$.product_id') as varchar) as product_id, cast(json_extract(message, '$.price') as varchar) as price from coinbase_ticker where message_type = 'ticker'",
        "description": "Keep the ticker frames and pull the two members this table needs out of the provider frame."
      },
      {
        "step_id": "ticker_frames",
        "sql": "select event_id, cast(received_at as timestamptz) as received_at, replace(product_id, '\"', '') as product_id, try_cast(replace(price, '\"', '') as decimal(24, 8)) as price from ticker_messages",
        "description": "Cast the recorder's text. json_extract returns the JSON value, so a string member still carries its quotes; strip them before the numeric cast."
      }
    ]
  },
  "checks": [
    {
      "check_id": "one_row_per_frame",
      "kind": "key_uniqueness",
      "enforcement": "required",
      "columns": [
        "event_id"
      ]
    },
    {
      "check_id": "price_present",
      "kind": "null_ceiling",
      "enforcement": "required",
      "columns": [
        "price"
      ],
      "max_null_ppm": 10000
    },
    {
      "check_id": "no_long_silence",
      "kind": "continuity",
      "enforcement": "advisory",
      "columns": [
        "received_at"
      ],
      "max_gap_seconds": 300,
      "description": "A gap longer than five minutes is a window nobody was listening for, not a quiet market."
    },
    {
      "check_id": "frames_present",
      "kind": "row_expectation",
      "enforcement": "required",
      "min_rows": 1
    }
  ],
  "units": [],
  "timezone": "UTC"
}

Part by part

The source

source_class is stream and locator.kind is stream, whose display_locator is the wss:// endpoint. credential_mode is none for an open venue. A keyed venue uses stream.wss@2.0.0 and a shaped credential enrolled with mr-data keys set, with the same document shape otherwise.

connector.stream_connector_document_digest is the fingerprint step 1 printed. It binds this source to the connector document the capture ran under.

stream.members names the members this source reads, comma separated. A source declares stream.members or stream.series, the rotating roster a subscription follows, never both. A source that names neither reads whatever recording binds to its connector document.

The source declares no window and no closed. A bookmark continues a stream.

The fixed relation

A stream source reaches the SQL as the recorder’s line, every column text, in this order. A connector that declares a member field adds a ninth column, member, after event_time. This recipe declares none, so the relation has these eight:

Column	Content
epoch	the connection this frame arrived on
event_id	the identity the recorder assigned the frame
event_time	the declared event time in the kind the connector declares, or the RFC 3339 arrival time when the recorder timed that frame by receipt
message	the provider frame, canonical JSON text
message_class	snapshot, delta, or whatever else the connector classifies
message_type	the venue’s own message type
received_at	RFC 3339 UTC receipt time
sequence	the frame’s position within its connection

Transform

Read the frame with the engine’s JSON functions:

select event_id, received_at,
       cast(json_extract(message, '$.product_id') as varchar) as product_id,
       cast(json_extract(message, '$.price') as varchar) as price
from coinbase_ticker
where message_type = 'ticker'

json_extract returns the JSON value, so a string member still carries its quotes. The second step strips them before the numeric cast:

select event_id,
       cast(received_at as timestamptz) as received_at,
       replace(product_id, '"', '') as product_id,
       try_cast(replace(price, '"', '') as decimal(24, 8)) as price
from ticker_messages

Run a sample and read mr-data peek before trusting that cast. The venue decides what goes in a member, and a try_cast returning null for every row looks like a quiet market.

event_time holds two kinds of value, row by row

A connector whose event_time_field is null times every frame by receipt. One whose event_time_fallback is received_at times only the frames the venue did not stamp. In one relation a row stamped by the venue carries the declared kind, say epoch milliseconds. The row beside it carries an RFC 3339 arrival time. A transform that casts the column blind gets nulls for half of it and no error.

Read received_at when the question wants arrival time. Guard the cast on the shape when the question wants the venue’s own time:

select case
         when event_time like '%-%' then cast(event_time as timestamp)
         else to_timestamp(cast(event_time as bigint) / 1000)
       end as stamped_at
from coinbase_ticker

A connector that declares an event_time_field and leaves event_time_fallback at refuse, the default, never mixes the two. Every frame carries the declared kind, or the capture refused it.

Checks

key_uniqueness on event_id is the grain. The continuity check on received_at with max_gap_seconds: 300 is advisory. A gap longer than five minutes is a window nobody was listening for rather than a quiet market, and the run records it without ending.

Units

This document declares none. A currency is not in the unit vocabulary, and the vocabulary refuses a code outside its subset, naming the construct that put it there. A column with no physical unit stays out of units.

Register, run, inspect

mr-data dataset create --name "Coinbase Exchange ticker frames" --json
mr-data recipe recipe.json --json

mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --sample --max-rows 2000 --json
mr-data peek RUN_ID --json
mr-data query RUN_ID "select product_id, count(*) from run_table group by product_id" --json
mr-data checks RUN_ID --json

mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --json
mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --refresh --json

run has no duration flag and will not grow one. A build reads recorded batches and stops at their end. The recorder fixed those bytes at capture time, not at build time.

What the run reports

The receipt names the recordings the run read and the bookmark it reached. Each run records that bookmark, and the next run continues from it.

A refresh of a recorded stream rewrites no part. It keeps every part of the previous version and adds this run’s. The table declares no materialization for a stream source, and the engine reads none. Such a refresh reads only the rows that arrived after the position the previous version committed at.

Traps

A missed window is a permanent hole. Recordings last thirty days, and a refresh continues from the bookmark the last build reached. Neither can recover messages that were never recorded. Report an unrecorded period as a gap in the series rather than promising a backfill, because there is nothing to fetch. Keep the subscription running where continuity matters.

A refresh of a stream table needs a backend that writes a v2 manifest for a version composed at a recorder’s checkpoint. Against one without it, every refresh of every stream table fails PREDECESSOR_UNAVAILABLE permanently. A full build does not converge either, because the next checkpoint writes a v1 manifest over it.

When no recording exists, the build fails even if the recipe is valid. Run mr-data stream recordings --document-digest D and check what exists and when it expires.

A keyed venue needs a shaped credential, stored by part:

mr-data keys set VENUE_KEY --part key_id=ABC123 --part private_key_pem=@signing.pem --json
mr-data stream capture --document-digest D --registry R --member M --seconds 300 \
  --secret VENUE_KEY --json

--part NAME=@PATH reads that part from disk. The venue’s document declares the parts its credential shape requires. The command names any set of parts that does not match. No command flag takes a secret as an argument.

A probe that found the key wrong has done its job. It reports an auth_outcome: authenticated, auth_rejected, auth_unconfirmed, clock_skew_suspected and more. Act on that outcome.