WITH
params AS (
  SELECT
    (now() AT TIME ZONE 'America/Chicago') AS now_local,
    /* lookbacks */
    INTERVAL '20 minutes' AS win_darkblock,
    INTERVAL '90 minutes' AS win_highs,
    INTERVAL '45 minutes' AS win_lows,
    INTERVAL '30 minutes' AS win_stealth,
    INTERVAL '20 minutes' AS win_shock,
    INTERVAL '10 minutes' AS win_options,
    INTERVAL '20 minutes' AS win_recent,
    /* opening hour (Central) */
    TIME '08:30:00' AS orb_start,
    TIME '09:30:00' AS orb_end,
    /* thresholds */
    1000000::numeric AS per_print_min_notional,  -- $1M per print
    5000000::numeric AS cluster_min_notional,    -- $5M cluster
    150::numeric     AS orb_min_rvol,
    2000000::numeric AS orb_min_notional,
    120::numeric     AS highs_min_rvol,
    3000000::numeric AS stealth_min_notional,
    100::numeric     AS stealth_max_rvol,
    2::int           AS shock_min_events,
    2::int           AS opt_pair_min,
    2000000::numeric AS opt_min_notional,
    3::int           AS min_score
),

/* ---------- Normalize: build event_ts_local from alert_date + alert_time_txt ---------- */
base AS (
  SELECT
    symbol      ::text    AS ticker,
    alert_type  ::text    AS evt_type,
    message     ::text    AS message,
    notional    ::numeric AS notional,
    pct_of_avg30::numeric AS pct_of_avg30,
    CASE
      WHEN alert_time_txt ~* '(AM|PM)\s*$'
        THEN to_timestamp(to_char(alert_date,'YYYY-MM-DD') || ' ' || alert_time_txt,
                          'YYYY-MM-DD HH12:MI:SS AM')::timestamp
      ELSE to_timestamp(to_char(alert_date,'YYYY-MM-DD') || ' ' || alert_time_txt,
                        'YYYY-MM-DD HH24:MI:SS')::timestamp
    END AS event_ts_local
  FROM public.alertstream_monthly
  WHERE symbol IS NOT NULL
),

/* ---------- S1: Dark/Block cluster (liquidity magnets) ---------- */
s1_agg AS (
  SELECT b.ticker,
         COUNT(*)                        AS prints,
         SUM(b.notional)                 AS notional_usd,
         ROUND(AVG(b.pct_of_avg30),1)    AS avg_rvol
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_darkblock
    AND b.evt_type IN ('DarkPool','Block')
    AND b.notional >= p.per_print_min_notional
  GROUP BY b.ticker
),
s1 AS (
  SELECT s.*
  FROM s1_agg s CROSS JOIN params p
  WHERE s.notional_usd >= p.cluster_min_notional
),

/* ---------- S2: ORB momentum (opening hour, today) ---------- */
s2_agg AS (
  SELECT b.ticker,
         COUNT(*)                 AS orb_triggers,
         SUM(b.notional)          AS notional_usd,
         MAX(b.pct_of_avg30)      AS rvol_peak
  FROM base b
  CROSS JOIN params p
  WHERE DATE(b.event_ts_local) = DATE(p.now_local)
    AND b.event_ts_local::time BETWEEN p.orb_start AND p.orb_end
    AND b.evt_type = 'Orb'
  GROUP BY b.ticker
),
s2 AS (
  SELECT s.*
  FROM s2_agg s CROSS JOIN params p
  WHERE s.rvol_peak >= p.orb_min_rvol
    AND s.notional_usd >= p.orb_min_notional
),

/* ---------- S3: 52W High + "$ volume met" confirmation ---------- */
s3_agg AS (
  SELECT b.ticker,
         COUNT(*)                 AS hits,
         MAX(b.pct_of_avg30)      AS rvol_peak,
         SUM(CASE WHEN b.message ILIKE 'Required Dollar Volume Met%' THEN 1 ELSE 0 END) AS dollar_vol_flags
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_highs
    AND b.evt_type = 'AnnualHigh'
  GROUP BY b.ticker
),
s3 AS (
  SELECT s.*
  FROM s3_agg s CROSS JOIN params p
  WHERE s.rvol_peak >= p.highs_min_rvol
    AND s.dollar_vol_flags >= 1
),

/* ---------- S4: Annual Lows + volume pop (reversal watch) ---------- */
s4_lows AS (
  SELECT b.ticker, COUNT(*) AS low_hits, MAX(b.pct_of_avg30) AS rvol
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_lows
    AND b.evt_type = 'AnnualLow'
  GROUP BY b.ticker
),
s4_volpop AS (
  SELECT DISTINCT b.ticker
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_lows
    AND b.evt_type IN ('VolumeActive','PriceSpike')
),
s4 AS (
  SELECT l.ticker, l.low_hits, l.rvol
  FROM s4_lows l
  JOIN s4_volpop v ON v.ticker = l.ticker
),

/* ---------- S5: Stealth Dark (big dark, low rVol) ---------- */
s5_agg AS (
  SELECT b.ticker,
         COUNT(*)                        AS prints,
         SUM(b.notional)                 AS notional_usd,
         ROUND(AVG(b.pct_of_avg30),1)    AS avg_rvol
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_stealth
    AND b.evt_type = 'DarkPool'
  GROUP BY b.ticker
),
s5 AS (
  SELECT s.*
  FROM s5_agg s CROSS JOIN params p
  WHERE s.notional_usd >= p.stealth_min_notional
    AND s.avg_rvol     <  p.stealth_max_rvol
),

/* ---------- S6: Shock tape (spikes/flushes) ---------- */
s6_agg AS (
  SELECT b.ticker,
         SUM(CASE WHEN b.evt_type='PriceSpike'   THEN 1 ELSE 0 END)::int AS spike_cnt,
         SUM(CASE WHEN b.evt_type='RapidDecline' THEN 1 ELSE 0 END)::int AS flush_cnt,
         SUM(b.notional) AS notional_usd
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_shock
    AND b.evt_type IN ('PriceSpike','RapidDecline')
  GROUP BY b.ticker
),
s6 AS (
  SELECT s.*
  FROM s6_agg s CROSS JOIN params p
  WHERE (s.spike_cnt + s.flush_cnt) >= p.shock_min_events
),

/* ---------- S7: Options + (Block/Dark) convergence ---------- */
s7_agg AS (
  SELECT b.ticker,
         SUM(CASE WHEN b.evt_type='Options'  THEN 1 ELSE 0 END)::int AS opt_hits,
         SUM(CASE WHEN b.evt_type='Block'    THEN 1 ELSE 0 END)::int AS block_hits,
         SUM(CASE WHEN b.evt_type='DarkPool' THEN 1 ELSE 0 END)::int AS dark_hits,
         SUM(b.notional) AS notional_usd
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_options
  GROUP BY b.ticker
),
s7 AS (
  SELECT s.*
  FROM s7_agg s CROSS JOIN params p
  WHERE s.opt_hits >= 1
    AND (s.block_hits + s.dark_hits) >= p.opt_pair_min
    AND s.notional_usd >= p.opt_min_notional
),

/* ---------- Recent context & last seen ---------- */
rr AS (
  SELECT b.ticker,
         SUM(b.notional)        AS recent_notional,
         MAX(b.pct_of_avg30)    AS recent_rvol_peak
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_recent
  GROUP BY b.ticker
),
last_seen AS (
  SELECT DISTINCT ON (b.ticker)
         b.ticker,
         b.event_ts_local AS last_seen_ts
  FROM base b
  CROSS JOIN params p
  WHERE b.event_ts_local >= p.now_local - p.win_recent
  ORDER BY b.ticker, b.event_ts_local DESC
),

/* ---------- Universe & Reasons ---------- */
universe AS (
  SELECT ticker FROM s1
  UNION SELECT ticker FROM s2
  UNION SELECT ticker FROM s3
  UNION SELECT ticker FROM s4
  UNION SELECT ticker FROM s5
  UNION SELECT ticker FROM s6
  UNION SELECT ticker FROM s7
),
signals AS (
  SELECT ticker, 'Dark/Block cluster' AS sig, 3 AS w FROM s1
  UNION ALL SELECT ticker, 'ORB momentum'     , 3 FROM s2
  UNION ALL SELECT ticker, '52W High + $Vol'  , 2 FROM s3
  UNION ALL SELECT ticker, 'Low + VolPop'     , 2 FROM s4
  UNION ALL SELECT ticker, 'Stealth Dark'     , 2 FROM s5
  UNION ALL SELECT ticker, 'Shock Tape'       , 1 FROM s6
  UNION ALL SELECT ticker, 'Options + Prints' , 3 FROM s7
),

/* ---------- Scoreboard ---------- */
scored AS (
  SELECT
    u.ticker,
    COALESCE(rr.recent_notional,0)::numeric  AS recent_notional,
    COALESCE(rr.recent_rvol_peak,0)::numeric AS rvol_peak,
    ls.last_seen_ts                           AS last_seen_ts,
    SUM(s.w)::int                             AS score,
    string_agg(s.sig, ' | ' ORDER BY s.w DESC, s.sig) AS reasons
  FROM universe u
  LEFT JOIN signals s ON s.ticker = u.ticker
  LEFT JOIN rr       ON rr.ticker = u.ticker
  LEFT JOIN last_seen ls ON ls.ticker = u.ticker
  GROUP BY u.ticker, rr.recent_notional, rr.recent_rvol_peak, ls.last_seen_ts
)

SELECT
  ticker,
  score,
  CASE
    WHEN recent_notional >= 1e9 THEN '$' || to_char(ROUND(recent_notional/1e9,1), 'FM999999990.0') || 'B'
    WHEN recent_notional >= 1e6 THEN '$' || to_char(ROUND(recent_notional/1e6,1), 'FM999999990.0') || 'M'
    WHEN recent_notional >= 1e3 THEN '$' || to_char(ROUND(recent_notional/1e3,0), 'FM999999990')   || 'K'
    ELSE '$' || to_char(ROUND(recent_notional,0), 'FM999999990')
  END AS notional_20m,
  ROUND(rvol_peak,1) AS rvol_peak,
  to_char(last_seen_ts, 'HH24:MI:SS') AS last_seen_ct,
  reasons
FROM scored, params p
WHERE score >= p.min_score
ORDER BY score DESC, recent_notional DESC, rvol_peak DESC
LIMIT 50;