momentum-system/research/ingestion/signals.py

"""Signal scanning: derive gapper candidates and scanner snapshots.

Reads minute/daily bars from the market database, computes gap %% and
relative volume, and writes ranked rows into scanner_snapshots so the
dashboard and research notebooks can replay what the scanner saw.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass
from datetime import date, datetime, timezone

SCANNER_VERSION = "gapper-v2"


@dataclass
class GapperCandidate:
    symbol: str
    price: float
    gap_pct: float               # DYNAMIC: latest close vs prev close (display/rank)
    cumulative_volume: int
    relative_volume: float
    vwap: float | None
    prev_close: float | None = None   # for the STATIC overnight gap (open vs prev close)
    rank: int = 0
    # Time-of-day rvol: cum volume vs MEDIAN(20d) prorated by session elapsed time.
    # The legacy relative_volume divides cum-so-far by the full-day MEAN, which is
    # (a) ~0.1-0.3 for a NORMAL heavy name minutes after the open (cum can't reach
    # a full-day average yet) and (b) poisoned by reverse-split / prior-squeeze
    # days in the mean (BIYA 07-20: legacy 1.27 vs tod 481 — median 1.5M, mean
    # 29.4M). None when no baseline. Legacy relative_volume is kept untouched —
    # validated boundaries (gap-squeeze rvol>=5) are calibrated in its units.
    relative_volume_tod: float | None = None


def _session_elapsed_fraction(now: datetime | None = None) -> float:
    """Fraction of the 390-min regular session elapsed, floored at 0.05 so a
    premarket/at-open scan doesn't divide by ~zero expected volume."""
    try:
        from zoneinfo import ZoneInfo
        et = (now or datetime.now(ZoneInfo("America/New_York")))
        if et.tzinfo is not None:
            et = et.astimezone(ZoneInfo("America/New_York"))
        elapsed = (et.hour - 9) * 60 + (et.minute - 30)
    except Exception:  # noqa: BLE001 — fall back to full-day (legacy behaviour)
        return 1.0
    return min(1.0, max(0.05, elapsed / 390.0))


def scan_gappers(
    con,
    session_date: date,
    min_gap_pct: float = 5.0,
    min_relative_volume: float = 2.0,
    price_min: float = 1.0,
    price_max: float = 20.0,
    limit: int = 25,
    now: datetime | None = None,
) -> list[GapperCandidate]:
    """Find symbols gapping up on elevated volume for a session."""
    rows = con.execute(
        """
        WITH latest AS (
            SELECT symbol, close, vwap,
                   SUM(volume) OVER (PARTITION BY symbol) AS cum_volume,
                   ROW_NUMBER() OVER (
                       PARTITION BY symbol ORDER BY timestamp DESC
                   ) AS rn
            FROM minute_bars
            WHERE session_date = ?
        ),
        prev AS (
            SELECT symbol, close AS prev_close,
                   ROW_NUMBER() OVER (
                       PARTITION BY symbol ORDER BY trade_date DESC
                   ) AS rn
            FROM daily_bars
            WHERE trade_date < ?
        ),
        avg_vol AS (
            SELECT symbol, AVG(volume) AS adv,
                   PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY volume) AS med_vol
            FROM (
                SELECT symbol, volume,
                       ROW_NUMBER() OVER (
                           PARTITION BY symbol ORDER BY trade_date DESC
                       ) AS rn
                FROM daily_bars
                WHERE trade_date < ?
            )
            WHERE rn <= 20
            GROUP BY symbol
        )
        SELECT l.symbol, l.close, l.vwap, l.cum_volume,
               p.prev_close, a.adv, a.med_vol
        FROM latest l
        JOIN prev p ON p.symbol = l.symbol AND p.rn = 1
        LEFT JOIN avg_vol a ON a.symbol = l.symbol
        WHERE l.rn = 1 AND l.close BETWEEN ? AND ?
        """,
        [session_date, session_date, session_date, price_min, price_max],
    ).fetchall()

    frac = _session_elapsed_fraction(now)
    candidates: list[GapperCandidate] = []
    for symbol, close, vwap, cum_volume, prev_close, adv, med_vol in rows:
        if not prev_close or prev_close <= 0:
            continue
        gap_pct = (float(close) - float(prev_close)) / float(prev_close) * 100.0
        rvol = (float(cum_volume) / float(adv)) if adv and adv > 0 else 0.0
        rvol_tod = (float(cum_volume) / (float(med_vol) * frac)
                    if med_vol and float(med_vol) > 0 else None)
        # either rvol flavour clears the bar: legacy is ~0.1-0.3 early in the
        # session even for a heavy name (cum-so-far vs full-day mean), so the
        # tod variant keeps real early movers from being silently dropped
        if gap_pct >= min_gap_pct and (
                rvol >= min_relative_volume
                or (rvol_tod or 0.0) >= min_relative_volume):
            candidates.append(
                GapperCandidate(
                    symbol=symbol,
                    price=float(close),
                    prev_close=float(prev_close),
                    gap_pct=gap_pct,
                    cumulative_volume=int(cum_volume),
                    relative_volume=rvol,
                    vwap=float(vwap) if vwap is not None else None,
                    relative_volume_tod=rvol_tod,
                )
            )
    candidates.sort(key=lambda c: (c.gap_pct * max(c.relative_volume, 0.1)), reverse=True)
    for i, candidate in enumerate(candidates[:limit], start=1):
        candidate.rank = i
    return candidates[:limit]


def store_scanner_snapshot(
    con, candidates: list[GapperCandidate], snapshot_time: datetime | None = None
) -> int:
    """Persist a scan into scanner_snapshots. Returns rows written."""
    ts = (snapshot_time or datetime.now(timezone.utc)).replace(tzinfo=None)
    rows = [
        (
            str(uuid.uuid4()),
            ts,
            c.symbol,
            c.rank,
            c.price,
            c.gap_pct,
            c.cumulative_volume,
            c.relative_volume,
            c.vwap,
            (c.price - c.vwap) / c.vwap * 100.0 if c.vwap else None,
            SCANNER_VERSION,
        )
        for c in candidates
    ]
    if not rows:
        return 0
    con.executemany(
        """
        INSERT INTO scanner_snapshots (
            id, snapshot_time, symbol, rank, price, gap_pct,
            cumulative_volume, relative_volume, vwap, distance_from_vwap,
            scanner_version
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        rows,
    )
    return len(rows)