feat(deploy): schedule fundamentals shadow imports

app/scheduler.py

@@ -41,6 +41,9 @@ from app.services import (
     settings_store,
     shadow_book_service,
 )
+from app.services.data_import import STATUS_FAILED, SourceImporter, run_import
+from app.services.dolt_earnings_importer import DoltEarningsImporter
+from app.services.sec_fundamentals_importer import SecFundamentalsImporter
 from app.services.alert_service import dispatch_alerts
 from app.services.backtest_service import (
     BACKTEST_TARGET_MODELS,
@@ -93,6 +96,8 @@ _JOB_NAMES = [
     "data_backfill",
     "sentiment_collector",
     "fundamental_collector",
+    "dolt_earnings_import",
+    "sec_fundamentals_import",
     "rr_scanner",
     "ticker_universe_sync",
     "alerts",
@@ -912,6 +917,70 @@ async def collect_fundamentals() -> None:
         _runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc))
 
 
+# ---------------------------------------------------------------------------
+# Jobs: shadow fundamentals sources
+# ---------------------------------------------------------------------------
+
+
+async def _run_shadow_import(job_name: str, importer: SourceImporter) -> None:
+    """Run one source importer and surface its audit result in Admin → Jobs."""
+    _log_event(logging.INFO, "job_start", job=job_name)
+    _runtime_start(job_name, total=1)
+
+    try:
+        async with async_session_factory() as db:
+            if not await _is_job_enabled(db, job_name):
+                _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled")
+                _runtime_finish(job_name, "skipped", processed=0, total=1, message="Disabled")
+                return
+
+        run = await run_import(importer)
+        if run is None:
+            message = "Another import for this source is already running"
+            _log_event(logging.INFO, "job_skipped", job=job_name, reason="source_locked")
+            _runtime_finish(job_name, "skipped", processed=0, total=1, message=message)
+            return
+
+        revision = f" · {run.revision[:12]}" if run.revision else ""
+        message = f"{run.status}{revision}"
+        if run.status == STATUS_FAILED:
+            message = run.error_details or message
+            _log_event(logging.ERROR, "job_error", job=job_name, message=message)
+            _runtime_finish(job_name, "error", processed=0, total=1, message=message)
+            return
+
+        _log_event(
+            logging.INFO,
+            "job_complete",
+            job=job_name,
+            import_status=run.status,
+            revision=run.revision,
+        )
+        _runtime_finish(job_name, "completed", processed=1, total=1, message=message)
+    except asyncio.CancelledError:
+        _runtime_finish(job_name, "error", processed=0, total=1, message="Cancelled")
+        raise
+    except Exception as exc:
+        _log_event(
+            logging.ERROR,
+            "job_error",
+            job=job_name,
+            error_type=type(exc).__name__,
+            message=str(exc),
+        )
+        _runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
+
+
+async def run_dolt_earnings_import() -> None:
+    """Pull and import the Dolt earnings calendar/results feed in shadow."""
+    await _run_shadow_import("dolt_earnings_import", DoltEarningsImporter())
+
+
+async def run_sec_fundamentals_import() -> None:
+    """Import tracked-universe SEC facts in shadow."""
+    await _run_shadow_import("sec_fundamentals_import", SecFundamentalsImporter())
+
+
 # ---------------------------------------------------------------------------
 # Job: R:R Scanner
 # ---------------------------------------------------------------------------
@@ -1452,6 +1521,9 @@ SCHEDULE_DEFAULTS: dict[str, str] = {
     "schedule_timezone": "America/New_York",
     # Morning data/display refresh (no qualifying R:R scan).
     "schedule_daily_pipeline_cron": "0 2 * * *",
+    # Shadow source imports. They never write legacy fundamental_data before A5.
+    "schedule_dolt_earnings_cron": "30 2 * * *",
+    "schedule_sec_fundamentals_cron": "0 4 * * *",
     # Fetch in-progress bars → scan → Telegram (manual MOC window).
     "schedule_near_close_pipeline_cron": "30 15 * * mon-fri",
     # Fetch final bars → outcome eval (must not run on the partial near-close bar).
@@ -1465,6 +1537,8 @@ SCHEDULE_DEFAULTS: dict[str, str] = {
 # job id -> schedule setting key
 _CRON_JOBS: dict[str, str] = {
     "daily_pipeline": "schedule_daily_pipeline_cron",
+    "dolt_earnings_import": "schedule_dolt_earnings_cron",
+    "sec_fundamentals_import": "schedule_sec_fundamentals_cron",
     "near_close_pipeline": "schedule_near_close_pipeline_cron",
     "after_close_pipeline": "schedule_after_close_pipeline_cron",
     "intraday_pipeline": "schedule_intraday_pipeline_cron",
@@ -1549,6 +1623,28 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         _cron_trigger(cfg["schedule_daily_pipeline_cron"], tz, "schedule_daily_pipeline_cron"),
         id="daily_pipeline", name="Morning Pipeline", replace_existing=True,
     )
+    scheduler.add_job(
+        run_dolt_earnings_import,
+        _cron_trigger(
+            cfg["schedule_dolt_earnings_cron"],
+            tz,
+            "schedule_dolt_earnings_cron",
+        ),
+        id="dolt_earnings_import",
+        name="Dolt Earnings Import (shadow)",
+        replace_existing=True,
+    )
+    scheduler.add_job(
+        run_sec_fundamentals_import,
+        _cron_trigger(
+            cfg["schedule_sec_fundamentals_cron"],
+            tz,
+            "schedule_sec_fundamentals_cron",
+        ),
+        id="sec_fundamentals_import",
+        name="SEC Fundamentals Import (shadow)",
+        replace_existing=True,
+    )
     scheduler.add_job(
         run_near_close_pipeline,
         _cron_trigger(
@@ -1622,6 +1718,8 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
             "cron": cfg["schedule_daily_pipeline_cron"],
             "steps": [name for name, _ in _DAILY_PIPELINE_STEPS],
         },
+        dolt_earnings_import={"cron": cfg["schedule_dolt_earnings_cron"]},
+        sec_fundamentals_import={"cron": cfg["schedule_sec_fundamentals_cron"]},
         near_close_pipeline={
             "cron": cfg["schedule_near_close_pipeline_cron"],
             "steps": [name for name, _ in _NEAR_CLOSE_PIPELINE_STEPS],

app/services/admin_service.py

@@ -612,6 +612,8 @@ VALID_JOB_NAMES = {
     "benchmark_collector",
     "sentiment_collector",
     "fundamental_collector",
+    "dolt_earnings_import",
+    "sec_fundamentals_import",
     "rr_scanner",
     "ticker_universe_sync",
     "outcome_evaluator",
@@ -633,6 +635,8 @@ JOB_LABELS = {
     "benchmark_collector": "Benchmark Collector",
     "sentiment_collector": "Sentiment Collector",
     "fundamental_collector": "Fundamental Collector",
+    "dolt_earnings_import": "Dolt Earnings Import (shadow)",
+    "sec_fundamentals_import": "SEC Fundamentals Import (shadow)",
     "rr_scanner": "R:R Scanner",
     "ticker_universe_sync": "Ticker Universe Sync",
     "outcome_evaluator": "Outcome Evaluator",