CREATE TABLE prices (
      ts TIMESTAMP,
      sym SYMBOL,
      price DOUBLE
  ) timestamp(ts) PARTITION BY DAY;
INSERT INTO prices
  SELECT
      dateadd('s', x::int, '2010-01-01T00:00:00.000000Z') as ts,
      rnd_symbol_zipf(5_000, 2.0),
      rnd_double() * 10.0 + 5.0
      FROM long_sequence(320_000_000);

CREATE TABLE orders (
    id LONG,
    order_ts TIMESTAMP,
    sym1 SYMBOL CAPACITY 1024,
    sym2 SYMBOL CAPACITY 1024,
    unit_price DOUBLE,
    volume DOUBLE
) TIMESTAMP(order_ts) PARTITION BY DAY;

WITH order_basics AS (SELECT * FROM (
  SELECT
    x AS id,
    dateadd('s', (x * 10)::int, '2020-01-01T00:00:00.000000Z') AS order_ts,
    rnd_symbol('sym4893','sym4774','sym4535', ... 500 rarest symbols in prices) AS sym1,
    rnd_symbol('sym4893','sym4774','sym4535', ... 500 rarest symbols in prices) AS sym2,
    rnd_double() * 20 + 10 AS volume
  FROM long_sequence(1000))
  TIMESTAMP(order_ts)
  ),
  join1 AS (
    SELECT ob.*, p.price price1 FROM order_basics ob ASOF JOIN prices p ON (ob.sym1 = p.sym)
  ),
  join2 AS (
    SELECT j1.*, p.price price2 FROM join1 j1 ASOF JOIN prices p ON (j1.sym2 = p.sym)
  )
INSERT INTO orders
SELECT id, order_ts, sym1, sym2, price1/price2 AS unit_price, volume FROM join2;

WITH
  offsets AS (
    SELECT sec_offs, 1_000_000 * sec_offs usec_offs 
    FROM (SELECT x-601 AS sec_offs FROM long_sequence(1201))
  ),
  points AS (SELECT * FROM (
    SELECT id, order_ts, sym1, sym2, unit_price, volume, sec_offs, order_ts + usec_offs AS ts
    FROM orders CROSS JOIN offsets
    ORDER BY order_ts + usec_offs
  ) TIMESTAMP(ts)),
  join1 AS (
    SELECT t.*, p.price AS sym1_price
    FROM points as t
    ASOF JOIN prices as p
    ON (t.sym1 = p.sym)
  ),
  join2 AS (
    SELECT t.*, p.price AS sym2_price
    FROM join1 as t
    ASOF JOIN prices as p
    ON (t.sym2 = p.sym)
  ),
  markouts AS (
  SELECT
    sec_offs, 
    volume,
    volume * (unit_price - (sym1_price/sym2_price)) AS weighted_markout
  FROM join2
  )
SELECT sec_offs, sum(weighted_markout) / sum(volume) AS avg_weigthed_markout 
FROM markouts;