"""Feature orchestrator: combines market, event, and financial features.""" from __future__ import annotations import datetime as dt from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from libs.common.file_store import read_exhibit from libs.common.logging import get_logger from libs.common.time_utils import filing_time_bucket as classify_time_bucket from libs.labeler.reaction_date import compute_reaction_date from libs.db.models import Document, Event, EventParse, FeatureSnapshot from libs.features.event_features import compute_event_features from libs.features.financial_features import compute_financial_features from libs.features.intraday_features import compute_intraday_features from libs.features.market_features import compute_market_features from libs.features.text_features import compute_text_features from libs.oracle_client.company import CompanyService from libs.oracle_client.financial import FinancialService from libs.oracle_client.price import PriceService logger = get_logger(__name__) SNAPSHOT_VERSION = "1.0.0" async def build_features_for_event( session: AsyncSession, event: Event, price_service: PriceService, financial_service: FinancialService | None = None, company_service: CompanyService | None = None, ) -> tuple[FeatureSnapshot, FeatureSnapshot] | None: """Build market_v1 and event_v1 feature snapshots for an event. Returns (market_snapshot, event_snapshot) or None on failure. """ # Get ticker from event (via symbol) if not event.symbol_id: logger.warning("event_no_symbol", event_id=event.event_id) return None # Extract ticker from symbol_id (format: SYM::{ticker}::{venue}) parts = event.symbol_id.split("::") ticker = parts[1] if len(parts) >= 2 else None if not ticker: logger.warning("event_bad_symbol_id", event_id=event.event_id, symbol_id=event.symbol_id) return None event_date_str = event.event_date.isoformat() # Compute reaction date for market feature alignment if isinstance(event.filed_at_utc, dt.datetime): ftb = classify_time_bucket(event.filed_at_utc, event.event_date) else: ftb = "unknown" reaction_date = compute_reaction_date(event.event_date, ftb) reaction_date_str = reaction_date.isoformat() # Fetch price bars from Stock Oracle try: # 30 trading days before event start_date = (event.event_date - dt.timedelta(days=45)).isoformat() end_date = (event.event_date + dt.timedelta(days=5)).isoformat() price_response = await price_service.get_daily_bars( ticker, start=start_date, end=end_date ) bars = price_response.bars except Exception as exc: logger.error("price_fetch_failed", event_id=event.event_id, error=str(exc)) return None # Compute market features mf = compute_market_features(bars, reaction_date_str) # Persist market_cap_proxy + exchange_proxy so EventDetector doesn't need # a live Oracle screener call during paper trading / backsim. if company_service is not None: try: company_info = await company_service.get_company(ticker) mf["market_cap_proxy"] = company_info.market_cap mf["exchange_proxy"] = company_info.exchange except Exception as exc: logger.debug("builder_company_info_failed", ticker=ticker, error=str(exc)) # Get latest valid event parse result = await session.execute( select(EventParse) .where(EventParse.event_id == event.event_id) .where(EventParse.validation_status == "valid") .order_by(EventParse.event_parse_id.desc()) .limit(1) ) parse = result.scalar_one_or_none() if parse is None: logger.warning("no_valid_parse", event_id=event.event_id) return None ef = compute_event_features(parse.output_json) if ftb != "unknown": ef["filing_time_bucket"] = ftb market_snapshot = FeatureSnapshot( event_id=event.event_id, snapshot_name="market_v1", snapshot_version=SNAPSHOT_VERSION, feature_json=mf, ) event_snapshot = FeatureSnapshot( event_id=event.event_id, snapshot_name="event_v1", snapshot_version=SNAPSHOT_VERSION, feature_json=ef, ) session.add(market_snapshot) session.add(event_snapshot) await session.flush() # Optional: text sentiment features (non-fatal if unavailable) try: doc_result = await session.execute( select(Document).where(Document.document_id == event.primary_document_id) ) doc = doc_result.scalar_one_or_none() if doc and doc.accession_no: exhibit_text = read_exhibit(doc.accession_no, "EX-99.1") tf = compute_text_features(exhibit_text) text_snapshot = FeatureSnapshot( event_id=event.event_id, snapshot_name="text_v1", snapshot_version=SNAPSHOT_VERSION, feature_json=tf, ) session.add(text_snapshot) await session.flush() logger.info("text_features_built", event_id=event.event_id, word_count=tf["lm_word_count"]) except FileNotFoundError: logger.debug("text_features_no_exhibit", event_id=event.event_id) except Exception as exc: logger.warning("text_features_skipped", event_id=event.event_id, error=str(exc)) # Optional: financial features (non-fatal if unavailable) if financial_service is not None: try: fin_response = await financial_service.get_financial_data(ticker) ff = compute_financial_features(fin_response) if ff: financial_snapshot = FeatureSnapshot( event_id=event.event_id, snapshot_name="financial_v1", snapshot_version=SNAPSHOT_VERSION, feature_json=ff, ) session.add(financial_snapshot) await session.flush() logger.info("financial_features_built", event_id=event.event_id, ticker=ticker) except Exception as exc: logger.warning( "financial_features_skipped", event_id=event.event_id, ticker=ticker, error=str(exc), ) # Optional: earnings surprise features (non-fatal if unavailable) if event.event_type == "earnings_release": try: from libs.oracle_client import EarningsSurpriseService from libs.oracle_client.client import OracleClient from libs.common.config import get_settings settings = get_settings() async with OracleClient(base_url=settings.stock_oracle_url) as surprise_client: svc = EarningsSurpriseService(surprise_client) quarters = await svc.get_surprise(ticker) # Match by reported_date closest to event_date import datetime as _dt best_match = None best_delta = 999 for q in quarters: rd = q.get("reported_date", "") if not rd: continue delta = abs((_dt.date.fromisoformat(rd) - event.event_date).days) if delta < best_delta and delta <= 5: best_delta = delta best_match = q if best_match: surprise_snapshot = FeatureSnapshot( event_id=event.event_id, snapshot_name="earnings_surprise_v1", snapshot_version=SNAPSHOT_VERSION, feature_json={ "reported_eps": best_match.get("reported_eps"), "estimated_eps": best_match.get("estimated_eps"), "earnings_surprise_pct": best_match.get("surprise_percentage"), "earnings_beat": best_match.get("beat"), }, ) session.add(surprise_snapshot) await session.flush() logger.info("earnings_surprise_built", event_id=event.event_id, ticker=ticker, surprise_pct=best_match.get("surprise_percentage")) except Exception as exc: logger.debug("earnings_surprise_skipped", event_id=event.event_id, error=str(exc)) # Optional: intraday volume profile features (non-fatal if unavailable) try: intraday_resp = await price_service.get_historical_intraday(ticker, reaction_date_str) intraday_bars = [b.model_dump() for b in intraday_resp.bars] idf = compute_intraday_features(intraday_bars) if idf: intraday_snapshot = FeatureSnapshot( event_id=event.event_id, snapshot_name="intraday_v1", snapshot_version=SNAPSHOT_VERSION, feature_json=idf, ) session.add(intraday_snapshot) await session.flush() logger.info("intraday_features_built", event_id=event.event_id, ticker=ticker) except Exception as exc: logger.debug("intraday_features_skipped", event_id=event.event_id, error=str(exc)) logger.info( "features_built", event_id=event.event_id, ticker=ticker, market_features=list(mf.keys()), ) return market_snapshot, event_snapshot