A stream aggregated into bars

Turn recorded trade frames into one-minute bars, and name the pair each product trades in.

The question

What were the open, high, low, close and traded size of each product in each minute a recorder was listening?

A venue pushes one message per trade. A bar is a minute of those messages reduced to five numbers. The reduction happens in the transform, because a recording holds frames and nothing else.

The sources

	The venue	The product list
Publisher	Coinbase Exchange market-data feed	Coinbase Exchange REST API
Address	wss://ws-feed.exchange.coinbase.com	https://api.exchange.coinbase.com/products
Format	JSON frames, sealed by a recorder	JSON array, application/json; charset=utf-8
Cadence	continuous while a subscription is open	changes when the venue lists or delists a product
Rights	read under the venue’s published market-data terms for its open feed, recorded as terms_of_service_asserted	the same terms, the same basis
Why	the trades themselves, with the venue’s own trade identifier on each	the pair each product prices, which no trade frame carries

The feed is open, so neither source needs a credential. The product list held 837 products when this page was written, and three of them are what this table records.

Record before you build

A build never reaches the venue. A run reads the batches a recorder sealed before the run began. Live streams has the four steps and every flag. What this recipe needs from them is the connector document’s digest and one recording:

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

The connector document decides which of the two relations below the build reads. A document that states member_field seals the member with each frame:

"member_field": "/product_id"

The recipe

recipe.json

{
  "dataset": {
    "name": "Coinbase crypto trade prices, one-minute bars",
    "description": "Each row is one product and one minute of the Coinbase Exchange match feed, carrying the first and last trade price of the minute, the highest and lowest, the contracts traded, and the currencies the venue prices that product in."
  },
  "question": {"text": "What were the open, high, low, close and traded size of each product in each minute a recorder was listening?"},
  "table": {
    "name": "minute_bars",
    "description": "One row per product and UTC minute, for the minutes a recording covered.",
    "grain": ["product_id", "minute"],
    "columns": [
      {
        "name": "product_id",
        "type": "string",
        "nullable": false,
        "description": "The market the trades happened in, as the venue names it.",
        "presentation": {"chart": "top_values"}
      },
      {
        "name": "base_currency",
        "type": "string",
        "nullable": true,
        "description": "The asset being bought and sold, from the venue's product list.",
        "presentation": {"chart": "top_values"}
      },
      {
        "name": "quote_currency",
        "type": "string",
        "nullable": true,
        "description": "The currency the price is quoted in, from the venue's product list.",
        "presentation": {"chart": "top_values"}
      },
      {
        "name": "minute",
        "type": "timestamp",
        "nullable": false,
        "description": "The UTC minute the trades fall in, by the venue's own trade time.",
        "presentation": {"chart": "timeline", "bucket": "hour", "story": "The bars stop where the recording stopped, not where the market went quiet."}
      },
      {
        "name": "open",
        "type": "decimal",
        "nullable": true,
        "description": "The price of the first trade in the minute, by the venue's trade identifier.",
        "presentation": {"chart": "histogram", "bins": 10}
      },
      {
        "name": "high",
        "type": "decimal",
        "nullable": true,
        "description": "The highest trade price in the minute.",
        "presentation": {"chart": "histogram", "bins": 10}
      },
      {
        "name": "low",
        "type": "decimal",
        "nullable": true,
        "description": "The lowest trade price in the minute.",
        "presentation": {"chart": "histogram", "bins": 10}
      },
      {
        "name": "close",
        "type": "decimal",
        "nullable": true,
        "description": "The price of the last trade in the minute, by the venue's trade identifier.",
        "presentation": {"chart": "histogram", "bins": 10}
      },
      {
        "name": "volume",
        "type": "decimal",
        "nullable": true,
        "description": "The base-currency size traded in the minute.",
        "presentation": {"chart": "histogram", "bins": 10}
      },
      {
        "name": "trades",
        "type": "integer",
        "nullable": false,
        "description": "How many trades the minute held.",
        "presentation": {"chart": "histogram", "bins": 8, "story": "Most minutes hold a handful of trades and a few hold hundreds."}
      }
    ]
  },
  "sources": [
    {
      "name": "coinbase_matches",
      "description": "Every trade the Coinbase Exchange market-data feed pushed while a recorder was subscribed.",
      "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,SOL-USD"}]
      },
      "limits": {"max_source_bytes": 268435456, "max_rows": 5000000}
    },
    {
      "name": "coinbase_products",
      "description": "The venue's own list of tradable products, with the pair each one prices.",
      "source_class": "user_api",
      "data_classification": "public",
      "locator": {"kind": "https_url", "display_locator": "https://api.exchange.coinbase.com/products"},
      "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": "public.https",
        "credential_mode": "none",
        "origin": "https://api.exchange.coinbase.com",
        "parameters": [
          {"name": "reader.family_id", "value": "json.tabular"},
          {"name": "reader.family_version", "value": "1.1.0"},
          {
            "name": "reader.decode_options",
            "value": "{\"columns\":[{\"name\":\"id\",\"pointer\":\"/id\",\"required\":true},{\"name\":\"base_currency\",\"pointer\":\"/base_currency\",\"required\":true},{\"name\":\"quote_currency\",\"pointer\":\"/quote_currency\",\"required\":true},{\"name\":\"status\",\"pointer\":\"/status\",\"required\":true}],\"document_format\":\"json\",\"expand\":[],\"records_pointer\":\"\"}"
          }
        ]
      },
      "limits": {"max_source_bytes": 4194304, "max_rows": 5000, "max_requests": 1}
    }
  ],
  "transform": {
    "engine": "duckdb_sql",
    "steps": [
      {
        "step_id": "trades",
        "sql": "select member, received_at, json_extract_string(message, '$.trade_id') as trade_id, json_extract_string(message, '$.time') as trade_time, json_extract_string(message, '$.price') as price, json_extract_string(message, '$.size') as size from coinbase_matches where message_type = 'match'",
        "description": "Keep the trade frames and pull the four members a bar needs out of the provider frame."
      },
      {
        "step_id": "bars",
        "sql": "select member as product_id, time_bucket(interval 1 minute, cast(trade_time as timestamp with time zone)) as minute, cast(arg_min(try_cast(price as decimal(24, 8)), cast(trade_id as bigint)) as decimal(24, 8)) as open, cast(max(try_cast(price as decimal(24, 8))) as decimal(24, 8)) as high, cast(min(try_cast(price as decimal(24, 8))) as decimal(24, 8)) as low, cast(arg_max(try_cast(price as decimal(24, 8)), cast(trade_id as bigint)) as decimal(24, 8)) as close, cast(sum(try_cast(size as decimal(24, 8))) as decimal(24, 8)) as volume, cast(count(*) as integer) as trades from trades group by member, time_bucket(interval 1 minute, cast(trade_time as timestamp with time zone))",
        "description": "Reduce the trades to one bar per member and minute. The member column comes from the recorder, so no bar depends on parsing the product out of the frame."
      },
      {
        "step_id": "minute_bars",
        "sql": "select b.product_id, p.base_currency, p.quote_currency, b.minute, b.open, b.high, b.low, b.close, b.volume, b.trades from bars b left join (select id as product_id, base_currency, quote_currency from coinbase_products qualify row_number() over (partition by id order by base_currency) = 1) p on p.product_id = b.product_id",
        "description": "Carry the pair the venue prices each product in. The left join keeps a bar whose product is missing from the list, and the qualify keeps a repeated product from multiplying it."
      }
    ]
  },
  "checks": [
    {
      "check_id": "one_bar_per_product_minute",
      "kind": "key_uniqueness",
      "enforcement": "advisory",
      "columns": ["product_id", "minute"],
      "description": "One row per product per minute. An explicit full resync can acquire frames that span a minute boundary, which is worth recording and is not worth ending a run for."
    },
    {
      "check_id": "every_bar_names_a_pair",
      "kind": "null_ceiling",
      "enforcement": "required",
      "columns": ["quote_currency"],
      "max_null_ppm": 0,
      "description": "Every recorded product appears in the venue's product list."
    },
    {
      "check_id": "close_parsed",
      "kind": "null_ceiling",
      "enforcement": "required",
      "columns": ["close"],
      "max_null_ppm": 0,
      "description": "A null close means the price member did not cast, not a quiet minute."
    },
    {
      "check_id": "volume_plausible",
      "kind": "value_range",
      "enforcement": "required",
      "columns": ["volume"],
      "min_value": "0",
      "max_value": "100000000"
    },
    {
      "check_id": "bars_present",
      "kind": "row_expectation",
      "enforcement": "required",
      "min_rows": 1
    },
    {
      "check_id": "no_long_silence",
      "kind": "continuity",
      "enforcement": "advisory",
      "columns": ["minute"],
      "max_gap_seconds": 900,
      "description": "A gap of more than fifteen minutes across all three products is a window nobody recorded."
    }
  ],
  "units": [],
  "timezone": "UTC"
}

Part by part

The two relations a recorder seals

A stream source presents to the SQL as the recorder’s sealed line, every column text. There are two such lines and the recipe declares neither.

Position	Eight columns	Nine columns
1	epoch	epoch
2	event_id	event_id
3	event_time	event_time
4	message	member
5	message_class	message
6	message_type	message_class
7	received_at	message_type
8	sequence	received_at
9		sequence

The connector document chooses. One that states a member_field writes nine columns, with member fourth, directly after event_time. One without attribution writes its original eight columns byte for byte. Adding the member elsewhere does not change a stored recording’s digest. Any other header causes RUN_RECIPE_INVALID.

This recipe groups on member, so it never parses the product out of a frame. Without connector attribution, the first step reads json_extract_string(message, '$.product_id'). Every row then depends on the venue keeping that field where it is.

The static member table

coinbase_products is an ordinary public.https source with a Reader pinned on connector.parameters:

"parameters": [
  {"name": "reader.family_id", "value": "json.tabular"},
  {"name": "reader.family_version", "value": "1.1.0"},
  {"name": "reader.decode_options", "value": "{\"columns\":[…],\"document_format\":\"json\",\"expand\":[],\"records_pointer\":\"\"}"}
]

Write all three names or none. reader.decode_options is the canonical JSON text of what the family itself admits, with every defaulted setting written out. A stated subset digests differently from the pin the worker reconstructs, and registration refuses it.

records_pointer: "" is the RFC 6901 pointer for a root-level array, which is what this endpoint returns. The four pointers name the members the transform needs. Asking for the other fourteen would cost bytes on every full resync acquisition.

The source declares no window and no closed. That makes it unwindowed: Studio refuses a normal table refresh before acquiring it with RESYNC_REQUIRED, rather than silently re-reading the whole product list. Use an explicit table resync, with one caller-held request UUID, when re-reading both the product list and the recorded-stream boundary is intended.

Transform

The first step keeps the trade frames and pulls four members out of each one:

select member, received_at,
       json_extract_string(message, '$.trade_id') as trade_id,
       json_extract_string(message, '$.time')     as trade_time,
       json_extract_string(message, '$.price')    as price,
       json_extract_string(message, '$.size')     as size
from coinbase_matches
where message_type = 'match'

json_extract_string returns the member’s text. json_extract returns the JSON value, so a string still carries its quotes and a numeric cast over it fails.

The second step reduces the trades to one bar per member and minute:

select member as product_id,
       time_bucket(interval 1 minute, cast(trade_time as timestamp with time zone)) as minute,
       cast(arg_min(try_cast(price as decimal(24, 8)), cast(trade_id as bigint)) as decimal(24, 8)) as open,
       cast(max(try_cast(price as decimal(24, 8))) as decimal(24, 8)) as high,
       cast(min(try_cast(price as decimal(24, 8))) as decimal(24, 8)) as low,
       cast(arg_max(try_cast(price as decimal(24, 8)), cast(trade_id as bigint)) as decimal(24, 8)) as close,
       cast(sum(try_cast(size as decimal(24, 8))) as decimal(24, 8)) as volume,
       cast(count(*) as integer) as trades
from trades
group by member, time_bucket(interval 1 minute, cast(trade_time as timestamp with time zone))

arg_min and arg_max order by the venue’s own trade identifier rather than by sequence. sequence is the frame’s position within one connection, so it restarts when the socket does.

The bucket reads $.time, the instant the venue stamped the trade with. A bar is then about when the trade happened. Bucketing received_at instead would put a frame the recorder read late into the minute it arrived in.

Every outer cast sets the result type. sum and min over exact decimals stay exact. A column declared decimal accepts DECIMAL(p,s) and nothing else, so the transform states the cast.

The third step carries the pair:

select b.product_id, p.base_currency, p.quote_currency, b.minute,
       b.open, b.high, b.low, b.close, b.volume, b.trades
from bars b
left join (select id as product_id, base_currency, quote_currency
           from coinbase_products
           qualify row_number() over (partition by id order by base_currency) = 1) p
  on p.product_id = b.product_id

The join is a left join, so a product the venue stopped listing keeps its bars and fails a check instead of disappearing. The qualify reduces the right side to one row per id, which is what stops a repeated product from multiplying the bars.

Checks

every_bar_names_a_pair is a null_ceiling on quote_currency with max_null_ppm: 0, and it is required. It is the check this shape needs. A product the list does not carry reaches the table as a bar with no currencies, and a left join says nothing about it.

close_parsed is the same ceiling on close. A null close means the price member did not cast, not a quiet minute. try_cast returning null for every row looks like a market nobody traded in.

volume_plausible bounds volume between "0" and "100000000", as decimal strings, because the document carries no fractional JSON number.

one_bar_per_product_minute is the grain, and it is advisory. An explicit resync is a full run over the material it reads now; it does not continue from a predecessor bookmark. A minute split in the material that run read produces one bar in each adjacent result row. Record that result without ending the run.

no_long_silence is a continuity check on minute with max_gap_seconds: 900, also advisory. A gap across all three products is a window nobody recorded rather than a quiet market.

Units

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

Register, run, inspect

mr-data dataset create --name "Coinbase crypto trade prices, one-minute bars" --json
mr-data recipe recipe.json --json

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

mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --json
mr-data table resync TABLE_ID --request-id UUID --json

run has no duration flag. A build reads sealed batches and stops at their end. The recorder fixed those bytes at capture time.

What the run reports

The receipt names the recordings the full run read and the bounds of that material. A later explicit resync does not continue from a predecessor bookmark; it acquires its own current material.

The receipt also carries the product list as an ordinary source entry. It names the rendered address, fetched bytes, byte digest and coverage disposition.

An explicit resync of a recorded stream is not an append. It can replace the live table with the material it acquires now. The table declares no materialization for a stream source and the engine reads none.

mr-data checks RUN_ID --json prints one result per declared check, with the recipe’s own check_id upper-cased, whether it passed, and what the worker measured in words.

Traps

The two sources age differently. The stream part is fixed at capture time, while the product list is re-read only by an explicit resync. That full resync does not append to the prior table; it can replace the live table with the material it acquired.

A missed recording window is a permanent hole. Recordings last thirty days, and an explicit resync cannot recover messages that were never recorded or use a predecessor bookmark to bridge them. Report an unrecorded period as a gap in the series rather than promising a backfill.

--max-rows applies to each source separately. A ceiling below the product count cuts the list, and EVERY_BAR_NAMES_A_PAIR fails on the products the cut removed. Set the ceiling above the largest source rather than at the size of the sample you wanted.

A connector that attributes nothing seals eight columns, and every step naming member then refuses. Read mr-data stream document show ID before writing the SQL, or read the first rows of a sample with mr-data peek.

A step may name only its sources and earlier steps. trades is both a step id here and a declared column name, which is allowed. A step id equal to a source name is not, and the engine refuses it rather than resolving it.