From 57d38ecfe065988f43b3d385663f8806f1d213fe Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Thu, 26 Mar 2026 20:14:57 -0700 Subject: [PATCH] Add earnings surprise feature pipeline and snapshot export improvements Adds earnings surprise extraction to parser/features/labeler pipeline, improves filing fetcher robustness, and extends snapshot export with new field support. Co-Authored-By: Claude Opus 4.6 (1M context) --- apps/pipeline/dataset_export/main.py | 18 ++ apps/pipeline/event_parser/main.py | 26 ++- apps/pipeline/feature_builder/main.py | 26 ++- apps/pipeline/filing_fetcher/main.py | 282 ++++++++++++++++---------- apps/pipeline/label_generator/main.py | 25 ++- libs/export/snapshot_export.py | 108 +++++++++- libs/features/event_features.py | 2 + libs/labeler/label_generator.py | 17 +- libs/oracle_client/price.py | 19 ++ libs/parser/rule_parser.py | 35 +++- libs/parser/text_normalizer.py | 7 + 11 files changed, 428 insertions(+), 137 deletions(-) diff --git a/apps/pipeline/dataset_export/main.py b/apps/pipeline/dataset_export/main.py index 85b6d3a..7187fe0 100644 --- a/apps/pipeline/dataset_export/main.py +++ b/apps/pipeline/dataset_export/main.py @@ -22,7 +22,12 @@ async def run_dataset_export( feature_versions: list[str] | None = None, label_version: str = "label-2.0.0", symbols: list[str] | None = None, + universe_profile: str | None = None, + start_date: str | None = None, + end_date: str | None = None, ) -> dict: + import datetime as dt + async with get_session() as session: manifest = await export_dataset_snapshot( session=session, @@ -32,6 +37,9 @@ async def run_dataset_export( feature_versions=feature_versions, label_version=label_version, symbols=symbols, + universe_profile=universe_profile, + start_date=dt.date.fromisoformat(start_date) if start_date else None, + end_date=dt.date.fromisoformat(end_date) if end_date else None, ) return manifest @@ -65,6 +73,13 @@ def main() -> None: default=None, help="YAML file with 'symbols' list to filter export (e.g. configs/symbols_midcap.yaml)", ) + parser.add_argument( + "--universe-profile", + default=None, + help="Named live screener profile to filter export (e.g. midlarge-liquid-long-v1)", + ) + parser.add_argument("--start-date", default=None, metavar="YYYY-MM-DD", help="Inclusive event_date lower bound") + parser.add_argument("--end-date", default=None, metavar="YYYY-MM-DD", help="Inclusive event_date upper bound") parser.add_argument("--json", action="store_true", help="Print manifest JSON to stdout") args = parser.parse_args() @@ -86,6 +101,9 @@ def main() -> None: feature_versions=args.feature_versions, label_version=args.label_version, symbols=symbols, + universe_profile=args.universe_profile, + start_date=args.start_date, + end_date=args.end_date, ) ) diff --git a/apps/pipeline/event_parser/main.py b/apps/pipeline/event_parser/main.py index 2773290..7b20c4e 100644 --- a/apps/pipeline/event_parser/main.py +++ b/apps/pipeline/event_parser/main.py @@ -24,13 +24,17 @@ from libs.db.session import get_session from libs.oracle_client import FilingsService, make_oracle_client from libs.parser.rule_parser import PARSER_VERSION, SCHEMA_VERSION, RuleBasedParser, _classify_event_type from libs.parser.schema_validator import validate_parser_output -from libs.parser.text_normalizer import normalize_text +from libs.parser.text_normalizer import looks_like_html, normalize_text logger = get_logger(__name__) _parser = RuleBasedParser() -async def run_event_parser(run_id: str) -> dict[str, int]: +async def run_event_parser( + run_id: str, + start_date: str | None = None, + end_date: str | None = None, +) -> dict[str, int]: settings = get_settings() app_config = settings.get_app_config() exhibit_types = app_config.get("pipeline", {}).get("exhibit_types", ["EX-99.1"]) @@ -48,9 +52,13 @@ async def run_event_parser(run_id: str) -> dict[str, int]: session.add(job) await session.flush() - result = await session.execute( - select(Document).where(Document.parsed_status == "ready_for_parse") - ) + stmt = select(Document).where(Document.parsed_status == "ready_for_parse") + if start_date: + stmt = stmt.where(Document.filing_date >= dt.date.fromisoformat(start_date)) + if end_date: + stmt = stmt.where(Document.filing_date <= dt.date.fromisoformat(end_date)) + + result = await session.execute(stmt.order_by(Document.filing_date, Document.document_id)) docs = result.scalars().all() stats["seen"] = len(docs) @@ -75,7 +83,7 @@ async def run_event_parser(run_id: str) -> dict[str, int]: stats["errors"] += 1 continue - normalized = normalize_text(text) + normalized = normalize_text(text, is_html=looks_like_html(text)) metadata = { "filing_date": doc.filing_date.isoformat(), @@ -304,7 +312,7 @@ async def reparse_events(run_id: str) -> dict[str, int]: continue if text: - normalized = normalize_text(text) + normalized = normalize_text(text, is_html=looks_like_html(text)) metadata = { "filing_date": doc.filing_date.isoformat(), "accepted_at_utc": ( @@ -357,6 +365,8 @@ async def reparse_events(run_id: str) -> dict[str, int]: def main() -> None: parser = argparse.ArgumentParser(description="Event Parser") parser.add_argument("--run-id", default=new_job_run_id()) + parser.add_argument("--start-date", default=None, metavar="YYYY-MM-DD") + parser.add_argument("--end-date", default=None, metavar="YYYY-MM-DD") parser.add_argument( "--reparse", action="store_true", @@ -371,7 +381,7 @@ def main() -> None: if args.reparse: asyncio.run(reparse_events(args.run_id)) else: - asyncio.run(run_event_parser(args.run_id)) + asyncio.run(run_event_parser(args.run_id, start_date=args.start_date, end_date=args.end_date)) if __name__ == "__main__": diff --git a/apps/pipeline/feature_builder/main.py b/apps/pipeline/feature_builder/main.py index 8d08f9e..83820ff 100644 --- a/apps/pipeline/feature_builder/main.py +++ b/apps/pipeline/feature_builder/main.py @@ -14,17 +14,22 @@ from libs.common.logging import bind_job_run_id, configure_logging, get_logger from libs.db.models import Event, JobRun from libs.db.session import get_session from libs.features.builder import build_features_for_event -from libs.oracle_client import FinancialService, PriceService, make_oracle_client +from libs.oracle_client import CompanyService, FinancialService, PriceService, make_oracle_client logger = get_logger(__name__) -async def run_feature_builder(run_id: str) -> dict[str, int]: +async def run_feature_builder( + run_id: str, + start_date: str | None = None, + end_date: str | None = None, +) -> dict[str, int]: stats = {"seen": 0, "built": 0, "skipped": 0, "errors": 0} async with make_oracle_client() as client: price_svc = PriceService(client) financial_svc = FinancialService(client) + company_svc = CompanyService(client) async with get_session() as session: job = JobRun( @@ -37,16 +42,21 @@ async def run_feature_builder(run_id: str) -> dict[str, int]: session.add(job) await session.flush() - result = await session.execute( - select(Event).where(Event.status == "pending") - ) + stmt = select(Event).where(Event.status == "pending") + if start_date: + stmt = stmt.where(Event.event_date >= dt.date.fromisoformat(start_date)) + if end_date: + stmt = stmt.where(Event.event_date <= dt.date.fromisoformat(end_date)) + + result = await session.execute(stmt.order_by(Event.event_date, Event.event_id)) events = result.scalars().all() stats["seen"] = len(events) for event in events: try: snapshots = await build_features_for_event( - session, event, price_svc, financial_service=financial_svc + session, event, price_svc, financial_service=financial_svc, + company_service=company_svc, ) if snapshots is None: event.status = "rejected" @@ -73,13 +83,15 @@ async def run_feature_builder(run_id: str) -> dict[str, int]: def main() -> None: parser = argparse.ArgumentParser(description="Feature Builder") parser.add_argument("--run-id", default=new_job_run_id()) + parser.add_argument("--start-date", default=None, metavar="YYYY-MM-DD") + parser.add_argument("--end-date", default=None, metavar="YYYY-MM-DD") args = parser.parse_args() settings = get_settings() configure_logging(settings.log_level) bind_job_run_id(args.run_id) - asyncio.run(run_feature_builder(args.run_id)) + asyncio.run(run_feature_builder(args.run_id, start_date=args.start_date, end_date=args.end_date)) if __name__ == "__main__": diff --git a/apps/pipeline/filing_fetcher/main.py b/apps/pipeline/filing_fetcher/main.py index 7bfcbab..9108646 100644 --- a/apps/pipeline/filing_fetcher/main.py +++ b/apps/pipeline/filing_fetcher/main.py @@ -5,6 +5,7 @@ import argparse import asyncio import datetime as dt import uuid +from dataclasses import dataclass from sqlalchemy import select @@ -20,14 +21,125 @@ from libs.oracle_client.exceptions import OracleNotFoundError logger = get_logger(__name__) -async def fetch_exhibits(run_id: str) -> dict[str, int]: +@dataclass(slots=True) +class _FetchedExhibit: + exhibit_type: str + checksum: str + cache_path: str + + +@dataclass(slots=True) +class _FetchResult: + doc_id: str + accession_no: str | None + fetched: list[_FetchedExhibit] + item_numbers: list[str] | None + errors: int + + +async def _fetch_doc( + svc: FilingsService, + doc: Document, + exhibit_types: list[str], +) -> _FetchResult: + if not doc.accession_no: + return _FetchResult( + doc_id=str(doc.document_id), + accession_no=None, + fetched=[], + item_numbers=None, + errors=0, + ) + + fetched: list[_FetchedExhibit] = [] + errors = 0 + + for exhibit_type in exhibit_types: + if exists_exhibit(doc.accession_no, exhibit_type): + logger.info( + "exhibit_already_cached", + accession_no=doc.accession_no, + exhibit_type=exhibit_type, + ) + continue + + try: + response = await asyncio.wait_for( + svc.get_exhibit(doc.accession_no, exhibit_type), + timeout=30.0, + ) + checksum = write_exhibit(doc.accession_no, exhibit_type, response.content) + + from libs.common.file_store import exhibit_path + + fetched.append( + _FetchedExhibit( + exhibit_type=exhibit_type, + checksum=checksum, + cache_path=str(exhibit_path(doc.accession_no, exhibit_type)), + ) + ) + logger.info( + "exhibit_fetched", + accession_no=doc.accession_no, + exhibit_type=exhibit_type, + ) + except OracleNotFoundError: + logger.warning( + "exhibit_not_found", + accession_no=doc.accession_no, + exhibit_type=exhibit_type, + ) + except Exception as exc: + logger.error( + "exhibit_fetch_error", + accession_no=doc.accession_no, + exhibit_type=exhibit_type, + error=str(exc), + ) + errors += 1 + + item_numbers: list[str] | None = None + if not doc.item_numbers: + try: + item_numbers = await asyncio.wait_for( + svc.get_filing_items(doc.accession_no), + timeout=15.0, + ) + if item_numbers: + logger.info( + "item_numbers_extracted", + accession_no=doc.accession_no, + items=item_numbers, + ) + except Exception as exc: + logger.debug( + "item_numbers_extraction_failed", + accession_no=doc.accession_no, + error=str(exc), + ) + + return _FetchResult( + doc_id=str(doc.document_id), + accession_no=doc.accession_no, + fetched=fetched, + item_numbers=item_numbers, + errors=errors, + ) + + +async def fetch_exhibits( + run_id: str, + start_date: str | None = None, + end_date: str | None = None, +) -> dict[str, int]: settings = get_settings() app_config = settings.get_app_config() exhibit_types = app_config.get("pipeline", {}).get("exhibit_types", ["EX-99.1"]) + concurrency = max(1, int(app_config.get("pipeline", {}).get("fetcher_concurrency", 20))) stats = {"seen": 0, "written": 0, "skipped": 0, "errors": 0} - - batch_size = 100 + batch_size = max(10, int(app_config.get("pipeline", {}).get("fetcher_batch_size", 25))) async with make_oracle_client() as client: svc = FilingsService(client) @@ -44,115 +156,67 @@ async def fetch_exhibits(run_id: str) -> dict[str, int]: await session.flush() await session.commit() - result = await session.execute( - select(Document).where(Document.parsed_status == "pending") - ) + stmt = select(Document).where(Document.parsed_status == "pending") + if start_date: + stmt = stmt.where(Document.filing_date >= dt.date.fromisoformat(start_date)) + if end_date: + stmt = stmt.where(Document.filing_date <= dt.date.fromisoformat(end_date)) + + result = await session.execute(stmt.order_by(Document.filing_date, Document.document_id)) docs = result.scalars().all() stats["seen"] = len(docs) + doc_map = {str(doc.document_id): doc for doc in docs} - for idx, doc in enumerate(docs, 1): - if not doc.accession_no: - stats["skipped"] += 1 - continue - - fetched_any = False - for exhibit_type in exhibit_types: - if exists_exhibit(doc.accession_no, exhibit_type): - logger.info( - "exhibit_already_cached", - accession_no=doc.accession_no, - exhibit_type=exhibit_type, - ) - fetched_any = True - continue - - try: - response = await asyncio.wait_for( - svc.get_exhibit(doc.accession_no, exhibit_type), - timeout=30.0, - ) - checksum = write_exhibit( - doc.accession_no, exhibit_type, response.content - ) - - from libs.common.file_store import exhibit_path - - cache_path = str(exhibit_path(doc.accession_no, exhibit_type)) - - existing_cache = await session.execute( - select(ExhibitCache).where( - ExhibitCache.accession_no == doc.accession_no, - ExhibitCache.exhibit_type == exhibit_type, - ) - ) - if existing_cache.scalar_one_or_none() is None: - cache_row = ExhibitCache( - accession_no=doc.accession_no, - exhibit_type=exhibit_type, - content_hash=checksum, - cache_path=cache_path, - ) - session.add(cache_row) - - fetched_any = True - stats["written"] += 1 - logger.info( - "exhibit_fetched", - accession_no=doc.accession_no, - exhibit_type=exhibit_type, - ) - - except OracleNotFoundError: - logger.warning( - "exhibit_not_found", - accession_no=doc.accession_no, - exhibit_type=exhibit_type, - ) - except Exception as exc: - logger.error( - "exhibit_fetch_error", - accession_no=doc.accession_no, - exhibit_type=exhibit_type, - error=str(exc), - ) - stats["errors"] += 1 - - # Extract item_numbers from SGML header if not already set - if not doc.item_numbers: - try: - items = await asyncio.wait_for( - svc.get_filing_items(doc.accession_no), - timeout=15.0, - ) - if items: - doc.item_numbers = items - logger.info( - "item_numbers_extracted", - accession_no=doc.accession_no, - items=items, - ) - except Exception as exc: - logger.debug( - "item_numbers_extraction_failed", - accession_no=doc.accession_no, - error=str(exc), - ) - - # Always advance to ready_for_parse (exhibit may not exist) - doc.parsed_status = "ready_for_parse" - doc.updated_at_utc = dt.datetime.now(tz=dt.UTC) - - # Commit in batches to preserve progress - if idx % batch_size == 0: - await session.commit() - logger.info( - "batch_committed", - processed=idx, - total=stats["seen"], - written=stats["written"], - errors=stats["errors"], + for batch_start in range(0, len(docs), batch_size): + batch_docs = docs[batch_start: batch_start + batch_size] + + for task_start in range(0, len(batch_docs), concurrency): + task_docs = batch_docs[task_start: task_start + concurrency] + results = await asyncio.gather( + *[_fetch_doc(svc, doc, exhibit_types) for doc in task_docs] ) + for result_row in results: + doc = doc_map[result_row.doc_id] + if not result_row.accession_no: + stats["skipped"] += 1 + continue + + for fetched in result_row.fetched: + existing_cache = await session.execute( + select(ExhibitCache).where( + ExhibitCache.accession_no == result_row.accession_no, + ExhibitCache.exhibit_type == fetched.exhibit_type, + ) + ) + if existing_cache.scalar_one_or_none() is None: + session.add( + ExhibitCache( + accession_no=result_row.accession_no, + exhibit_type=fetched.exhibit_type, + content_hash=fetched.checksum, + cache_path=fetched.cache_path, + ) + ) + stats["written"] += 1 + + if result_row.item_numbers: + doc.item_numbers = result_row.item_numbers + + doc.parsed_status = "ready_for_parse" + doc.updated_at_utc = dt.datetime.now(tz=dt.UTC) + stats["errors"] += result_row.errors + + processed = min(batch_start + len(batch_docs), len(docs)) + await session.commit() + logger.info( + "batch_committed", + processed=processed, + total=stats["seen"], + written=stats["written"], + errors=stats["errors"], + ) + job.status = "succeeded" if stats["errors"] == 0 else "partial" job.finished_at_utc = dt.datetime.now(tz=dt.UTC) job.records_seen = stats["seen"] @@ -167,13 +231,15 @@ async def fetch_exhibits(run_id: str) -> dict[str, int]: def main() -> None: parser = argparse.ArgumentParser(description="Filing Fetcher") parser.add_argument("--run-id", default=new_job_run_id()) + parser.add_argument("--start-date", default=None, metavar="YYYY-MM-DD") + parser.add_argument("--end-date", default=None, metavar="YYYY-MM-DD") args = parser.parse_args() settings = get_settings() configure_logging(settings.log_level) bind_job_run_id(args.run_id) - asyncio.run(fetch_exhibits(args.run_id)) + asyncio.run(fetch_exhibits(args.run_id, start_date=args.start_date, end_date=args.end_date)) if __name__ == "__main__": diff --git a/apps/pipeline/label_generator/main.py b/apps/pipeline/label_generator/main.py index f003caa..822888a 100644 --- a/apps/pipeline/label_generator/main.py +++ b/apps/pipeline/label_generator/main.py @@ -23,6 +23,8 @@ async def run_label_generator( run_id: str, entry_convention: str = "next_open_after_reaction_close", event_id_filter: str | None = None, + start_date: str | None = None, + end_date: str | None = None, ) -> dict[str, int]: stats = {"seen": 0, "labeled": 0, "skipped": 0, "errors": 0} @@ -46,8 +48,12 @@ async def run_label_generator( if event_id_filter: stmt = stmt.where(Event.event_id == event_id_filter) + if start_date: + stmt = stmt.where(Event.event_date >= dt.date.fromisoformat(start_date)) + if end_date: + stmt = stmt.where(Event.event_date <= dt.date.fromisoformat(end_date)) - result = await session.execute(stmt) + result = await session.execute(stmt.order_by(Event.event_date, Event.event_id)) rows = result.all() stats["seen"] = len(rows) @@ -57,7 +63,7 @@ async def run_label_generator( stats["skipped"] += 1 continue - # Skip if label already exists + # Skip if label already exists (unless it's pending — regenerate those) existing = await session.execute( select(EventLabel).where( EventLabel.event_id == event.event_id, @@ -65,9 +71,14 @@ async def run_label_generator( EventLabel.label_version == LABEL_VERSION, ) ) - if existing.scalar_one_or_none() is not None: - stats["skipped"] += 1 - continue + existing_label = existing.scalar_one_or_none() + if existing_label is not None: + if existing_label.label_status != "pending": + stats["skipped"] += 1 + continue + # Pending label — market has since closed; delete and regenerate + await session.delete(existing_label) + await session.flush() try: label = await generate_labels( @@ -109,6 +120,8 @@ def main() -> None: help="Entry price convention", ) parser.add_argument("--event-id", default=None, help="Process a single event by ID") + parser.add_argument("--start-date", default=None, metavar="YYYY-MM-DD") + parser.add_argument("--end-date", default=None, metavar="YYYY-MM-DD") args = parser.parse_args() settings = get_settings() @@ -120,6 +133,8 @@ def main() -> None: run_id=args.run_id, entry_convention=args.entry_convention, event_id_filter=args.event_id, + start_date=args.start_date, + end_date=args.end_date, ) ) diff --git a/libs/export/snapshot_export.py b/libs/export/snapshot_export.py index 31366f5..cf16820 100644 --- a/libs/export/snapshot_export.py +++ b/libs/export/snapshot_export.py @@ -14,6 +14,7 @@ import pyarrow.parquet as pq from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from libs.common.config import get_settings from libs.common.logging import get_logger from libs.common.time_utils import ( filing_time_bucket as classify_time_bucket, @@ -25,6 +26,14 @@ logger = get_logger(__name__) MANIFEST_FILENAME = "manifest.json" +_EXPORT_ENRICHMENTS = [ + "market_bar_backfill_v1", + "pre_event_momentum_v1", + "price_vs_sma20_v1", + "prior_event_drift_v1", + "macro_regime_v1", +] + _UNIVERSE_PROFILE_MIDLARGE_LIQUID_LONG_V1 = "midlarge-liquid-long-v1" _UNIVERSE_PROFILE_MIDPLUS_LIQUID_LONG_V1 = "midplus-liquid-long-v1" @@ -64,6 +73,68 @@ _UNIVERSE_PROFILES: dict[str, dict[str, Any]] = { } +def _resolve_local_universe_profile_fallback( + universe_profile: str, + *, + exclude_snapshot_id: str | None = None, +) -> dict[str, dict[str, Any]]: + """Load deterministic symbol metadata from existing local snapshots. + + This is used only when the live screener is unavailable. We intentionally + avoid falling through to another live Oracle path here because that would + make exports silently depend on a different remote endpoint. + """ + settings = get_settings() + candidate_roots = [Path(settings.parquet_dir), Path("data/datasets/snapshots")] + seen_roots: set[Path] = set() + manifests: list[tuple[str, Path]] = [] + + for root in candidate_roots: + root = root.resolve() + if root in seen_roots or not root.exists(): + continue + seen_roots.add(root) + for manifest_path in root.glob("*/manifest.json"): + try: + manifest = json.loads(manifest_path.read_text()) + except Exception: + continue + if manifest.get("universe_profile") != universe_profile: + continue + if exclude_snapshot_id and manifest.get("snapshot_id") == exclude_snapshot_id: + continue + manifests.append((str(manifest.get("created_at_utc") or ""), manifest_path.parent)) + + symbol_meta: dict[str, dict[str, Any]] = {} + for _created_at, snap_dir in sorted(manifests, reverse=True): + for split_name in ("train", "valid", "test"): + parquet_path = snap_dir / f"{split_name}.parquet" + if not parquet_path.exists(): + continue + try: + parquet_file = pq.ParquetFile(str(parquet_path)) + available = set(parquet_file.schema_arrow.names) + required = {"ticker", "market_cap_proxy", "exchange_proxy"} + if not required.issubset(available): + continue + table = parquet_file.read(columns=["ticker", "market_cap_proxy", "exchange_proxy"]) + except Exception: + continue + for row in table.to_pylist(): + ticker = str(row.get("ticker") or "").upper() + if not ticker or ticker in symbol_meta: + continue + market_cap_proxy = row.get("market_cap_proxy") + exchange_proxy = row.get("exchange_proxy") + if market_cap_proxy is None and exchange_proxy is None: + continue + symbol_meta[ticker] = { + "market_cap_proxy": market_cap_proxy, + "exchange_proxy": exchange_proxy, + } + return symbol_meta + + def _get_git_commit_hash() -> str: """Return the current git commit hash (short), or 'unknown'.""" try: @@ -346,6 +417,7 @@ async def _backfill_market_fields(rows: list[dict[str, Any]]) -> None: row.get("avg_dollar_volume_20d") is None or row.get("reaction_day_low") is None or row.get("reaction_day_high") is None + or row.get("pre_event_momentum_20d") is None ) ] if not pending_rows: @@ -394,6 +466,20 @@ async def _backfill_market_fields(rows: list[dict[str, Any]]) -> None: float(date_bars[d]["close"]) * float(date_bars[d]["volume"]) for d in prior ) / len(prior) + # Pre-event momentum: 20d return and SMA20 position + if row.get("pre_event_momentum_20d") is None and idx >= 20: + close_now = float(event_bar.get("close", 0)) + close_20d = float(date_bars[sorted_dates[idx - 20]].get("close", 0)) + if close_20d > 0 and close_now > 0: + row["pre_event_momentum_20d"] = (close_now - close_20d) / close_20d + if row.get("price_vs_sma20") is None and idx >= 20: + close_now = float(event_bar.get("close", 0)) + sma20 = sum( + float(date_bars[sorted_dates[idx - i]].get("close", 0)) + for i in range(20) + ) / 20 + if sma20 > 0 and close_now > 0: + row["price_vs_sma20"] = (close_now - sma20) / sma20 async def export_dataset_snapshot( @@ -445,7 +531,23 @@ async def export_dataset_snapshot( out_path = Path(output_dir) / snapshot_id out_path.mkdir(parents=True, exist_ok=True) - profile_symbol_meta = await _resolve_universe_profile(universe_profile) + profile_symbol_meta: dict[str, dict[str, Any]] = {} + if universe_profile: + try: + profile_symbol_meta = await _resolve_universe_profile(universe_profile) + except Exception as exc: + profile_symbol_meta = _resolve_local_universe_profile_fallback( + universe_profile, + exclude_snapshot_id=snapshot_id, + ) + logger.warning( + "snapshot_export_universe_profile_resolution_failed", + universe_profile=universe_profile, + error=str(exc), + fallback_symbols=len(profile_symbol_meta), + ) + if not profile_symbol_meta: + raise explicit_symbols = {s.upper() for s in symbols} if symbols else None if profile_symbol_meta: profile_symbols = set(profile_symbol_meta.keys()) @@ -457,7 +559,8 @@ async def export_dataset_snapshot( if profile_symbol_meta else explicit_symbols ) - explicit_symbol_meta = await _resolve_symbol_metadata(missing_meta_symbols) + if missing_meta_symbols: + explicit_symbol_meta = await _resolve_symbol_metadata(missing_meta_symbols) # Query: JOIN feature_snapshots + event_labels via event_id # When merging multiple versions, query all and group by event_id @@ -607,6 +710,7 @@ async def export_dataset_snapshot( "created_at_utc": utc_now().isoformat(), "code_commit_hash": _get_git_commit_hash(), "feature_version": "+".join(versions), + "export_enrichments": list(_EXPORT_ENRICHMENTS), "parser_version": parser_version, "label_version": label_version, "split_policy": split_policy, diff --git a/libs/features/event_features.py b/libs/features/event_features.py index 5b62c8a..6ed9352 100644 --- a/libs/features/event_features.py +++ b/libs/features/event_features.py @@ -85,6 +85,8 @@ def compute_event_features(parser_output: dict[str, Any]) -> dict[str, Any]: "event_type": parser_output.get("event_type", "unknown"), "event_direction": parser_output.get("event_direction", "unknown"), "parse_confidence_overall": confidence.overall, + "parse_confidence_event_direction": confidence.event_direction, + "parse_confidence_guidance": confidence.guidance, "filing_time_bucket": parser_output.get("filing_time_bucket", "unknown"), "event_date": parser_output.get("event_date", ""), } diff --git a/libs/labeler/label_generator.py b/libs/labeler/label_generator.py index 8565330..04f975e 100644 --- a/libs/labeler/label_generator.py +++ b/libs/labeler/label_generator.py @@ -136,8 +136,13 @@ async def generate_labels( event_date: dt.date = event.event_date reaction_date = compute_reaction_date(event_date, filing_time_bucket) - # 2. Compute entry_date = next trading day after reaction_date - entry_date = next_trading_day(reaction_date) + # 2. Compute entry_date + # reaction_close: enter at close of reaction_date itself + # next_open_after_reaction_close: enter at open of next trading day + if entry_convention == "reaction_close": + entry_date = reaction_date + else: + entry_date = next_trading_day(reaction_date) # 3. Fetch price bars (entry_date + _LOOK_AHEAD_DAYS trading days) fetch_start = entry_date @@ -170,12 +175,18 @@ async def generate_labels( ) if not bars: + # For reaction_close, the bar may not exist yet if market hasn't closed. + # Return "pending" so the pipeline can regenerate after market close. + if entry_convention == "reaction_close" and entry_date >= dt.date.today(): + label_status_no_bar = "pending" + else: + label_status_no_bar = "unavailable" return EventLabel( event_id=event.event_id, entry_convention=entry_convention, reaction_date=reaction_date, entry_date=entry_date, - label_status="unavailable", + label_status=label_status_no_bar, invalid_event_for_labeling=False, label_version=LABEL_VERSION, ) diff --git a/libs/oracle_client/price.py b/libs/oracle_client/price.py index b035ea5..e9c7f31 100644 --- a/libs/oracle_client/price.py +++ b/libs/oracle_client/price.py @@ -51,6 +51,25 @@ class PriceService: bars = [IntradayBar.model_validate(b) for b in data.get("data", [])] return IntradayResponse(ticker=data.get("ticker", ticker), bars=bars) + async def get_historical_intraday( + self, + ticker: str, + date: str, + interval: str = "5min", + ) -> IntradayResponse: + """Fetch historical intraday bars for a specific date via Alpaca endpoint. + + Uses: GET /api/v1/alpaca/intraday/{ticker}?interval=5min&start_date=DATE&end_date=DATE + """ + params: dict[str, str] = { + "interval": interval, + "start_date": date, + "end_date": date, + } + data = await self._client.get(f"/api/v1/alpaca/intraday/{ticker}", params=params) + bars = [IntradayBar.model_validate(b) for b in data.get("data", [])] + return IntradayResponse(ticker=data.get("ticker", ticker), bars=bars) + async def get_today(self, ticker: str) -> PriceDataResponse: data = await self._client.get(f"/api/v1/price/today/{ticker}") bars = [ diff --git a/libs/parser/rule_parser.py b/libs/parser/rule_parser.py index b018393..c0b9791 100644 --- a/libs/parser/rule_parser.py +++ b/libs/parser/rule_parser.py @@ -23,8 +23,9 @@ _ITEM_RE = re.compile(r"item\s+(\d+\.\d+)", re.IGNORECASE) # Guidance _GUIDANCE_RAISED = re.compile( - r"(guidance raised|above prior outlook|raised.*guidance|increased.*guidance" - r"|raised.*forecast|above.*expectations|above.*consensus|raised.*outlook" + r"(guidance raised|above prior outlook|raised.*guidance|raising.*guidance" + r"|increased.*guidance|raised.*forecast|raising.*forecast" + r"|above.*expectations|above.*consensus|raised.*outlook|raising.*outlook" r"|above.*prior.*guidance|exceed.*guidance)", re.IGNORECASE, ) @@ -38,12 +39,24 @@ _GUIDANCE_MAINTAINED = re.compile( r"(reaffirm|maintain.*guidance|in.line.*outlook|inline.*guidance|on.track)", re.IGNORECASE, ) -_GUIDANCE_WITHDRAWN = re.compile(r"(withdraw.*guidance|suspend.*guidance)", re.IGNORECASE) +_GUIDANCE_WITHDRAWN = re.compile( + r"\b(?:withdrawn|withdraw|withdrawing|suspended|suspend|suspending)\b" + r"(?:\W+\w+){0,3}\W+guidance\b", + re.IGNORECASE, +) + +_EARNINGS_RELEASE_HINT = re.compile( + r"((?:reports?|reported|announced)\s+(?:its\s+)?(?:fiscal\s+)?" + r"(?:first|second|third|fourth|q[1-4]|[1-4](?:st|nd|rd|th))?.{0,40}financial results)" + r"|financial results for (?:its\s+)?(?:fiscal\s+)?" + r"(?:first|second|third|fourth|q[1-4]|[1-4](?:st|nd|rd|th)).{0,40}quarter", + re.IGNORECASE, +) # Signals _DEMAND_STRONG = re.compile( r"(demand remains strong|strong demand|robust demand|record demand" - r"|demand acceleration|pipeline.*strong|strong.*pipeline)", + r"|demand acceleration|pipeline.*strong|strong.*pipeline|broad.based demand)", re.IGNORECASE, ) _DEMAND_WEAK = re.compile( @@ -132,6 +145,12 @@ def _classify_event_type(items: list[str]) -> str: return "unknown" +def _infer_event_type_from_text(text: str) -> str: + if _EARNINGS_RELEASE_HINT.search(text): + return "earnings_release" + return "unknown" + + def _detect_guidance(text: str) -> GuidanceOutput: if _GUIDANCE_WITHDRAWN.search(text): return GuidanceOutput(status="withdrawn", scope="unknown", notes="Guidance withdrawn.") @@ -232,6 +251,12 @@ def _classify_direction( elif signals.margin_quality == "deteriorating": bearish_signals += 1 + if signals.backlog_or_bookings == "present": + bullish_signals += 1 + + if signals.customer_expansion == "present": + bullish_signals += 1 + if risk_flags.financing_related: bearish_signals += 1 @@ -346,6 +371,8 @@ class RuleBasedParser: items = metadata.get("item_numbers") or _extract_item_numbers(text) event_type = _classify_event_type(items) + if event_type == "unknown": + event_type = _infer_event_type_from_text(text) guidance = _detect_guidance(text) signals = _detect_signals(text) risk_flags = _detect_risk_flags(text) diff --git a/libs/parser/text_normalizer.py b/libs/parser/text_normalizer.py index f0b3e33..2f30feb 100644 --- a/libs/parser/text_normalizer.py +++ b/libs/parser/text_normalizer.py @@ -14,6 +14,7 @@ _BOILERPLATE_PATTERNS = [ ] _WHITESPACE = re.compile(r"\s{3,}") +_HTML_MARKERS = re.compile(r"<(?:!doctype|html|body|div|table|p|br|font|span)\b", re.IGNORECASE) def html_to_text(html: str) -> str: @@ -25,6 +26,12 @@ def html_to_text(html: str) -> str: return soup.get_text(separator="\n") +def looks_like_html(text: str) -> bool: + """Return True when SEC exhibit text appears to still be HTML markup.""" + sample = text.lstrip()[:2048] + return bool(_HTML_MARKERS.search(sample)) + + def normalize_unicode(text: str) -> str: """Normalize unicode to NFC and replace fancy quotes/dashes.""" text = unicodedata.normalize("NFC", text)