2s10s inversion episodes: first close, last close, length, and depth

WITH inverted_closes AS
(
    SELECT
        date,
        round((yield_10_year - yield_2_year) * 100) AS spread_bps
    FROM global_markets.treasury_yields
    WHERE isNotNull(yield_10_year) AND isNotNull(yield_2_year)
      AND yield_10_year < yield_2_year
),
flagged AS
(
    SELECT
        date,
        spread_bps,
        if(dateDiff('day', lagInFrame(date, 1, date) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW), date) > 60, 1, 0) AS starts_new_episode
    FROM inverted_closes
),
episodes AS
(
    SELECT
        date,
        spread_bps,
        sum(starts_new_episode) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS episode_id
    FROM flagged
)
SELECT
    min(date) AS first_inverted_close,
    max(date) AS last_inverted_close,
    count() AS inverted_closes,
    min(spread_bps) AS deepest_bps
FROM episodes
GROUP BY episode_id
ORDER BY first_inverted_close