Top-decile put/call readings per month, grouped by how the S&P 500 tracker moved

WITH raw AS (
    SELECT date AS day,
           sumIf(volume, upper(substring(ticker, length(ticker) - 8, 1)) = 'P') AS put_volume,
           sumIf(volume, upper(substring(ticker, length(ticker) - 8, 1)) = 'C') AS call_volume
    FROM global_markets.options_greeks
    WHERE date >= toDate('2022-01-01')
      AND date <= toDate('2026-07-31')
      AND volume > 0
      AND underlying_symbol IN ('AAPL', 'MSFT', 'NVDA', 'AMZN', 'META',
                                'TSLA', 'GOOGL', 'JPM', 'KO', 'XOM')
    GROUP BY day
    HAVING call_volume > 0
),
equity_pcr AS (
    SELECT day, toFloat64(put_volume) / toFloat64(call_volume) AS pcr
    FROM raw
),
cut AS (
    SELECT quantileDeterministic(0.90)(pcr, cityHash64(day)) AS p90
    FROM equity_pcr
),
by_month AS (
    SELECT toStartOfMonth(day) AS month_start,
           countIf(pcr >= cut.p90) AS extreme_reads,
           quantileDeterministic(0.5)(pcr, cityHash64(day)) AS median_pcr
    FROM equity_pcr CROSS JOIN cut
    GROUP BY month_start
),
spy_daily AS (
    SELECT toDate(toTimeZone(window_start, 'America/New_York')) AS day,
           toFloat64(argMax(close, window_start)) AS close
    FROM global_markets.delayed_stocks_minute_aggs
    WHERE ticker = 'SPY'
      AND toDate(toTimeZone(window_start, 'America/New_York')) >= toDate('2022-01-01')
      AND toDate(toTimeZone(window_start, 'America/New_York')) <= toDate('2026-07-31')
      AND (toHour(toTimeZone(window_start, 'America/New_York')) * 60
           + toMinute(toTimeZone(window_start, 'America/New_York'))) BETWEEN 570 AND 959
    GROUP BY day
),
spy_month AS (
    SELECT toStartOfMonth(day) AS month_start,
           100 * (argMax(close, day) / argMin(close, day) - 1) AS move_pct
    FROM spy_daily
    GROUP BY month_start
)
SELECT multiIf(s.move_pct <= -3, 'index down 3% or more',
               s.move_pct < 0, 'index down under 3%',
               s.move_pct < 3, 'index up under 3%',
               'index up 3% or more') AS index_move_bucket,
       count() AS observations,
       round(avg(m.extreme_reads), 1) AS top_decile_reads_avg,
       round(quantileDeterministic(0.5)(m.median_pcr, cityHash64(m.month_start)), 2) AS typical_pcr_ratio
FROM by_month AS m
INNER JOIN spy_month AS s ON m.month_start = s.month_start
GROUP BY index_move_bucket
ORDER BY min(s.move_pct) ASC