diff --git a/.env.example b/.env.example index 3092ca4..f74c548 100644 --- a/.env.example +++ b/.env.example @@ -32,7 +32,8 @@ ALPHA_VANTAGE_API_KEY= # e.g. Windows: C:\Program Files\Dolt\bin\dolt.exe). DOLT_DATA_DIR holds the # clones; in PRODUCTION it MUST be outside the deploy tree (deploy is # rsync --delete) — e.g. /var/lib/signal-platform/dolt. The earnings clone lives -# at /. Set up: dolt clone post-no-preference/earnings . +# at /. Production setup is automated by +# deploy/provision_fundamentals.sh; see docs/fundamentals-deployment.md. DOLT_BINARY=dolt DOLT_DATA_DIR=dolt-data DOLT_EARNINGS_SUBDIR=earnings diff --git a/app/scheduler.py b/app/scheduler.py index 4f7f9a3..76b6df5 100644 --- a/app/scheduler.py +++ b/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], diff --git a/app/services/admin_service.py b/app/services/admin_service.py index 6797f44..2cf3f59 100644 --- a/app/services/admin_service.py +++ b/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", diff --git a/deploy/provision_fundamentals.sh b/deploy/provision_fundamentals.sh new file mode 100755 index 0000000..dcd2ee1 --- /dev/null +++ b/deploy/provision_fundamentals.sh @@ -0,0 +1,100 @@ +#!/usr/bin/env bash +set -euo pipefail + +# One-time production provisioning for the shadow fundamentals sources. +# Run as root to install; run with --check as the deploy user for a read-only +# preflight. Version upgrades are intentional code changes, never "latest". +DOLT_VERSION="2.2.0" +DOLT_BINARY="${DOLT_BINARY:-/usr/local/bin/dolt}" +DOLT_DATA_DIR="${DOLT_DATA_DIR:-/var/lib/signal-platform/dolt}" +DOLT_EARNINGS_SUBDIR="${DOLT_EARNINGS_SUBDIR:-earnings}" +APP_USER="${APP_USER:-deploy}" +APP_GROUP="${APP_GROUP:-deploy}" +ENV_FILE="${ENV_FILE:-/opt/signalplatform/.env}" +MIN_FREE_GB="${DOLT_MIN_FREE_DISK_GB:-5}" +EARNINGS_DIR="${DOLT_DATA_DIR}/${DOLT_EARNINGS_SUBDIR}" + +fail() { + echo "ERROR: $*" >&2 + exit 1 +} + +version_ok() { + local output + output="$("$DOLT_BINARY" version 2>/dev/null || true)" + grep -Eq "(^|[[:space:]])v?${DOLT_VERSION}([[:space:]]|$)" <<<"$output" +} + +check_free_space() { + local available_kb + available_kb="$(df -Pk "$DOLT_DATA_DIR" | awk 'NR == 2 {print $4}')" + [[ "$available_kb" =~ ^[0-9]+$ ]] || fail "could not read free space for $DOLT_DATA_DIR" + if ! awk -v available="$available_kb" -v minimum_gb="$MIN_FREE_GB" \ + 'BEGIN { exit !(available >= minimum_gb * 1024 * 1024) }'; then + fail "$DOLT_DATA_DIR has less than ${MIN_FREE_GB} GB free" + fi +} + +check_env() { + [[ -f "$ENV_FILE" ]] || fail "missing environment file: $ENV_FILE" + grep -Fqx "DOLT_BINARY=$DOLT_BINARY" "$ENV_FILE" \ + || fail "set DOLT_BINARY=$DOLT_BINARY in $ENV_FILE" + grep -Fqx "DOLT_DATA_DIR=$DOLT_DATA_DIR" "$ENV_FILE" \ + || fail "set DOLT_DATA_DIR=$DOLT_DATA_DIR in $ENV_FILE" + grep -Fqx "DOLT_EARNINGS_SUBDIR=$DOLT_EARNINGS_SUBDIR" "$ENV_FILE" \ + || fail "set DOLT_EARNINGS_SUBDIR=$DOLT_EARNINGS_SUBDIR in $ENV_FILE" + grep -Eq '^SEC_USER_AGENT=.*@.*' "$ENV_FILE" \ + || fail "SEC_USER_AGENT in $ENV_FILE must contain a real contact email" +} + +check_all() { + id "$APP_USER" >/dev/null 2>&1 || fail "missing service user: $APP_USER" + [[ -x "$DOLT_BINARY" ]] || fail "missing Dolt binary: $DOLT_BINARY" + version_ok || fail "expected Dolt $DOLT_VERSION at $DOLT_BINARY" + [[ -d "$EARNINGS_DIR/.dolt" ]] \ + || fail "missing earnings clone: $EARNINGS_DIR" + if [[ "$(id -un)" == "$APP_USER" ]]; then + [[ -r "$EARNINGS_DIR/.dolt" ]] \ + || fail "earnings clone is not readable by $APP_USER" + elif command -v runuser >/dev/null 2>&1; then + runuser -u "$APP_USER" -- test -r "$EARNINGS_DIR/.dolt" \ + || fail "earnings clone is not readable by $APP_USER" + else + fail "run --check as $APP_USER (or install runuser)" + fi + check_free_space + check_env + echo "OK: Dolt $DOLT_VERSION and earnings clone are provisioned" +} + +if [[ "${1:-}" == "--check" ]]; then + check_all + exit 0 +fi + +[[ "$EUID" -eq 0 ]] || fail "run provisioning as root (or use --check)" +command -v curl >/dev/null 2>&1 || fail "curl is required" +command -v runuser >/dev/null 2>&1 || fail "runuser is required" +id "$APP_USER" >/dev/null 2>&1 || fail "missing service user: $APP_USER" + +if ! version_ok; then + installer="$(mktemp)" + trap 'rm -f "$installer"' EXIT + curl -fsSL \ + "https://github.com/dolthub/dolt/releases/download/v${DOLT_VERSION}/install.sh" \ + -o "$installer" + bash "$installer" +fi +version_ok || fail "Dolt $DOLT_VERSION installation failed" + +install -d -o "$APP_USER" -g "$APP_GROUP" -m 0750 "$DOLT_DATA_DIR" +check_free_space + +if [[ ! -d "$EARNINGS_DIR/.dolt" ]]; then + [[ ! -e "$EARNINGS_DIR" ]] \ + || fail "$EARNINGS_DIR exists but is not a Dolt clone" + runuser -u "$APP_USER" -- \ + "$DOLT_BINARY" clone post-no-preference/earnings "$EARNINGS_DIR" +fi + +check_all diff --git a/docs/dolt-integration-plan.md b/docs/dolt-integration-plan.md index 50dff31..71d6b2a 100644 --- a/docs/dolt-integration-plan.md +++ b/docs/dolt-integration-plan.md @@ -164,8 +164,11 @@ calls. Check free disk space before pulling; alert and skip if below threshold. **Deployment constraints:** deploy is `rsync --delete` of the repo tree (`.gitea/workflows/deploy.yml:127`), so clones and archives must live **outside the deployment path** — an env-configured persistent directory (e.g. -`DOLT_DATA_DIR=/var/lib/signal-platform/dolt`), backed up. The dolt binary is a new -prod runtime dependency: install with a pinned version in the deploy workflow. +`DOLT_DATA_DIR=/var/lib/signal-platform/dolt`). The dolt binary is a new prod +runtime dependency: install it once with the version-pinned provisioner in +`deploy/provision_fundamentals.sh`; operational steps are in +`docs/fundamentals-deployment.md`. The clone is reproducible from DoltHub; the +normalized PostgreSQL rows remain part of the normal database backup. **Validation gates (block promotion, raise an alert via the existing system-events path):** source freshness as expected; tracked-universe coverage; no duplicate @@ -193,13 +196,14 @@ Workstream A: - Dolt earnings import: daily ~02:30 ET (with the future-row replacement above). - **SEC fundamentals job: daily ~04:00 ET.** One job, three steps: - (a) freshness check **using conditional HTTP metadata (ETag/Last-Modified / - If-None-Match) before downloading** — an unchanged archive is a `no_op` without - pulling the multi-GB file; verify by SHA-256 after any actual download; - (b) when changed, **parse and write only tracked-universe CIKs** through - staging→promotion (the tracked universe is `ticker_universe_service`'s set; - CIKs are resolved from `company_tickers.json` each run, so a newly added ticker - self-resolves on its next SEC run — until then its metrics are simply null); + (a) detect a composite revision from the latest EDGAR daily-index date, the + exact tracked index rows, and the tracked-universe fingerprint — an unchanged + revision is a `no_op` before Company Facts are fetched; + (b) when changed, fetch and parse Company Facts only for tracked-universe CIKs + that filed, plus full available history for the first run or a newly added + issuer, through validation→atomic promotion. Universe resolution and the exact + index inputs are cached during revision detection and reused during staging; + CIKs are resolved from `company_tickers.json` without writes until promotion; (c) **always, locally, and only after production activation** (the phase-A5 parity approval): refresh the legacy `fundamental_data` fields and mark affected cached fundamental scores stale. **Sources differ per field** — do not assume all @@ -405,9 +409,9 @@ workstream B — Alpaca remains the price source throughout. **Workstream A:** - A0. License review **DONE** (earnings approved for private/internal use under - CC BY-SA 4.0, no redistribution — see Licensing above); still pending at deploy - time: dolt binary pinned in deploy; `DOLT_DATA_DIR` outside the rsync tree - (earnings clone only — small), provisioned and backed up. + CC BY-SA 4.0, no redistribution — see Licensing above). The Dolt version, + persistent `DOLT_DATA_DIR`, clone, and production checks are captured in + `deploy/provision_fundamentals.sh` and `docs/fundamentals-deployment.md`. - A1. Migration 026, import-run framework. - A2. Earnings ingestion in shadow (writes `earnings_events`, prod untouched); verify forward-calendar coverage and rescheduling behavior. diff --git a/docs/fundamentals-deployment.md b/docs/fundamentals-deployment.md new file mode 100644 index 0000000..fa67d34 --- /dev/null +++ b/docs/fundamentals-deployment.md @@ -0,0 +1,133 @@ +# Fundamentals production deployment + +This is the one-time production setup for the Dolt earnings and SEC fundamentals +imports. Both imports remain shadow inputs until the separate A5 scoring-cutover +approval. Do not add OS cron entries: the application scheduler owns both jobs. + +## What the deployment adds + +- `Dolt Earnings Import (shadow)` runs daily at 02:30 America/New_York. +- `SEC Fundamentals Import (shadow)` runs daily at 04:00 America/New_York. +- Both jobs are visible, toggleable, and manually triggerable in Admin → Jobs. +- Cron expressions are editable in Admin → Schedule. +- Every attempt is recorded in `data_import_runs`; failures also create a system + event. A failed validation does not promote partial data. + +The systemd service uses one application worker. The import framework also holds +a PostgreSQL advisory lock per source, so an overlapping manual/scheduled run is +skipped safely. + +## Prerequisites + +The production `.env` at `/opt/signalplatform/.env` must contain: + +```dotenv +DOLT_BINARY=/usr/local/bin/dolt +DOLT_DATA_DIR=/var/lib/signal-platform/dolt +DOLT_EARNINGS_SUBDIR=earnings +DOLT_MIN_FREE_DISK_GB=5.0 +SEC_USER_AGENT=signal-platform/1.0 (contact: real-address@example.com) +SEC_REQUEST_SPACING_SECONDS=0.2 +``` + +Use a real monitored contact address. Keep at least 5 GB free at the Dolt data +path; 8–10 GB gives comfortable growth headroom. The data directory must stay +outside `/opt/signalplatform`, because deployments use `rsync --delete` there. + +## One-time provisioning + +First deploy the commit containing this bundle to production. Then SSH to the +server and run: + +```bash +cd /opt/signalplatform +sudo bash ./deploy/provision_fundamentals.sh +sudo -u deploy bash ./deploy/provision_fundamentals.sh --check +sudo systemctl restart signalplatform.service +curl -fsS http://127.0.0.1:8998/api/v1/health +``` + +The provisioner is idempotent. It installs the pinned Dolt version, creates the +persistent directory as `deploy:deploy`, clones +`post-no-preference/earnings`, verifies free space and `.env`, and refuses an +unexpected Dolt version. It does not modify PostgreSQL or start an import. + +Do not replace the pinned version with `latest`. A future Dolt upgrade should be +a reviewed change to `DOLT_VERSION`, followed by the same provision/check flow. + +## First-run verification + +In Admin → Jobs, wait until no other job is running, then: + +1. Trigger **Dolt Earnings Import (shadow)**. Expect `completed` with import + status `promoted`; a repeat without an upstream change should report `no_op`. +2. Trigger **SEC Fundamentals Import (shadow)**. The first run performs the + tracked-universe history backfill and can take materially longer than a daily + incremental run. Expect `completed` with import status `promoted`. +3. Check Admin → System Events. There should be no new import error. +4. Confirm the next-run times correspond to 02:30 and 04:00 New York time. +5. Open several ticker pages and confirm the fundamentals panel has populated + data and still handles partial/missing issuers cleanly. + +Optional database verification: + +```sql +SELECT source, status, revision, source_max_date, started_at, completed_at, + validation_json +FROM data_import_runs +WHERE source IN ('dolt_earnings', 'sec_facts') +ORDER BY id DESC +LIMIT 10; + +SELECT count(*) FROM earnings_events WHERE source = 'dolt_earnings'; +SELECT count(*), count(DISTINCT cik) FROM fundamental_snapshots; +``` + +During the longer first SEC run, execute the following in a second SSH session. +It opens an independent database connection and attempts the same source lock: + +```bash +cd /opt/signalplatform +sudo -u deploy .venv/bin/python - <<'PY' +import asyncio + +from sqlalchemy import text + +from app.database import engine +from app.services.data_import import _advisory_key + + +async def main(): + key = _advisory_key("sec_facts") + async with engine.connect() as connection: + acquired = await connection.scalar( + text("SELECT pg_try_advisory_lock(:key)"), {"key": key} + ) + print("UNEXPECTED: lock acquired" if acquired else "OK: source lock is busy") + if acquired: + await connection.execute( + text("SELECT pg_advisory_unlock(:key)"), {"key": key} + ) + + +asyncio.run(main()) +PY +``` + +Expect `OK: source lock is busy`. This is the remaining live-PostgreSQL +mutual-exclusion check; SQLite unit tests cannot exercise PostgreSQL advisory +locks. A second Admin trigger should independently report the job as busy. + +## Failure and rollback + +- Disable the failing shadow job in Admin → Jobs. This stops scheduled imports + without changing existing data or the legacy scoring path. +- Inspect the job runtime, latest `data_import_runs.validation_json`, service + logs, and Admin → System Events before retrying. +- Re-run `sudo -u deploy bash ./deploy/provision_fundamentals.sh --check` for + binary, clone, permission, disk, or environment failures. +- The Dolt clone is a reproducible cache and does not need a bespoke backup. + PostgreSQL (including `earnings_events`, `fundamental_snapshots`, and import + audit rows) must remain covered by the normal production database backup. +- Do not proceed to A5 while either shadow feed is unhealthy or the parity gate + has not received explicit approval. diff --git a/frontend/src/components/admin/ScheduleSettings.tsx b/frontend/src/components/admin/ScheduleSettings.tsx index c2bc3a9..0dd2472 100644 --- a/frontend/src/components/admin/ScheduleSettings.tsx +++ b/frontend/src/components/admin/ScheduleSettings.tsx @@ -6,10 +6,12 @@ import { SkeletonTable } from '../ui/Skeleton'; const DEFAULTS: ScheduleConfig = { schedule_timezone: 'America/New_York', schedule_daily_pipeline_cron: '0 2 * * *', - schedule_near_close_pipeline_cron: '30 15 * * 1-5', - schedule_after_close_pipeline_cron: '45 16 * * 1-5', - schedule_intraday_pipeline_cron: '0 10-15 * * 1-5', - schedule_fundamentals_cron: '0 1 * * 1', + schedule_dolt_earnings_cron: '30 2 * * *', + schedule_sec_fundamentals_cron: '0 4 * * *', + schedule_near_close_pipeline_cron: '30 15 * * mon-fri', + schedule_after_close_pipeline_cron: '45 16 * * mon-fri', + schedule_intraday_pipeline_cron: '0 10-15 * * mon-fri', + schedule_fundamentals_cron: '0 1 * * mon', }; const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: boolean }[] = [ @@ -24,6 +26,18 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b hint: 'OHLCV → benchmark → sentiment → regime → alerts (no R:R scan). Default 02:00 ET so regime-quadrant changes hit Telegram in the morning.', mono: true, }, + { + key: 'schedule_dolt_earnings_cron', + label: 'Dolt earnings (shadow)', + hint: 'Pull and import earnings dates/results daily at 02:30 ET. Live scoring remains untouched before A5.', + mono: true, + }, + { + key: 'schedule_sec_fundamentals_cron', + label: 'SEC fundamentals (shadow)', + hint: 'Import tracked-universe SEC facts daily at 04:00 ET. Unchanged revisions become no-op runs.', + mono: true, + }, { key: 'schedule_near_close_pipeline_cron', label: 'Near-close pipeline (scan + alert)', @@ -44,8 +58,8 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b }, { key: 'schedule_fundamentals_cron', - label: 'Fundamentals (weekly)', - hint: 'Slow, rate-limited. Default early Monday ET.', + label: 'Legacy fundamentals (weekly)', + hint: 'Existing provider chain retained until the A5 parity approval and A6 removal.', mono: true, }, ]; diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 2f5af55..bd15b43 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -191,6 +191,8 @@ export interface ActivationConfig { export interface ScheduleConfig { schedule_timezone: string; schedule_daily_pipeline_cron: string; + schedule_dolt_earnings_cron: string; + schedule_sec_fundamentals_cron: string; schedule_near_close_pipeline_cron: string; schedule_after_close_pipeline_cron: string; schedule_intraday_pipeline_cron: string; diff --git a/tests/unit/test_schedule_config.py b/tests/unit/test_schedule_config.py index cb2dde3..bc745a2 100644 --- a/tests/unit/test_schedule_config.py +++ b/tests/unit/test_schedule_config.py @@ -80,6 +80,28 @@ class TestTradingDayCrons: ) assert fire.strftime("%a") == "Mon" + @pytest.mark.parametrize( + ("key", "hour", "minute"), + ( + ("schedule_dolt_earnings_cron", 2, 30), + ("schedule_sec_fundamentals_cron", 4, 0), + ), + ) + def test_shadow_imports_run_daily_at_expected_et_time( + self, key: str, hour: int, minute: int + ): + from datetime import datetime + + from apscheduler.triggers.cron import CronTrigger + + trigger = CronTrigger.from_crontab( + SCHEDULE_DEFAULTS[key], timezone=SCHEDULE_DEFAULTS["schedule_timezone"] + ) + fire = trigger.get_next_fire_time( + None, datetime(2026, 7, 19, tzinfo=trigger.timezone) + ) + assert (fire.hour, fire.minute) == (hour, minute) + class TestScheduleConfig: async def test_defaults_when_unset(self, session: AsyncSession): diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index b9ae8a6..bb69087 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -1,5 +1,7 @@ """Unit tests for app.scheduler module.""" +from types import SimpleNamespace + import pytest from app.scheduler import ( @@ -8,7 +10,9 @@ from app.scheduler import ( _parse_frequency, _resume_tickers, _last_successful, + _run_shadow_import, configure_scheduler, + get_job_runtime_snapshot, queue_backtest_options, queue_backtest_target_model, scheduler, @@ -106,6 +110,8 @@ class TestConfigureScheduler: "benchmark_collector", "sentiment_collector", "fundamental_collector", + "dolt_earnings_import", + "sec_fundamentals_import", "rr_scanner", "shadow_book", "ticker_universe_sync", @@ -137,6 +143,8 @@ class TestConfigureScheduler: "data_collector", "data_backfill", "fundamental_collector", + "dolt_earnings_import", + "sec_fundamentals_import", "market_regime", "near_close_pipeline", "regime_monitor", @@ -147,3 +155,91 @@ class TestConfigureScheduler: "shadow_book", "ticker_universe_sync", ]) + + +class _SessionContext: + async def __aenter__(self): + return object() + + async def __aexit__(self, *exc): + return None + + +class TestShadowImportJobs: + @staticmethod + def _session_factory(): + return _SessionContext() + + async def test_promoted_run_surfaces_completion(self, monkeypatch): + async def enabled(db, job_name): + return True + + async def imported(importer): + return SimpleNamespace( + status="promoted", revision="abcdef1234567890", error_details=None + ) + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) + monkeypatch.setattr("app.scheduler.run_import", imported) + + await _run_shadow_import("dolt_earnings_import", object()) + + runtime = get_job_runtime_snapshot("dolt_earnings_import") + assert runtime["status"] == "completed" + assert runtime["processed"] == 1 + assert runtime["message"] == "promoted · abcdef123456" + + async def test_failed_run_surfaces_error(self, monkeypatch): + async def enabled(db, job_name): + return True + + async def imported(importer): + return SimpleNamespace( + status="failed", revision=None, error_details="validation failed" + ) + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) + monkeypatch.setattr("app.scheduler.run_import", imported) + + await _run_shadow_import("sec_fundamentals_import", object()) + + runtime = get_job_runtime_snapshot("sec_fundamentals_import") + assert runtime["status"] == "error" + assert runtime["processed"] == 0 + assert runtime["message"] == "validation failed" + + async def test_source_lock_surfaces_skipped(self, monkeypatch): + async def enabled(db, job_name): + return True + + async def imported(importer): + return None + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", enabled) + monkeypatch.setattr("app.scheduler.run_import", imported) + + await _run_shadow_import("dolt_earnings_import", object()) + + runtime = get_job_runtime_snapshot("dolt_earnings_import") + assert runtime["status"] == "skipped" + assert "already running" in runtime["message"] + + async def test_disabled_job_never_runs_importer(self, monkeypatch): + async def disabled(db, job_name): + return False + + async def should_not_run(importer): + raise AssertionError("disabled job ran importer") + + monkeypatch.setattr("app.scheduler.async_session_factory", self._session_factory) + monkeypatch.setattr("app.scheduler._is_job_enabled", disabled) + monkeypatch.setattr("app.scheduler.run_import", should_not_run) + + await _run_shadow_import("sec_fundamentals_import", object()) + + runtime = get_job_runtime_snapshot("sec_fundamentals_import") + assert runtime["status"] == "skipped" + assert runtime["message"] == "Disabled"