Completed SPY drawdowns since 2016: depth, days falling, days climbing back

WITH daily AS (
    SELECT toDate(toTimeZone(window_start, 'America/New_York')) AS session_date,
           argMax(toFloat64(close), window_start) AS close_px
    FROM global_markets.delayed_stocks_minute_aggs
    WHERE ticker = 'SPY'
      AND toDate(toTimeZone(window_start, 'America/New_York')) >= toDate('2016-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 session_date
),
runs AS (
    SELECT session_date,
           close_px,
           max(close_px) OVER (ORDER BY session_date
                               ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_peak
    FROM daily
),
episodes AS (
    SELECT running_peak AS peak_px,
           min(session_date) AS peak_date,
           argMin(session_date, close_px) AS trough_date,
           min(close_px) AS trough_px
    FROM runs
    GROUP BY running_peak
),
sequenced AS (
    SELECT peak_px,
           peak_date,
           trough_date,
           trough_px,
           leadInFrame(peak_date, 1) OVER (ORDER BY peak_date
                                           ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) AS recovery_date
    FROM episodes
)
SELECT formatDateTime(peak_date, '%b %Y') AS episode_label,
       round(100 * (1 - trough_px / peak_px), 1) AS fall_from_peak_pct,
       dateDiff('day', peak_date, trough_date) AS days_to_trough,
       dateDiff('day', trough_date, recovery_date) AS days_to_new_high,
       dateDiff('day', peak_date, recovery_date) AS days_underwater
FROM sequenced
WHERE recovery_date > trough_date
  AND round(100 * (1 - trough_px / peak_px), 1) >= 5
ORDER BY fall_from_peak_pct DESC
LIMIT 8