"""Same-day catalyst helpers for intraday ORB research. This module intentionally starts with the narrowest reliable catalyst source we have today: Oracle's parsed filing events. The goal is to stop treating premarket attention as if it were a true catalyst signal. """ from __future__ import annotations import asyncio import gzip import json import os from pathlib import Path from typing import Any, Callable from libs.oracle_client import AttentionService, FilingsService, OracleClient class FilingEventCache: """Disk cache for same-day filing events keyed by ticker. Layout: {cache_dir}/{TICKER}.json.gz Each file stores a coverage window plus the raw event payloads returned by Oracle. Reads support exact or subset date ranges when coverage is present. """ def __init__(self, cache_dir: str = "data/cache/orb_catalyst") -> None: self._root = Path(cache_dir) def _path(self, ticker: str) -> Path: return self._root / f"{ticker.upper()}.json.gz" def get( self, ticker: str, start_date: str, end_date: str, ) -> list[dict[str, Any]] | None: p = self._path(ticker) if not p.exists(): return None try: with gzip.open(p, "rt", encoding="utf-8") as fh: payload = json.load(fh) except Exception: p.unlink(missing_ok=True) return None coverage_start = str(payload.get("coverage_start") or "") coverage_end = str(payload.get("coverage_end") or "") if not coverage_start or not coverage_end: p.unlink(missing_ok=True) return None if start_date < coverage_start or end_date > coverage_end: return None events = payload.get("events", []) return [ e for e in events if start_date <= str(e.get("filing_date", ""))[:10] <= end_date ] def put( self, ticker: str, start_date: str, end_date: str, events: list[dict[str, Any]], ) -> None: p = self._path(ticker) p.parent.mkdir(parents=True, exist_ok=True) existing_events: list[dict[str, Any]] = [] coverage_start = start_date coverage_end = end_date if p.exists(): try: with gzip.open(p, "rt", encoding="utf-8") as fh: payload = json.load(fh) existing_events = list(payload.get("events", [])) old_start = str(payload.get("coverage_start") or "") old_end = str(payload.get("coverage_end") or "") if old_start: coverage_start = min(coverage_start, old_start) if old_end: coverage_end = max(coverage_end, old_end) except Exception: p.unlink(missing_ok=True) existing_events = [] merged: dict[tuple[str, str, str, str], dict[str, Any]] = {} for event in existing_events + list(events): key = ( str(event.get("filing_date", ""))[:10], str(event.get("accession_number", "")), str(event.get("event_type", "")), str(event.get("item_number", "")), ) merged[key] = event payload = { "coverage_start": coverage_start, "coverage_end": coverage_end, "events": [merged[k] for k in sorted(merged)], } tmp = p.with_suffix(".tmp") try: with gzip.open(tmp, "wt", encoding="utf-8") as fh: json.dump(payload, fh) os.replace(tmp, p) except Exception: if tmp.exists(): tmp.unlink(missing_ok=True) raise class AttentionEventCache: """Disk cache for same-day attention features keyed by ticker + event_date.""" def __init__(self, cache_dir: str = "data/cache/orb_attention") -> None: self._root = Path(cache_dir) def _path(self, ticker: str) -> Path: return self._root / f"{ticker.upper()}.json.gz" def get(self, ticker: str, event_date: str) -> dict[str, Any] | None: p = self._path(ticker) if not p.exists(): return None try: with gzip.open(p, "rt", encoding="utf-8") as fh: payload = json.load(fh) except Exception: p.unlink(missing_ok=True) return None day_map = payload.get("days") if not isinstance(day_map, dict): p.unlink(missing_ok=True) return None data = day_map.get(event_date) return dict(data) if isinstance(data, dict) else None def put(self, ticker: str, event_date: str, features: dict[str, Any]) -> None: p = self._path(ticker) p.parent.mkdir(parents=True, exist_ok=True) payload: dict[str, Any] = {"days": {}} if p.exists(): try: with gzip.open(p, "rt", encoding="utf-8") as fh: existing = json.load(fh) if isinstance(existing, dict) and isinstance(existing.get("days"), dict): payload = existing except Exception: p.unlink(missing_ok=True) payload = {"days": {}} payload.setdefault("days", {}) payload["days"][event_date] = dict(features) tmp = p.with_suffix(".tmp") try: with gzip.open(tmp, "wt", encoding="utf-8") as fh: json.dump(payload, fh) os.replace(tmp, p) except Exception: if tmp.exists(): tmp.unlink(missing_ok=True) raise def _event_type_weight(event_type: str) -> float: et = event_type.lower() if any(key in et for key in ("earnings", "results", "guidance", "regulation_fd", "material")): return 1.0 if any(key in et for key in ("acquisition", "merger", "agreement", "contract", "asset")): return 0.95 if any(key in et for key in ("management", "director", "officer", "analyst")): return 0.60 return 0.75 def _build_event_feature_map( events_by_ticker: dict[str, list[dict[str, Any]]], start_date: str, end_date: str, ) -> dict[str, dict[str, dict[str, Any]]]: features: dict[str, dict[str, dict[str, Any]]] = {} for ticker, events in events_by_ticker.items(): ticker_map: dict[str, list[dict[str, Any]]] = {} for event in events: filing_date = str(event.get("filing_date", ""))[:10] if not filing_date or filing_date < start_date or filing_date > end_date: continue ticker_map.setdefault(filing_date, []).append(event) if not ticker_map: continue day_features: dict[str, dict[str, Any]] = {} for day, day_events in ticker_map.items(): event_types = sorted({ str(event.get("event_type", "")).strip() for event in day_events if str(event.get("event_type", "")).strip() }) score = max((_event_type_weight(t) for t in event_types), default=0.0) if len(day_events) > 1: score = min(1.25, score + 0.10 * (len(day_events) - 1)) day_features[day] = { "event_flag": True, "event_count": len(day_events), "event_types": event_types, "event_score": round(score, 4), } features[ticker] = day_features return features def _build_attention_feature_map( rows: dict[tuple[str, str], dict[str, Any]], ) -> dict[str, dict[str, dict[str, Any]]]: out: dict[str, dict[str, dict[str, Any]]] = {} for (ticker, day), features in rows.items(): out.setdefault(ticker, {})[day] = dict(features) return out def _extract_attention_features(payload: Any) -> dict[str, Any]: wiki_spike = payload.wiki.spike_10d wiki_zscore = payload.wiki.zscore_20d article_count_3d = payload.news.article_count_3d us_article_count_3d = payload.news.us_article_count_3d resolver_confidence = payload.entity.resolver_confidence return { "attention_wiki_spike_10d": wiki_spike, "attention_wiki_zscore_20d": wiki_zscore, "attention_article_count_3d": article_count_3d, "attention_us_article_count_3d": us_article_count_3d, "attention_resolver_confidence": resolver_confidence, } async def fetch_filing_event_features_bulk( tickers: list[str], start_date: str, end_date: str, client: OracleClient, cache: FilingEventCache | None = None, concurrency: int = 6, progress_callback: Callable[[int, int], None] | None = None, ) -> dict[str, dict[str, dict[str, Any]]]: """Fetch and cache same-day filing event features for a ticker universe.""" svc = FilingsService(client) semaphore = asyncio.Semaphore(concurrency) lock = asyncio.Lock() events_by_ticker: dict[str, list[dict[str, Any]]] = {} completed = 0 def _emit_progress() -> None: if progress_callback: progress_callback(completed, len(tickers)) async def fetch_one(ticker: str) -> None: nonlocal completed cached = cache.get(ticker, start_date, end_date) if cache else None if cached is None: async with semaphore: try: resp = await svc.get_filing_events( ticker, start_date=start_date, end_date=end_date, ) cached = [ { "ticker": event.ticker, "filing_date": event.filing_date, "accession_number": event.accession_number, "event_type": event.event_type, "item_number": event.item_number, "form_type": event.form_type, "title": event.title, } for event in resp.events ] if cache is not None: await asyncio.to_thread(cache.put, ticker, start_date, end_date, cached) except Exception: cached = [] async with lock: events_by_ticker[ticker] = list(cached) completed += 1 _emit_progress() await asyncio.gather(*(fetch_one(ticker) for ticker in tickers)) return _build_event_feature_map(events_by_ticker, start_date, end_date) async def fetch_attention_features_bulk( ticker_days: list[tuple[str, str]], client: OracleClient, cache: AttentionEventCache | None = None, concurrency: int = 8, progress_callback: Callable[[int, int], None] | None = None, ) -> dict[str, dict[str, dict[str, Any]]]: """Fetch same-day attention features for a set of (ticker, date) pairs.""" svc = AttentionService(client) semaphore = asyncio.Semaphore(concurrency) lock = asyncio.Lock() completed = 0 rows: dict[tuple[str, str], dict[str, Any]] = {} unique_pairs = sorted({(ticker.upper(), day) for ticker, day in ticker_days}) def _emit_progress() -> None: if progress_callback: progress_callback(completed, len(unique_pairs)) async def fetch_one(ticker: str, event_date: str) -> None: nonlocal completed cached = cache.get(ticker, event_date) if cache else None if cached is None: async with semaphore: try: payload = await svc.get_event_attention(ticker, event_date) cached = _extract_attention_features(payload) if cache is not None: await asyncio.to_thread(cache.put, ticker, event_date, cached) except Exception: cached = { "attention_wiki_spike_10d": None, "attention_wiki_zscore_20d": None, "attention_article_count_3d": 0, "attention_us_article_count_3d": 0, "attention_resolver_confidence": 0.0, } async with lock: rows[(ticker, event_date)] = dict(cached) completed += 1 _emit_progress() await asyncio.gather(*(fetch_one(ticker, day) for ticker, day in unique_pairs)) return _build_attention_feature_map(rows)