Market-wide options volume by session, with monthly expirations labelled

WITH tape AS (
    SELECT toDate(toTimeZone(window_start, 'America/New_York')) AS d,
           sum(toFloat64(volume)) AS vol
    FROM global_markets.options_minute_aggs
    WHERE window_start >= toDateTime(today() - 45, 'America/New_York')
    GROUP BY d
),
ranked AS (
    SELECT d, vol, row_number() OVER (ORDER BY d DESC) AS raw_rn
    FROM tape
),
cal AS (
    SELECT d, vol, rn, sum(if(rn BETWEEN 2 AND 21, 1, 0)) OVER () AS baseline_sessions
    FROM (
        SELECT d, vol, row_number() OVER (ORDER BY d DESC) AS rn
        FROM ranked
        WHERE vol >= 0.75 * (SELECT quantileExact(0.5)(vol) FROM ranked WHERE raw_rn > 1)
    )
),
w AS (
    SELECT d, vol,
           toStartOfMonth(d) + toIntervalDay(((5 - toDayOfWeek(toStartOfMonth(d)) + 7) % 7) + 14) AS third_friday
    FROM cal
    WHERE rn <= 25
),
marked AS (
    SELECT d, vol,
           (d = max(if(d <= third_friday, d, toDate('1970-01-01'))) OVER (PARTITION BY toStartOfMonth(d)))
             AND (third_friday <= max(d) OVER ()) AS is_expiry
    FROM w
),
latest AS (
    SELECT d, vol, is_expiry,
           max(if(is_expiry, d, toDate('1970-01-01'))) OVER () AS last_expiry_d
    FROM marked
)
SELECT formatDateTime(d, '%b %e') AS session,
       round(vol / 1e6, 1) AS contracts_m,
       multiIf(is_expiry, 'monthly expiration', 'ordinary') AS session_type,
       round(max(if(d = last_expiry_d, vol, 0)) OVER () / 1e6, 1) AS monthly_expiry_m,
       toYYYYMMDD(d) AS session_id
FROM latest
ORDER BY d ASC