-- v3 vs v3.1 battle test, one paired hour. DuckDB 1.5.5 or later, httpfs, straight off dl.
-- Change the two SET lines for another hour (day folder and hh).
INSTALL httpfs; LOAD httpfs; SET TimeZone = 'UTC';
-- dl rate-limits one address (HTTP 429): retry with backoff, and cache what is read.
SET http_retries = 8; SET http_retry_wait_ms = 2000; SET http_retry_backoff = 2;
SET enable_http_metadata_cache = true; SET threads = 4;
SET VARIABLE day = '2026-10-29';
SET VARIABLE hh  = '18';

SET VARIABLE h    = getvariable('day') || 'T' || getvariable('hh');
SET VARIABLE v3   = 'https://dl.pendulumflow.com/v3/'   || getvariable('day') || '/' || getvariable('hh') || '/';
SET VARIABLE v31  = 'https://dl.pendulumflow.com/v3.1/' || getvariable('day') || '/' || getvariable('hh') || '/';
CREATE OR REPLACE TABLE manifest AS
  SELECT * FROM read_json(getvariable('v31') || getvariable('h') || '.manifest.json', maximum_object_size = 268435456);
CREATE OR REPLACE TABLE files AS SELECT unnest(files, recursive := true) FROM manifest;
SET VARIABLE lean = (SELECT list(getvariable('v31') || file ORDER BY file) FROM files WHERE role = 'lean');
SET VARIABLE prov = (SELECT list(getvariable('v31') || file ORDER BY file) FROM files WHERE role = 'provenance');
SET VARIABLE v3file = getvariable('v3') || getvariable('h') || '.parquet';
SET VARIABLE t0     = (getvariable('day') || ' ' || getvariable('hh') || ':00:00+00')::timestamptz;

-- 1. Rows per event type: v3 = v3.1 stored + mirror_rows_dropped + book_rows_dropped (book only).
--    A FULL JOIN, so an event type present on only one side shows up (reconciles = false).
WITH a AS (SELECT event_type, count(*) n FROM read_parquet(getvariable('v3file')) GROUP BY ALL),
     b AS (SELECT event_type, count(*) n FROM read_parquet(getvariable('lean')) GROUP BY ALL),
     d AS (SELECT unnest(map_keys(m)) et, unnest(map_values(m)) n
           FROM (SELECT json_transform(to_json(mirror_rows_dropped), '"MAP(VARCHAR, BIGINT)"') m FROM manifest)),
     r AS (SELECT coalesce(a.event_type, b.event_type) event_type, coalesce(a.n, 0) v3_rows, coalesce(b.n, 0) v31_rows,
                  coalesce(d.n, 0) mirror_dropped,
                  CASE WHEN coalesce(a.event_type, b.event_type) = 'book' THEN k.book_rows_dropped ELSE 0 END book_dropped
           FROM a FULL JOIN b USING (event_type) LEFT JOIN d ON d.et = coalesce(a.event_type, b.event_type),
                (SELECT book_rows_dropped FROM manifest) k)
SELECT *, v3_rows = v31_rows + mirror_dropped + book_dropped AS reconciles FROM r ORDER BY 1;

-- 2. Schema diff, names AND logical types (lean + provenance vs v3). A column in both with a different type
--    shows as 'TYPE DIFFERS'. Expected: no TYPE DIFFERS row.
CREATE OR REPLACE TABLE s3 AS SELECT column_name AS col, column_type AS ctype FROM (DESCRIBE SELECT * FROM read_parquet(getvariable('v3file')));
CREATE OR REPLACE TABLE s31 AS
  SELECT column_name AS col, column_type AS ctype, 'lean' AS f FROM (DESCRIBE SELECT * FROM read_parquet(getvariable('lean')))
  UNION ALL SELECT column_name, column_type, 'provenance' FROM (DESCRIBE SELECT * FROM read_parquet(getvariable('prov')));
SELECT coalesce(a.col, b.col) AS col,
       CASE WHEN b.col IS NULL THEN 'only in v3' WHEN a.col IS NULL THEN 'only in v3.1 (' || b.f || ')'
            WHEN a.ctype <> b.ctype THEN 'TYPE DIFFERS: v3 ' || a.ctype || ', v3.1 (' || b.f || ') ' || b.ctype
            ELSE 'both (' || b.f || ')' END AS where_
FROM s3 a FULL JOIN s31 b USING (col)
WHERE a.col IS NULL OR b.col IS NULL OR b.f = 'provenance' OR a.ctype <> b.ctype ORDER BY 2, 1;

-- 3. Duplicate rows in the first 5 minutes (by venue timestamp), the SAME key on both sides:
--    (event_type, sequence, source_witness). v3.1 keeps sequence and source_witness in the provenance
--    file, so its lean and provenance files are joined on row_ordinal first (table v31, reused by test 10).
--    Expected 0 and 0. (Rows the previous hour already had are dropped at the hour's start in BOTH archives;
--    the manifest's ab_accounting.*.boundary_dropped counts them.)
CREATE OR REPLACE TABLE v31 AS
  SELECT * FROM read_parquet(getvariable('lean')) AS l JOIN read_parquet(getvariable('prov')) AS pr USING (row_ordinal);
SELECT 'v3' side, count(*) - count(DISTINCT (event_type, sequence, source_witness)) dup_rows
FROM read_parquet(getvariable('v3file')) WHERE timestamp >= getvariable('t0') AND timestamp < getvariable('t0') + INTERVAL 5 MINUTE
UNION ALL
SELECT 'v3.1', count(*) - count(DISTINCT (event_type, sequence, source_witness))
FROM v31 WHERE timestamp >= getvariable('t0') AND timestamp < getvariable('t0') + INTERVAL 5 MINUTE
UNION ALL
SELECT 'v3.1 lean rows without a provenance row', (SELECT count(*) FROM read_parquet(getvariable('lean'))) - (SELECT count(*) FROM v31);

-- 4. Best bid/ask per minute for the 3 busiest best_bid_ask tokens: last value each minute, v3 vs v3.1.
CREATE OR REPLACE TABLE top3 AS
  SELECT asset_id FROM read_parquet(getvariable('lean')) WHERE event_type = 'best_bid_ask'
  GROUP BY 1 ORDER BY count(*) DESC LIMIT 3;
WITH a AS (SELECT asset_id, date_trunc('minute', timestamp) m, arg_max(best_bid, (timestamp, sequence)) bid, arg_max(best_ask, (timestamp, sequence)) ask
           FROM read_parquet(getvariable('v3file')) WHERE event_type = 'best_bid_ask' AND asset_id IN (SELECT asset_id FROM top3) GROUP BY ALL),
     b AS (SELECT asset_id, date_trunc('minute', timestamp) m, arg_max(best_bid, (timestamp, row_ordinal)) bid, arg_max(best_ask, (timestamp, row_ordinal)) ask
           FROM read_parquet(getvariable('lean')) WHERE event_type = 'best_bid_ask' AND asset_id IN (SELECT asset_id FROM top3) GROUP BY ALL)
SELECT hex(asset_id)[1:12] token, count(*) n_minutes, count(*) FILTER (WHERE a.bid IS DISTINCT FROM b.bid OR a.ask IS DISTINCT FROM b.ask) differ
FROM a FULL JOIN b USING (asset_id, m) GROUP BY 1 ORDER BY 1;

-- 5. Trades (last_trade_price), row for row: every trade's (market, asset_id, timestamp, price, size, side,
--    fee_rate_bps, transaction_hash) on one side and not the other, as a multiset (EXCEPT ALL both ways),
--    plus the per-market count and volume. Expected: only_in_v3 = 0, only_in_v31 = 0, markets_differ = 0.
--    An hour with no trades says 'not applicable': run this test on a busy hour.
CREATE OR REPLACE TABLE t3 AS SELECT market, asset_id, timestamp, price, size, side, fee_rate_bps, transaction_hash
  FROM read_parquet(getvariable('v3file')) WHERE event_type = 'last_trade_price';
CREATE OR REPLACE TABLE t31 AS SELECT market, asset_id, timestamp, price, size, side, fee_rate_bps, transaction_hash
  FROM read_parquet(getvariable('lean')) WHERE event_type = 'last_trade_price';
WITH a AS (SELECT market, count(*) n, sum(size) vol FROM t3 GROUP BY 1), b AS (SELECT market, count(*) n, sum(size) vol FROM t31 GROUP BY 1)
SELECT CASE WHEN (SELECT count(*) FROM t3) + (SELECT count(*) FROM t31) = 0 THEN 'not applicable: no trades in this hour'
            ELSE 'compared' END AS status,
       (SELECT count(*) FROM t3) v3_trades, (SELECT count(*) FROM t31) v31_trades,
       (SELECT sum(size) FROM t3) v3_volume, (SELECT sum(size) FROM t31) v31_volume,
       (SELECT count(*) FROM (SELECT * FROM t3 EXCEPT ALL SELECT * FROM t31)) only_in_v3,
       (SELECT count(*) FROM (SELECT * FROM t31 EXCEPT ALL SELECT * FROM t3)) only_in_v31,
       (SELECT count(*) FROM a FULL JOIN b USING (market) WHERE a.n IS DISTINCT FROM b.n OR a.vol IS DISTINCT FROM b.vol) markets_differ;

-- 6. Time order. v3 is grouped by event_type, not sorted within one, and cut by receive time: no row is
--    received outside the hour. v3.1 is the same hour: timestamp + recv_lag_ms (the receive time) is inside
--    the hour on every row, recv_lag_ms is never empty, and each lean file is sorted by its manifest order_by,
--    so timestamp never goes back within one (file, event_type, market, asset_id). Expected 0, 0, 0, 0.
SELECT 'v3: received outside the hour' chk, count_if(timestamp_received < getvariable('t0') OR timestamp_received >= getvariable('t0') + INTERVAL 1 HOUR) n
FROM read_parquet(getvariable('v3file'))
UNION ALL
SELECT 'v3.1: received outside the hour', count_if(timestamp + to_milliseconds(recv_lag_ms) < getvariable('t0') OR timestamp + to_milliseconds(recv_lag_ms) >= getvariable('t0') + INTERVAL 1 HOUR)
FROM read_parquet(getvariable('lean'))
UNION ALL
SELECT 'v3.1: recv_lag_ms empty', count_if(recv_lag_ms IS NULL) FROM read_parquet(getvariable('lean'))
UNION ALL
SELECT 'v3.1: timestamp goes back within a token', count_if(d < INTERVAL 0 SECOND) FROM (
  SELECT timestamp - lag(timestamp) OVER (PARTITION BY filename, event_type, market, asset_id ORDER BY file_row_number) d
  FROM read_parquet(getvariable('lean'), filename = true, file_row_number = true));

-- 8. Opening state = the previous hour's end state, for the 3 busiest best_bid_ask tokens (top3 from test 4).
--    For each token, the last book or price_change stamped at or before hh:00 is found among the previous
--    hour's rows and this hour's (rows received after hh:00 but stamped before it count), in the format's
--    token order (timestamp, sequence, row_ordinal; this hour's row wins a tie across the two hours).
--    Expected, verdict 'equal' on each token: book_as_of = that update's time, and the opening best bid/ask =
--    the price_change's best_bid/best_ask, or, when the last update is a book, its highest bid and lowest ask.
--    The opening-state file is taken from the manifest (role 'opening_state').
SET VARIABLE prev  = getvariable('t0') - INTERVAL 1 HOUR;
SET VARIABLE pv31  = 'https://dl.pendulumflow.com/v3.1/' || strftime(getvariable('prev'), '%Y-%m-%d/%H/');
SET VARIABLE ph    = strftime(getvariable('prev'), '%Y-%m-%dT%H');
SET VARIABLE opening = (SELECT getvariable('v31') || file FROM files WHERE role = 'opening_state');
CREATE OR REPLACE TABLE pman AS SELECT * FROM read_json(getvariable('pv31') || getvariable('ph') || '.manifest.json', maximum_object_size = 268435456);
SET VARIABLE plean = (SELECT list(getvariable('pv31') || f.file ORDER BY f.file) FROM (SELECT unnest(files) f FROM pman) WHERE f.role = 'lean');
SET VARIABLE pprov = (SELECT list(getvariable('pv31') || f.file ORDER BY f.file) FROM (SELECT unnest(files) f FROM pman) WHERE f.role = 'provenance');
WITH o AS (SELECT asset_id, best_bid, best_ask, book_as_of FROM read_parquet(getvariable('opening'))
           WHERE asset_id IN (SELECT asset_id FROM top3)),
     r AS (SELECT 1 AS hour_rank, asset_id, event_type, timestamp, sequence, row_ordinal, best_bid, best_ask, bids, asks
           FROM (SELECT * FROM read_parquet(getvariable('plean')) WHERE event_type IN ('book', 'price_change') AND asset_id IN (SELECT asset_id FROM top3)) AS pl
           JOIN read_parquet(getvariable('pprov')) AS pp USING (row_ordinal)
           WHERE timestamp <= getvariable('t0')
           UNION ALL
           SELECT 2, asset_id, event_type, timestamp, sequence, row_ordinal, best_bid, best_ask, bids, asks FROM v31
           WHERE event_type IN ('book', 'price_change') AND asset_id IN (SELECT asset_id FROM top3) AND timestamp <= getvariable('t0')),
     l AS (SELECT asset_id, arg_max(r, (timestamp, hour_rank, sequence, row_ordinal)) x FROM r GROUP BY 1),
     p AS (SELECT asset_id, x.timestamp t, x.event_type et,
                  CASE WHEN x.event_type = 'book' THEN list_max(list_transform(x.bids, lambda b: b.price)) ELSE x.best_bid END bid,
                  CASE WHEN x.event_type = 'book' THEN list_min(list_transform(x.asks, lambda a: a.price)) ELSE x.best_ask END ask
           FROM l)
SELECT hex(o.asset_id)[1:12] token, o.book_as_of, p.t last_stamped, p.et, o.best_bid, p.bid, o.best_ask, p.ask,
       CASE WHEN p.t IS NULL THEN 'no book or price_change before hh:00'
            WHEN o.book_as_of <> p.t THEN 'FAIL: book_as_of is not the last update'
            WHEN o.best_bid IS NOT DISTINCT FROM p.bid AND o.best_ask IS NOT DISTINCT FROM p.ask THEN 'equal'
            ELSE 'FAIL: best bid/ask differ' END verdict
FROM o LEFT JOIN p USING (asset_id) ORDER BY 1;

-- 10. Same values, row for row. Every v3.1 stored row (lean + provenance, table v31 from test 3) is matched to
--     its v3 row on (event_type, sequence, source_witness), and every shared column is compared exactly.
--     Two v3 columns are checked by the format page's rules, since v3.1 does not store them:
--       timestamp_received = timestamp + recv_lag_ms, to the millisecond, rounded down (it can be negative lag);
--       spread = CASE WHEN spread_differs THEN spread_raw WHEN event_type = 'best_bid_ask' THEN best_ask - best_bid END.
--     arrival_skew is in microseconds in both, but v3.1 floors it to a whole millisecond (the manifest's
--     receive_time_precision), so it is compared with v3's value floored the same way.
--     Expected: unmatched = 0 and every *_differ = 0. (v3 rows with no v3.1 row are the mirror twins and repeat
--     books of test 1; this test checks the rows v3.1 stores.)
CREATE OR REPLACE TABLE m10 AS
  SELECT b.*, a.timestamp AS t3, a.market AS market3, a.asset_id AS asset3, a.best_bid AS bid3, a.best_ask AS ask3,
         a.bids AS bids3, a.asks AS asks3, a.price AS price3, a.size AS size3, a.side AS side3, a.fee_rate_bps AS fee3,
         a.transaction_hash AS tx3, a.id AS id3, a.assets_ids AS aids3, a.winning_asset_id AS win3,
         a.winning_outcome AS wout3, a.outcomes AS outs3, a.question AS q3, a.slug AS slug3,
         a.old_tick_size AS ots3, a.new_tick_size AS nts3, a.witness_set AS ws3, a.arrival_skew AS skew3,
         a.timestamp_received AS recv3, a.spread AS spread3, a.event_type IS NOT NULL AS matched
  FROM v31 b LEFT JOIN read_parquet(getvariable('v3file')) a USING (event_type, sequence, source_witness);
SELECT count(*) v31_rows, count_if(NOT matched) unmatched,
  count_if(matched AND (timestamp IS DISTINCT FROM t3 OR market IS DISTINCT FROM market3 OR asset_id IS DISTINCT FROM asset3)) key_differ,
  count_if(matched AND (best_bid IS DISTINCT FROM bid3 OR best_ask IS DISTINCT FROM ask3 OR bids IS DISTINCT FROM bids3 OR asks IS DISTINCT FROM asks3)) book_differ,
  count_if(matched AND (price IS DISTINCT FROM price3 OR size IS DISTINCT FROM size3 OR side IS DISTINCT FROM side3
                        OR fee_rate_bps IS DISTINCT FROM fee3 OR transaction_hash IS DISTINCT FROM tx3)) trade_differ,
  count_if(matched AND (id IS DISTINCT FROM id3 OR assets_ids IS DISTINCT FROM aids3 OR winning_asset_id IS DISTINCT FROM win3
                        OR winning_outcome IS DISTINCT FROM wout3 OR outcomes IS DISTINCT FROM outs3 OR question IS DISTINCT FROM q3
                        OR slug IS DISTINCT FROM slug3 OR old_tick_size IS DISTINCT FROM ots3 OR new_tick_size IS DISTINCT FROM nts3)) market_differ,
  count_if(matched AND witness_set IS DISTINCT FROM ws3) witness_set_differ,
  count_if(matched AND arrival_skew IS DISTINCT FROM CAST(floor(skew3 / 1000) * 1000 AS BIGINT)) arrival_skew_differ,
  count_if(matched AND timestamp + to_milliseconds(recv_lag_ms) IS DISTINCT FROM date_trunc('millisecond', recv3)) recv_differ,
  count_if(matched AND CASE WHEN spread_differs THEN spread_raw WHEN event_type = 'best_bid_ask' THEN best_ask - best_bid END
                       IS DISTINCT FROM spread3) spread_differ
FROM m10;