From e46395bfeb09995b71f58536f3b93b02844e57ea Mon Sep 17 00:00:00 2001 From: I Luk Kim Date: Sat, 9 May 2026 02:21:48 -0700 Subject: [PATCH] Backfill: renormalize 1,069 historical oracle-fallback rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Apply libs.parser.event_type_normalizer (added in commit 722e5cf) to existing events whose parser_version LIKE 'oracle-fallback%' so historical rows match the forward-going normalization wired into the parser. Implementation: - New renormalize_oracle_fallback_events() in apps/pipeline/event_parser/main.py - SELECT filter Event.parser_version.like("oracle-fallback%") — broader than a hardcoded IN list, so already-aligned values skip naturally and future additions to _ORACLE_TO_STRATEGY get picked up automatically - New --renormalize-oracle-fallback CLI flag, chainable with --reparse - JobRun row written (job_name=event_parser_renormalize_oracle) - Per-row renormalize_event_updated INFO log + final renormalize_done summary with transition counters Live DB run: seen=2937 / updated=1069 / skipped=1868 / errors=0. Wall ~2 sec (pure DB UPDATEs, no Oracle calls). Transitions: earnings_result -> earnings_release : 412 shareholder_vote -> other_material_event : 409 regulation_fd -> guidance_update : 237 acquisition_disposition -> other_material_event : 7 other -> other_material_event : 4 Unmapped Oracle values (financial_obligation 225, articles_amendment 81, contract_termination 48, etc.) preserved verbatim — honest filter-drop. Stale-by-design (mirrors existing reparse_events convention): - Event.event_id PK still embeds old raw event_type substring - EventParse.output_json["event_type"] still carries raw Oracle value Strategies read Event.event_type, not those fields. Avoids cascading PK rewrites across event_parses/feature_snapshots/event_labels tables. Integration test: tests/integration/test_renormalize_oracle_fallback.py inserts 4 fixtures, drives _apply_oracle_renormalization() against the rolled-back db_session, asserts updated/skipped/error counts and final row state. Snapshot rebuild not run — nightly auto-rebuild picks up normalized values incrementally. Co-Authored-By: Claude Opus 4.7 --- apps/pipeline/event_parser/main.py | 122 +++++++++++++++++- .../test_renormalize_oracle_fallback.py | 93 +++++++++++++ 2 files changed, 213 insertions(+), 2 deletions(-) create mode 100644 tests/integration/test_renormalize_oracle_fallback.py diff --git a/apps/pipeline/event_parser/main.py b/apps/pipeline/event_parser/main.py index ae52598..417a6e6 100644 --- a/apps/pipeline/event_parser/main.py +++ b/apps/pipeline/event_parser/main.py @@ -23,7 +23,7 @@ from libs.db.models import Document, Event, EventParse, JobRun, SymbolMaster from libs.db.session import get_session from libs.oracle_client import FilingsService, make_oracle_client from libs.oracle_client.models import FilingEventEntry -from libs.parser.event_type_normalizer import normalize_oracle_event +from libs.parser.event_type_normalizer import normalize_oracle_event, normalize_oracle_event_type 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 looks_like_html, normalize_text @@ -529,6 +529,111 @@ async def reparse_events(run_id: str) -> dict[str, int]: return stats +async def _apply_oracle_renormalization(session) -> dict[str, int | dict[str, int]]: + """Inner helper: scan oracle-fallback rows, apply the string normalizer, update. + + Extracted so tests can drive it against an injected session without forking + a separate ``get_session()`` context. Caller owns commit/rollback. + """ + stats: dict[str, int | dict[str, int]] = { + "seen": 0, + "updated": 0, + "skipped": 0, + "errors": 0, + "transitions": {}, # "raw -> normalized" -> count + } + + result = await session.execute( + select(Event).where(Event.parser_version.like("oracle-fallback%")) + ) + events = result.scalars().all() + stats["seen"] = len(events) + + if not events: + logger.info("renormalize_nothing_to_do") + return stats + + transitions: dict[str, int] = stats["transitions"] # type: ignore[assignment] + + for event in events: + try: + raw = event.event_type + normalized = normalize_oracle_event_type(raw) + if normalized == raw: + stats["skipped"] = int(stats["skipped"]) + 1 # type: ignore[arg-type] + continue + + event.event_type = normalized + event.updated_at_utc = dt.datetime.now(tz=dt.UTC) + stats["updated"] = int(stats["updated"]) + 1 # type: ignore[arg-type] + key = f"{raw} -> {normalized}" + transitions[key] = transitions.get(key, 0) + 1 + logger.info( + "renormalize_event_updated", + event_id=event.event_id, + old_type=raw, + new_type=normalized, + ) + except Exception as exc: + logger.error( + "renormalize_error", + event_id=event.event_id, + error=str(exc), + ) + stats["errors"] = int(stats["errors"]) + 1 # type: ignore[arg-type] + + return stats + + +async def renormalize_oracle_fallback_events(run_id: str) -> dict[str, int | dict[str, int]]: + """Apply the Oracle vocabulary normalizer to historical oracle-fallback rows. + + Scope: rows where ``parser_version LIKE 'oracle-fallback%'``. The string-only + ``normalize_oracle_event_type`` is applied to each ``event_type`` and the row + is updated only when the value changes. Already-aligned values + (``earnings_release``, ``other_material_event``, etc.) are skipped naturally + because the normalizer returns them unchanged. + + Out of scope (intentional, mirrors ``reparse_events()`` behavior): + - ``event_id`` rewrite — the PK embeds the original ``event_type`` string but + changing it would cascade across ``event_parses``, ``feature_snapshots``, + and ``event_labels``. Left stale; harmless to downstream consumers. + - ``EventParse.output_json['event_type']`` — also left stale. ``Event.event_type`` + is the column the strategy reads. + """ + stats: dict[str, int | dict[str, int]] = { + "seen": 0, + "updated": 0, + "skipped": 0, + "errors": 0, + "transitions": {}, + } + + async with get_session() as session: + job = JobRun( + job_run_id=uuid.UUID(run_id), + job_name="event_parser_renormalize_oracle", + source_name="oracle", + run_date=dt.date.today(), + status="running", + ) + session.add(job) + await session.flush() + + stats = await _apply_oracle_renormalization(session) + + errors = int(stats["errors"]) # type: ignore[arg-type] + job.status = "succeeded" if errors == 0 else "partial" + job.finished_at_utc = dt.datetime.now(tz=dt.UTC) + job.records_seen = int(stats["seen"]) # type: ignore[arg-type] + job.records_written = int(stats["updated"]) # type: ignore[arg-type] + job.error_count = errors + + logger.info("renormalize_done", **{k: v for k, v in stats.items() if k != "transitions"}) + logger.info("renormalize_transitions", transitions=stats["transitions"]) + return stats + + def main() -> None: parser = argparse.ArgumentParser(description="Event Parser") parser.add_argument("--run-id", default=new_job_run_id()) @@ -539,13 +644,26 @@ def main() -> None: action="store_true", help="Re-classify existing events with event_type='unknown' using Oracle API item numbers", ) + parser.add_argument( + "--renormalize-oracle-fallback", + action="store_true", + help=( + "One-shot: apply libs.parser.event_type_normalizer to historical " + "oracle-fallback rows. Pure DB UPDATEs (no Oracle API calls)." + ), + ) args = parser.parse_args() settings = get_settings() configure_logging(settings.log_level) bind_job_run_id(args.run_id) - if args.reparse: + if args.renormalize_oracle_fallback: + asyncio.run(renormalize_oracle_fallback_events(args.run_id)) + if args.reparse: + # Allow chaining: renormalize first, then run the unknown reparse. + asyncio.run(reparse_events(new_job_run_id())) + elif args.reparse: asyncio.run(reparse_events(args.run_id)) else: asyncio.run(run_event_parser(args.run_id, start_date=args.start_date, end_date=args.end_date)) diff --git a/tests/integration/test_renormalize_oracle_fallback.py b/tests/integration/test_renormalize_oracle_fallback.py new file mode 100644 index 0000000..8d75dde --- /dev/null +++ b/tests/integration/test_renormalize_oracle_fallback.py @@ -0,0 +1,93 @@ +"""Integration test: --renormalize-oracle-fallback applies the normalizer in DB. + +Drives ``_apply_oracle_renormalization()`` against the test session so the +helper executes against a real Postgres while the surrounding transaction +rolls back at the end of the test. +""" +from __future__ import annotations + +import datetime as dt + +import pytest +from sqlalchemy import select + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_renormalize_updates_oracle_fallback_event_type(db_session): + from apps.pipeline.event_parser.main import _apply_oracle_renormalization + from libs.db.models import Document, Event, IssuerMaster, SymbolMaster + + issuer = IssuerMaster( + issuer_id="ISSUER::RNRM::0000000001", + issuer_name="Renormalize Test Co.", + ticker="RNRM", + ) + db_session.add(issuer) + + symbol = SymbolMaster( + symbol_id="SYM::RNRM::XNYS", + issuer_id="ISSUER::RNRM::0000000001", + ticker="RNRM", + venue="XNYS", + ) + db_session.add(symbol) + + doc = Document( + document_id="DOC::test::ISSUER::RNRM::2026-04-01::RNRMACC001", + source_name="sec", + form_type="8-K", + filing_date=dt.date(2026, 4, 1), + accession_no="RNRMACC001", + parsed_status="succeeded", + ) + db_session.add(doc) + await db_session.flush() + + # Three rows covering: a stale Oracle synonym (earnings_result), + # an aggregation case (shareholder_vote), and an already-aligned row + # that must NOT change. + targets = [ + ("EVT::test::earnings_result::0", "earnings_result", "earnings_release"), + ("EVT::test::shareholder_vote::0", "shareholder_vote", "other_material_event"), + ("EVT::test::other_material_event::0", "other_material_event", "other_material_event"), + # Honest pass-through case — unmapped Oracle vocab stays as-is. + ("EVT::test::financial_obligation::0", "financial_obligation", "financial_obligation"), + ] + + for evt_id, raw_type, _expected in targets: + db_session.add( + Event( + event_id=evt_id, + primary_document_id=doc.document_id, + symbol_id="SYM::RNRM::XNYS", + event_type=raw_type, + event_direction="unknown", + event_date=dt.date(2026, 4, 1), + filed_at_utc=dt.datetime(2026, 4, 1, 22, 0, tzinfo=dt.UTC), + parser_version="oracle-fallback-rule-1.0.0", + status="pending", + ) + ) + await db_session.flush() + + stats = await _apply_oracle_renormalization(db_session) + + assert stats["seen"] >= 4 + # 2 of our 4 fixture rows should flip. + assert stats["updated"] >= 2 + # Both already-aligned and pass-through rows count as skipped. + assert stats["skipped"] >= 2 + assert stats["errors"] == 0 + + transitions = stats["transitions"] + assert transitions.get("earnings_result -> earnings_release", 0) >= 1 + assert transitions.get("shareholder_vote -> other_material_event", 0) >= 1 + + # Verify the actual row state in the DB matches expectations. + for evt_id, _raw_type, expected in targets: + row = await db_session.execute(select(Event).where(Event.event_id == evt_id)) + evt = row.scalar_one() + assert evt.event_type == expected, ( + f"{evt_id}: expected {expected}, got {evt.event_type}" + )