Every 2s10s un-inversion: the crossing and the spread 3, 6 and 12 months later

WITH
daily AS
(
    SELECT
        date,
        dateDiff('day', toDate('1970-01-01'), date)   AS dnum,
        ifNull(toFloat64(max(yield_2_year)), 0.)      AS y2,
        ifNull(toFloat64(max(yield_10_year)), 0.)     AS y10,
        y10 - y2                                      AS spread
    FROM global_markets.treasury_yields
    WHERE yield_2_year > 0 AND yield_10_year > 0
    GROUP BY date
),
scored AS
(
    SELECT
        date, dnum, y2, y10, spread,
        count()         OVER (ORDER BY date ROWS BETWEEN 60 PRECEDING AND 1 PRECEDING) AS rows_prior,
        sum(spread < 0) OVER (ORDER BY date ROWS BETWEEN 60 PRECEDING AND 1 PRECEDING) AS neg_prior,
        sum(spread > 0) OVER (ORDER BY date ROWS BETWEEN CURRENT ROW AND 20 FOLLOWING) AS pos_next
    FROM daily
),
candidates AS
(
    SELECT
        date, dnum, y2, y10, spread,
        lagInFrame(dnum, 1, toInt64(-100000)) OVER (ORDER BY dnum ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS prev_dnum
    FROM scored
    WHERE spread > 0 AND rows_prior = 60 AND neg_prior >= 35 AND pos_next >= 19
),
episodes AS
(
    SELECT date AS cross_date, y2, y10, spread
    FROM candidates
    WHERE dnum - prev_dnum > 180
      AND date <= today() - 380
)
SELECT
    formatDateTime(e.cross_date, '%b %Y')                                       AS episode,
    toString(toDate(e.cross_date))                                              AS crossed_on,
    concat(formatDateTime(e.cross_date, '%b '), toString(toDayOfMonth(e.cross_date)), ', ', toString(toYear(e.cross_date))) AS crossed_on_label,
    countIf(d.spread < 0 AND d.date < e.cross_date)                             AS negative_sessions_prior,
    round(minIf(d.spread, d.date < e.cross_date), 2)                            AS trough_spread,
    round(argMinIf(d.spread, d.date, d.date >= addDays(e.cross_date, 91)), 2)   AS spread_3m_later,
    round(argMinIf(d.spread, d.date, d.date >= addDays(e.cross_date, 182)), 2)  AS spread_6m_later,
    round(argMinIf(d.spread, d.date, d.date >= addDays(e.cross_date, 365)), 2)  AS spread_12m_later
FROM episodes AS e
CROSS JOIN daily AS d
WHERE d.date BETWEEN addDays(e.cross_date, -800) AND addDays(e.cross_date, 380)
GROUP BY e.cross_date
HAVING countIf(d.date >= addDays(e.cross_date, 365)) > 0
ORDER BY e.cross_date