Revert "feat: Phase B fip_id liquid-breadth research tooling"

This reverts commit 9704e0d85a.

app/services/backtest_service.py

@@ -30,11 +30,6 @@ Environment variables (see also run_backtest_snapshot.py):
     BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1
     BACKTEST_RESEARCH_EXITS=1
     BACKTEST_MIN_RR_SWEEP=1
-
-  Broad-universe signal research (local snapshots only; inert when unset):
-    BACKTEST_LIQUID_BREADTH=1500   # PIT top-N by 63d median $vol, price floor
-    BACKTEST_LIQUID_MIN_PRICE=5    # USD close floor at as-of (default 5)
-    BACKTEST_SIGNAL_EVAL_ONLY=1   # skip portfolio_sim / monitor (signal IC only)
 """
 
 from __future__ import annotations
@@ -881,63 +876,6 @@ def _signal_values(
     return out
 
 
-def _liquid_breadth_top_n() -> int:
-    """0 = off (production path). N > 0 enables PIT top-N $vol mask for signal IC."""
-    raw = os.getenv("BACKTEST_LIQUID_BREADTH", "").strip()
-    if not raw:
-        return 0
-    try:
-        return max(0, int(raw))
-    except ValueError:
-        return 0
-
-
-def _liquid_min_price() -> float:
-    raw = os.getenv("BACKTEST_LIQUID_MIN_PRICE", "5").strip() or "5"
-    try:
-        return max(0.0, float(raw))
-    except ValueError:
-        return 5.0
-
-
-def _signal_eval_only() -> bool:
-    return os.getenv("BACKTEST_SIGNAL_EVAL_ONLY", "").strip() in ("1", "true", "yes")
-
-
-async def _load_research_rank_only_symbols(db: AsyncSession) -> set[str]:
-    """Symbols that feed signal IC only (no GTL/candidate replay).
-
-    Optional side table ``research_rank_only`` on research snapshots. Missing
-    table → empty set (production path unchanged).
-    """
-    from sqlalchemy import text
-
-    try:
-        result = await db.execute(text("SELECT symbol FROM research_rank_only"))
-        return {str(row[0]).upper() for row in result.fetchall() if row[0]}
-    except Exception:
-        return set()
-
-
-def _median_dollar_vol_63(
-    closes: list[float], volumes: list[float], i: int, lookback: int = 63
-) -> float | None:
-    """Rolling median of close×volume over ``lookback`` bars ending at ``i`` (inclusive)."""
-    if i + 1 < lookback or lookback < 2:
-        return None
-    dvs: list[float] = []
-    for k in range(i - lookback + 1, i + 1):
-        if closes[k] > 0 and volumes[k] >= 0:
-            dvs.append(closes[k] * float(volumes[k]))
-    if len(dvs) < max(20, lookback // 2):
-        return None
-    dvs_sorted = sorted(dvs)
-    mid = len(dvs_sorted) // 2
-    if len(dvs_sorted) % 2:
-        return dvs_sorted[mid]
-    return 0.5 * (dvs_sorted[mid - 1] + dvs_sorted[mid])
-
-
 def _accumulate_signal_series(
     records: list,
     collected: dict,
@@ -945,20 +883,13 @@ def _accumulate_signal_series(
 ) -> None:
     """For each weekly as-of bar, emit (signal, forward-return) pairs keyed by ISO
     week into ``collected[name][week_key]``. Forward return is close-to-close over
-    HORIZON trading days. Mutates ``collected`` (a dict of dict of list).
-
-    When ``BACKTEST_LIQUID_BREADTH`` is set, observations are dicts with PIT
-    liquidity fields for the mask; otherwise plain ``(val, fwd)`` tuples so the
-    production signal path stays unchanged.
-    """
+    HORIZON trading days. Mutates ``collected`` (a dict of dict of list)."""
     n = len(records)
     if n < HORIZON + 21:
         return
     closes = [float(r.close) for r in records]
     highs = [float(r.high) for r in records]
-    volumes = [float(getattr(r, "volume", 0) or 0) for r in records]
     dates = [r.date for r in records]
-    liquid_mode = _liquid_breadth_top_n() > 0
     for i in _weekly_asof_indices(records):
         j = i + HORIZON
         if j >= n or closes[i] <= 0:
@@ -966,17 +897,8 @@ def _accumulate_signal_series(
         fwd = closes[j] / closes[i] - 1.0
         iso = records[i].date.isocalendar()
         week_key = (iso[0], iso[1])
-        dvol = _median_dollar_vol_63(closes, volumes, i) if liquid_mode else None
         for name, val in _signal_values(dates, closes, highs, i, benchmark_closes).items():
-            if liquid_mode:
-                collected[name][week_key].append({
-                    "val": val,
-                    "fwd": fwd,
-                    "close": closes[i],
-                    "median_dvol_63": dvol,
-                })
-            else:
-                collected[name][week_key].append((val, fwd))
+            collected[name][week_key].append((val, fwd))
 
 
 def _rank(xs: list[float]) -> list[float]:
@@ -1015,54 +937,6 @@ def _spearman(xs: list[float], ys: list[float]) -> float | None:
     return _pearson(_rank(xs), _rank(ys))
 
 
-def _obs_val_fwd(rec: object) -> tuple[float, float] | None:
-    """Unpack a signal observation: ``(val, fwd)`` or research dict form."""
-    if isinstance(rec, dict):
-        try:
-            return float(rec["val"]), float(rec["fwd"])
-        except (KeyError, TypeError, ValueError):
-            return None
-    if isinstance(rec, (tuple, list)) and len(rec) >= 2:
-        try:
-            return float(rec[0]), float(rec[1])
-        except (TypeError, ValueError):
-            return None
-    return None
-
-
-def _filter_liquid_breadth_week(
-    recs: list,
-    *,
-    top_n: int,
-    min_price: float,
-) -> list[tuple[float, float]]:
-    """Point-in-time top-N by median $vol among names with price ≥ floor.
-
-    Ranking is relative (IEX volume undercount is OK for order stats). Membership
-    is recomputed every week from as-of bars — never frozen from today's liquidity.
-    """
-    ranked: list[tuple[float, float, float]] = []  # (-dvol, val, fwd)
-    for rec in recs:
-        if not isinstance(rec, dict):
-            pair = _obs_val_fwd(rec)
-            if pair is not None:
-                ranked.append((0.0, pair[0], pair[1]))
-            continue
-        close = rec.get("close")
-        dvol = rec.get("median_dvol_63")
-        if close is None or float(close) < min_price:
-            continue
-        if dvol is None or float(dvol) <= 0:
-            continue
-        pair = _obs_val_fwd(rec)
-        if pair is None:
-            continue
-        ranked.append((-float(dvol), pair[0], pair[1]))
-    ranked.sort(key=lambda row: row[0])
-    kept = ranked[:top_n]
-    return [(val, fwd) for _, val, fwd in kept]
-
-
 def _quintile_spread(pairs: list[tuple[float, float]]) -> float | None:
     """Mean forward return of the top signal-quintile minus the bottom quintile."""
     n = len(pairs)
@@ -1108,16 +982,10 @@ def _signal_evaluation(collected: dict) -> list[dict]:
 
     IC is measured on NON-OVERLAPPING forward windows (weeks thinned to ~HORIZON
     apart) so the t-stat isn't inflated by autocorrelation. A signal with no edge
-    lands near IC 0 / score 0; one with too few independent windows is flagged
+    lands near IC 0 / spread 0; one with too few independent windows is flagged
     unreliable rather than trusted on a lucky handful.
-
-    When ``BACKTEST_LIQUID_BREADTH=N`` is set, each week's cross-section is first
-    restricted to the top-N names by point-in-time 63d median dollar volume
-    (price ≥ BACKTEST_LIQUID_MIN_PRICE). Production path (flag unset) is unchanged.
     """
     stride = max(1, round(HORIZON / 5))  # ISO weeks spanned by the forward window
-    top_n = _liquid_breadth_top_n()
-    min_price = _liquid_min_price()
     rows: list[dict] = []
     for name in sorted(collected):
         weeks_map = collected[name]
@@ -1128,25 +996,13 @@ def _signal_evaluation(collected: dict) -> list[dict]:
         sizes: list[int] = []
         for wk in kept:
             recs = weeks_map[wk]
-            if top_n > 0:
-                pairs = _filter_liquid_breadth_week(
-                    recs, top_n=top_n, min_price=min_price
-                )
-            else:
-                pairs = []
-                for rec in recs:
-                    pair = _obs_val_fwd(rec)
-                    if pair is not None:
-                        pairs.append(pair)
-            if len(pairs) < MIN_CROSS_SECTION:
-                continue
-            ic = _spearman([p[0] for p in pairs], [p[1] for p in pairs])
+            ic = _spearman([r[0] for r in recs], [r[1] for r in recs])
             if ic is not None:
                 ics.append(ic)
-            spread = _quintile_spread(pairs)
+            spread = _quintile_spread(recs)
             if spread is not None:
                 spreads.append(spread)
-            sizes.append(len(pairs))
+            sizes.append(len(recs))
         if not ics:
             continue
         mean_ic = sum(ics) / len(ics)
@@ -1155,7 +1011,7 @@ def _signal_evaluation(collected: dict) -> list[dict]:
         else:
             std = 0.0
         t_stat = mean_ic / std * math.sqrt(len(ics)) if std > 0 else None
-        row = {
+        rows.append({
             "signal": name,
             "weeks": len(ics),
             "avg_cross_section": round(sum(sizes) / len(sizes), 1) if sizes else None,
@@ -1164,11 +1020,7 @@ def _signal_evaluation(collected: dict) -> list[dict]:
             "ic_positive_pct": round(sum(1 for x in ics if x > 0) / len(ics) * 100, 1),
             "mean_quintile_spread": round(sum(spreads) / len(spreads), 4) if spreads else None,
             "reliable": len(ics) >= MIN_RELIABLE_PERIODS,
-        }
-        if top_n > 0:
-            row["liquid_breadth_top_n"] = top_n
-            row["liquid_min_price"] = min_price
-        rows.append(row)
+        })
     rows.sort(key=lambda r: r["mean_ic"], reverse=True)
     return rows
 
@@ -1189,15 +1041,10 @@ def _replay_and_signals(
     benchmark_closes: dict[date, float] | None = None,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
     cadence: str = DEFAULT_BACKTEST_CADENCE,
-    signal_only: bool = False,
 ) -> tuple[list[dict], dict]:
     """The CPU-bound per-ticker work, as a top-level (picklable) function so it can
     run in a worker process. Takes primitive column arrays (cheap to pickle),
-    rebuilds bar objects, and returns (candidates, signal_series).
-
-    ``signal_only=True`` (research rank-only names): skip GTL/candidate replay so
-    the production portfolio book is never polluted by broad-universe tickers.
-    """
+    rebuilds bar objects, and returns (candidates, signal_series)."""
     date_ords, opens, highs, lows, closes, volumes = columns
     bars = [
         SimpleNamespace(
@@ -1205,9 +1052,8 @@ def _replay_and_signals(
         )
         for o, op, hi, lo, cl, vo in zip(date_ords, opens, highs, lows, closes, volumes)
     ]
-    candidates: list[dict] = []
-    if not signal_only:
-        candidates = _replay_ticker(
+    return (
+        _replay_ticker(
             symbol,
             bars,
             config,
@@ -1215,9 +1061,7 @@ def _replay_and_signals(
             benchmark_closes,
             target_model,
             cadence,
-        )
-    return (
-        candidates,
+        ),
         _signal_series(bars, benchmark_closes),
     )
 
@@ -3945,12 +3789,6 @@ async def run_backtest(
     result = await db.execute(select(Ticker).order_by(Ticker.symbol))
     tickers = list(result.scalars().all())
     total = len(tickers)
-    rank_only_symbols = await _load_research_rank_only_symbols(db)
-    if rank_only_symbols:
-        logger.info(json.dumps({
-            "event": "backtest_rank_only_loaded",
-            "count": len(rank_only_symbols),
-        }))
 
     candidates: list[dict] = []
     # Signal IC remains a weekly, non-overlapping diagnostic regardless of the
@@ -4009,16 +3847,10 @@ async def run_backtest(
                         continue
                     if columns is not None:
                         futures.append(loop.run_in_executor(
-                            pool,
-                            _replay_and_signals,
-                            ticker.symbol,
-                            columns,
-                            config,
-                            activation,
+                            pool, _replay_and_signals, ticker.symbol, columns, config, activation,
                             benchmark_closes,
                             target_model,
                             cadence,
-                            ticker.symbol in rank_only_symbols,
                         ))
                 for result in await asyncio.gather(*futures, return_exceptions=True):
                     if isinstance(result, Exception):
@@ -4038,15 +3870,10 @@ async def run_backtest(
                 columns = await _fetch_columns(db, ticker.symbol)
                 if columns is not None:
                     _merge(await asyncio.to_thread(
-                        _replay_and_signals,
-                        ticker.symbol,
-                        columns,
-                        config,
-                        activation,
+                        _replay_and_signals, ticker.symbol, columns, config, activation,
                         benchmark_closes,
                         target_model,
                         cadence,
-                        ticker.symbol in rank_only_symbols,
                     ))
             except Exception:
                 logger.exception("Backtest replay failed for %s", ticker.symbol)
@@ -4089,75 +3916,73 @@ async def run_backtest(
     portfolio_monitor_report: dict | None = None
     holdout_report: dict | None = None
     min_rr_sweep_report: dict | None = None
-    if not _signal_eval_only():
+    try:
+        qual_symbols = sorted({
+            c["symbol"]
+            for c in candidates
+            if c.get("qualified")
+            or any(_qualifies_strategy_variant(c, cfg) for cfg in STRATEGY_VARIANTS)
+        })
+        price_columns: dict[str, tuple] = {}
+        for sym in qual_symbols:
+            cols = await _fetch_columns(db, sym)
+            if cols is not None:
+                price_columns[sym] = cols
+
+        spy_closes: dict | None = None
         try:
-            qual_symbols = sorted({
-                c["symbol"]
-                for c in candidates
-                if c.get("qualified")
-                or any(_qualifies_strategy_variant(c, cfg) for cfg in STRATEGY_VARIANTS)
-            })
-            price_columns: dict[str, tuple] = {}
-            for sym in qual_symbols:
-                cols = await _fetch_columns(db, sym)
-                if cols is not None:
-                    price_columns[sym] = cols
-
-            spy_closes: dict | None = None
-            try:
-                oldest = min((cols[0][0] for cols in price_columns.values()), default=None)
-                days_needed = None
-                if oldest is not None and not _offline_snapshot_mode():
-                    days_needed = (date.today() - date.fromordinal(oldest)).days + 30
-                spy_closes = await _load_benchmark_closes_for_backtest(
-                    db, days=days_needed, refresh=oldest is not None
-                )
-            except Exception:
-                logger.exception("Benchmark load for the portfolio sim failed")
-
-            for policy in ("target", "hold"):
-                sim = _simulate_portfolio(
-                    candidates, price_columns, spy_closes, policy, hold_horizon
-                )
-                if sim is not None:
-                    sim_policies.append({"policy": policy, **sim})
-            strategy_variant_rows = _strategy_variant_sims(
-                candidates, price_columns, spy_closes, hold_horizon
+            oldest = min((cols[0][0] for cols in price_columns.values()), default=None)
+            days_needed = None
+            if oldest is not None and not _offline_snapshot_mode():
+                days_needed = (date.today() - date.fromordinal(oldest)).days + 30
+            spy_closes = await _load_benchmark_closes_for_backtest(
+                db, days=days_needed, refresh=oldest is not None
             )
-            exit_policy_rows = _exit_policy_sims(
-                candidates, price_columns, spy_closes, hold_horizon
-            )
-            live_exit_policy: dict | None = None
-            try:
-                from app.services.paper_trade_service import get_exit_policy
+        except Exception:
+            logger.exception("Benchmark load for the portfolio sim failed")
 
-                live_exit_policy = await get_exit_policy(db)
-            except Exception:
-                logger.exception("Live exit policy load failed; monitor uses defaults")
-            portfolio_monitor_report = _portfolio_monitor(
-                candidates, price_columns, spy_closes, hold_horizon,
+        for policy in ("target", "hold"):
+            sim = _simulate_portfolio(
+                candidates, price_columns, spy_closes, policy, hold_horizon
+            )
+            if sim is not None:
+                sim_policies.append({"policy": policy, **sim})
+        strategy_variant_rows = _strategy_variant_sims(
+            candidates, price_columns, spy_closes, hold_horizon
+        )
+        exit_policy_rows = _exit_policy_sims(
+            candidates, price_columns, spy_closes, hold_horizon
+        )
+        live_exit_policy: dict | None = None
+        try:
+            from app.services.paper_trade_service import get_exit_policy
+
+            live_exit_policy = await get_exit_policy(db)
+        except Exception:
+            logger.exception("Live exit policy load failed; monitor uses defaults")
+        portfolio_monitor_report = _portfolio_monitor(
+            candidates, price_columns, spy_closes, hold_horizon,
+            live_exit_policy=live_exit_policy,
+            cadence=cadence,
+        )
+        split = _holdout_split()
+        if split is not None:
+            holdout_report = _holdout_evaluation(
+                candidates, price_columns, spy_closes, hold_horizon, split,
                 live_exit_policy=live_exit_policy,
                 cadence=cadence,
             )
-            split = _holdout_split()
-            if split is not None:
-                holdout_report = _holdout_evaluation(
-                    candidates, price_columns, spy_closes, hold_horizon, split,
-                    live_exit_policy=live_exit_policy,
-                    cadence=cadence,
-                )
-            if _min_rr_sweep_enabled():
-                min_rr_sweep_report = _min_rr_sweep(
-                    candidates, price_columns, spy_closes, activation, current_min_pct,
-                    hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence,
-                )
-        except Exception:
-            logger.exception("Portfolio simulation failed")
+        if _min_rr_sweep_enabled():
+            min_rr_sweep_report = _min_rr_sweep(
+                candidates, price_columns, spy_closes, activation, current_min_pct,
+                hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence,
+            )
+    except Exception:
+        logger.exception("Portfolio simulation failed")
 
     report = {
         "generated_at": datetime.now(timezone.utc).isoformat(),
         "tickers": total,
-        "rank_only_tickers": len(rank_only_symbols),
         "candidates": len(candidates),
         "qualified": len(qualified),
         "params": {
@@ -4174,9 +3999,6 @@ async def run_backtest(
             "target_model_label": BACKTEST_TARGET_MODELS[target_model],
             "is_production_target_model": target_model == PRODUCTION_GTL_TARGET_MODEL,
             "production_reentry_policy": PRODUCTION_REENTRY_POLICY,
-            "liquid_breadth_top_n": _liquid_breadth_top_n() or None,
-            "liquid_min_price": _liquid_min_price() if _liquid_breadth_top_n() else None,
-            "signal_eval_only": _signal_eval_only(),
         },
         "activation": activation,
         "overall_qualified": _bucket_stats(qualified),