/* ==================== STOCKS — PANEL2 with price + alert context (Postgres/Supabase) ==================== */
/* Assumptions (as per your schema):
   - public."OptionsFlow_monthly" has columns: Symbol, CallPut, Side, Strike, Spot, Premium, Volume, OI,
     ExpirationDate, CreatedDate, CreatedTime, StockEtf, Price, etc. (strings/numbers)
   - public.prices_intraday_1m(Symbol text, ts timestamptz, open numeric, high numeric, low numeric, close numeric, volume numeric)
   - public.prices_daily(Symbol text, "Date" date, close numeric)
   - public."AlertStream_monthly" has lowercase columns: "date" text, "timestamp" text, "ticker" text,
     "type" text, "message" text, "price" numeric, "volume" numeric
*/

WITH cfg AS (
  SELECT
    /* <-- adjust these if desired */
    (CURRENT_DATE - INTERVAL '1 day')::date AS start_day,
    (CURRENT_DATE)::date                   AS end_day,
    0.20::numeric                         AS tol_pct  -- tape-alignment tolerance (pct points)
),

/* 1) FLOW: normalize fields + clean numerics */
base AS (
  SELECT
    ofm.ctid                               AS rid,                          -- stable row identity within a monthly table
    ofm.*,
    UPPER(TRIM(ofm."Symbol"))              AS symbol_norm,

    /* cp_norm */
    CASE WHEN UPPER(TRIM(ofm."CallPut")) IN ('C','CALL','CALLS','CE') THEN 'CALL'
         WHEN UPPER(TRIM(ofm."CallPut")) IN ('P','PUT','PUTS','PE')   THEN 'PUT'
    END                                    AS cp_norm,

    /* side_norm */
    CASE
      WHEN UPPER(TRIM(ofm."Side")) IN ('A','AA','ASK','BUY','BOT','BTO','AT_ASK')  THEN 'BUY'
      WHEN UPPER(TRIM(ofm."Side")) IN ('B','BB','BID','SELL','SLD','STO','AT_BID') THEN 'SELL'
      ELSE NULL
    END                                    AS side_norm,

    /* numerics: cast via ::text → regexp_replace → ::numeric so it works whether input is text or numeric */
    NULLIF(REGEXP_REPLACE(ofm."Strike"::text,  '[\s$,]', '', 'g'),'')::numeric  AS strike_num,
    NULLIF(REGEXP_REPLACE(ofm."Spot"::text,    '[\s$,]', '', 'g'),'')::numeric  AS spot_num,
    NULLIF(REGEXP_REPLACE(ofm."Premium"::text, '[\s$,]', '', 'g'),'')::numeric  AS premium_num,
    NULLIF(REGEXP_REPLACE(ofm."Volume"::text,  '[\s$,]', '', 'g'),'')::numeric  AS vol_num,
    NULLIF(REGEXP_REPLACE(ofm."OI"::text,      '[\s$,]', '', 'g'),'')::numeric  AS oi_num,

    /* parse expiration date to DATE (supports YYYY-MM-DD or MM/DD/YYYY) */
    CASE
      WHEN ofm."ExpirationDate"::text ~ '^\d{4}-\d{2}-\d{2}$' THEN ofm."ExpirationDate"::date
      WHEN ofm."ExpirationDate"::text ~ '^\d{1,2}/\d{1,2}/\d{2,4}$' THEN to_date(ofm."ExpirationDate"::text, 'MM/DD/YYYY')
      ELSE NULL
    END                                    AS exp_date,

    /* build FLOW timestamp in local feed time (assume CST = America/Chicago) */
    /* Normalize date format first, then parse time */
    CASE
      WHEN ofm."CreatedTime"::text ~* '(AM|PM)'
        THEN CASE
          /* Normalize date to YYYY-MM-DD format first */
          /* Has seconds: "9:43:09 AM" or "09:43:09 AM" */
          WHEN ofm."CreatedTime"::text ~ ':\d{2}:\d{2}\s*(AM|PM)'
            THEN COALESCE(
              -- Normalize date and parse with normalized format
              NULLIF(to_timestamp(
                CASE 
                  WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$' 
                    THEN ofm."CreatedDate"::text
                  WHEN ofm."CreatedDate"::text ~ '^\d{1,2}/\d{1,2}/\d{4}$'
                    THEN to_date(ofm."CreatedDate"::text, 'MM/DD/YYYY')::text
                  ELSE NULL
                END || ' ' || btrim(ofm."CreatedTime"::text), 
                'YYYY-MM-DD HH12:MI:SS AM'), NULL),
              -- Fallback: try with MM/DD/YYYY format directly
              NULLIF(to_timestamp(ofm."CreatedDate"::text || ' ' || btrim(ofm."CreatedTime"::text), 'MM/DD/YYYY HH12:MI:SS AM'), NULL)
            )
          /* Has minutes: "9:43 AM" or "09:43 AM" */
          WHEN ofm."CreatedTime"::text ~ ':\d{2}\s*(AM|PM)'
            THEN COALESCE(
              -- Normalize date and parse
              NULLIF(to_timestamp(
                CASE 
                  WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$' 
                    THEN ofm."CreatedDate"::text
                  WHEN ofm."CreatedDate"::text ~ '^\d{1,2}/\d{1,2}/\d{4}$'
                    THEN to_date(ofm."CreatedDate"::text, 'MM/DD/YYYY')::text
                  ELSE NULL
                END || ' ' || btrim(ofm."CreatedTime"::text), 
                'YYYY-MM-DD HH12:MI AM'), NULL),
              -- Fallback: try with MM/DD/YYYY format directly
              NULLIF(to_timestamp(ofm."CreatedDate"::text || ' ' || btrim(ofm."CreatedTime"::text), 'MM/DD/YYYY HH12:MI AM'), NULL),
              -- Try with seconds format too
              NULLIF(to_timestamp(
                CASE 
                  WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$' 
                    THEN ofm."CreatedDate"::text
                  WHEN ofm."CreatedDate"::text ~ '^\d{1,2}/\d{1,2}/\d{4}$'
                    THEN to_date(ofm."CreatedDate"::text, 'MM/DD/YYYY')::text
                  ELSE NULL
                END || ' ' || btrim(ofm."CreatedTime"::text) || ':00', 
                'YYYY-MM-DD HH12:MI:SS AM'), NULL),
              NULLIF(to_timestamp(ofm."CreatedDate"::text || ' ' || btrim(ofm."CreatedTime"::text) || ':00', 'MM/DD/YYYY HH12:MI:SS AM'), NULL)
            )
          /* Just hour: "9 AM" or "09 AM" - convert to "9:00:00 AM" */
          WHEN ofm."CreatedTime"::text ~ '^\d{1,2}\s*(AM|PM)$'
            THEN COALESCE(
              NULLIF(to_timestamp(
                CASE 
                  WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$' 
                    THEN ofm."CreatedDate"::text
                  WHEN ofm."CreatedDate"::text ~ '^\d{1,2}/\d{1,2}/\d{4}$'
                    THEN to_date(ofm."CreatedDate"::text, 'MM/DD/YYYY')::text
                  ELSE NULL
                END || ' ' || REGEXP_REPLACE(btrim(ofm."CreatedTime"::text), '^(\d{1,2})\s*(AM|PM)$', '\1:00:00 \2', 'i'), 
                'YYYY-MM-DD HH12:MI:SS AM'), NULL),
              NULLIF(to_timestamp(ofm."CreatedDate"::text || ' ' || REGEXP_REPLACE(btrim(ofm."CreatedTime"::text), '^(\d{1,2})\s*(AM|PM)$', '\1:00:00 \2', 'i'), 'MM/DD/YYYY HH12:MI:SS AM'), NULL)
            )
          /* Fallback: try various formats */
          ELSE COALESCE(
            NULLIF(to_timestamp(
              CASE 
                WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$' 
                  THEN ofm."CreatedDate"::text
                WHEN ofm."CreatedDate"::text ~ '^\d{1,2}/\d{1,2}/\d{4}$'
                  THEN to_date(ofm."CreatedDate"::text, 'MM/DD/YYYY')::text
                ELSE NULL
              END || ' ' || btrim(ofm."CreatedTime"::text), 
              'YYYY-MM-DD HH12:MI AM'), NULL),
            NULLIF(to_timestamp(ofm."CreatedDate"::text || ' ' || btrim(ofm."CreatedTime"::text), 'MM/DD/YYYY HH12:MI AM'), NULL)
          )
        END
      -- Non-AM/PM times (24-hour format)
      WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$'
        THEN to_timestamp(ofm."CreatedDate"::text || ' ' || btrim(ofm."CreatedTime"::text), 'YYYY-MM-DD HH24:MI:SS')
      ELSE to_timestamp(
        CASE 
          WHEN ofm."CreatedDate"::text ~ '^\d{4}-\d{2}-\d{2}$' 
            THEN ofm."CreatedDate"::text
          WHEN ofm."CreatedDate"::text ~ '^\d{1,2}/\d{1,2}/\d{4}$'
            THEN to_date(ofm."CreatedDate"::text, 'MM/DD/YYYY')::text
          ELSE ofm."CreatedDate"::text
        END || ' ' || btrim(ofm."CreatedTime"::text),
        'YYYY-MM-DD HH24:MI:SS')
    END                                    AS flow_ts_local   -- timestamp without time zone (CST clock)

  FROM public."OptionsFlow_monthly" ofm
  WHERE ofm."Premium" IS NOT NULL
    AND btrim(ofm."Premium"::text) <> ''
    -- AND ofm."StockEtf" = 'STOCK'
    -- AND ofm."Symbol" NOT IN ('TSLA','NVDA')  -- your panel excludes these
),

/* 2) Window restriction (CST date window) and compute UTC + DTE */
flow AS (
  SELECT
    b.*,
    (b.flow_ts_local)::date                                         AS flow_date_cst,
    /* Convert local CST clock to a real point in time (timestamptz) */
    (b.flow_ts_local AT TIME ZONE 'America/Chicago')                AS flow_ts_utc,
    /* Days to Expiration (DTE) */
    CASE
      WHEN b.exp_date IS NOT NULL AND (b.flow_ts_local)::date IS NOT NULL
        THEN (b.exp_date - (b.flow_ts_local)::date)
      ELSE NULL
    END                                                             AS dte
  FROM base b
  WHERE (b.flow_ts_local)::date BETWEEN (SELECT start_day FROM cfg) AND (SELECT end_day FROM cfg)
),

/* 3) Direction + moneyness % */
mny AS (
  SELECT
    f.*,
    CASE
      WHEN cp_norm='CALL' AND strike_num >  spot_num THEN 'OTM'
      WHEN cp_norm='CALL' AND strike_num <= spot_num THEN 'ITM'
      WHEN cp_norm='PUT'  AND strike_num <  spot_num THEN 'OTM'
      WHEN cp_norm='PUT'  AND strike_num >= spot_num THEN 'ITM'
    END AS moneyness,
    CASE
      WHEN (cp_norm='CALL' AND side_norm='BUY') OR (cp_norm='PUT' AND side_norm='SELL') THEN 'BULL'
      WHEN (cp_norm='PUT'  AND side_norm='BUY') OR (cp_norm='CALL' AND side_norm='SELL') THEN 'BEAR'
      ELSE NULL
    END AS direction,
    /* +mny_pct means ITM depth; -mny_pct means OTM distance */
    CASE
      WHEN spot_num IS NOT NULL AND spot_num <> 0 AND cp_norm='CALL'
        THEN ((spot_num - strike_num) * 100.0) / NULLIF(spot_num,0)
      WHEN spot_num IS NOT NULL AND spot_num <> 0 AND cp_norm='PUT'
        THEN ((strike_num - spot_num) * 100.0) / NULLIF(spot_num,0)
      ELSE NULL
    END AS mny_pct
  FROM flow f
),

/* 4) Running sums by (exp_date, symbol) — keep your original analytics */
agg AS (
  SELECT
    m.*,

    SUM(CASE WHEN cp_norm='CALL' AND side_norm='BUY'  AND moneyness='OTM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_cb_otm,
    SUM(CASE WHEN cp_norm='CALL' AND side_norm='BUY'  AND moneyness='ITM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_cb_itm,
    SUM(CASE WHEN cp_norm='CALL' AND side_norm='SELL' AND moneyness='OTM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_cs_otm,
    SUM(CASE WHEN cp_norm='CALL' AND side_norm='SELL' AND moneyness='ITM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_cs_itm,

    SUM(CASE WHEN cp_norm='PUT'  AND side_norm='BUY'  AND moneyness='OTM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_pb_otm,
    SUM(CASE WHEN cp_norm='PUT'  AND side_norm='BUY'  AND moneyness='ITM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_pb_itm,
    SUM(CASE WHEN cp_norm='PUT'  AND side_norm='SELL' AND moneyness='OTM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_ps_otm,
    SUM(CASE WHEN cp_norm='PUT'  AND side_norm='SELL' AND moneyness='ITM' THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_ps_itm,

    SUM(vol_num) OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS vol_all,
    SUM(oi_num)  OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS oi_all,

    SUM(CASE WHEN (cp_norm='CALL' AND side_norm='BUY') OR (cp_norm='PUT'  AND side_norm='SELL') THEN vol_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS bull_vol,
    SUM(CASE WHEN (cp_norm='PUT'  AND side_norm='BUY') OR (cp_norm='CALL' AND side_norm='SELL') THEN vol_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS bear_vol,

    SUM(CASE WHEN (cp_norm='CALL' AND side_norm='BUY') OR (cp_norm='PUT'  AND side_norm='SELL') THEN oi_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS bull_oi,
    SUM(CASE WHEN (cp_norm='PUT'  AND side_norm='BUY') OR (cp_norm='CALL' AND side_norm='SELL') THEN oi_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS bear_oi,

    /* OI runs for 💰 badge */
    SUM(CASE WHEN cp_norm='CALL' AND side_norm='BUY' AND moneyness='OTM' THEN oi_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS oi_cb_otm,
    SUM(CASE WHEN cp_norm='PUT'  AND side_norm='BUY' AND moneyness='OTM' THEN oi_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS oi_pb_otm,

    /* UNKNOWN SIDE tracking - premium and count for flows where direction is NULL (running sum) */
    SUM(CASE WHEN direction IS NULL THEN premium_num ELSE 0 END)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS prem_unknown_sym,
    COUNT(*) FILTER (WHERE direction IS NULL)
      OVER (PARTITION BY exp_date, symbol_norm ORDER BY flow_ts_utc, rid) AS cnt_unknown_sym,

    /* Total premium by symbol/expiration for rule-based rockets (total sum, not running) */
    SUM(premium_num) OVER (PARTITION BY exp_date, symbol_norm) AS prem_total_sym,
    /* Directional premium totals by symbol/expiration (for rule-based rockets) */
    SUM(CASE WHEN direction='BULL' THEN premium_num ELSE 0 END) OVER (PARTITION BY exp_date, symbol_norm) AS prem_bull_sym,
    SUM(CASE WHEN direction='BEAR' THEN premium_num ELSE 0 END) OVER (PARTITION BY exp_date, symbol_norm) AS prem_bear_sym,

    SUM(1) OVER (PARTITION BY exp_date, symbol_norm, direction ORDER BY flow_ts_utc, rid) AS dir_count
  FROM mny m
),

/* 5) Netting + badges (kept from your logic) */
final1 AS (
  SELECT
    a.*,
    (bull_vol - bear_vol) AS net_vol_raw,
    (bull_oi  - bear_oi ) AS net_oi_raw,
    (prem_cb_itm + prem_ps_itm) AS bull_prem_itm,
    (prem_cs_itm + prem_pb_itm) AS bear_prem_itm,
    (prem_cb_otm + prem_ps_otm) AS otm_bull_raw,
    (prem_pb_otm + prem_cs_otm) AS otm_bear_raw,
    (prem_cb_itm + prem_ps_itm + prem_cb_otm + prem_ps_otm) AS bull_total,
    (prem_cs_itm + prem_pb_itm + prem_cs_otm + prem_pb_otm) AS bear_total
  FROM agg a
),
deco AS (
  SELECT
    f.*,
    CASE
      WHEN (bull_prem_itm) > (bear_prem_itm) THEN '🟢'
      WHEN (bull_prem_itm) < (bear_prem_itm) THEN '🔴'
      ELSE ''
    END AS badge_round,
    (
      CASE WHEN (bull_prem_itm) > (bear_prem_itm) AND prem_cb_itm > prem_ps_itm THEN '💎' ELSE '' END
      || CASE WHEN (bull_prem_itm) < (bear_prem_itm) AND prem_pb_itm > prem_cs_itm THEN '💎' ELSE '' END
      || CASE WHEN (bull_prem_itm) > (bear_prem_itm) AND (otm_bull_raw - otm_bear_raw) > 10000 THEN '⭐' ELSE '' END
      || CASE WHEN (bull_prem_itm) < (bear_prem_itm) AND (otm_bear_raw - otm_bull_raw) > 10000 THEN '⭐' ELSE '' END
      || CASE WHEN (bull_prem_itm) > (bear_prem_itm) AND oi_cb_otm > 100000 THEN '💰' ELSE '' END
      || CASE WHEN (bull_prem_itm) < (bear_prem_itm) AND oi_pb_otm > 100000 THEN '💰' ELSE '' END
      || CASE WHEN COALESCE(vol_num,0) > COALESCE(oi_num,0) THEN '✔' ELSE '' END
    ) AS badge_more,
    CASE
      WHEN premium_num > 10000 AND
           (
             ( (bull_prem_itm) > (bear_prem_itm)
               AND ( (cp_norm='CALL' AND POSITION('AA' IN UPPER("Side"))>0)
                     OR  (cp_norm='PUT'  AND POSITION('BB' IN UPPER("Side"))>0) ) )
             OR
             ( (bull_prem_itm) < (bear_prem_itm)
               AND ( (cp_norm='CALL' AND POSITION('BB' IN UPPER("Side"))>0)
                     OR  (cp_norm='PUT'  AND POSITION('AA' IN UPPER("Side"))>0) ) )
           )
      THEN '⚡'
      ELSE ''
    END AS flash
  FROM final1 f
),
rocketize AS (
  SELECT
    d.*,
    CASE WHEN d.badge_round = '🟢' THEN 1 ELSE 0 END                           AS has_round_green,
    CASE WHEN POSITION('💎' IN d.badge_more) > 0 THEN 1 ELSE 0 END             AS has_diamond,
    CASE WHEN POSITION('⭐' IN d.badge_more) > 0 THEN 1 ELSE 0 END              AS has_star,
    CASE WHEN POSITION('💰' IN d.badge_more) > 0 THEN 1 ELSE 0 END             AS has_money,
    CASE WHEN d.flash = '⚡' THEN 1 ELSE 0 END                                  AS has_flash,
    CASE WHEN COALESCE(d.vol_num,0) > COALESCE(d.oi_num,0) THEN 1 ELSE 0 END   AS has_check,
    CASE WHEN d.premium_num > 1000000 THEN 1 ELSE 0 END                         AS has_fire,
    CASE WHEN d.premium_num >  500000 THEN 1 ELSE 0 END                         AS has_cash,
    CASE
      WHEN ( d.badge_round = '🟢'
             AND POSITION('💎' IN d.badge_more) > 0
             AND POSITION('⭐' IN d.badge_more) > 0
             AND POSITION('💰' IN d.badge_more) > 0
             AND d.flash = '⚡'
             AND d.premium_num > 1000000 )
        THEN '🚀'
      WHEN (COALESCE(d.vol_num,0) > COALESCE(d.oi_num,0))
           AND d.flash = '⚡'
           AND POSITION('💰' IN d.badge_more) > 0
           AND d.premium_num > 500000
        THEN '🚀🚀🚀'
      WHEN (COALESCE(d.vol_num,0) > COALESCE(d.oi_num,0))
           AND d.flash = '⚡'
           AND POSITION('💰' IN d.badge_more) > 0
        THEN '🚀🚀'
      WHEN (COALESCE(d.vol_num,0) > COALESCE(d.oi_num,0))
           AND ( d.flash = '⚡' OR d.premium_num > 500000 )
        THEN '🚀'
      ELSE ''
    END AS rocket
  FROM deco d
),

/* 6) PRICE CONTEXT (use LATERAL for fast top-1 seeks; convert bar times to CST for session bucketing) */
price_ctx AS (
  SELECT
    c.rid,
    c.symbol_norm,
    c.flow_ts_utc,
    c.flow_ts_local,
    c.flow_date_cst,

    /* Session bucket in CST based on flow time */
    CASE
      WHEN EXTRACT(HOUR FROM c.flow_ts_local)::int BETWEEN 4 AND 9
           AND NOT (EXTRACT(HOUR FROM c.flow_ts_local)::int = 9 AND EXTRACT(MINUTE FROM c.flow_ts_local)::int >= 30) THEN 'PRE'
      WHEN (EXTRACT(HOUR FROM c.flow_ts_local)::int = 9 AND EXTRACT(MINUTE FROM c.flow_ts_local)::int >= 30)
        OR (EXTRACT(HOUR FROM c.flow_ts_local)::int > 9 AND EXTRACT(HOUR FROM c.flow_ts_local)::int < 16) THEN 'RTH'
      WHEN EXTRACT(HOUR FROM c.flow_ts_local)::int BETWEEN 16 AND 20 THEN 'POST'
      ELSE 'OFF'
    END AS session_bucket,

    /* last 1m bar <= flow time */
    p1.close  AS u_close,
    p1.high   AS u_high,
    p1.low    AS u_low,
    p1.volume AS u_vol_1m,
    p1.ts     AS u_ts,

    /* first RTH bar >= 09:30:00 CST for that flow date */
    rth.open  AS rth_open,
    rth.ts    AS rth_ts,

    /* prior daily close (CST prior day) */
    pd.close  AS prior_close,

    /* 5m / 15m ago prices */
    p5.close  AS close_5m_ago,
    p15.close AS close_15m_ago

  FROM rocketize c

  /* nearest <= flow */
  LEFT JOIN LATERAL (
    SELECT p.close, p.high, p.low, p.volume, p.ts
    FROM public.prices_intraday_1m p
    WHERE UPPER(p.symbol)=c.symbol_norm
      AND p.ts <= c.flow_ts_utc
    ORDER BY p.ts DESC
    LIMIT 1
  ) p1 ON TRUE

  /* first RTH bar of same CST date (convert bar ts to CST on the fly) */
  LEFT JOIN LATERAL (
    SELECT p.open, p.ts
    FROM public.prices_intraday_1m p
    WHERE UPPER(p.symbol)=c.symbol_norm
      AND (timezone('America/Chicago', p.ts))::date = c.flow_date_cst
      AND (timezone('America/Chicago', p.ts))::time >= time '09:30:00'
    ORDER BY p.ts ASC
    LIMIT 1
  ) rth ON TRUE

  /* prior close (CST prior calendar day) */
  LEFT JOIN LATERAL (
    SELECT d.close
    FROM public.prices_daily d
    WHERE UPPER(d.symbol)=c.symbol_norm
      AND d."Date" = c.flow_date_cst - INTERVAL '1 day'
    ORDER BY d."Date" DESC
    LIMIT 1
  ) pd ON TRUE

  /* 5m ago */
  LEFT JOIN LATERAL (
    SELECT p.close
    FROM public.prices_intraday_1m p
    WHERE UPPER(p.symbol)=c.symbol_norm
      AND p.ts <= c.flow_ts_utc - INTERVAL '5 minutes'
    ORDER BY p.ts DESC
    LIMIT 1
  ) p5 ON TRUE

  /* 15m ago */
  LEFT JOIN LATERAL (
    SELECT p.close
    FROM public.prices_intraday_1m p
    WHERE UPPER(p.symbol)=c.symbol_norm
      AND p.ts <= c.flow_ts_utc - INTERVAL '15 minutes'
    ORDER BY p.ts DESC
    LIMIT 1
  ) p15 ON TRUE
),

/* 7) Price features with proper NUMERIC rounding (avoid double-precision round()) */
price_feats AS (
  SELECT
    px.*,
    /* vs prior close is available in all sessions; vs RTH open only valid during/after RTH */
    CASE
      WHEN px.prior_close IS NOT NULL AND px.u_close IS NOT NULL
        THEN ROUND( (((px.u_close - px.prior_close) * 100.0) / NULLIF(px.prior_close,0))::numeric, 2 )
      ELSE NULL
    END AS pct_vs_prior_close,

    CASE
      WHEN px.rth_open IS NOT NULL
           AND px.u_close IS NOT NULL
           AND px.session_bucket = 'RTH'
        THEN ROUND( (((px.u_close - px.rth_open) * 100.0) / NULLIF(px.rth_open,0))::numeric, 2 )
      ELSE NULL
    END AS pct_vs_rth_open,

    CASE WHEN px.close_5m_ago  IS NOT NULL AND px.u_close IS NOT NULL
      THEN ROUND( (((px.u_close - px.close_5m_ago)  * 100.0) / NULLIF(px.close_5m_ago,0))::numeric, 2 )
      ELSE NULL END AS pct_5m_momo,

    CASE WHEN px.close_15m_ago IS NOT NULL AND px.u_close IS NOT NULL
      THEN ROUND( (((px.u_close - px.close_15m_ago) * 100.0) / NULLIF(px.close_15m_ago,0))::numeric, 2 )
      ELSE NULL END AS pct_15m_momo
  FROM price_ctx px
),

/* 8) ALERTS: normalize Date + AM/PM time → CST local ts → UTC; find nearest within ±15m via LATERAL */
alert_nearest AS (
  SELECT
    c.rid,
    a1.near_alert_type,
    a1.near_alert_msg,
    a1.near_alert_price,
    a1.near_alert_volume,
    a1.event_ts_utc
  FROM rocketize c
  LEFT JOIN LATERAL (
    SELECT q."type"  AS near_alert_type,
           q."message" AS near_alert_msg,
           q."price" AS near_alert_price,
           q."volume" AS near_alert_volume,
           q.event_ts_utc
    FROM (
      SELECT
        a."type",
        a."message",
        a."price",
        a."volume",
        /* Build local event timestamp (CST clock) from text columns */
        CASE
          WHEN a."date" ~ '^\d{4}-\d{2}-\d{2}$'
            THEN to_timestamp(a."date" || ' ' || btrim(a."timestamp"), 'YYYY-MM-DD HH12:MI AM')
          ELSE to_timestamp(a."date" || ' ' || btrim(a."timestamp"), 'MM/DD/YYYY HH12:MI AM')
        END                                            AS event_ts_local,
        /* Convert to a real point in time (UTC) */
        (CASE
          WHEN a."date" ~ '^\d{4}-\d{2}-\d{2}$'
            THEN to_timestamp(a."date" || ' ' || btrim(a."timestamp"), 'YYYY-MM-DD HH12:MI AM')
          ELSE to_timestamp(a."date" || ' ' || btrim(a."timestamp"), 'MM/DD/YYYY HH12:MI AM')
         END AT TIME ZONE 'America/Chicago')           AS event_ts_utc
      FROM public."AlertStream_monthly" a
      WHERE UPPER(a."ticker") = c.symbol_norm
    ) q
    WHERE q.event_ts_utc BETWEEN c.flow_ts_utc - INTERVAL '15 minutes'
                             AND c.flow_ts_utc + INTERVAL '15 minutes'
    ORDER BY ABS(EXTRACT(EPOCH FROM (q.event_ts_utc - c.flow_ts_utc))) ASC
    LIMIT 1
  ) a1 ON TRUE
),

/* 9) Enrich + tape-alignment with tolerance + score + Rocket label with [ITM/OTM %] */
flow_enriched AS (
  SELECT
    r.*,
    pf.session_bucket,
    pf.pct_vs_prior_close,
    pf.pct_vs_rth_open,
    pf.pct_5m_momo,
    pf.pct_15m_momo,

    an.near_alert_type,
    an.near_alert_msg,

    /* Tape alignment with tolerance band from cfg */
    CASE
      WHEN pf.session_bucket='PRE'  AND pf.pct_vs_prior_close IS NOT NULL THEN
        CASE WHEN (r.direction='BULL' AND pf.pct_vs_prior_close >= (SELECT tol_pct FROM cfg))
               OR (r.direction='BEAR' AND pf.pct_vs_prior_close <= -(SELECT tol_pct FROM cfg)) THEN 1 ELSE 0 END
      WHEN pf.session_bucket='RTH'  AND pf.pct_vs_rth_open   IS NOT NULL THEN
        CASE WHEN (r.direction='BULL' AND pf.pct_vs_rth_open >= (SELECT tol_pct FROM cfg))
               OR (r.direction='BEAR' AND pf.pct_vs_rth_open <= -(SELECT tol_pct FROM cfg)) THEN 1 ELSE 0 END
      WHEN pf.session_bucket='POST' AND pf.pct_vs_prior_close IS NOT NULL THEN
        CASE WHEN (r.direction='BULL' AND pf.pct_vs_prior_close >= (SELECT tol_pct FROM cfg))
               OR (r.direction='BEAR' AND pf.pct_vs_prior_close <= -(SELECT tol_pct FROM cfg)) THEN 1 ELSE 0 END
      ELSE 0
    END AS tape_alignment,

    CASE WHEN an.near_alert_type IS NOT NULL THEN 1 ELSE 0 END AS catalyst_flag
  FROM rocketize r
  LEFT JOIN price_feats    pf ON pf.rid = r.rid
  LEFT JOIN alert_nearest  an ON an.rid = r.rid
),
scored AS (
  SELECT
    fe.*,
    (CASE
       WHEN fe.premium_num >= 2000000 THEN 3.0
       WHEN fe.premium_num >=  800000 THEN 2.0
       WHEN fe.premium_num >=  200000 THEN 1.0
       ELSE 0.0
     END)
    + 1.5 * CASE
              WHEN (fe.bull_total + fe.bear_total) IS NOT NULL AND (fe.bull_total + fe.bear_total) <> 0
                THEN (fe.bull_total - fe.bear_total) / NULLIF(ABS(fe.bull_total + fe.bear_total),0)
              ELSE 0 END
    + 1.2 * CASE WHEN COALESCE(fe.vol_num,0) > COALESCE(fe.oi_num,0) THEN 1 ELSE 0 END
    + 1.0 * CASE fe.session_bucket WHEN 'RTH' THEN 1 WHEN 'POST' THEN 0.5 WHEN 'PRE' THEN 0.3 ELSE 0 END
    + 1.0 * fe.catalyst_flag
    + 0.8 * CASE
              WHEN fe.cp_norm='CALL' AND fe.strike_num > fe.spot_num THEN 1
              WHEN fe.cp_norm='PUT'  AND fe.strike_num < fe.spot_num THEN 1
              ELSE 0 END
    + 0.5 * fe.tape_alignment
    AS rocket_score,
    /* Rule-based rockets (SQLite style) */
    CASE
      WHEN fe.premium_num >= 250000
       AND ABS(fe.prem_bull_sym - fe.prem_bear_sym) / NULLIF(fe.prem_total_sym, 0) >= 0.6
       AND COALESCE(fe.vol_num, 0) > COALESCE(fe.oi_num, 0)
       AND fe.tape_alignment = 1
      THEN '🚀🚀🚀'
      WHEN fe.premium_num >= 150000
       AND fe.tape_alignment = 1
      THEN '🚀🚀'
      WHEN fe.premium_num >= 80000
       AND fe.tape_alignment = 1
      THEN '🚀'
      ELSE NULL
    END AS rocket_rule
  FROM flow_enriched fe
),
rocket2 AS (
  SELECT
    s.*,
    /* rocket label core - prefer rule-based, fallback to score-based, then original rocket */
    COALESCE(
      s.rocket_rule,
      CASE
        WHEN s.rocket_score >= 5.0 THEN '🚀🚀🚀'
        WHEN s.rocket_score >= 3.5 THEN '🚀🚀'
        WHEN s.rocket_score >= 2.0 THEN '🚀'
        ELSE COALESCE(s.rocket, '')
      END
    ) AS rocket_base,
    /* append [ITM/OTM %] */
    CASE
      WHEN s.mny_pct IS NULL THEN
        COALESCE(
          s.rocket_rule,
          CASE
            WHEN s.rocket_score >= 5.0 THEN '🚀🚀🚀'
            WHEN s.rocket_score >= 3.5 THEN '🚀🚀'
            WHEN s.rocket_score >= 2.0 THEN '🚀'
            ELSE COALESCE(s.rocket, '')
          END
        )
      ELSE
        COALESCE(
          s.rocket_rule,
          CASE
            WHEN s.rocket_score >= 5.0 THEN '🚀🚀🚀'
            WHEN s.rocket_score >= 3.5 THEN '🚀🚀'
            WHEN s.rocket_score >= 2.0 THEN '🚀'
            ELSE COALESCE(s.rocket, '')
          END
        )
        || ' [' || CASE WHEN s.mny_pct >= 0 THEN 'ITM ' ELSE 'OTM ' END
        || TRIM(TO_CHAR(ABS(s.mny_pct)::numeric, 'FM999990.0')) || '%]'
    END AS Rocket_with_mny
  FROM scored s
)

/* ==================== FINAL SELECT ==================== */
SELECT
  (fe.flow_ts_local)::date AS "CreatedDate",
  TO_CHAR(fe.flow_ts_local, 'HH12:MI:SS AM') AS "CreatedTime",

  /* Symbol line with session + catalyst + your badges */
  (
    CASE
      WHEN fe.direction = 'BULL' THEN '(' || fe.dir_count || '🟩)  '
      WHEN fe.direction = 'BEAR' THEN '(' || fe.dir_count || '🟥)  '
      ELSE ''
    END
    || fe."Symbol"
    || ' · ' || COALESCE(fe.session_bucket,'?')
    || CASE WHEN fe.near_alert_type IS NOT NULL THEN ' ⚡' ELSE '' END
    || CASE WHEN (fe.badge_round || fe.badge_more) <> '' THEN ' ' || fe.badge_round || fe.badge_more ELSE '' END
    || fe.flash
    || CASE
         WHEN fe.premium_num > 1000000 THEN ' 🔥'
         WHEN fe.premium_num >  500000 THEN ' 💵'
         WHEN fe.premium_num >  100000 THEN ' '
         ELSE ''
       END
  ) AS "Symbol",

  r2.Rocket_with_mny AS "Rocket",

  /* NetPremium and Premium abbreviations as before */
  CASE
    WHEN ABS(fe.bull_total - fe.bear_total) >= 1e6
      THEN (CASE WHEN fe.bull_total - fe.bear_total < 0 THEN '-' ELSE '' END)
           || TRIM(TO_CHAR(ABS(fe.bull_total - fe.bear_total)/1e6, 'FM999990.00')) || ' M'
    WHEN ABS(fe.bull_total - fe.bear_total) >= 1e3
      THEN (CASE WHEN fe.bull_total - fe.bear_total < 0 THEN '-' ELSE '' END)
           || TRIM(TO_CHAR(ROUND(ABS(fe.bull_total - fe.bear_total)/1e3)::numeric, 'FM999990')) || ' K'
    ELSE TRIM(TO_CHAR(ROUND(fe.bull_total - fe.bear_total)::numeric, 'FM999990'))
  END AS "NetPremium",

  CASE
    WHEN fe.premium_num IS NULL THEN NULL
    WHEN ABS(fe.premium_num) >= 1e6
      THEN (CASE WHEN fe.premium_num < 0 THEN '-' ELSE '' END)
           || TRIM(TO_CHAR(ABS(fe.premium_num)/1e6, 'FM999990.00')) || ' M'
    WHEN ABS(fe.premium_num) >= 1e3
      THEN (CASE WHEN fe.premium_num < 0 THEN '-' ELSE '' END)
           || TRIM(TO_CHAR(ROUND(ABS(fe.premium_num)/1e3)::numeric, 'FM999990')) || ' K'
    ELSE TRIM(TO_CHAR(ROUND(fe.premium_num)::numeric, 'FM999990'))
  END AS "Premium",

  fe."ExpirationDate", fe."Price", fe."CallPut", fe."Side", fe."Strike", fe."Spot", fe."Volume", fe."OI",

  /* DTE and unknown-side tracking */
  fe.dte AS "DTE",
  fe.prem_unknown_sym AS "PremUnknownSym",
  fe.cnt_unknown_sym AS "CntUnknownSym",

  /* trader-facing context — now meaningful */
  fe.pct_vs_prior_close AS "PctVsPriorClose",
  fe.pct_vs_rth_open    AS "PctVsRthOpen",
  fe.pct_5m_momo        AS "Pct5m",
  fe.pct_15m_momo       AS "Pct15m",

  /* Tape alignment arrows with tolerance */
  CASE
    WHEN fe.tape_alignment=1 AND fe.direction='BULL' THEN '↗︎'
    WHEN fe.tape_alignment=1 AND fe.direction='BEAR' THEN '↘︎'
    ELSE ''
  END AS "TapeAlign",

  fe.near_alert_type    AS "NearAlert"

FROM rocket2 r2
JOIN flow_enriched fe ON fe.rid = r2.rid
WHERE
  fe.premium_num > 80000
  AND ( fe.badge_round IN ('🟢','🔴')
        AND POSITION('💎' IN fe.badge_more)>0
        AND POSITION('⭐' IN fe.badge_more)>0 )
  AND (
        (fe.direction='BULL' AND fe.badge_round='🟢' AND (fe.bull_total - fe.bear_total) > 0)
     OR (fe.direction='BEAR' AND fe.badge_round='🔴' AND (fe.bull_total - fe.bear_total) < 0)
      )
ORDER BY fe.flow_ts_utc DESC, fe.rid DESC;