labeler(009): market context + thin-name RVOL fallback (feature population)

Steps 1-2 of plan 009 — fill two fillable feature gaps so a conversion model
gets a fair, instrumented test:

- market context: _spy_context computes spy_intraday_return (SPY RTH open ->
  trigger-minute return) + a coarse tape_regime {trend_up,trend_down,chop} from
  SPY's sign + higher-highs structure. Replaces the hardcoded
  tape_regime=None/spy_intraday_return=None; wired into ALL five build paths
  (v1/vr1/run1/liq1/pm1) via _preload_spy. Null when SPY has no RTH bars at/before
  the trigger (pre-30d history, or a premarket fire before SPY's RTH open).
- thin-name RVOL fallback (v1): when avg_vol is 0/null (thin IPO-class, no 20d
  baseline) fall back to the median of prior sessions' total volume (real
  daily_bars.volume, knowable pre-open); flagged in new setup_events column
  relative_volume_source ('avg_vol_20d' | 'prior_days_median_volume' | NULL).
  Never fabricates a baseline.

A feature to TEST, not a signal to trust. No live gate/sizing/exit change.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

research/labeler.py

@@ -193,6 +193,81 @@ def _catalyst_at(cats: dict, symbol: str, setup_ts) -> tuple:
     return (best[1], round(fresh, 1))
 
 
+# ---------------------------------------------------------------------------
+# Market context (plan 009): the "is the whole tape with or against the trade"
+# feature. spy_intraday_return = SPY's RTH open -> setup-minute return; tape_regime
+# = a coarse {trend_up, trend_down, chop} from SPY's sign + higher-highs structure.
+# A FEATURE TO TEST, not a signal to trust — deliberately simple. Null (both) when
+# SPY has no RTH bars at/before the setup (pre-30d history, or a premarket setup
+# whose fire precedes SPY's RTH open — premarket is a separate regime, left null).
+# ---------------------------------------------------------------------------
+_TAPE_THRESH = 0.001   # +/-0.1% dead-band around flat
+
+
+def _spy_context(spy_rth, setup_ts) -> tuple:
+    """(spy_intraday_return, tape_regime) from SPY's RTH path up to setup_ts, or
+    (None, None). spy_rth = SPY regular-hours bars for the session (DataFrame with
+    timestamp/open/high/low/close), chronological."""
+    if spy_rth is None or len(spy_rth) == 0:
+        return (None, None)
+    up = spy_rth[spy_rth["timestamp"] <= setup_ts]
+    if len(up) == 0:
+        return (None, None)
+    open_px = float(up.iloc[0]["open"])
+    if open_px <= 0:
+        return (None, None)
+    ret = round((float(up.iloc[-1]["close"]) - open_px) / open_px, 5)
+    highs = up["high"].astype(float).to_numpy()
+    lows = up["low"].astype(float).to_numpy()
+    mid = len(up) // 2
+    hh = ll = False
+    if mid:   # >=2 bars: compare 2nd-half extremes to 1st-half (structure)
+        hh = bool(highs[mid:].max() > highs[:mid].max())
+        ll = bool(lows[mid:].min() < lows[:mid].min())
+    if ret >= _TAPE_THRESH and hh and not ll:
+        regime = "trend_up"
+    elif ret <= -_TAPE_THRESH and ll and not hh:
+        regime = "trend_down"
+    else:
+        regime = "chop"
+    return (ret, regime)
+
+
+def _preload_spy(con, symbol: str = "SPY") -> dict:
+    """{session_date: SPY RTH DataFrame} for market context. tape_regime keys off
+    SPY alone (QQQ is ingested for completeness/gap baselines, not used here)."""
+    rows = con.execute(
+        "SELECT session_date, timestamp, open, high, low, close FROM minute_bars "
+        "WHERE symbol=? AND is_regular_hours ORDER BY session_date, timestamp",
+        [symbol]).fetchall()
+    grouped: dict = {}
+    for d, ts, o, h, lo, c in rows:
+        grouped.setdefault(d, []).append((ts, o, h, lo, c))
+    return {d: pd.DataFrame(v, columns=["timestamp", "open", "high", "low", "close"])
+            for d, v in grouped.items()}
+
+
+def _preload_vol_baseline(con) -> dict:
+    """{(symbol, date): (fallback_volume, kind)} — a REAL prior-volume baseline for
+    thin IPO-class names whose rolling_avg_volume_20d is 0/null (documented blind
+    spot). Baseline = median of the symbol's available PRIOR-session total volumes
+    (grounded in real daily_bars.volume, knowable before the session opens); None
+    when the symbol has no prior day. Scoped to labeled symbols to match _preload_daily."""
+    rows = con.execute(
+        "SELECT symbol, trade_date, volume FROM daily_bars "
+        "WHERE symbol IN (SELECT DISTINCT symbol FROM minute_bars) "
+        "ORDER BY symbol, trade_date").fetchall()
+    out: dict = {}
+    prior: dict = {}
+    for sym, d, vol in rows:
+        pv = prior.get(sym)
+        if pv:
+            out[(sym, d)] = (_median(pv), "prior_days_median_volume")
+        if vol is not None:
+            prior.setdefault(sym, []).append(float(vol))
+    return out
+
+
 def _forward_labels(rth, setup_ts, entry_ref, invalidation, session_open) -> dict:
     """Forward-only outcome labels from a decision point. STRICT time split: uses
     only RTH bars strictly AFTER setup_ts; entry_ref/invalidation define R. Shared by
@@ -252,7 +327,8 @@ def _forward_labels(rth, setup_ts, entry_ref, invalidation, session_open) -> dic
                 time_to_max_upside_minutes=t_up, time_to_max_drawdown_minutes=t_dd)
 
 
-def build_one(con, symbol, session_date, daily: dict, cats: dict):
+def build_one(con, symbol, session_date, daily: dict, cats: dict,
+              spy: dict | None = None, vol_fb: dict | None = None):
     """Fetch one symbol-session's bars + baselines, then compute. None if empty."""
     rows = con.execute(
         f"SELECT {', '.join(_MB_COLS)} FROM minute_bars "
@@ -262,13 +338,21 @@ def build_one(con, symbol, session_date, daily: dict, cats: dict):
         return None
     df = pd.DataFrame(rows, columns=_MB_COLS)
     prior_close, avg_vol = daily.get((symbol, session_date), (None, None))
-    return compute_setup(symbol, session_date, df, prior_close, avg_vol, cats)
+    spy_session = (spy or {}).get(session_date)
+    vol_fallback = (vol_fb or {}).get((symbol, session_date))
+    return compute_setup(symbol, session_date, df, prior_close, avg_vol, cats,
+                         spy_session=spy_session, vol_fallback=vol_fallback)
 
 
-def compute_setup(symbol, session_date, df, prior_close, avg_vol, cats: dict):
+def compute_setup(symbol, session_date, df, prior_close, avg_vol, cats: dict,
+                  *, spy_session=None, vol_fallback=None):
     """PURE (no DB): (setup, features, labels) for one session's bars, or None.
     Strict time split — features use bars at/before the decision point, labels
-    only after. Testable on synthetic bars."""
+    only after. Testable on synthetic bars.
+
+    spy_session: SPY RTH bars for this session (market context; see _spy_context).
+    vol_fallback: (baseline_volume, kind) for the thin-name RVOL fallback when
+    avg_vol is 0/null — a real prior-volume baseline, never a fabricated one."""
     df = df.copy()
     df["is_regular_hours"] = df["is_regular_hours"].astype(bool)
     df["is_premarket"] = df["is_premarket"].astype(bool)
@@ -299,9 +383,17 @@ def compute_setup(symbol, session_date, df, prior_close, avg_vol, cats: dict):
     minutes_elapsed = max(1, int((df["is_regular_hours"] & (df["timestamp"] <= setup_ts)).sum()))
     rvol = (vol_upto / avg_vol) if avg_vol else None
     tod_rvol = calculate_time_of_day_rvol(vol_upto, avg_vol, minutes_elapsed) if avg_vol else None
+    # thin-name RVOL fallback (plan 009): thin IPO-class names have avg_vol=0/null
+    # (no 20d history), so rvol was always None for them. Fall back to a REAL prior-
+    # volume baseline (median of prior sessions' total volume) and flag which was used.
+    rvol_source = "avg_vol_20d" if avg_vol else None
+    if not avg_vol and vol_fallback and vol_fallback[0]:
+        rvol = vol_upto / float(vol_fallback[0])
+        rvol_source = vol_fallback[1]
     dq = calculate_data_quality_score(upto)
     dist_vwap = ((last_close - vwap) / vwap) if vwap else None
     cat_type, cat_fresh = _catalyst_at(cats, symbol, setup_ts)
+    spy_ret, tape = _spy_context(spy_session, setup_ts)
     above_vwap = bool(vwap is not None and last_close >= vwap)
 
     # ---- LABELS: only bars strictly AFTER the decision point (shared helper) --
@@ -313,13 +405,15 @@ def compute_setup(symbol, session_date, df, prior_close, avg_vol, cats: dict):
                  entry_reference_price=entry_ref, invalidation_price=invalidation,
                  target_r_multiple=TARGET_R, above_vwap_flag=above_vwap,
                  vwap_at_trigger=vwap, gap_pct=gap_pct, relative_volume=rvol,
-                 catalyst_freshness_at_trigger=cat_fresh, session_minute_number=minutes_elapsed)
+                 relative_volume_source=rvol_source,
+                 catalyst_freshness_at_trigger=cat_fresh, session_minute_number=minutes_elapsed,
+                 tape_regime_at_trigger=tape, spy_intraday_return_pct_at_trigger=spy_ret)
     feat = dict(id=sid, timestamp=setup_ts, symbol=symbol, session_date=session_date,
                 feature_version=FEATURE_VERSION, gap_pct=gap_pct, premarket_gap_pct=pm_gap,
                 relative_volume=rvol, time_of_day_adjusted_relative_volume=tod_rvol,
                 float_rotation_pct=None, vwap=vwap, distance_from_vwap=dist_vwap,
                 ema9=levels.ema_9, ema20=levels.ema_20, catalyst_freshness_minutes=cat_fresh,
-                catalyst_type=cat_type, tape_regime=None, spy_intraday_return=None,
+                catalyst_type=cat_type, tape_regime=tape, spy_intraday_return=spy_ret,
                 vix_level=None,
                 metadata_json=json.dumps({"dq_score": round(dq.score, 3), "dq_grade": dq.grade,
                                           "bar_count_at_setup": len(upto), "R": round(R, 5)}))
@@ -340,13 +434,15 @@ def build(con, limit=None, rebuild=False):
         con.execute("DELETE FROM setup_events WHERE setup_version=?", [SETUP_VERSION])
     daily = _preload_daily(con)
     cats = _preload_catalysts(con)
+    spy = _preload_spy(con)
+    vol_fb = _preload_vol_baseline(con)
     sessions = con.execute(
         "SELECT DISTINCT symbol, session_date FROM minute_bars ORDER BY session_date, symbol"
         + (f" LIMIT {int(limit)}" if limit else "")).fetchall()
     built = skipped = 0
     for sym, d in sessions:
         try:
-            res = build_one(con, sym, d, daily, cats)
+            res = build_one(con, sym, d, daily, cats, spy=spy, vol_fb=vol_fb)
         except Exception as exc:  # noqa: BLE001 — one bad session can't stop the build
             print(f"  ! {sym} {d}: {exc}")
             res = None
@@ -371,7 +467,7 @@ def build(con, limit=None, rebuild=False):
 # ----------------------------------------------------------------------------
 
 def compute_vwap_reclaim_setups(symbol, session_date, df, prior_close, avg_vol,
-                                cats: dict, *, cooldown=10) -> list:
+                                cats: dict, *, cooldown=10, spy_session=None) -> list:
     """0..N shadow vwap_reclaim setups for one session. Slides detect_vwap_reclaim
     over the RTH bars and emits on each RISING EDGE (not-valid -> valid) with a bar
     cooldown so one curl episode is one setup. Each emit carries the detector's OWN
@@ -408,6 +504,7 @@ def compute_vwap_reclaim_setups(symbol, session_date, df, prior_close, avg_vol,
         sv = sig.signal_values
         vwap_at = round(entry_ref - float(sv.get("dist_from_vwap") or 0.0), 5)
         _, cat_fresh = _catalyst_at(cats, symbol, setup_ts)
+        spy_ret, tape = _spy_context(spy_session, setup_ts)
         # key on the RTH bar INDEX (unique per session) -- NOT the wall-clock minute:
         # some sessions have sub-minute/irregular timestamps, so two fires a cooldown
         # apart can share an HH:MM and collide on the primary key.
@@ -419,7 +516,8 @@ def compute_vwap_reclaim_setups(symbol, session_date, df, prior_close, avg_vol,
                      impulse_pct=sv.get("run_pct"), pullback_low=invalidation,
                      pullback_depth_pct=sv.get("pullback_held_frac"),
                      above_vwap_flag=True, vwap_at_trigger=vwap_at,
-                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=i + 1)
+                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=i + 1,
+                     tape_regime_at_trigger=tape, spy_intraday_return_pct_at_trigger=spy_ret)
         lab = _forward_labels(rth, setup_ts, entry_ref, invalidation, session_open)
         out.append((setup, dict(setup_id=sid, label_version=LABEL_VERSION, **lab)))
     return out
@@ -432,6 +530,7 @@ def build_vwap_reclaim(con, limit=None, rebuild=False, cooldown=10):
         con.execute("DELETE FROM setup_events WHERE setup_version=?", [VR_SETUP_VERSION])
     daily = _preload_daily(con)
     cats = _preload_catalysts(con)
+    spy = _preload_spy(con)
     sessions = con.execute(
         "SELECT DISTINCT symbol, session_date FROM minute_bars ORDER BY session_date, symbol"
         + (f" LIMIT {int(limit)}" if limit else "")).fetchall()
@@ -445,7 +544,8 @@ def build_vwap_reclaim(con, limit=None, rebuild=False, cooldown=10):
         dfb = pd.DataFrame(rows, columns=_MB_COLS)
         pc, av = daily.get((sym, d), (None, None))
         try:
-            setups = compute_vwap_reclaim_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown)
+            setups = compute_vwap_reclaim_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown,
+                                                 spy_session=spy.get(d))
         except Exception as exc:  # noqa: BLE001 — one bad session can't stop the build
             print(f"  ! {sym} {d}: {exc}")
             continue
@@ -561,7 +661,7 @@ def validate_vwap_reclaim(con, min_r_frac=0.015, min_r_abs=0.02):
 # ----------------------------------------------------------------------------
 
 def compute_runner_setups(symbol, session_date, df, prior_close, avg_vol,
-                          cats: dict, *, cooldown=10) -> list:
+                          cats: dict, *, cooldown=10, spy_session=None) -> list:
     """0..N shadow leading-gainer-runner setups for one session. Slides
     detect_leading_gainer_runner over the session (PM-INCLUSIVE: the detector at RTH
     bar k sees pm+rth[:k+1], so run%/VWAP are full-session), emits on each RISING EDGE
@@ -613,6 +713,7 @@ def compute_runner_setups(symbol, session_date, df, prior_close, avg_vol,
         # signal_values carry numpy scalars; psycopg2 can't adapt np.float64 -> cast native
         nf = lambda key: (float(sv[key]) if sv.get(key) is not None else None)
         _, cat_fresh = _catalyst_at(cats, symbol, setup_ts)
+        spy_ret, tape = _spy_context(spy_session, setup_ts)
         sid = f"{symbol}:{session_date}:b{k}:{RUNNER_SETUP_VERSION}"
         setup = dict(setup_id=sid, symbol=symbol, setup_time=setup_ts, session_date=session_date,
                      setup_name=RUNNER_NAME, setup_version=RUNNER_SETUP_VERSION,
@@ -621,7 +722,8 @@ def compute_runner_setups(symbol, session_date, df, prior_close, avg_vol,
                      gap_pct=nf("total_run"), relative_volume=nf("volume_burst_ratio"),
                      above_vwap_flag=True, vwap_at_trigger=nf("session_vwap"),
                      pullback_depth_pct=nf("pm_capture"),
-                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=j + 1)
+                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=j + 1,
+                     tape_regime_at_trigger=tape, spy_intraday_return_pct_at_trigger=spy_ret)
         lab = _forward_labels(rth, setup_ts, entry_ref, invalidation, session_open)
         out.append((setup, dict(setup_id=sid, label_version=LABEL_VERSION, **lab)))
     return out
@@ -634,6 +736,7 @@ def build_runner(con, limit=None, rebuild=False, cooldown=10):
         con.execute("DELETE FROM setup_events WHERE setup_version=?", [RUNNER_SETUP_VERSION])
     daily = _preload_daily(con)
     cats = _preload_catalysts(con)
+    spy = _preload_spy(con)
     sessions = con.execute(
         "SELECT DISTINCT symbol, session_date FROM minute_bars ORDER BY session_date, symbol"
         + (f" LIMIT {int(limit)}" if limit else "")).fetchall()
@@ -647,7 +750,8 @@ def build_runner(con, limit=None, rebuild=False, cooldown=10):
         dfb = pd.DataFrame(rows, columns=_MB_COLS)
         pc, av = daily.get((sym, d), (None, None))
         try:
-            setups = compute_runner_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown)
+            setups = compute_runner_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown,
+                                           spy_session=spy.get(d))
         except Exception as exc:  # noqa: BLE001 — one bad session can't stop the build
             print(f"  ! {sym} {d}: {exc}")
             continue
@@ -814,7 +918,7 @@ def validate_runner(con):
 # ----------------------------------------------------------------------------
 
 def compute_liquid_setups(symbol, session_date, df, prior_close, avg_vol,
-                          cats: dict, *, cooldown=10) -> list:
+                          cats: dict, *, cooldown=10, spy_session=None) -> list:
     """0..N shadow liquid_intraday_breakout setups for one session. Slides the detector over
     RTH bars (RTH-cumulative VWAP; these are intraday breakouts, not PM gappers), emits on each
     RISING EDGE with a cooldown. Stores rvol + cumulative dollar-volume as STRATIFICATION
@@ -848,6 +952,7 @@ def compute_liquid_setups(symbol, session_date, df, prior_close, avg_vol,
         sv = sig.signal_values
         nf = lambda key: (float(sv[key]) if sv.get(key) is not None else None)
         _, cat_fresh = _catalyst_at(cats, symbol, setup_ts)
+        spy_ret, tape = _spy_context(spy_session, setup_ts)
         sid = f"{symbol}:{session_date}:b{i}:{LIQUID_SETUP_VERSION}"
         setup = dict(setup_id=sid, symbol=symbol, setup_time=setup_ts, session_date=session_date,
                      setup_name=LIQUID_NAME, setup_version=LIQUID_SETUP_VERSION,
@@ -855,7 +960,8 @@ def compute_liquid_setups(symbol, session_date, df, prior_close, avg_vol,
                      target_r_multiple=0.0, relative_volume=nf("rvol"),
                      pullback_volume_ratio=nf("cum_dollar_vol"),   # cumulative $-volume (liquidity)
                      above_vwap_flag=True, vwap_at_trigger=nf("session_vwap"),
-                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=i + 1)
+                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=i + 1,
+                     tape_regime_at_trigger=tape, spy_intraday_return_pct_at_trigger=spy_ret)
         lab = _forward_labels(rth, setup_ts, entry_ref, invalidation, session_open)
         out.append((setup, dict(setup_id=sid, label_version=LABEL_VERSION, **lab)))
     return out
@@ -868,6 +974,7 @@ def build_liquid(con, limit=None, rebuild=False, cooldown=10):
         con.execute("DELETE FROM setup_events WHERE setup_version=?", [LIQUID_SETUP_VERSION])
     daily = _preload_daily(con)
     cats = _preload_catalysts(con)
+    spy = _preload_spy(con)
     sessions = con.execute(
         "SELECT DISTINCT symbol, session_date FROM minute_bars ORDER BY session_date, symbol"
         + (f" LIMIT {int(limit)}" if limit else "")).fetchall()
@@ -881,7 +988,8 @@ def build_liquid(con, limit=None, rebuild=False, cooldown=10):
         dfb = pd.DataFrame(rows, columns=_MB_COLS)
         pc, av = daily.get((sym, d), (None, None))
         try:
-            setups = compute_liquid_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown)
+            setups = compute_liquid_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown,
+                                           spy_session=spy.get(d))
         except Exception as exc:  # noqa: BLE001 — one bad session can't stop the build
             print(f"  ! {sym} {d}: {exc}")
             continue
@@ -996,7 +1104,7 @@ def validate_liquid(con):
 # ----------------------------------------------------------------------------
 
 def compute_premarket_setups(symbol, session_date, df, prior_close, avg_vol,
-                             cats: dict, *, cooldown=10) -> list:
+                             cats: dict, *, cooldown=10, spy_session=None) -> list:
     """0..N shadow premarket_setup fires for one session, on PREMARKET bars only. Forward labels
     use the SAME _forward_labels but over the premarket bars after the fire (outcome is measured
     pre-9:30, where the trader enters+exits). Stores gap + premarket volume as stratification. PURE."""
@@ -1031,6 +1139,9 @@ def compute_premarket_setups(symbol, session_date, df, prior_close, avg_vol,
         sv = sig.signal_values
         nf = lambda key: (float(sv[key]) if sv.get(key) is not None else None)
         _, cat_fresh = _catalyst_at(cats, symbol, setup_ts)
+        # market context uses SPY's RTH open->trigger path; a premarket fire precedes
+        # SPY's RTH open, so this is null for pm1 (premarket is a separate regime).
+        spy_ret, tape = _spy_context(spy_session, setup_ts)
         sid = f"{symbol}:{session_date}:b{i}:{PREMARKET_SETUP_VERSION}"
         setup = dict(setup_id=sid, symbol=symbol, setup_time=setup_ts, session_date=session_date,
                      setup_name=PREMARKET_NAME, setup_version=PREMARKET_SETUP_VERSION,
@@ -1038,7 +1149,8 @@ def compute_premarket_setups(symbol, session_date, df, prior_close, avg_vol,
                      target_r_multiple=0.0, gap_pct=nf("gap"), impulse_pct=nf("velocity"),
                      relative_volume=nf("cum_pm_volume"), above_vwap_flag=True,
                      vwap_at_trigger=nf("pm_vwap"),
-                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=i + 1)
+                     catalyst_freshness_at_trigger=cat_fresh, session_minute_number=i + 1,
+                     tape_regime_at_trigger=tape, spy_intraday_return_pct_at_trigger=spy_ret)
         lab = _forward_labels(pm, setup_ts, entry_ref, invalidation, pm_open)
         out.append((setup, dict(setup_id=sid, label_version=LABEL_VERSION, **lab)))
     return out
@@ -1051,6 +1163,7 @@ def build_premarket(con, limit=None, rebuild=False, cooldown=10):
         con.execute("DELETE FROM setup_events WHERE setup_version=?", [PREMARKET_SETUP_VERSION])
     daily = _preload_daily(con)
     cats = _preload_catalysts(con)
+    spy = _preload_spy(con)
     sessions = con.execute(
         "SELECT DISTINCT symbol, session_date FROM minute_bars ORDER BY session_date, symbol"
         + (f" LIMIT {int(limit)}" if limit else "")).fetchall()
@@ -1064,7 +1177,8 @@ def build_premarket(con, limit=None, rebuild=False, cooldown=10):
         dfb = pd.DataFrame(rows, columns=_MB_COLS)
         pc, av = daily.get((sym, d), (None, None))
         try:
-            setups = compute_premarket_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown)
+            setups = compute_premarket_setups(sym, d, dfb, pc, av, cats, cooldown=cooldown,
+                                              spy_session=spy.get(d))
         except Exception as exc:  # noqa: BLE001
             print(f"  ! {sym} {d}: {exc}")
             continue

storage/pg_schema.sql

@@ -193,7 +193,8 @@ CREATE TABLE IF NOT EXISTS setup_events (
     catalyst_freshness_at_trigger DOUBLE PRECISION,
     session_minute_number INTEGER,
     tape_regime_at_trigger VARCHAR,
-    spy_intraday_return_pct_at_trigger DOUBLE PRECISION
+    spy_intraday_return_pct_at_trigger DOUBLE PRECISION,
+    relative_volume_source VARCHAR
 );
 
 CREATE TABLE IF NOT EXISTS outcome_labels (
@@ -334,3 +335,6 @@ CREATE TABLE IF NOT EXISTS scanner_snapshots (
 -- =====================================================================
 ALTER TABLE setup_events ADD COLUMN IF NOT EXISTS tape_regime_at_trigger VARCHAR;
 ALTER TABLE setup_events ADD COLUMN IF NOT EXISTS spy_intraday_return_pct_at_trigger DOUBLE PRECISION;
+-- which volume baseline produced relative_volume (plan 009 thin-name RVOL fallback):
+-- 'avg_vol_20d' (normal), 'prior_days_median_volume' (thin IPO fallback), or NULL.
+ALTER TABLE setup_events ADD COLUMN IF NOT EXISTS relative_volume_source VARCHAR;

tests/unit/research/test_labeler.py

@@ -4,7 +4,8 @@ from datetime import date, datetime
 
 import pandas as pd
 
-from research.labeler import compute_setup, compute_vwap_reclaim_setups, _wilson_lower
+from research.labeler import (
+    compute_setup, compute_vwap_reclaim_setups, _wilson_lower, _spy_context)
 
 COLS = ["timestamp", "session_date", "is_premarket", "is_regular_hours",
         "is_afterhours", "open", "high", "low", "close", "volume", "vwap"]
@@ -108,6 +109,84 @@ def test_vwap_reclaim_silent_on_one_way_fade():
                                        avg_vol=50_000, cats={}) == []
 
 
+# --- market context (plan 009) --------------------------------------------
+
+def _spy(seq):
+    """seq = list of (minute_offset, open, high, low, close) -> SPY RTH frame
+    with exactly the columns _spy_context reads."""
+    rows = []
+    for m, o, h, lo, c in seq:
+        ts = datetime(2026, 6, 1, 9, 30) + pd.Timedelta(minutes=m)
+        rows.append([ts, o, h, lo, c])
+    return pd.DataFrame(rows, columns=["timestamp", "open", "high", "low", "close"])
+
+
+def _setup_ts():
+    # compute_setup's decision point = the 5th RTH bar (minute 4 = 09:34)
+    return datetime(2026, 6, 1, 9, 34)
+
+
+def test_market_context_trend_up():
+    up = _spy([(m, 400 + m, 400.5 + m, 399.8 + m, 400.3 + m) for m in range(6)])
+    ret, regime = _spy_context(up, _setup_ts())
+    assert ret > 0 and regime == "trend_up"
+
+
+def test_market_context_trend_down():
+    down = _spy([(m, 405 - m, 405.2 - m, 404.5 - m, 404.7 - m) for m in range(6)])
+    ret, regime = _spy_context(down, _setup_ts())
+    assert ret < 0 and regime == "trend_down"
+
+
+def test_market_context_chop_is_not_just_sign():
+    # net up a hair, but oscillating (2nd half sets both a higher high AND a lower
+    # low) -> chop, proving regime uses structure, not only the return sign.
+    chop = _spy([(0, 400, 401, 399, 400.0), (1, 400, 401, 399, 400.2),
+                 (2, 400.2, 400.5, 399.5, 400.0), (3, 400, 402, 398, 400.1),
+                 (4, 400.1, 401.5, 398.5, 400.3)])
+    ret, regime = _spy_context(chop, _setup_ts())
+    assert regime == "chop"
+    # no SPY bars at/before the trigger -> null (pre-30d history / premarket fire)
+    assert _spy_context(_spy([(20, 400, 401, 399, 400)]), _setup_ts()) == (None, None)
+    assert _spy_context(None, _setup_ts()) == (None, None)
+
+
+def test_market_context_lands_on_setup_event():
+    up = _spy([(m, 400 + m, 400.5 + m, 399.8 + m, 400.3 + m) for m in range(6)])
+    setup, feat, _ = compute_setup("RUN", D, _runner_bars(), prior_close=0.80,
+                                   avg_vol=50_000, cats={}, spy_session=up)
+    assert setup["tape_regime_at_trigger"] == "trend_up"
+    assert setup["spy_intraday_return_pct_at_trigger"] > 0
+    # also mirrored into engineered_features (the old hardcoded None is gone)
+    assert feat["tape_regime"] == "trend_up" and feat["spy_intraday_return"] > 0
+
+
+# --- thin-name RVOL fallback (plan 009) -----------------------------------
+
+def test_rvol_fallback_thin_name():
+    # avg_vol=0 (thin IPO, no 20d baseline) + a real prior-volume baseline ->
+    # non-null relative_volume, flagged as the fallback source.
+    setup, feat, _ = compute_setup("THIN", D, _runner_bars(), prior_close=0.80,
+                                   avg_vol=0, cats={},
+                                   vol_fallback=(100_000.0, "prior_days_median_volume"))
+    assert setup["relative_volume"] is not None
+    assert setup["relative_volume"] == feat["relative_volume"]
+    assert setup["relative_volume_source"] == "prior_days_median_volume"
+
+
+def test_rvol_source_flags_normal_and_absent_baseline():
+    # normal path (avg_vol present) is flagged avg_vol_20d
+    s_norm, _, _ = compute_setup("OK", D, _runner_bars(), prior_close=0.80,
+                                 avg_vol=50_000, cats={})
+    assert s_norm["relative_volume"] is not None
+    assert s_norm["relative_volume_source"] == "avg_vol_20d"
+    # thin name with NO baseline available -> stays null, never fabricated
+    s_none, _, _ = compute_setup("NOBASE", D, _runner_bars(), prior_close=0.80,
+                                 avg_vol=0, cats={}, vol_fallback=None)
+    assert s_none["relative_volume"] is None
+    assert s_none["relative_volume_source"] is None
+
+
 def test_wilson_lower_bound():
     # the honest small-sample lower bound: below the point estimate, 0 at n=0,
     # monotonic in successes, and tighter (higher) as n grows at a fixed rate.