momentum-system/youtube_claims/recap/ourside.py

"""What OUR system saw/did for a (date, symbol): discovery, state changes, gate verdicts,
signals, orders, closes, risk rules -- from BOTH stacks' event logs (paper DB :5433 and live
Tradier DB :5434) -- plus the day's SIP tape stats, float and news. Ported from the pilot's
ourside.py; read-only against the trading DBs. Event timestamps are ET-naive; minute_bars are
UTC-naive."""

from __future__ import annotations

import re
from collections import Counter

import psycopg2

from youtube_claims.recap import settings
from youtube_claims.recap.tape import BarSource, tape_stats

_EVENT_TYPES = ["symbol_discovered", "symbol_state_changed", "signal_blocked", "entry_gate_evaluated",
                "signal_ready", "order_submitted", "position_closed", "risk_rule_triggered"]


def norm_reason(m: str | None) -> str:
    """Collapse numbers so '3.2x < 5x' and '4.1x < 5x' count as one reason."""
    return re.sub(r"\d+(\.\d+)?", "#", m or "")[:120]


def summarize_events(ev: list[tuple]) -> dict:
    """ev: [(ts, event_type, message)] ordered by time -> the compact per-stack view."""
    hm = lambda t: t.strftime("%H:%M")  # noqa: E731
    blocks = Counter(norm_reason(e[2].split("blocked:", 1)[-1]) for e in ev if e[1] == "signal_blocked")
    gates = Counter(norm_reason(e[2].split("rejected:", 1)[-1]) for e in ev
                    if e[1] == "entry_gate_evaluated" and "rejected" in (e[2] or ""))
    states: list[tuple] = []
    for e in ev:
        if e[1] == "symbol_state_changed" and (not states or states[-1][1] != e[2][:160]):
            states.append((hm(e[0]), (e[2] or "")[:160]))
    pick = lambda typ, n, w: [(hm(e[0]), (e[2] or "")[:w]) for e in ev if e[1] == typ][:n]  # noqa: E731
    return {"discovered": pick("symbol_discovered", 2, 200), "states": states[:8],
            "top_blocks": blocks.most_common(3), "gate_rejects": gates.most_common(3),
            "signals": pick("signal_ready", 3, 120), "orders": pick("order_submitted", 3, 100),
            "closes": pick("position_closed", 3, 120), "risk": pick("risk_rule_triggered", 4, 140)}


class OurSide:
    def __init__(self):
        self.paper = psycopg2.connect(settings.database_url(), connect_timeout=10)
        self.paper.autocommit = True
        try:
            self.live = psycopg2.connect(settings.live_database_url(), connect_timeout=10)
            self.live.autocommit = True
        except Exception as e:  # noqa: BLE001 -- the paper view alone is still useful
            print(f"[ourside] live DB unavailable: {e}")
            self.live = None
        self.bars = BarSource(settings.dirs()["cache"] / "bars", paper_con=self.paper)

    def close(self):
        for c in (self.paper, self.live):
            if c is not None:
                c.close()

    def _q(self, con, sql, args=()):
        with con.cursor() as cur:
            cur.execute(sql, args)
            return cur.fetchall()

    def events(self, con, d: str, sym: str) -> list[tuple]:
        return self._q(con, "select timestamp, event_type, message from events where event_type = any(%s) "
                            "and timestamp between %s and %s and payload_json like %s order by timestamp",
                       (_EVENT_TYPES, f"{d} 04:00", f"{d} 20:00", f'%"symbol": "{sym}"%'))

    def prev_close(self, d: str, sym: str) -> float | None:
        r = self._q(self.paper, "select close from daily_bars where symbol=%s and trade_date<%s "
                                "order by trade_date desc limit 1", (sym, d))
        return float(r[0][0]) if r and r[0][0] else None

    def view(self, d: str, sym: str) -> dict:
        d, sym = str(d), sym.upper()
        out: dict = {"date": d, "symbol": sym}
        for name, con in (("paper", self.paper), ("live", self.live)):
            out[name] = summarize_events(self.events(con, d, sym)) if con is not None else None
        bars = self.bars.bars(sym, d)
        out["tape"] = tape_stats(bars, self.prev_close(d, sym))
        local = self._q(self.paper, "select min(timestamp), count(*) from minute_bars where symbol=%s "
                                    "and session_date=%s", (sym, d))
        if local and local[0][0]:
            import datetime as dt
            from zoneinfo import ZoneInfo
            first = local[0][0].replace(tzinfo=dt.timezone.utc).astimezone(ZoneInfo("America/New_York"))
            out["tape"]["our_first_bar"] = first.strftime("%H:%M")
        out["tape"]["our_bar_count"] = local[0][1] if local else 0
        fl = self._q(self.paper, "select float_shares from float_cache where symbol=%s", (sym,))
        out["float"] = float(fl[0][0]) if fl and fl[0][0] else None
        out["news"] = [r[0][:140] for r in self._q(
            self.paper, "select headline from news_events where symbol=%s and published_at::date "
                        "between %s::date - 1 and %s::date order by published_at limit 3", (sym, d, d))]
        return out

    def day_universe(self, d: str) -> list[tuple]:
        """Symbols our scanner ranked that day: [(symbol, gap %, price)] (the bundle's list)."""
        rows = self._q(self.paper, "select symbol, max(gap_pct), max(price) from scanner_snapshots "
                                   "where snapshot_time::date=%s group by symbol "
                                   "order by max(gap_pct) desc nulls last limit 40", (str(d),))
        if not rows:
            rows = self._q(self.paper, "select symbol, gap_pct, null from historical_gappers "
                                       "where session_date=%s order by rank limit 40", (str(d),))
        return [(s, round(float(g or 0) * (100 if g and abs(g) < 20 else 1), 1),
                 round(float(p), 4) if p else None) for s, g, p in rows]


def discovery_time(view: dict | None) -> str | None:
    """Earliest HH:MM either stack discovered the symbol."""
    ts = [t for side in ("paper", "live") for t, _ in ((view or {}).get(side) or {}).get("discovered", [])]
    return min(ts) if ts else None