# A stream aggregated into bars Source: https://mostlyright.md/docs/recipes/stream-to-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](/docs/build/live-streams/) has the four steps and every flag. What this recipe needs from them is the connector document's digest and one recording: ```sh 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: ```json "member_field": "/product_id" ``` ## The recipe ```json title="recipe.json" { "dataset": { "name": "Coinbase crypto trade prices, one-minute bars" }, "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`: ```json "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: ```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' ``` `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: ```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)) ``` `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: ```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 ``` 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 Use the real table for one full build. For a bounded test first, follow the [scratch-table workflow](/docs/build/run-and-inspect/#test-before-the-full-build). A successful sample on the real table can publish truncated data. Stop if the forecast refuses; size the recipe before running. ```sh mr-data dataset create --name "Coinbase crypto trade prices, one-minute bars" --json mr-data recipe recipe.json --json mr-data recipe explain RECIPE_ID --json mr-data recipe forecast RECIPE_ID --digest RECIPE_DIGEST --json mr-data run --recipe RECIPE_ID --digest RECIPE_DIGEST --full --json mr-data watch RUN_ID --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 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.