window

WITH
bars AS
(
    SELECT
        date,
        toFloat64(argMax(close, _ingest_time)) AS px
    FROM global_markets.stocks_daily_aggs
    WHERE ticker = 'SPY'
    GROUP BY date
),
smas AS
(
    SELECT
        date,
        px,
        row_number() OVER (ORDER BY date) AS rn,
        avg(px) OVER (ORDER BY date ROWS BETWEEN 49 PRECEDING AND CURRENT ROW) AS avg_50,
        avg(px) OVER (ORDER BY date ROWS BETWEEN 199 PRECEDING AND CURRENT ROW) AS avg_200,
        max(px) OVER (ORDER BY date ROWS BETWEEN 251 PRECEDING AND CURRENT ROW) AS high_1y
    FROM bars
),
states AS
(
    SELECT
        *,
        (avg_50 < avg_200) AS below,
        lagInFrame((avg_50 < avg_200), 1) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS prev_below
    FROM smas
),
flags AS
(
    SELECT
        *,
        (rn >= 201 AND below = 1 AND prev_below = 0) AS is_death,
        (rn >= 201 AND below = 0 AND prev_below = 1) AS is_golden
    FROM states
),
events AS
(
    SELECT date, rn, is_death
    FROM flags
    WHERE is_death OR is_golden
),
paired AS
(
    SELECT
        rn,
        is_death,
        leadInFrame(rn, 1) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS next_rn
    FROM events
),
cross_stats AS
(
    SELECT
        countIf(is_death)                                                           AS death_cross_count,
        countIf(NOT is_death)                                                       AS golden_cross_count,
        minIf(toInt64(next_rn) - toInt64(rn), is_death AND next_rn > 0)            AS min_sessions_to_golden,
        quantileExactIf(toInt64(next_rn) - toInt64(rn), is_death AND next_rn > 0)  AS median_sessions_to_golden,
        maxIf(toInt64(next_rn) - toInt64(rn), is_death AND next_rn > 0)            AS max_sessions_to_golden
    FROM paired
),
span AS
(
    SELECT
        concat(formatDateTime(min(date), '%b'), ' ', toString(toDayOfMonth(min(date))), ', ', toString(toYear(min(date)))) AS first_session,
        concat(formatDateTime(max(date), '%b'), ' ', toString(toDayOfMonth(max(date))), ', ', toString(toYear(max(date)))) AS last_session,
        count()                                                        AS session_count,
        round(quantileExactIf((1 - px / high_1y) * 100, is_death), 1)  AS median_off_high_pct,
        round(minIf((1 - px / high_1y) * 100, is_death), 1)            AS min_off_high_pct,
        round(maxIf((1 - px / high_1y) * 100, is_death), 1)            AS max_off_high_pct
    FROM flags
)
SELECT *
FROM span, cross_stats