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) <noreply@anthropic.com>

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,
         )
     )
 

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(
+        stmt = select(Document).where(Document.parsed_status == "ready_for_parse")
-            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__":

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(
+            stmt = select(Event).where(Event.status == "pending")
-                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__":

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 = max(10, int(app_config.get("pipeline", {}).get("fetcher_batch_size", 25)))
-    batch_size = 100
 
     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(
+            stmt = select(Document).where(Document.parsed_status == "pending")
-                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):
+            for batch_start in range(0, len(docs), batch_size):
-                if not doc.accession_no:
+                batch_docs = docs[batch_start: batch_start + batch_size]
-                    stats["skipped"] += 1
+
-                    continue
+                for task_start in range(0, len(batch_docs), concurrency):
-
+                    task_docs = batch_docs[task_start: task_start + concurrency]
-                fetched_any = False
+                    results = await asyncio.gather(
-                for exhibit_type in exhibit_types:
+                        *[_fetch_doc(svc, doc, exhibit_types) for doc in task_docs]
-                    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 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__":

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:
+            existing_label = existing.scalar_one_or_none()
-                stats["skipped"] += 1
+            if existing_label is not None:
-                continue
+                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,
         )
     )
 

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,

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", ""),
     }

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
+    # 2. Compute entry_date
-    entry_date = next_trading_day(reaction_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,
         )

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 = [

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"(guidance raised|above prior outlook|raised.*guidance|raising.*guidance"
-    r"|raised.*forecast|above.*expectations|above.*consensus|raised.*outlook"
+    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)

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)