forward_record

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,
        leadInFrame(px, 21) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS px_1m,
        leadInFrame(px, 63) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS px_3m,
        leadInFrame(px, 126) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS px_6m,
        leadInFrame(px, 252) OVER (ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) AS px_12m
    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
),
unrolled AS
(
    SELECT
        is_death,
        tupleElement(h, 1) AS horizon_order,
        tupleElement(h, 2) AS horizon,
        tupleElement(h, 3) AS fwd_px,
        px
    FROM flags
    ARRAY JOIN [(1, '1 month', px_1m), (2, '3 months', px_3m), (3, '6 months', px_6m), (4, '12 months', px_12m)] AS h
    WHERE rn >= 201
      AND tupleElement(h, 3) > 0
)
SELECT
    horizon,
    countIf(is_death)                                                            AS case_count,
    countIf(is_death AND fwd_px > px)                                            AS higher_count,
    toInt32(round(100 * countIf(is_death AND fwd_px > px) / countIf(is_death)))  AS higher_pct,
    toInt32(round(100 * countIf(fwd_px > px) / count()))                         AS all_sessions_higher_pct,
    round(avgIf((fwd_px / px - 1) * 100, is_death), 1)                           AS avg_return_pct,
    round(quantileExactIf((fwd_px / px - 1) * 100, is_death), 1)                 AS median_return_pct,
    round(minIf((fwd_px / px - 1) * 100, is_death), 1)                           AS worst_pct,
    round(maxIf((fwd_px / px - 1) * 100, is_death), 1)                           AS best_pct
FROM unrolled
GROUP BY horizon_order, horizon
HAVING countIf(is_death) > 0
ORDER BY horizon_order