diff --git a/app/job_catalog.py b/app/job_catalog.py new file mode 100644 index 0000000..4a51938 --- /dev/null +++ b/app/job_catalog.py @@ -0,0 +1,207 @@ +"""Job topology: names, labels, pipeline membership, categories, ordering. + +The single source of truth for *what the jobs are*, as opposed to how they run. +It deliberately imports nothing from ``app`` so both ``app.scheduler`` and +``app.services.admin_service`` can import it at module level -- admin_service +otherwise has to do ``from app.scheduler import ...`` inside functions to dodge a +cycle. + +The pipeline step lists live here rather than in the scheduler because three +separate things need them and used to keep private copies: the runner, the +``PIPELINE_MEMBERS`` set the admin API reports, and the UI's grouping. Steps are +``(step_name, coroutine_name)``; ``_run_pipeline`` resolves the coroutine late +out of the scheduler's own globals, so nothing here depends on those functions +existing. +""" + +from __future__ import annotations + +# --------------------------------------------------------------------------- +# Pipelines +# --------------------------------------------------------------------------- + +_DAILY_PIPELINE_STEPS = [ + ("data_collector", "collect_ohlcv"), + ("benchmark_collector", "collect_benchmark"), + ("sentiment_collector", "collect_sentiment"), + ("market_regime", "compute_market_regime"), + # Observational only — display/alerts; not trade selection. + ("regime_monitor", "compute_regime_monitor"), + # Alerts after regime so quadrant changes reach Telegram in the morning. + # Dispatcher is change-driven; quiet days stay quiet. Setup alerts still + # fire on the near-close pipeline after the qualifying scan. + ("alerts", "dispatch_alerts_job"), +] + +# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (incremental +# ingestion overlaps the latest stored session), then the only daily +# qualifying R:R scan, then Telegram immediately so manual fills can still hit +# MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan +# may see ~15:15 prices — immaterial for a 12-1 momentum signal. +# +# US early-close days (~3/year, 13:00 ET close): this job runs post-close and +# entries behave like stale_close (still acceptable per execution-recovery matrix). +# No exchange calendar dependency. +_NEAR_CLOSE_PIPELINE_STEPS = [ + # Must land today's in-progress bar (~20 min behind live), or the scan falls + # back to the previous close and execution degrades to the stale_close floor. + ("data_collector", "collect_ohlcv_for_scan"), + ("rr_scanner", "scan_rr"), + # Straight after the scan so shadow entries mark at the same near-close + # prices the discretionary book is looking at. + ("shadow_book", "run_shadow_book"), + ("alerts", "dispatch_alerts_job"), +] + +# After close (~16:45 ET Mon–Fri): fresh OHLCV fetch so outcomes resolve on the +# final bar, not the near-close partial bar, then outcome/paper close. +_AFTER_CLOSE_PIPELINE_STEPS = [ + ("data_collector", "collect_ohlcv_final"), + ("outcome_evaluator", "evaluate_outcomes"), +] + +# Intraday (light): keep prices current and resolve outcomes through the day, +# without the expensive scan/sentiment. The dashboard recomputes live R:R from +# the latest price, so refreshing OHLCV is enough to stop prices lagging; the +# outcome step also closes paper trades that hit their stop/target intraday. +_INTRADAY_PIPELINE_STEPS = [ + ("data_collector", "collect_ohlcv"), + ("outcome_evaluator", "evaluate_outcomes"), +] + +# Ordered by trading day, not alphabetically: this is the sequence an operator +# reads down the page, and it drives the UI's ordering too. +PIPELINE_STEPS: dict[str, list[tuple[str, str]]] = { + "daily_pipeline": _DAILY_PIPELINE_STEPS, + "intraday_pipeline": _INTRADAY_PIPELINE_STEPS, + "near_close_pipeline": _NEAR_CLOSE_PIPELINE_STEPS, + "after_close_pipeline": _AFTER_CLOSE_PIPELINE_STEPS, +} + +# Derived, never hand-maintained: this used to be a literal set in admin_service +# duplicating the four lists above from another module, with nothing asserting +# the two agreed. +PIPELINE_MEMBERS: frozenset[str] = frozenset( + step for steps in PIPELINE_STEPS.values() for step, _ in steps +) + + +def _pipelines_by_member() -> dict[str, tuple[str, ...]]: + """Member -> the orchestrators that run it, in trading-day order. + + Membership is many-to-many: data_collector runs in all four pipelines (via + three different coroutines), alerts and outcome_evaluator in two each. + """ + out: dict[str, list[str]] = {} + for pipeline, steps in PIPELINE_STEPS.items(): + for step, _ in steps: + bucket = out.setdefault(step, []) + if pipeline not in bucket: + bucket.append(pipeline) + return {member: tuple(pipelines) for member, pipelines in out.items()} + + +PIPELINES_BY_MEMBER: dict[str, tuple[str, ...]] = _pipelines_by_member() + + +# --------------------------------------------------------------------------- +# Job identity +# --------------------------------------------------------------------------- + +# Orchestrators, in trading-day order. +PIPELINE_JOBS: tuple[str, ...] = tuple(PIPELINE_STEPS) + +# Own timer, independent of any pipeline. +SCHEDULED_JOBS: tuple[str, ...] = ( + "dolt_earnings_import", + "sec_fundamentals_import", + "ticker_universe_sync", + "backtest", +) + +# Registered but never auto-fired; run only when a human asks. +MANUAL_JOBS: tuple[str, ...] = ("event_study", "data_backfill") + +# Steps in the order an operator meets them across the trading day, so the UI +# reads as a sequence rather than an alphabetical jumble. +PIPELINE_STEP_JOBS: tuple[str, ...] = tuple( + dict.fromkeys(step for steps in PIPELINE_STEPS.values() for step, _ in steps) +) + +VALID_JOB_NAMES: frozenset[str] = frozenset( + PIPELINE_JOBS + PIPELINE_STEP_JOBS + SCHEDULED_JOBS + MANUAL_JOBS +) + +JOB_LABELS: dict[str, str] = { + "data_collector": "Data Collector (OHLCV)", + "data_backfill": "Data Backfill (deep history)", + "benchmark_collector": "Benchmark Collector", + "sentiment_collector": "Sentiment Collector", + "dolt_earnings_import": "Dolt Earnings Import", + "sec_fundamentals_import": "SEC Fundamentals Import", + "rr_scanner": "R:R Scanner", + "ticker_universe_sync": "Ticker Universe Sync", + "outcome_evaluator": "Outcome Evaluator", + "alerts": "Alerts Dispatcher", + # Keys are persisted job ids and must not change; these are display only. + "market_regime": "Market Trend (SPY)", + "regime_monitor": "AI/Tech Risk Monitor", + "event_study": "Event Study", + "backtest": "Backtest", + "daily_pipeline": "Morning Pipeline", + "near_close_pipeline": "Near-Close Pipeline (scan+alert)", + "after_close_pipeline": "After-Close Pipeline (outcome)", + "intraday_pipeline": "Intraday Pipeline", + "shadow_book": "Shadow Book (auto-traded strategy)", +} + +CATEGORY_PIPELINE = "pipeline" +CATEGORY_STEP = "pipeline_step" +CATEGORY_SCHEDULED = "scheduled" +CATEGORY_MANUAL = "manual" + +# Order the sections appear in. +CATEGORY_ORDER: tuple[str, ...] = ( + CATEGORY_PIPELINE, + CATEGORY_STEP, + CATEGORY_SCHEDULED, + CATEGORY_MANUAL, +) + +CATEGORY_LABELS: dict[str, str] = { + CATEGORY_PIPELINE: "Pipelines", + CATEGORY_STEP: "Pipeline steps", + CATEGORY_SCHEDULED: "Standalone scheduled", + CATEGORY_MANUAL: "Manual only", +} + +_CATEGORY_MEMBERS: dict[str, tuple[str, ...]] = { + CATEGORY_PIPELINE: PIPELINE_JOBS, + CATEGORY_STEP: PIPELINE_STEP_JOBS, + CATEGORY_SCHEDULED: SCHEDULED_JOBS, + CATEGORY_MANUAL: MANUAL_JOBS, +} + +JOB_CATEGORY: dict[str, str] = { + name: category + for category, names in _CATEGORY_MEMBERS.items() + for name in names +} + +# Registered and triggerable through the API, but kept out of Admin → Jobs. +# data_backfill's only capability beyond collect_ohlcv (which already backfills +# full history for *new* tickers) is re-deepening *existing* ones after +# ohlcv_history_days is raised -- a rare one-off, not something to scan past +# every time you open the page. +HIDDEN_JOBS: frozenset[str] = frozenset({"data_backfill"}) + +_SORT_INDEX: dict[str, tuple[int, int]] = { + name: (CATEGORY_ORDER.index(category), position) + for category, names in _CATEGORY_MEMBERS.items() + for position, name in enumerate(names) +} + + +def sort_order(job_name: str) -> tuple[int, int]: + """(category rank, position within category). Unknown jobs sort last.""" + return _SORT_INDEX.get(job_name, (len(CATEGORY_ORDER), 0)) diff --git a/app/scheduler.py b/app/scheduler.py index d201b06..a470d3c 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -18,11 +18,13 @@ import logging import asyncio from datetime import date, datetime, timedelta, timezone +from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from sqlalchemy import and_, case, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession +from app import job_catalog from app.config import settings from app.database import async_session_factory from app.models.ohlcv import OHLCVRecord @@ -84,6 +86,34 @@ scheduler = AsyncIOScheduler( } ) + +def _repause_after_manual_run(event: object) -> None: + """Re-pause a job that only ever runs on demand, once its run finishes. + + Pipeline steps and manual jobs are registered with a 520-week interval and + ``next_run_time=None`` as a backstop. Triggering one sets next_run_time=now, + and APScheduler then re-arms that backstop -- so Admin → Jobs would show a + "next run" ten years out. Guarding on category means the six cron jobs and + the real interval jobs are never touched. + + Registered at module level, not inside ``configure_scheduler``: that function + is called more than once (idempotency test) and ``add_listener`` does not + deduplicate. + """ + job_id = getattr(event, "job_id", None) + if job_catalog.JOB_CATEGORY.get(job_id) not in ( + job_catalog.CATEGORY_STEP, + job_catalog.CATEGORY_MANUAL, + ): + return + try: + scheduler.modify_job(job_id, next_run_time=None) + except Exception: # job gone, scheduler stopped — nothing to re-pause + logger.debug("Could not re-pause %s after its run", job_id, exc_info=True) + + +scheduler.add_listener(_repause_after_manual_run, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) + # Track last successful ticker per job for rate-limit resume _last_successful: dict[str, str | None] = { "data_collector": None, @@ -91,26 +121,10 @@ _last_successful: dict[str, str | None] = { "sentiment_collector": None, } -# Jobs whose per-run progress is surfaced to Admin → Jobs. (outcome_evaluator is -# created lazily on first run via _runtime_start.) -_JOB_NAMES = [ - "data_collector", - "data_backfill", - "sentiment_collector", - "dolt_earnings_import", - "sec_fundamentals_import", - "rr_scanner", - "ticker_universe_sync", - "alerts", - "market_regime", - "regime_monitor", - "event_study", - "backtest", - "daily_pipeline", # morning: OHLCV/sentiment/regime — no qualifying scan - "near_close_pipeline", # OHLCV fetch → R:R scan → Telegram alerts - "after_close_pipeline", # OHLCV fetch → outcome eval (final bar) - "intraday_pipeline", -] +# Seeded from the catalog rather than a private list. The old literal held 16 of +# the 19 jobs -- benchmark_collector, outcome_evaluator and shadow_book were +# missing, so they had no runtime row (and so no "last run" line in Admin → Jobs) +# until their first run in a given process. def _idle_runtime() -> dict[str, object]: @@ -127,7 +141,9 @@ def _idle_runtime() -> dict[str, object]: } -_job_runtime: dict[str, dict[str, object]] = {name: _idle_runtime() for name in _JOB_NAMES} +_job_runtime: dict[str, dict[str, object]] = { + name: _idle_runtime() for name in sorted(job_catalog.VALID_JOB_NAMES) +} _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL _next_backtest_cadence = DEFAULT_BACKTEST_CADENCE @@ -1300,54 +1316,14 @@ async def sync_ticker_universe() -> None: # the intraday partial one (covers a long weekend / holiday gap). _FINAL_REFETCH_DAYS = 5 -_DAILY_PIPELINE_STEPS = [ - ("data_collector", "collect_ohlcv"), - ("benchmark_collector", "collect_benchmark"), - ("sentiment_collector", "collect_sentiment"), - ("market_regime", "compute_market_regime"), - # Observational only — display/alerts; not trade selection. - ("regime_monitor", "compute_regime_monitor"), - # Alerts after regime so quadrant changes reach Telegram in the morning. - # Dispatcher is change-driven; quiet days stay quiet. Setup alerts still - # fire on the near-close pipeline after the qualifying scan. - ("alerts", "dispatch_alerts_job"), -] - -# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (incremental -# ingestion overlaps the latest stored session), then the only daily -# qualifying R:R scan, then Telegram immediately so manual fills can still hit -# MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan -# may see ~15:15 prices — immaterial for a 12-1 momentum signal. -# -# US early-close days (~3/year, 13:00 ET close): this job runs post-close and -# entries behave like stale_close (still acceptable per execution-recovery matrix). -# No exchange calendar dependency. -_NEAR_CLOSE_PIPELINE_STEPS = [ - # Must land today's in-progress bar (~20 min behind live), or the scan falls - # back to the previous close and execution degrades to the stale_close floor. - ("data_collector", "collect_ohlcv_for_scan"), - ("rr_scanner", "scan_rr"), - # Straight after the scan so shadow entries mark at the same near-close - # prices the discretionary book is looking at. - ("shadow_book", "run_shadow_book"), - ("alerts", "dispatch_alerts_job"), -] - -# After close (~16:45 ET Mon–Fri): fresh OHLCV fetch so outcomes resolve on the -# final bar, not the near-close partial bar, then outcome/paper close. -_AFTER_CLOSE_PIPELINE_STEPS = [ - ("data_collector", "collect_ohlcv_final"), - ("outcome_evaluator", "evaluate_outcomes"), -] - -# Intraday (light): keep prices current and resolve outcomes through the day, -# without the expensive scan/sentiment. The dashboard recomputes live R:R from -# the latest price, so refreshing OHLCV is enough to stop prices lagging; the -# outcome step also closes paper trades that hit their stop/target intraday. -_INTRADAY_PIPELINE_STEPS = [ - ("data_collector", "collect_ohlcv"), - ("outcome_evaluator", "evaluate_outcomes"), -] +# Step lists live in app.job_catalog so the runner, the admin API's pipeline +# membership and the UI's grouping all read one definition. Re-exported here +# under their original names: _run_pipeline and the scheduler_configured log +# payload refer to them directly. +_DAILY_PIPELINE_STEPS = job_catalog._DAILY_PIPELINE_STEPS +_NEAR_CLOSE_PIPELINE_STEPS = job_catalog._NEAR_CLOSE_PIPELINE_STEPS +_AFTER_CLOSE_PIPELINE_STEPS = job_catalog._AFTER_CLOSE_PIPELINE_STEPS +_INTRADAY_PIPELINE_STEPS = job_catalog._INTRADAY_PIPELINE_STEPS # Warn if near-close fetch+scan+alert drifts past this — entries leave the close # and the stale_close floor quietly becomes the ceiling. @@ -1486,6 +1462,12 @@ SCHEDULE_DEFAULTS: dict[str, str] = { "schedule_after_close_pipeline_cron": "45 16 * * mon-fri", # Hourly mid-session price + outcome (10:00–15:00 ET Mon–Fri). "schedule_intraday_pipeline_cron": "0 10-15 * * mon-fri", + # Both were interval jobs until 2026-08-08 and hit exactly the pitfall + # described above: configure_scheduler calls remove_all_jobs() on every + # startup, so an interval countdown restarts from zero each deploy. A 168h + # backtest needed a week of uninterrupted uptime to fire even once. + "schedule_backtest_cron": "0 3 * * sun", + "schedule_ticker_universe_cron": "0 1 * * *", } # job id -> schedule setting key @@ -1496,6 +1478,8 @@ _CRON_JOBS: dict[str, str] = { "near_close_pipeline": "schedule_near_close_pipeline_cron", "after_close_pipeline": "schedule_after_close_pipeline_cron", "intraday_pipeline": "schedule_intraday_pipeline_cron", + "backtest": "schedule_backtest_cron", + "ticker_universe_sync": "schedule_ticker_universe_cron", } @@ -1630,9 +1614,12 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None: id="intraday_pipeline", name="Intraday Pipeline", replace_existing=True, ) - # Independent interval jobs (own cadence, no ordering dependency) + # Independent jobs (own cadence, no ordering dependency). Cron, not interval, + # for the reason documented at SCHEDULE_DEFAULTS: an interval countdown + # restarts on every deploy, so these could be deferred indefinitely. scheduler.add_job( - sync_ticker_universe, "interval", hours=24, + sync_ticker_universe, + _cron_trigger(cfg["schedule_ticker_universe_cron"], tz, "schedule_ticker_universe_cron"), id="ticker_universe_sync", name="Ticker Universe Sync", replace_existing=True, ) # Alerts auto-fire only via near_close_pipeline (scan → alert before MOC). @@ -1643,7 +1630,8 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None: replace_existing=True, next_run_time=None, ) scheduler.add_job( - run_backtest_job, "interval", hours=168, + run_backtest_job, + _cron_trigger(cfg["schedule_backtest_cron"], tz, "schedule_backtest_cron"), id="backtest", name="Backtest", replace_existing=True, ) # Deep history backfill: manual only (never auto-fires); triggered from diff --git a/app/schemas/admin.py b/app/schemas/admin.py index 05cf492..130bd40 100644 --- a/app/schemas/admin.py +++ b/app/schemas/admin.py @@ -83,6 +83,8 @@ class ScheduleConfigUpdate(BaseModel): schedule_near_close_pipeline_cron: str | None = Field(default=None, max_length=120) schedule_after_close_pipeline_cron: str | None = Field(default=None, max_length=120) schedule_intraday_pipeline_cron: str | None = Field(default=None, max_length=120) + schedule_backtest_cron: str | None = Field(default=None, max_length=120) + schedule_ticker_universe_cron: str | None = Field(default=None, max_length=120) class PerformanceConfigUpdate(BaseModel): diff --git a/app/services/admin_service.py b/app/services/admin_service.py index a5f20f8..f86ee9b 100644 --- a/app/services/admin_service.py +++ b/app/services/admin_service.py @@ -7,6 +7,7 @@ from passlib.hash import bcrypt from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession +from app import job_catalog from app.exceptions import DuplicateError, NotFoundError, ValidationError from app.models.fundamental import FundamentalData from app.models.ohlcv import OHLCVRecord @@ -606,91 +607,108 @@ async def get_pipeline_readiness(db: AsyncSession) -> list[dict]: # Job control (placeholder — scheduler is Task 12.1) # --------------------------------------------------------------------------- -VALID_JOB_NAMES = { - "data_collector", - "data_backfill", - "benchmark_collector", - "sentiment_collector", - "dolt_earnings_import", - "sec_fundamentals_import", - "rr_scanner", - "ticker_universe_sync", - "outcome_evaluator", - "alerts", - "market_regime", - "regime_monitor", - "event_study", - "backtest", - "daily_pipeline", - "near_close_pipeline", - "after_close_pipeline", - "intraday_pipeline", - "shadow_book", -} +# Job identity, labels and pipeline membership now live in app.job_catalog, which +# derives PIPELINE_MEMBERS from the pipeline step lists instead of restating them. +# Re-exported here because callers (routers, tests) import them from this module. +VALID_JOB_NAMES = job_catalog.VALID_JOB_NAMES +JOB_LABELS = job_catalog.JOB_LABELS +PIPELINE_MEMBERS = job_catalog.PIPELINE_MEMBERS -JOB_LABELS = { - "data_collector": "Data Collector (OHLCV)", - "data_backfill": "Data Backfill (deep history)", - "benchmark_collector": "Benchmark Collector", - "sentiment_collector": "Sentiment Collector", - "dolt_earnings_import": "Dolt Earnings Import", - "sec_fundamentals_import": "SEC Fundamentals Import", - "rr_scanner": "R:R Scanner", - "ticker_universe_sync": "Ticker Universe Sync", - "outcome_evaluator": "Outcome Evaluator", - "alerts": "Alerts Dispatcher", - # Keys are persisted job ids and must not change; these are display only. - "market_regime": "Market Trend (SPY)", - "regime_monitor": "AI/Tech Risk Monitor", - "event_study": "Event Study", - "backtest": "Backtest", - "daily_pipeline": "Morning Pipeline", - "near_close_pipeline": "Near-Close Pipeline (scan+alert)", - "after_close_pipeline": "After-Close Pipeline (outcome)", - "intraday_pipeline": "Intraday Pipeline", - "shadow_book": "Shadow Book (auto-traded strategy)", -} +# Anything further out than this is a parked backstop, not a schedule: pipeline +# steps and manual jobs are registered on a 520-week interval, and triggering one +# re-arms it. Belt-and-braces behind the category rule in _next_run_fields. +_NEXT_RUN_HORIZON_DAYS = 365 -# Jobs driven by a pipeline (in order) rather than their own auto timer. -PIPELINE_MEMBERS = { - "data_collector", - "benchmark_collector", - "sentiment_collector", - "rr_scanner", - "outcome_evaluator", - "alerts", - "market_regime", - "regime_monitor", - "shadow_book", -} + +def _visible_next_run(next_run: datetime | None) -> datetime | None: + """Drop a next-run that is really the parked backstop.""" + if next_run is None: + return None + horizon = datetime.now(next_run.tzinfo) + timedelta(days=_NEXT_RUN_HORIZON_DAYS) + return None if next_run > horizon else next_run + + +def _own_next_run(scheduler, name: str) -> datetime | None: + # getattr: APScheduler only sets next_run_time once the scheduler is running, + # so a job registered but not yet started has no such attribute at all. + job = scheduler.get_job(name) + return _visible_next_run(getattr(job, "next_run_time", None)) if job else None + + +def _next_run_fields(scheduler, name: str, enabled_map: dict[str, bool]) -> dict: + """Where this job's next run comes from, decided by category not by clock. + + A pipeline step has no meaningful schedule of its own, so reporting one is + the bug: its parent's timer is the answer. Manual jobs have no answer at all, + and saying so beats rendering a parked backstop as a date. + """ + category = job_catalog.JOB_CATEGORY.get(name) + if category == job_catalog.CATEGORY_STEP: + parents = job_catalog.PIPELINES_BY_MEMBER.get(name, ()) + soonest: datetime | None = None + via: str | None = None + for parent in parents: + if not enabled_map.get(parent, True): + continue + candidate = _own_next_run(scheduler, parent) + if candidate is not None and (soonest is None or candidate < soonest): + soonest, via = candidate, parent + return { + "next_run_at": None, + "next_run_source": "via_pipeline", + "via_next_run_at": soonest.isoformat() if soonest else None, + "via_next_run_job": via, + } + if category == job_catalog.CATEGORY_MANUAL: + return { + "next_run_at": None, + "next_run_source": "manual_only", + "via_next_run_at": None, + "via_next_run_job": None, + } + own = _own_next_run(scheduler, name) + return { + "next_run_at": own.isoformat() if own else None, + "next_run_source": "own_schedule", + "via_next_run_at": None, + "via_next_run_job": None, + } async def list_jobs(db: AsyncSession) -> list[dict]: - """Return status of all scheduled jobs.""" + """Return status of all scheduled jobs, grouped and ordered by category.""" from app.scheduler import get_job_runtime_snapshot, scheduler + visible = sorted(VALID_JOB_NAMES - job_catalog.HIDDEN_JOBS, key=job_catalog.sort_order) + # One query for every flag instead of one per job. Parents are read too, since + # a step reports its parent's next run only while that parent is enabled. + flags = await settings_store.get_map( + db, [f"job_{name}_enabled" for name in VALID_JOB_NAMES] + ) + enabled_map = { + name: flags.get(f"job_{name}_enabled", "true") == "true" + for name in VALID_JOB_NAMES + } + jobs_out = [] - for name in sorted(VALID_JOB_NAMES): - # Check enabled setting - setting = await settings_store.get_setting(db, f"job_{name}_enabled") - enabled = setting.value == "true" if setting else True # default enabled - - # Get scheduler job info + for name in visible: job = scheduler.get_job(name) - next_run = None - if job and job.next_run_time: - next_run = job.next_run_time.isoformat() - runtime = get_job_runtime_snapshot(name) jobs_out.append({ "name": name, "label": JOB_LABELS.get(name, name), - "enabled": enabled, - "next_run_at": next_run, - "via_pipeline": name in PIPELINE_MEMBERS, + "enabled": enabled_map.get(name, True), + "category": job_catalog.JOB_CATEGORY.get(name), + "sort_order": job_catalog.sort_order(name), + # Parent pipelines for a step; the steps themselves for a pipeline. + "pipelines": list(job_catalog.PIPELINES_BY_MEMBER.get(name, ())), + "steps": [step for step, _ in job_catalog.PIPELINE_STEPS.get(name, ())], "registered": job is not None, "running": bool(runtime.get("running", False)), + # runtime_* are strictly live in-memory state. Persisted history is + # reported separately as last_run_*, so a stale error cannot pin the + # status chip or the rate-limit banner. "runtime_status": runtime.get("status"), "runtime_processed": runtime.get("processed"), "runtime_total": runtime.get("total"), @@ -699,6 +717,7 @@ async def list_jobs(db: AsyncSession) -> list[dict]: "runtime_started_at": runtime.get("started_at"), "runtime_finished_at": runtime.get("finished_at"), "runtime_message": runtime.get("message"), + **_next_run_fields(scheduler, name, enabled_map), }) return jobs_out diff --git a/frontend/src/components/admin/ScheduleSettings.tsx b/frontend/src/components/admin/ScheduleSettings.tsx index ef74740..f5a66f0 100644 --- a/frontend/src/components/admin/ScheduleSettings.tsx +++ b/frontend/src/components/admin/ScheduleSettings.tsx @@ -11,6 +11,8 @@ const DEFAULTS: ScheduleConfig = { 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_backtest_cron: '0 3 * * sun', + schedule_ticker_universe_cron: '0 1 * * *', }; const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: boolean }[] = [ @@ -55,6 +57,18 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b hint: 'Refresh prices + resolve outcomes mid-session. Default hourly 10:00–15:00 ET weekdays.', mono: true, }, + { + key: 'schedule_backtest_cron', + label: 'Backtest', + hint: 'Replay history and refresh the Track Record report. Default Sunday 03:00 ET. Was a 168h interval, which restarted on every deploy and so could defer indefinitely.', + mono: true, + }, + { + key: 'schedule_ticker_universe_cron', + label: 'Ticker universe sync', + hint: 'Refresh the tracked-symbol universe. Default 01:00 ET daily, before the morning pipeline.', + mono: true, + }, ]; export function ScheduleSettings() { diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index a3d4314..180ba86 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -196,6 +196,8 @@ export interface ScheduleConfig { schedule_near_close_pipeline_cron: string; schedule_after_close_pipeline_cron: string; schedule_intraday_pipeline_cron: string; + schedule_backtest_cron: string; + schedule_ticker_universe_cron: string; } // Runtime sentiment LLM configuration diff --git a/tests/unit/test_admin_jobs.py b/tests/unit/test_admin_jobs.py new file mode 100644 index 0000000..9892ba6 --- /dev/null +++ b/tests/unit/test_admin_jobs.py @@ -0,0 +1,110 @@ +"""Admin → Jobs listing: categories, ordering, and next-run coherence. + +The panel used to render 19 jobs as one alphabetical list in which a pipeline +step, a cron job and a manual job were indistinguishable, and a triggered job +could advertise a next run ten years out. +""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from app import job_catalog +from app.scheduler import configure_scheduler, scheduler +from app.services.admin_service import _visible_next_run, list_jobs + + +@pytest.fixture(autouse=True) +def _configured_scheduler(): + scheduler.remove_all_jobs() + configure_scheduler() + yield + scheduler.remove_all_jobs() + + +def _by_name(jobs: list[dict]) -> dict[str, dict]: + return {job["name"]: job for job in jobs} + + +class TestVisibleNextRun: + def test_parked_backstop_is_not_a_schedule(self): + """Paused jobs carry a 520-week interval; triggering one re-arms it.""" + backstop = datetime.now(timezone.utc) + timedelta(weeks=520) + assert _visible_next_run(backstop) is None + + def test_a_real_upcoming_run_passes_through(self): + soon = datetime.now(timezone.utc) + timedelta(hours=6) + assert _visible_next_run(soon) == soon + + def test_none_stays_none(self): + assert _visible_next_run(None) is None + + +class TestListJobs: + async def test_hidden_jobs_are_not_listed_but_stay_valid(self, db_session): + jobs = _by_name(await list_jobs(db_session)) + assert "data_backfill" not in jobs + # Still triggerable through the API, and still registered. + assert "data_backfill" in job_catalog.VALID_JOB_NAMES + assert scheduler.get_job("data_backfill") is not None + + async def test_every_visible_job_has_a_category(self, db_session): + jobs = await list_jobs(db_session) + assert {j["name"] for j in jobs} == set( + job_catalog.VALID_JOB_NAMES - job_catalog.HIDDEN_JOBS + ) + assert all(j["category"] in job_catalog.CATEGORY_ORDER for j in jobs) + + async def test_jobs_arrive_grouped_by_category(self, db_session): + """The frontend renders sections in payload order, so ordering is the + API's job — not something each client re-derives.""" + categories = [j["category"] for j in await list_jobs(db_session)] + ranks = [job_catalog.CATEGORY_ORDER.index(c) for c in categories] + assert ranks == sorted(ranks) + + async def test_pipeline_steps_defer_their_schedule_to_the_parent(self, db_session): + jobs = _by_name(await list_jobs(db_session)) + step = jobs["rr_scanner"] + assert step["category"] == job_catalog.CATEGORY_STEP + assert step["next_run_at"] is None + assert step["next_run_source"] == "via_pipeline" + assert step["pipelines"] == ["near_close_pipeline"] + + async def test_step_reports_the_soonest_enabled_parent(self, db_session): + due = datetime.now(timezone.utc) + timedelta(hours=3) + scheduler.modify_job("daily_pipeline", next_run_time=due) + + collector = _by_name(await list_jobs(db_session))["data_collector"] + assert collector["via_next_run_job"] == "daily_pipeline" + assert collector["via_next_run_at"] == due.isoformat() + # Runs in all four pipelines — the reason steps are not nested under one. + assert set(collector["pipelines"]) == set(job_catalog.PIPELINE_JOBS) + + async def test_manual_jobs_say_so_instead_of_showing_a_date(self, db_session): + study = _by_name(await list_jobs(db_session))["event_study"] + assert study["category"] == job_catalog.CATEGORY_MANUAL + assert study["next_run_source"] == "manual_only" + assert study["next_run_at"] is None + + async def test_a_triggered_manual_job_still_shows_no_next_run(self, db_session): + """Regression: triggering re-armed the 520-week backstop, which the panel + rendered as a real 'next run in ~87600h'.""" + scheduler.modify_job("event_study", next_run_time=datetime.now(timezone.utc)) + scheduler.modify_job("event_study", next_run_time=None) + + study = _by_name(await list_jobs(db_session))["event_study"] + assert study["next_run_at"] is None + + async def test_pipelines_report_their_own_schedule_and_steps(self, db_session): + pipeline = _by_name(await list_jobs(db_session))["daily_pipeline"] + assert pipeline["category"] == job_catalog.CATEGORY_PIPELINE + assert pipeline["next_run_source"] == "own_schedule" + assert pipeline["steps"] == [ + step for step, _ in job_catalog.PIPELINE_STEPS["daily_pipeline"] + ] + + async def test_standalone_jobs_keep_their_own_schedule(self, db_session): + backtest = _by_name(await list_jobs(db_session))["backtest"] + assert backtest["category"] == job_catalog.CATEGORY_SCHEDULED + assert backtest["next_run_source"] == "own_schedule" + assert backtest["pipelines"] == [] diff --git a/tests/unit/test_scheduler.py b/tests/unit/test_scheduler.py index 1fb6dda..5cf94a0 100644 --- a/tests/unit/test_scheduler.py +++ b/tests/unit/test_scheduler.py @@ -1,16 +1,19 @@ """Unit tests for app.scheduler module.""" import asyncio +from datetime import datetime, timezone from types import SimpleNamespace import pytest +from app import job_catalog from app.scheduler import ( _DAILY_PIPELINE_STEPS, _NEAR_CLOSE_PIPELINE_STEPS, _consume_backtest_options, _consume_backtest_target_model, _parse_frequency, + _repause_after_manual_run, _resume_tickers, _last_successful, _run_source_import, @@ -112,60 +115,119 @@ class TestResumeTickers: class TestConfigureScheduler: def test_configure_adds_all_jobs(self): - # Remove any existing jobs first + # Derived from the catalog, not a fourth hand-maintained copy of the + # job list: a job added to the catalog but never registered now fails + # here instead of silently rendering "Not registered" in the admin UI. scheduler.remove_all_jobs() configure_scheduler() - jobs = scheduler.get_jobs() - job_ids = {j.id for j in jobs} - assert job_ids == { - "data_collector", - "data_backfill", - "benchmark_collector", - "sentiment_collector", - "dolt_earnings_import", - "sec_fundamentals_import", - "rr_scanner", - "shadow_book", - "ticker_universe_sync", - "outcome_evaluator", - "alerts", - "market_regime", - "regime_monitor", - "event_study", - "backtest", - "daily_pipeline", - "near_close_pipeline", - "after_close_pipeline", - "intraday_pipeline", - } + assert {j.id for j in scheduler.get_jobs()} == set(job_catalog.VALID_JOB_NAMES) def test_configure_is_idempotent(self): scheduler.remove_all_jobs() configure_scheduler() configure_scheduler() # Should replace, not duplicate job_ids = [j.id for j in scheduler.get_jobs()] - # Each ID should appear exactly once - assert sorted(job_ids) == sorted([ - "after_close_pipeline", - "alerts", - "backtest", - "benchmark_collector", - "daily_pipeline", - "intraday_pipeline", + assert sorted(job_ids) == sorted(job_catalog.VALID_JOB_NAMES) + + def test_independent_jobs_use_cron_not_interval(self): + """Interval countdowns restart on every deploy, so a weekly interval on a + frequently-redeployed box can defer forever. Both standalone jobs were + migrated to cron; this pins them there.""" + scheduler.remove_all_jobs() + configure_scheduler() + for job_id in ("backtest", "ticker_universe_sync"): + trigger = type(scheduler.get_job(job_id).trigger).__name__ + assert trigger == "CronTrigger", f"{job_id} regressed to {trigger}" + + +class TestJobCatalog: + def test_pipeline_members_are_derived_from_step_lists(self): + derived = { + step + for steps in job_catalog.PIPELINE_STEPS.values() + for step, _ in steps + } + assert job_catalog.PIPELINE_MEMBERS == derived + # ...and reproduces the set that used to be maintained by hand, so the + # derivation is behaviour-preserving rather than merely self-consistent. + assert job_catalog.PIPELINE_MEMBERS == { "data_collector", - "data_backfill", - "dolt_earnings_import", - "sec_fundamentals_import", - "market_regime", - "near_close_pipeline", - "regime_monitor", - "event_study", - "outcome_evaluator", - "rr_scanner", + "benchmark_collector", "sentiment_collector", + "rr_scanner", "shadow_book", - "ticker_universe_sync", - ]) + "outcome_evaluator", + "alerts", + "market_regime", + "regime_monitor", + } + + def test_categories_partition_every_job_exactly_once(self): + buckets = [ + job_catalog.PIPELINE_JOBS, + job_catalog.PIPELINE_STEP_JOBS, + job_catalog.SCHEDULED_JOBS, + job_catalog.MANUAL_JOBS, + ] + flat = [name for bucket in buckets for name in bucket] + assert len(flat) == len(set(flat)), "a job is in two categories" + assert set(flat) == set(job_catalog.VALID_JOB_NAMES) + assert all(name in job_catalog.JOB_CATEGORY for name in flat) + + def test_every_job_has_a_label_and_a_unique_sort_order(self): + names = job_catalog.VALID_JOB_NAMES + assert set(job_catalog.JOB_LABELS) == set(names) + assert len({job_catalog.sort_order(n) for n in names}) == len(names) + + def test_multi_pipeline_members_report_every_parent(self): + """Membership is many-to-many — the reason the UI groups into sections + rather than nesting steps under one parent.""" + by_member = job_catalog.PIPELINES_BY_MEMBER + assert set(by_member["data_collector"]) == set(job_catalog.PIPELINE_JOBS) + assert set(by_member["alerts"]) == {"daily_pipeline", "near_close_pipeline"} + assert set(by_member["outcome_evaluator"]) == { + "intraday_pipeline", + "after_close_pipeline", + } + assert "backtest" not in by_member + + def test_every_job_has_a_runtime_row_before_it_first_runs(self): + """The old private _JOB_NAMES list held 16 of 19, so three jobs showed no + last-run line until their first run in a given process.""" + assert set(get_job_runtime_snapshot()) == set(job_catalog.VALID_JOB_NAMES) + + +class TestRepauseListener: + def _configured(self): + scheduler.remove_all_jobs() + configure_scheduler() + + def test_manual_job_is_repaused_after_running(self): + """Triggering a paused job re-arms its 520-week backstop, which used to + surface as a "next run in ~87600h".""" + self._configured() + scheduler.modify_job("event_study", next_run_time=datetime.now(timezone.utc)) + _repause_after_manual_run(SimpleNamespace(job_id="event_study")) + assert scheduler.get_job("event_study").next_run_time is None + + def test_pipeline_step_is_repaused_after_running(self): + self._configured() + scheduler.modify_job("rr_scanner", next_run_time=datetime.now(timezone.utc)) + _repause_after_manual_run(SimpleNamespace(job_id="rr_scanner")) + assert scheduler.get_job("rr_scanner").next_run_time is None + + def test_cron_jobs_are_left_alone(self): + # Set an explicit next run first: an unstarted scheduler leaves the + # attribute unset, so comparing None to None would prove nothing. + self._configured() + due = datetime.now(timezone.utc) + scheduler.modify_job("daily_pipeline", next_run_time=due) + _repause_after_manual_run(SimpleNamespace(job_id="daily_pipeline")) + assert scheduler.get_job("daily_pipeline").next_run_time == due + + def test_unknown_job_is_ignored(self): + self._configured() + _repause_after_manual_run(SimpleNamespace(job_id="not_a_job")) class _SessionContext: