From 4de842ba896610b7273f3955cf012124921237df Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Tue, 24 Mar 2026 03:02:11 -0700 Subject: [PATCH] Fix paper trader volume_ratio field name mismatch with selector MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EventDetector computed volume_ratio as fallback but selector checks volume_ratio_20d. When DB feature_json was missing this field, the volume gate was silently bypassed in paper trading — allowing trades like LKQ (vol=0.8) that the backtest correctly blocks. Now sets both volume_ratio_20d and volume_ratio for consistency. Co-Authored-By: Claude Opus 4.6 (1M context) --- apps/paper_trader/event_detector.py | 498 ++++++++++++++++++++++++++++ 1 file changed, 498 insertions(+) create mode 100644 apps/paper_trader/event_detector.py diff --git a/apps/paper_trader/event_detector.py b/apps/paper_trader/event_detector.py new file mode 100644 index 0000000..32b645f --- /dev/null +++ b/apps/paper_trader/event_detector.py @@ -0,0 +1,498 @@ +"""Detect new events from the pipeline DB and enrich them for paper trading. + +Mirrors the enrichment pattern in SnapshotStore._async_load() but targets +a single execution_date rather than loading an entire parquet file. +""" +from __future__ import annotations + +import asyncio +import datetime as dt +from typing import Any + +from libs.backtest.domain import BacktestConfig +from libs.common.logging import get_logger + +logger = get_logger(__name__) + + +class EventDetector: + """Queries the pipeline DB for events executing on a given date and enriches them.""" + + def __init__( + self, + db_dsn: str, + oracle_url: str, + ) -> None: + self._db_dsn = db_dsn + self._oracle_url = oracle_url + self._db_unavailable: bool = False # circuit breaker: skip after first failure + self._screener_unavailable: bool = False # circuit breaker for screener API + + async def get_candidates_for_date( + self, + execution_date: dt.date, + config: BacktestConfig, + convention: str | None = None, + ) -> list[dict[str, Any]]: + """Return enriched candidate rows for execution_date. + + convention: 'reaction_close' for same-day close entries, + 'next_open_after_reaction_close' for next-open entries. + None fetches both (legacy / debugging only). + + Steps: + 1. Query Event table for events with entry_date == execution_date + 2. Fetch SymbolMaster tickers + 3. Enrich with Alpaca price data (bars, avg_dollar_volume, ATR-14) + 4. Enrich with Oracle sector data + 5. Compute entry_price_est and score + """ + raw_rows = await self._fetch_events_for_date(execution_date, convention=convention) + if not raw_rows: + logger.debug("event_detector_no_events", date=execution_date.isoformat()) + return [] + + unique_symbols = sorted({ + str(r.get("symbol", "")).upper() + for r in raw_rows + if r.get("symbol") + }) + + # Fetch bars for the last 30 days for ADV + ATR computation + bar_start = execution_date - dt.timedelta(days=45) + bars_by_symbol, avg_dvol, atr_by_symbol = await self._fetch_enrichment_data( + unique_symbols, bar_start, execution_date + ) + company_info = await self._fetch_company_info(unique_symbols) + screener_mcaps = await self._fetch_screener_market_caps(unique_symbols) + + enriched_rows: list[dict[str, Any]] = [] + for row in raw_rows: + sym = str(row.get("symbol", "")).upper() + if not sym: + continue + + info = company_info.get(sym, {}) + enriched = dict(row) + enriched["symbol"] = sym + # Use Oracle-computed avg_dvol; fall back to stored avg_dollar_volume_20d + # from feature_json when Oracle enrichment returned nothing. + oracle_adv = avg_dvol.get(sym) + enriched["avg_dollar_volume"] = ( + oracle_adv if oracle_adv + else float(row.get("avg_dollar_volume_20d") or 0.0) + ) + # Prefer atr_14 stored in DB feature_json (computed by feature builder with + # event_date+5d bars, giving stable post-event ATR). Fall back to Oracle-computed + # value only if not already present — avoids inflating ATR with the large + # earnings reaction-day range when bars end exactly at execution_date. + if not enriched.get("atr_14"): + enriched["atr_14"] = atr_by_symbol.get(sym) + # Enrich reaction_day_low / reaction_day_high from Oracle bars. + # These are NOT stored in DB feature_json but ARE used by compute_stop_price() + # to tighten stops when reaction_day_low > atr_stop (matches snapshot pipeline). + if not enriched.get("reaction_day_low") or not enriched.get("reaction_day_high"): + sym_bars = bars_by_symbol.get(sym, {}) + reaction_date_raw = enriched.get("reaction_date") + if reaction_date_raw: + rd = _parse_date(reaction_date_raw) + if rd and rd in sym_bars: + if not enriched.get("reaction_day_low"): + enriched["reaction_day_low"] = sym_bars[rd].get("low") + if not enriched.get("reaction_day_high"): + enriched["reaction_day_high"] = sym_bars[rd].get("high") + + # Compute market features from Oracle bars if missing in DB feature_json. + # These can be None when the feature builder ran before reaction-day bars settled. + sym_bars = bars_by_symbol.get(sym, {}) + rd = _parse_date(enriched.get("reaction_date")) + if rd and rd in sym_bars: + sorted_dates = sorted(sym_bars.keys()) + try: + rd_idx = sorted_dates.index(rd) + except ValueError: + rd_idx = -1 + if rd_idx > 0: + prior_dates = sorted_dates[max(0, rd_idx - 20):rd_idx] + prev_bar = sym_bars[sorted_dates[rd_idx - 1]] + reaction_bar = sym_bars[rd] + + if not enriched.get("volume_ratio_20d") and not enriched.get("volume_ratio") and prior_dates: + avg_vol = sum(sym_bars[d]["volume"] for d in prior_dates) / len(prior_dates) + if avg_vol > 0: + vr = reaction_bar["volume"] / avg_vol + enriched["volume_ratio_20d"] = vr + enriched["volume_ratio"] = vr # backward compat + + if not enriched.get("reaction_day_return") and prev_bar["close"]: + enriched["reaction_day_return"] = ( + (reaction_bar["close"] - prev_bar["close"]) / prev_bar["close"] + ) + + if not enriched.get("close_location"): + rng = reaction_bar["high"] - reaction_bar["low"] + if rng > 0: + enriched["close_location"] = ( + (reaction_bar["close"] - reaction_bar["low"]) / rng + ) + + if not enriched.get("gap_size") and prev_bar["close"]: + enriched["gap_size"] = ( + (reaction_bar["open"] - prev_bar["close"]) / prev_bar["close"] + ) + + enriched["sector"] = info.get("sector", "UNKNOWN") + if "market_cap_proxy" not in enriched or enriched["market_cap_proxy"] is None: + # Prefer screener market_cap (same source as snapshot export pipeline) + market_cap = screener_mcaps.get(sym) + if market_cap is None: + market_cap = info.get("market_cap") + enriched["market_cap_proxy"] = market_cap + + # entry_price_est: use reaction-day close price + if "entry_price_est" not in enriched or not enriched["entry_price_est"]: + event_close = enriched.get("event_close") + if event_close: + enriched["entry_price_est"] = float(event_close) + else: + # Fall back to latest available close from bars + sym_bars = bars_by_symbol.get(sym, {}) + reaction_date_raw = enriched.get("reaction_date") + if reaction_date_raw: + rd = _parse_date(reaction_date_raw) + if rd and rd in sym_bars: + enriched["entry_price_est"] = sym_bars[rd].get("close", 0.0) + + if not enriched.get("entry_price_est"): + # Final fallback: use the most recent available close + # (needed for same-day pending events where reaction_date bar not yet available) + sym_bars = bars_by_symbol.get(sym, {}) + if sym_bars: + latest_d = max(sym_bars.keys()) + enriched["entry_price_est"] = sym_bars[latest_d].get("close", 0.0) + + if not enriched.get("entry_price_est"): + logger.debug( + "event_detector_skip_no_price", + symbol=sym, + event_id=enriched.get("event_id"), + ) + continue + + # Compute score if missing + if "score" not in enriched or enriched.get("score") is None: + from libs.backtest.scoring import compute_entry_score + enriched["score"] = compute_entry_score(enriched) + + enriched_rows.append(enriched) + + logger.info( + "event_detector_candidates_ready", + date=execution_date.isoformat(), + count=len(enriched_rows), + ) + return enriched_rows + + # ------------------------------------------------------------------ # + # DB queries + # ------------------------------------------------------------------ # + + async def _fetch_events_for_date( + self, + execution_date: dt.date, + convention: str | None = None, + ) -> list[dict[str, Any]]: + """Query EventLabel + FeatureSnapshot + Event + SymbolMaster for execution_date. + + entry_date lives in EventLabel, not Event. Features live in FeatureSnapshot.feature_json. + convention: filter by entry_convention ('reaction_close' or 'next_open_after_reaction_close'). + """ + if self._db_unavailable: + return [] + try: + from sqlalchemy import select + from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + + from libs.db.models import Event, EventLabel, FeatureSnapshot, SymbolMaster + + engine = create_async_engine( + self._db_dsn, echo=False, + connect_args={"timeout": 5}, + ) + async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + _UTC = __import__("zoneinfo").ZoneInfo("UTC") + + async with async_session() as session: + # Join: Event + EventLabel (entry_date) + FeatureSnapshot (features) + SymbolMaster + # Order by snapshot created_at ASC so newer snapshots override older ones when merging + stmt = ( + select(Event, EventLabel, FeatureSnapshot, SymbolMaster) + .join(EventLabel, Event.event_id == EventLabel.event_id) + .join(FeatureSnapshot, Event.event_id == FeatureSnapshot.event_id) + .outerjoin(SymbolMaster, Event.symbol_id == SymbolMaster.symbol_id) + .where(EventLabel.entry_date == execution_date) + .where(EventLabel.label_status.in_(["ok", "truncated", "pending"])) + .order_by(FeatureSnapshot.created_at_utc.asc()) + ) + if convention is not None: + stmt = stmt.where(EventLabel.entry_convention == convention) + rows = (await session.execute(stmt)).all() + + await engine.dispose() + + # Group by event_id — merge ALL FeatureSnapshot.feature_json per event. + # The feature builder creates multiple snapshots per event with different feature types + # (market features, NLP features, fundamental features). We need all of them merged. + # Ordered ASC so newer keys override older ones. + event_meta: dict[str, tuple] = {} # event_id -> (event, label, sym) + event_features: dict[str, dict] = {} # event_id -> merged feature_json + + for event, label, snapshot, sym in rows: + eid = event.event_id + if sym is None or not sym.ticker: + continue + if eid not in event_meta: + event_meta[eid] = (event, label, sym) + event_features[eid] = {} + # Merge: newer snapshot keys override older ones (ASC order) + event_features[eid].update(snapshot.feature_json or {}) + + result: list[dict[str, Any]] = [] + for eid, (event, label, sym) in event_meta.items(): + ts = event.filed_at_utc + if ts is None and event.event_date is not None: + ts = dt.datetime.combine( + event.event_date, dt.time(21, 0), tzinfo=_UTC + ) + + row: dict[str, Any] = { + "event_id": eid, + "symbol": sym.ticker, + "issuer_id": event.issuer_id, + "event_type": event.event_type or "", + "event_direction": event.event_direction or "", + "event_date": event.event_date, + "event_timestamp": ts, + "reaction_date": label.reaction_date, + "entry_date": label.entry_date, + "execution_date": execution_date, + "label_status": label.label_status, + # Merged features from ALL FeatureSnapshots for this event + **event_features[eid], + } + result.append(row) + + logger.debug( + "event_detector_db_fetched", + date=execution_date.isoformat(), + count=len(result), + ) + return result + + except Exception as exc: + if not self._db_unavailable: + err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ + logger.warning("event_detector_db_fetch_failed", error=err_msg) + self._db_unavailable = True + return [] + + # ------------------------------------------------------------------ # + # Enrichment helpers + # ------------------------------------------------------------------ # + + async def _fetch_enrichment_data( + self, + symbols: list[str], + start: dt.date, + end: dt.date, + concurrency: int = 16, + ) -> tuple[ + dict[str, dict[dt.date, dict[str, Any]]], # bars_by_symbol + dict[str, float], # avg_dvol + dict[str, float | None], # atr_14 + ]: + if not symbols: + return {}, {}, {} + + try: + from libs.oracle_client import OracleClient, PriceService + + bars_by_symbol: dict[str, dict[dt.date, dict[str, Any]]] = {} + avg_dvol: dict[str, float] = {} + atr_14_map: dict[str, float | None] = {} + semaphore = asyncio.Semaphore(concurrency) + + async with OracleClient(base_url=self._oracle_url) as client: + svc = PriceService(client) + + async def _fetch_one(sym: str) -> None: + async with semaphore: + try: + resp = await svc.get_daily_bars( + sym, + start=start.isoformat(), + end=end.isoformat(), + ) + date_map: dict[dt.date, dict[str, Any]] = {} + dollar_vols: list[float] = [] + closes: list[float] = [] + for bar in resp.bars: + d = dt.date.fromisoformat(bar.date) + b = { + "date": d, + "open": bar.open, + "high": bar.high, + "low": bar.low, + "close": bar.close, + "volume": bar.volume, + } + date_map[d] = b + dollar_vols.append(bar.close * bar.volume) + closes.append(bar.close) + + bars_by_symbol[sym] = date_map + last_20 = dollar_vols[-20:] + avg_dvol[sym] = sum(last_20) / len(last_20) if last_20 else 0.0 + atr_14_map[sym] = _compute_atr14(resp.bars) + except Exception as sym_exc: + err_msg = f"{type(sym_exc).__name__}: {sym_exc}" if str(sym_exc) else type(sym_exc).__name__ + # "Not found" is expected for delisted/renamed symbols — debug only + if "not found" in str(sym_exc).lower(): + logger.debug("event_detector_price_fetch_skipped", symbol=sym, error=err_msg) + else: + logger.warning("event_detector_price_fetch_failed", symbol=sym, error=err_msg) + bars_by_symbol[sym] = {} + avg_dvol[sym] = 0.0 + atr_14_map[sym] = None + + await asyncio.gather(*(_fetch_one(sym) for sym in symbols)) + + return bars_by_symbol, avg_dvol, atr_14_map + + except Exception as exc: + err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ + logger.warning("event_detector_oracle_failed", error=err_msg) + return {}, {}, {} + + async def _fetch_screener_market_caps( + self, + symbols: list[str], + ) -> dict[str, float | None]: + """Fetch market_cap for symbols via Oracle screener (batch). + + Mirrors snapshot_export._resolve_universe_profile() which uses + ScreenerService to get market_cap_proxy for all universe stocks. + """ + if not symbols or self._screener_unavailable: + return {} + + symbol_set = {s.upper() for s in symbols} + result: dict[str, float | None] = {s: None for s in symbol_set} + + try: + from libs.oracle_client import OracleClient, ScreenerService + + async with OracleClient(base_url=self._oracle_url) as client: + svc = ScreenerService(client) + # market_cap_min=500M matches snapshot_export pattern: reduces result set + # from ~10k to ~3k stocks, avoiding HTTP 500 on large responses. + stocks = await svc.search_all_stocks( + market_cap_min=500_000_000, + exchange="NYSE,NASDAQ,AMEX", + exclude_types="ETF,FUND,ADR,SPAC", + ) + + for stock in stocks: + sym = (stock.symbol or "").upper() + if sym in symbol_set: + result[sym] = stock.market_cap + + except Exception as exc: + err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ + logger.warning("event_detector_screener_mcap_failed", error=err_msg) + self._screener_unavailable = True + + return result + + async def _fetch_company_info( + self, + symbols: list[str], + concurrency: int = 16, + ) -> dict[str, dict[str, Any]]: + """Fetch sector and market_cap for each symbol from Oracle.""" + if not symbols: + return {} + + result: dict[str, dict[str, Any]] = {} + semaphore = asyncio.Semaphore(concurrency) + + try: + from libs.oracle_client import CompanyService, OracleClient + + async with OracleClient(base_url=self._oracle_url) as client: + company_svc = CompanyService(client) + + async def _fetch_one(sym: str) -> tuple[str, dict[str, Any]]: + async with semaphore: + try: + info = await company_svc.get_company(sym) + return sym, { + "sector": info.sector or "UNKNOWN", + "market_cap": info.market_cap, + } + except Exception: + return sym, {"sector": "UNKNOWN", "market_cap": None} + + fetched = await asyncio.gather(*(_fetch_one(sym) for sym in symbols)) + for sym, info in fetched: + result[sym] = info + + except Exception as exc: + err_msg = f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__ + logger.warning("event_detector_company_fetch_failed", error=err_msg) + + for sym in symbols: + result.setdefault(sym, {"sector": "UNKNOWN", "market_cap": None}) + return result + + +# ------------------------------------------------------------------ # +# Helpers +# ------------------------------------------------------------------ # + +def _compute_atr14(bars: list[Any]) -> float | None: + """Compute 14-day ATR from a list of bar objects.""" + if len(bars) < 2: + return None + + true_ranges: list[float] = [] + for i in range(1, len(bars)): + prev_close = float(bars[i - 1].close) + high = float(bars[i].high) + low = float(bars[i].low) + tr = max( + high - low, + abs(high - prev_close), + abs(low - prev_close), + ) + true_ranges.append(tr) + + if not true_ranges: + return None + + last_14 = true_ranges[-14:] if len(true_ranges) >= 14 else true_ranges + return sum(last_14) / len(last_14) + + +def _parse_date(raw: Any) -> dt.date | None: + if isinstance(raw, dt.datetime): + return raw.date() + if isinstance(raw, dt.date): + return raw + if isinstance(raw, str): + try: + return dt.date.fromisoformat(raw[:10]) + except ValueError: + return None + return None