dennisthiessen
ebff19940b
A new /regime tab scoring how far the AI/Tech bull regime has deteriorated toward a re-rating as a single 0-100 index with per-signal breakdown and a 7/30-day trend. Intentionally decoupled: nothing reads its output to gate or score trades — the daily-pipeline membership is scheduling only. - regime_monitor_service: price sub-scores (P1-P6 via Alpaca, like market_regime), VIX + HY credit spreads via a small FRED helper, weighted aggregation over available signals (missing source -> n/a, dropped from the denominator), one snapshot row/day, and a ~90-day history backfill by replaying the already-fetched series as-of each past day. - F1/F3 fundamentals proposed by the configured grounded LLM (reuses sentiment_provider_service config resolution), with a manual override + lock. - regime_snapshots table (migration 011); endpoints on the existing market router; admin-editable weights/threshold; standalone /regime page. Data needs: prices via Alpaca, VIX/credit via FRED (optional key — signals show n/a without it). No LLM needed for history. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1138 lines
50 KiB
Python
1138 lines
50 KiB
Python
"""APScheduler job definitions and FastAPI lifespan integration.
|
|
|
|
Defines four scheduled jobs:
|
|
- Data Collector (OHLCV fetch for all tickers)
|
|
- Sentiment Collector (sentiment for all tickers)
|
|
- Fundamental Collector (fundamentals for all tickers)
|
|
- R:R Scanner (trade setup scan for all tickers)
|
|
|
|
Each job processes tickers independently, logs errors as structured JSON,
|
|
handles rate limits by recording the last successful ticker, and checks
|
|
SystemSetting for enabled/disabled state.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import asyncio
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from sqlalchemy import case, func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import async_session_factory
|
|
from app.models.fundamental import FundamentalData
|
|
from app.models.ohlcv import OHLCVRecord
|
|
from app.models.sentiment import SentimentScore
|
|
from app.models.ticker import Ticker
|
|
from app.exceptions import ProviderError
|
|
from app.providers.alpaca import AlpacaOHLCVProvider
|
|
from app.providers.fundamentals_chain import build_fundamental_provider_chain
|
|
from app.providers.protocol import SentimentData
|
|
from app.services import fundamental_service, ingestion_service, sentiment_service, settings_store
|
|
from app.services.alert_service import dispatch_alerts
|
|
from app.services.backtest_service import run_and_store as run_backtest_and_store
|
|
from app.services.market_regime_service import update_market_regime
|
|
from app.services.regime_monitor_service import update_regime_monitor
|
|
from app.services.outcome_service import evaluate_pending_setups
|
|
from app.services.rr_scanner_service import scan_all_tickers
|
|
from app.services.sentiment_provider_service import build_sentiment_provider
|
|
from app.services.ticker_universe_service import bootstrap_universe
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Module-level scheduler instance.
|
|
#
|
|
# job_defaults matter a lot here: this is a single-process app, so the scheduler
|
|
# shares one event loop with the API and every other job. APScheduler's default
|
|
# misfire_grace_time is just 1 second — if the loop is busy at the instant a
|
|
# daily job is due (e.g. the scanner is mid-run), the fire is processed late,
|
|
# flagged a misfire, and SILENTLY SKIPPED while next_run still advances 24h. So
|
|
# we grant a generous grace window, coalesce missed runs into one catch-up, and
|
|
# cap each job at a single concurrent instance.
|
|
scheduler = AsyncIOScheduler(
|
|
job_defaults={
|
|
"coalesce": True,
|
|
"max_instances": 1,
|
|
"misfire_grace_time": 3600, # tolerate a busy loop; a daily job up to 1h late is fine
|
|
}
|
|
)
|
|
|
|
# Track last successful ticker per job for rate-limit resume
|
|
_last_successful: dict[str, str | None] = {
|
|
"data_collector": None,
|
|
"data_backfill": None,
|
|
"sentiment_collector": None,
|
|
"fundamental_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",
|
|
"fundamental_collector",
|
|
"rr_scanner",
|
|
"ticker_universe_sync",
|
|
"alerts",
|
|
"market_regime",
|
|
"regime_monitor",
|
|
"backtest",
|
|
"daily_pipeline",
|
|
"intraday_pipeline",
|
|
]
|
|
|
|
|
|
def _idle_runtime() -> dict[str, object]:
|
|
return {
|
|
"running": False,
|
|
"status": "idle",
|
|
"processed": 0,
|
|
"total": None,
|
|
"progress_pct": None,
|
|
"current_ticker": None,
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
"message": None,
|
|
}
|
|
|
|
|
|
_job_runtime: dict[str, dict[str, object]] = {name: _idle_runtime() for name in _JOB_NAMES}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _log_event(level: int, event: str, **fields: object) -> None:
|
|
"""Emit a structured JSON log line: {"event": ..., **fields}."""
|
|
logger.log(level, json.dumps({"event": event, **fields}))
|
|
|
|
|
|
def _log_job_error(job_name: str, ticker: str, error: Exception) -> None:
|
|
"""Log a per-ticker job error as structured JSON."""
|
|
_log_event(
|
|
logging.ERROR, "job_error", job=job_name, ticker=ticker,
|
|
error_type=type(error).__name__, message=str(error),
|
|
)
|
|
|
|
|
|
def _runtime_start(job_name: str, total: int | None = None, message: str | None = None) -> None:
|
|
_job_runtime[job_name] = {
|
|
**_idle_runtime(),
|
|
"running": True,
|
|
"status": "running",
|
|
"total": total,
|
|
"progress_pct": 0.0 if total and total > 0 else None,
|
|
"started_at": datetime.now(timezone.utc).isoformat(),
|
|
"message": message,
|
|
}
|
|
|
|
|
|
def _runtime_progress(
|
|
job_name: str,
|
|
processed: int,
|
|
total: int | None,
|
|
current_ticker: str | None = None,
|
|
message: str | None = None,
|
|
) -> None:
|
|
progress_pct: float | None = None
|
|
if total and total > 0:
|
|
progress_pct = round((processed / total) * 100.0, 1)
|
|
runtime = _job_runtime.get(job_name, {})
|
|
runtime.update({
|
|
"running": True,
|
|
"status": "running",
|
|
"processed": processed,
|
|
"total": total,
|
|
"progress_pct": progress_pct,
|
|
"current_ticker": current_ticker,
|
|
"message": message,
|
|
})
|
|
_job_runtime[job_name] = runtime
|
|
|
|
|
|
def _runtime_finish(
|
|
job_name: str,
|
|
status: str,
|
|
processed: int,
|
|
total: int | None,
|
|
message: str | None = None,
|
|
) -> None:
|
|
runtime = _job_runtime.get(job_name, {})
|
|
runtime.update({
|
|
"running": False,
|
|
"status": status,
|
|
"processed": processed,
|
|
"total": total,
|
|
"progress_pct": 100.0 if total and processed >= total else runtime.get("progress_pct"),
|
|
"current_ticker": None,
|
|
"finished_at": datetime.now(timezone.utc).isoformat(),
|
|
"message": message,
|
|
})
|
|
_job_runtime[job_name] = runtime
|
|
|
|
|
|
def get_job_runtime_snapshot(job_name: str | None = None) -> dict[str, dict[str, object]] | dict[str, object]:
|
|
if job_name is not None:
|
|
return dict(_job_runtime.get(job_name, {}))
|
|
return {name: dict(meta) for name, meta in _job_runtime.items()}
|
|
|
|
|
|
async def _is_job_enabled(db: AsyncSession, job_name: str) -> bool:
|
|
"""Check SystemSetting for job enabled state. Defaults to True."""
|
|
setting = await settings_store.get_setting(db, f"job_{job_name}_enabled")
|
|
return setting is None or setting.value.lower() == "true"
|
|
|
|
|
|
async def _get_all_tickers(db: AsyncSession) -> list[str]:
|
|
"""Return all tracked ticker symbols sorted alphabetically."""
|
|
result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _get_ohlcv_priority_tickers(db: AsyncSession) -> list[str]:
|
|
"""Return symbols prioritized for OHLCV collection.
|
|
|
|
Priority:
|
|
1) Tickers with no OHLCV bars
|
|
2) Tickers with data, oldest latest OHLCV date first
|
|
3) Alphabetical tiebreaker
|
|
"""
|
|
latest_date = func.max(OHLCVRecord.date)
|
|
missing_first = case((latest_date.is_(None), 0), else_=1)
|
|
result = await db.execute(
|
|
select(Ticker.symbol)
|
|
.outerjoin(OHLCVRecord, OHLCVRecord.ticker_id == Ticker.id)
|
|
.group_by(Ticker.id, Ticker.symbol)
|
|
.order_by(missing_first.asc(), latest_date.asc(), Ticker.symbol.asc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _get_sentiment_priority_tickers(db: AsyncSession) -> list[str]:
|
|
"""Symbols to fetch sentiment for, budgeted to stay in the free search tier.
|
|
|
|
Scope: only tickers that matter — watchlist + open paper trades + top-N by
|
|
composite score + the momentum leaders the activation gate qualifies on. Skip
|
|
any refreshed within ``sentiment_fresh_hours``. Cap the run at
|
|
``sentiment_max_per_run``, oldest/missing first. Once the relevant set is
|
|
fresh, runs make zero grounded searches until it ages out.
|
|
"""
|
|
from app.models.paper_trade import PaperTrade
|
|
from app.models.score import CompositeScore
|
|
from app.models.watchlist import WatchlistEntry
|
|
|
|
relevant: set[int] = set()
|
|
wl = await db.execute(
|
|
select(WatchlistEntry.ticker_id)
|
|
.where(WatchlistEntry.entry_type != "dismissed")
|
|
.distinct()
|
|
)
|
|
relevant.update(r[0] for r in wl.all())
|
|
pt = await db.execute(
|
|
select(PaperTrade.ticker_id).where(PaperTrade.status == "open").distinct()
|
|
)
|
|
relevant.update(r[0] for r in pt.all())
|
|
top = await db.execute(
|
|
select(CompositeScore.ticker_id)
|
|
.order_by(CompositeScore.score.desc())
|
|
.limit(settings.sentiment_top_composite)
|
|
)
|
|
relevant.update(r[0] for r in top.all())
|
|
|
|
# Momentum leaders: the tickers that can clear the activation gate, which
|
|
# selects the top ``min_momentum_percentile`` slice by 12-1 momentum — a
|
|
# different axis than composite score. The gate qualifies setups on this
|
|
# percentile, so without including them a freshly-qualifying ticker carries no
|
|
# sentiment and gets enhanced as neutral. Pre-fetching their sentiment here (in
|
|
# the daily pipeline, sentiment runs right after the OHLCV refresh) means the
|
|
# following R:R scan reads real sentiment for the setups it qualifies.
|
|
# Best-effort: a momentum/config failure must not stop sentiment collection.
|
|
try:
|
|
from app.services import momentum_service
|
|
from app.services.admin_service import get_activation_config
|
|
|
|
activation = await get_activation_config(db)
|
|
min_pct = float(activation.get("min_momentum_percentile", 0.0))
|
|
if min_pct > 0:
|
|
percentiles = await momentum_service.compute_momentum_percentiles(db)
|
|
leaders = [sym for sym, pct in percentiles.items() if pct >= min_pct]
|
|
if leaders:
|
|
rows = await db.execute(
|
|
select(Ticker.id).where(Ticker.symbol.in_(leaders))
|
|
)
|
|
relevant.update(r[0] for r in rows.all())
|
|
except Exception:
|
|
logger.exception("Sentiment momentum-leader scoping failed; using base relevant set")
|
|
|
|
if not relevant:
|
|
return []
|
|
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=settings.sentiment_fresh_hours)
|
|
latest_ts = func.max(SentimentScore.timestamp)
|
|
missing_first = case((latest_ts.is_(None), 0), else_=1)
|
|
result = await db.execute(
|
|
select(Ticker.symbol)
|
|
.outerjoin(SentimentScore, SentimentScore.ticker_id == Ticker.id)
|
|
.where(Ticker.id.in_(relevant))
|
|
.group_by(Ticker.id, Ticker.symbol)
|
|
.having(or_(latest_ts.is_(None), latest_ts < cutoff))
|
|
.order_by(missing_first.asc(), latest_ts.asc(), Ticker.symbol.asc())
|
|
.limit(settings.sentiment_max_per_run)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _get_fundamental_priority_tickers(db: AsyncSession) -> list[str]:
|
|
"""Return symbols prioritized for fundamentals refresh.
|
|
|
|
Priority:
|
|
1) Tickers with no fundamentals snapshot yet
|
|
2) Tickers with existing fundamentals, oldest fetched_at first
|
|
3) Alphabetical tiebreaker
|
|
"""
|
|
missing_first = case((FundamentalData.fetched_at.is_(None), 0), else_=1)
|
|
result = await db.execute(
|
|
select(Ticker.symbol)
|
|
.outerjoin(FundamentalData, FundamentalData.ticker_id == Ticker.id)
|
|
.order_by(missing_first.asc(), FundamentalData.fetched_at.asc(), Ticker.symbol.asc())
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
def _resume_tickers(symbols: list[str], job_name: str) -> list[str]:
|
|
"""Reorder tickers to resume after the last successful one (rate-limit resume).
|
|
|
|
If a previous run was rate-limited, start from the ticker after the last
|
|
successful one. Otherwise return the full list.
|
|
"""
|
|
last = _last_successful.get(job_name)
|
|
if last is None or last not in symbols:
|
|
return symbols
|
|
idx = symbols.index(last)
|
|
# Start from the next ticker, then wrap around
|
|
return symbols[idx + 1:] + symbols[:idx + 1]
|
|
|
|
|
|
def _chunked(symbols: list[str], chunk_size: int) -> list[list[str]]:
|
|
size = max(1, chunk_size)
|
|
return [symbols[i:i + size] for i in range(0, len(symbols), size)]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Data Collector (OHLCV)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def collect_ohlcv(full_backfill: bool = False, job_name: str = "data_collector") -> None:
|
|
"""Fetch latest daily OHLCV for all tracked tickers.
|
|
|
|
Uses AlpacaOHLCVProvider. Processes each ticker independently.
|
|
On rate limit, records last successful ticker for resume.
|
|
Start date is resolved by ingestion progress:
|
|
- existing ticker: resume from last_ingested_date + 1
|
|
- new ticker: backfill the configured history window
|
|
|
|
``full_backfill`` forces every ticker to re-fetch the full
|
|
``settings.ohlcv_history_days`` window (ignoring incremental resume) — used by
|
|
the manual data_backfill job to deepen shallow histories. ``job_name`` lets the
|
|
backfill report its own runtime/resume state separate from data_collector.
|
|
"""
|
|
_log_event(logging.INFO, "job_start", job=job_name)
|
|
_runtime_start(job_name)
|
|
processed = 0
|
|
total: int | None = None
|
|
|
|
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=0, message="Disabled")
|
|
return
|
|
|
|
symbols = await _get_ohlcv_priority_tickers(db)
|
|
if not symbols:
|
|
_log_event(logging.INFO, "job_complete", job=job_name, tickers=0)
|
|
_runtime_finish(job_name, "completed", processed=0, total=0, message="No tickers")
|
|
return
|
|
|
|
total = len(symbols)
|
|
_runtime_progress(job_name, processed=0, total=total)
|
|
|
|
# Build provider (skip if keys not configured)
|
|
if not settings.alpaca_api_key or not settings.alpaca_api_secret:
|
|
_log_event(logging.WARNING, "job_skipped", job=job_name, reason="alpaca keys not configured")
|
|
_runtime_finish(job_name, "skipped", processed=0, total=total, message="Alpaca keys not configured")
|
|
return
|
|
|
|
try:
|
|
provider = AlpacaOHLCVProvider(settings.alpaca_api_key, settings.alpaca_api_secret)
|
|
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=total, message=str(exc))
|
|
return
|
|
|
|
end_date = date.today()
|
|
# Full backfill: pass an explicit start_date so fetch_and_ingest re-pulls
|
|
# the whole window instead of resuming from the last stored bar.
|
|
backfill_start = (
|
|
end_date - timedelta(days=settings.ohlcv_history_days) if full_backfill else None
|
|
)
|
|
|
|
for symbol in symbols:
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
|
|
async with async_session_factory() as db:
|
|
try:
|
|
result = await ingestion_service.fetch_and_ingest(
|
|
db, provider, symbol, start_date=backfill_start, end_date=end_date,
|
|
)
|
|
_last_successful[job_name] = symbol
|
|
processed += 1
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
|
|
_log_event(logging.INFO, "ticker_collected", job=job_name, ticker=symbol, status=result.status, records=result.records_ingested)
|
|
if result.status == "partial":
|
|
# Rate limited — stop and resume next run
|
|
_log_event(logging.WARNING, "rate_limited", job=job_name, ticker=symbol, processed=processed)
|
|
_runtime_finish(job_name, "rate_limited", processed=processed, total=total, message=f"Rate limited at {symbol}")
|
|
return
|
|
except Exception as exc:
|
|
_log_job_error(job_name, symbol, exc)
|
|
|
|
# Reset resume pointer on full completion
|
|
_last_successful[job_name] = None
|
|
_log_event(logging.INFO, "job_complete", job=job_name, tickers=processed)
|
|
_runtime_finish(job_name, "completed", processed=processed, total=total, message=f"Processed {processed} tickers")
|
|
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=processed, total=total, message=str(exc))
|
|
|
|
|
|
async def backfill_ohlcv() -> None:
|
|
"""Deep historical backfill: re-fetch the full ``settings.ohlcv_history_days``
|
|
window for every ticker, ignoring incremental resume.
|
|
|
|
Manual/triggered job (Admin → Jobs). Run once to deepen the ~1-year histories
|
|
so long-lookback factors (12-month momentum, 52-week high) and multi-regime
|
|
backtests become computable. Idempotent (upsert); resumes after rate limits.
|
|
"""
|
|
await collect_ohlcv(full_backfill=True, job_name="data_backfill")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Sentiment Collector
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def collect_sentiment() -> None:
|
|
"""Fetch sentiment for all tracked tickers via OpenAI.
|
|
|
|
Processes each ticker independently. On rate limit, records last
|
|
successful ticker for resume.
|
|
"""
|
|
job_name = "sentiment_collector"
|
|
_log_event(logging.INFO, "job_start", job=job_name)
|
|
_runtime_start(job_name)
|
|
processed = 0
|
|
total: int | None = None
|
|
|
|
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=0, message="Disabled")
|
|
return
|
|
|
|
symbols = await _get_sentiment_priority_tickers(db)
|
|
if not symbols:
|
|
_log_event(logging.INFO, "job_complete", job=job_name, tickers=0)
|
|
_runtime_finish(job_name, "completed", processed=0, total=0, message="No tickers")
|
|
return
|
|
|
|
total = len(symbols)
|
|
_runtime_progress(job_name, processed=0, total=total)
|
|
|
|
try:
|
|
async with async_session_factory() as cfg_db:
|
|
provider = await build_sentiment_provider(cfg_db)
|
|
except ProviderError as exc:
|
|
_log_event(logging.WARNING, "job_skipped", job=job_name, reason=str(exc))
|
|
_runtime_finish(job_name, "skipped", processed=0, total=total, message=str(exc))
|
|
return
|
|
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=total, message=str(exc))
|
|
return
|
|
|
|
batch_size = max(1, settings.openai_sentiment_batch_size)
|
|
batches = _chunked(symbols, batch_size)
|
|
|
|
for batch in batches:
|
|
current_hint = batch[0] if len(batch) == 1 else f"{batch[0]} (+{len(batch) - 1})"
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=current_hint)
|
|
|
|
batch_results: dict[str, SentimentData] = {}
|
|
if len(batch) > 1 and hasattr(provider, "fetch_sentiment_batch"):
|
|
try:
|
|
batch_results = await provider.fetch_sentiment_batch(batch)
|
|
except Exception as exc:
|
|
msg = str(exc).lower()
|
|
if "rate" in msg or "quota" in msg or "429" in msg:
|
|
_log_event(logging.WARNING, "rate_limited", job=job_name, ticker=batch[0], processed=processed)
|
|
_runtime_finish(job_name, "rate_limited", processed=processed, total=total, message=f"Rate limited at {batch[0]}")
|
|
return
|
|
_log_event(logging.WARNING, "batch_fallback", job=job_name, batch=batch, reason=str(exc))
|
|
|
|
for symbol in batch:
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
|
|
data = batch_results.get(symbol) if batch_results else None
|
|
|
|
if data is None:
|
|
try:
|
|
data = await provider.fetch_sentiment(symbol)
|
|
except Exception as exc:
|
|
msg = str(exc).lower()
|
|
if "rate" in msg or "quota" in msg or "429" in msg:
|
|
_log_event(logging.WARNING, "rate_limited", job=job_name, ticker=symbol, processed=processed)
|
|
_runtime_finish(job_name, "rate_limited", processed=processed, total=total, message=f"Rate limited at {symbol}")
|
|
return
|
|
_log_job_error(job_name, symbol, exc)
|
|
continue
|
|
|
|
async with async_session_factory() as db:
|
|
try:
|
|
await sentiment_service.store_sentiment(
|
|
db,
|
|
symbol=symbol,
|
|
classification=data.classification,
|
|
confidence=data.confidence,
|
|
source=data.source,
|
|
timestamp=data.timestamp,
|
|
reasoning=data.reasoning,
|
|
citations=data.citations,
|
|
recommendation=data.recommendation,
|
|
)
|
|
_last_successful[job_name] = symbol
|
|
processed += 1
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
|
|
_log_event(logging.INFO, "ticker_collected", job=job_name, ticker=symbol, classification=data.classification, confidence=data.confidence)
|
|
except Exception as exc:
|
|
_log_job_error(job_name, symbol, exc)
|
|
|
|
_last_successful[job_name] = None
|
|
_log_event(logging.INFO, "job_complete", job=job_name, tickers=processed)
|
|
_runtime_finish(job_name, "completed", processed=processed, total=total, message=f"Processed {processed} tickers")
|
|
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=processed, total=total, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Fundamental Collector
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def collect_fundamentals() -> None:
|
|
"""Fetch fundamentals for all tracked tickers via FMP.
|
|
|
|
Processes each ticker independently. On rate limit, records last
|
|
successful ticker for resume.
|
|
"""
|
|
job_name = "fundamental_collector"
|
|
_log_event(logging.INFO, "job_start", job=job_name)
|
|
_runtime_start(job_name)
|
|
processed = 0
|
|
total: int | None = None
|
|
|
|
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=0, message="Disabled")
|
|
return
|
|
|
|
symbols = await _get_fundamental_priority_tickers(db)
|
|
if not symbols:
|
|
_log_event(logging.INFO, "job_complete", job=job_name, tickers=0)
|
|
_runtime_finish(job_name, "completed", processed=0, total=0, message="No tickers")
|
|
return
|
|
|
|
total = len(symbols)
|
|
_runtime_progress(job_name, processed=0, total=total)
|
|
|
|
if not (settings.fmp_api_key or settings.finnhub_api_key or settings.alpha_vantage_api_key):
|
|
_log_event(logging.WARNING, "job_skipped", job=job_name, reason="no fundamentals provider keys configured")
|
|
_runtime_finish(job_name, "skipped", processed=0, total=total, message="No fundamentals provider keys configured")
|
|
return
|
|
|
|
try:
|
|
provider = build_fundamental_provider_chain()
|
|
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=total, message=str(exc))
|
|
return
|
|
|
|
max_retries = max(0, settings.fundamental_rate_limit_retries)
|
|
base_backoff = max(1, settings.fundamental_rate_limit_backoff_seconds)
|
|
spacing = max(0.0, settings.fundamental_request_spacing_seconds)
|
|
|
|
async def _store(symbol: str, data) -> None:
|
|
async with async_session_factory() as db:
|
|
await fundamental_service.store_fundamental(
|
|
db,
|
|
symbol=symbol,
|
|
pe_ratio=data.pe_ratio,
|
|
revenue_growth=data.revenue_growth,
|
|
earnings_surprise=data.earnings_surprise,
|
|
market_cap=data.market_cap,
|
|
next_earnings_date=data.next_earnings_date,
|
|
unavailable_fields=data.unavailable_fields,
|
|
)
|
|
|
|
for symbol in symbols:
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
|
|
attempt = 0
|
|
while True:
|
|
try:
|
|
data = await provider.fetch_fundamentals(symbol)
|
|
await _store(symbol, data)
|
|
_last_successful[job_name] = symbol
|
|
processed += 1
|
|
_runtime_progress(job_name, processed=processed, total=total, current_ticker=symbol)
|
|
_log_event(logging.INFO, "ticker_collected", job=job_name, ticker=symbol)
|
|
break
|
|
except Exception as exc:
|
|
msg = str(exc).lower()
|
|
if "rate" in msg or "429" in msg:
|
|
if attempt < max_retries:
|
|
wait_seconds = base_backoff * (2 ** attempt)
|
|
attempt += 1
|
|
_log_event(logging.WARNING, "rate_limited_retry", job=job_name, ticker=symbol, attempt=attempt, max_retries=max_retries, wait_seconds=wait_seconds, processed=processed)
|
|
_runtime_progress(
|
|
job_name,
|
|
processed=processed,
|
|
total=total,
|
|
current_ticker=symbol,
|
|
message=f"Rate-limited at {symbol}; retry {attempt}/{max_retries} in {wait_seconds}s",
|
|
)
|
|
await asyncio.sleep(wait_seconds)
|
|
continue
|
|
|
|
# Retries exhausted: store whatever partial data we can
|
|
# still get (e.g. FMP market cap) and move on, rather than
|
|
# aborting the whole run and leaving every later ticker
|
|
# untouched.
|
|
_log_event(logging.WARNING, "rate_limited_partial", job=job_name, ticker=symbol, processed=processed)
|
|
try:
|
|
data = await provider.fetch_fundamentals(symbol, allow_partial=True)
|
|
await _store(symbol, data)
|
|
processed += 1
|
|
except Exception as exc2:
|
|
_log_job_error(job_name, symbol, exc2)
|
|
break
|
|
_log_job_error(job_name, symbol, exc)
|
|
break
|
|
|
|
if spacing:
|
|
await asyncio.sleep(spacing)
|
|
|
|
_last_successful[job_name] = None
|
|
_log_event(logging.INFO, "job_complete", job=job_name, tickers=processed)
|
|
_runtime_finish(job_name, "completed", processed=processed, total=total, message=f"Processed {processed} tickers")
|
|
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=processed, total=total, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: R:R Scanner
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def scan_rr() -> None:
|
|
"""Scan all tickers for trade setups meeting the R:R threshold.
|
|
|
|
Uses rr_scanner_service.scan_all_tickers which already handles
|
|
per-ticker error isolation internally.
|
|
"""
|
|
job_name = "rr_scanner"
|
|
_log_event(logging.INFO, "job_start", job=job_name)
|
|
_runtime_start(job_name)
|
|
processed = 0
|
|
total: int | None = None
|
|
|
|
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=0, message="Disabled")
|
|
return
|
|
|
|
symbols = await _get_all_tickers(db)
|
|
total = len(symbols)
|
|
_runtime_progress(job_name, processed=0, total=total)
|
|
|
|
def _on_progress(done: int, count: int, symbol: str) -> None:
|
|
_runtime_progress(
|
|
job_name, processed=done, total=count, current_ticker=symbol or None
|
|
)
|
|
|
|
try:
|
|
setups = await scan_all_tickers(
|
|
db, rr_threshold=settings.default_rr_threshold,
|
|
progress_callback=_on_progress,
|
|
)
|
|
processed = total or 0
|
|
_runtime_finish(job_name, "completed", processed=processed, total=total, message=f"Found {len(setups)} setups")
|
|
_log_event(logging.INFO, "job_complete", job=job_name, setups_found=len(setups))
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=processed, total=total, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
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=processed, total=total, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Outcome Evaluator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def evaluate_outcomes() -> None:
|
|
"""Evaluate unresolved trade setups against OHLCV data collected since.
|
|
|
|
Writes actual_outcome / outcome_date / evaluated_at on each decided setup.
|
|
Undecided setups stay pending and are re-checked on the next run.
|
|
"""
|
|
job_name = "outcome_evaluator"
|
|
_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
|
|
|
|
summary = await evaluate_pending_setups(
|
|
db, max_bars=settings.outcome_evaluation_max_bars
|
|
)
|
|
from app.services import paper_trade_service
|
|
closed_trades = await paper_trade_service.resolve_open_trades(db)
|
|
|
|
_runtime_progress(job_name, processed=1, total=1)
|
|
_runtime_finish(
|
|
job_name, "completed", processed=1, total=1,
|
|
message=f"Evaluated {summary['evaluated']}, pending {summary['still_pending']}, "
|
|
f"{closed_trades} paper trade(s) closed",
|
|
)
|
|
_log_event(logging.INFO, "job_complete", job=job_name, summary=summary)
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Alerts Dispatcher
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def dispatch_alerts_job() -> None:
|
|
"""Push Telegram alerts for qualified setups, S/R proximity, score drops, digest."""
|
|
job_name = "alerts"
|
|
_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
|
|
|
|
result = await dispatch_alerts(db)
|
|
|
|
_runtime_progress(job_name, processed=1, total=1)
|
|
_runtime_finish(
|
|
job_name, "completed", processed=1, total=1,
|
|
message=f"{result.get('status')}, sent {result.get('sent', 0)}",
|
|
)
|
|
_log_event(logging.INFO, "job_complete", job=job_name, result=result)
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Market Regime
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def compute_market_regime() -> None:
|
|
"""Refresh the cached benchmark (SPY) trend regime."""
|
|
job_name = "market_regime"
|
|
_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
|
|
|
|
regime = await update_market_regime(db)
|
|
|
|
_runtime_progress(job_name, processed=1, total=1)
|
|
_runtime_finish(
|
|
job_name, "completed", processed=1, total=1,
|
|
message=f"Regime: {regime.get('label')}",
|
|
)
|
|
_log_event(logging.INFO, "job_complete", job=job_name, label=regime.get("label"))
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Regime Monitor
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def compute_regime_monitor() -> None:
|
|
"""Refresh the standalone AI/Tech regime-change index (observational only).
|
|
|
|
Pulls sector/benchmark prices via Alpaca + VIX/credit spreads via FRED,
|
|
computes the 0-100 index, and persists a daily snapshot. Output feeds nothing
|
|
else — it only powers its own tab. Pipeline membership is scheduling only.
|
|
"""
|
|
job_name = "regime_monitor"
|
|
_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
|
|
|
|
result = await update_regime_monitor(db)
|
|
|
|
_runtime_progress(job_name, processed=1, total=1)
|
|
_runtime_finish(
|
|
job_name, "completed", processed=1, total=1,
|
|
message=f"Index: {result.get('total_score')} ({result.get('band')})",
|
|
)
|
|
_log_event(logging.INFO, "job_complete", job=job_name, score=result.get("total_score"))
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Backtest
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def run_backtest_job() -> None:
|
|
"""Replay the price-derived engine over history and cache the report."""
|
|
job_name = "backtest"
|
|
_log_event(logging.INFO, "job_start", job=job_name)
|
|
_runtime_start(job_name)
|
|
|
|
def _on_progress(done: int, count: int, symbol: str) -> None:
|
|
_runtime_progress(job_name, processed=done, total=count, current_ticker=symbol or None)
|
|
|
|
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=0, message="Disabled")
|
|
return
|
|
|
|
report = await run_backtest_and_store(db, _on_progress)
|
|
|
|
_runtime_finish(
|
|
job_name, "completed",
|
|
processed=report.get("tickers", 0), total=report.get("tickers", 0),
|
|
message=f"{report.get('candidates', 0)} setups, {report.get('qualified', 0)} qualified",
|
|
)
|
|
_log_event(logging.INFO, "job_complete", job=job_name, candidates=report.get("candidates"))
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=0, total=None, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Ticker Universe Sync
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def sync_ticker_universe() -> None:
|
|
"""Sync tracked tickers from configured default universe.
|
|
|
|
Setting key: ticker_universe_default (sp500 | nasdaq100 | nasdaq_all)
|
|
"""
|
|
job_name = "ticker_universe_sync"
|
|
_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
|
|
|
|
universe = (await settings_store.get_value(db, "ticker_universe_default", "sp500")).strip().lower()
|
|
|
|
async with async_session_factory() as db:
|
|
summary = await bootstrap_universe(db, universe, prune_missing=False)
|
|
_runtime_progress(job_name, processed=1, total=1)
|
|
_runtime_finish(job_name, "completed", processed=1, total=1, message=f"Synced {universe}")
|
|
_log_event(logging.INFO, "job_complete", job=job_name, universe=universe, summary=summary)
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Job: Daily Pipeline (orchestrator)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Steps run in dependency order: each uses fresh output from the previous one.
|
|
# (name, coroutine) — the names match the individual jobs so each step still
|
|
# updates its own runtime status while the pipeline runs.
|
|
#
|
|
# Daily (full): the complete data→signal refresh, once a day.
|
|
_DAILY_PIPELINE_STEPS = [
|
|
("data_collector", "collect_ohlcv"),
|
|
("sentiment_collector", "collect_sentiment"),
|
|
("rr_scanner", "scan_rr"),
|
|
("outcome_evaluator", "evaluate_outcomes"),
|
|
("market_regime", "compute_market_regime"),
|
|
# Observational only — runs here for scheduling; its output feeds nothing else.
|
|
("regime_monitor", "compute_regime_monitor"),
|
|
]
|
|
|
|
# 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"),
|
|
]
|
|
|
|
|
|
async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
|
|
"""Run an ordered list of (step_name, coroutine_name) steps.
|
|
|
|
Each step respects its own enable flag and manages its own runtime status; a
|
|
failing step is logged and the pipeline continues with the next one.
|
|
"""
|
|
_log_event(logging.INFO, "job_start", job=job_name)
|
|
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=0, message="Disabled")
|
|
return
|
|
|
|
total = len(steps)
|
|
_runtime_start(job_name, total=total)
|
|
|
|
funcs = globals()
|
|
done = 0
|
|
try:
|
|
for step_name, func_name in steps:
|
|
_runtime_progress(job_name, processed=done, total=total, current_ticker=step_name)
|
|
try:
|
|
await funcs[func_name]()
|
|
except Exception:
|
|
logger.exception("%s step %s failed", job_name, step_name)
|
|
done += 1
|
|
_runtime_finish(job_name, "completed", processed=done, total=total, message="Pipeline complete")
|
|
_log_event(logging.INFO, "job_complete", job=job_name)
|
|
except Exception as exc:
|
|
_runtime_finish(job_name, "error", processed=done, total=total, message=str(exc))
|
|
_log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
|
|
|
|
|
|
async def run_daily_pipeline() -> None:
|
|
"""Full daily flow: OHLCV → sentiment → R:R scan → outcome eval (+paper
|
|
close) → market regime."""
|
|
await _run_pipeline("daily_pipeline", _DAILY_PIPELINE_STEPS)
|
|
|
|
|
|
async def run_intraday_pipeline() -> None:
|
|
"""Light intraday flow: refresh OHLCV → evaluate outcomes (+paper close)."""
|
|
await _run_pipeline("intraday_pipeline", _INTRADAY_PIPELINE_STEPS)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Frequency helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_FREQUENCY_MAP: dict[str, dict[str, int]] = {
|
|
"hourly": {"hours": 1},
|
|
"daily": {"hours": 24},
|
|
"weekly": {"weeks": 1},
|
|
}
|
|
|
|
|
|
def _parse_frequency(freq: str) -> dict[str, int]:
|
|
"""Convert a frequency string to APScheduler interval kwargs."""
|
|
return _FREQUENCY_MAP.get(freq.lower(), {"hours": 24})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schedule config (cron, admin-configurable)
|
|
# ---------------------------------------------------------------------------
|
|
#
|
|
# The cron-driven jobs read their schedule from SystemSettings so it can be
|
|
# tuned from Admin → Jobs without a redeploy. A wall-clock CronTrigger also fixes
|
|
# the interval-trigger pitfall: an interval job resets its countdown to now+N on
|
|
# every process restart, so on a box that's redeployed often it can keep being
|
|
# deferred and never fire. Cron fires at a fixed local time regardless.
|
|
|
|
SCHEDULE_DEFAULTS: dict[str, str] = {
|
|
"schedule_timezone": "Europe/Berlin",
|
|
"schedule_daily_pipeline_cron": "0 7 * * *", # full refresh, ready by ~8am
|
|
"schedule_intraday_pipeline_cron": "0 14-22 * * 1-5", # hourly across the US session
|
|
"schedule_fundamentals_cron": "0 4 * * 1", # weekly, early Monday (slow job)
|
|
}
|
|
|
|
# job id -> schedule setting key
|
|
_CRON_JOBS: dict[str, str] = {
|
|
"daily_pipeline": "schedule_daily_pipeline_cron",
|
|
"intraday_pipeline": "schedule_intraday_pipeline_cron",
|
|
"fundamental_collector": "schedule_fundamentals_cron",
|
|
}
|
|
|
|
|
|
def validate_cron(expr: str, timezone: str) -> None:
|
|
"""Raise ValueError if the cron expression or timezone is invalid."""
|
|
CronTrigger.from_crontab((expr or "").strip(), timezone=(timezone or "").strip())
|
|
|
|
|
|
def _cron_trigger(expr: str, timezone: str, fallback_key: str) -> CronTrigger:
|
|
"""Build a CronTrigger, falling back to the default (UTC) on a bad value."""
|
|
try:
|
|
return CronTrigger.from_crontab(expr.strip(), timezone=timezone.strip())
|
|
except Exception:
|
|
_log_event(logging.WARNING, "invalid_cron", expr=expr, timezone=timezone, fallback=SCHEDULE_DEFAULTS[fallback_key])
|
|
return CronTrigger.from_crontab(SCHEDULE_DEFAULTS[fallback_key], timezone="UTC")
|
|
|
|
|
|
async def load_schedule_config(db: AsyncSession) -> dict[str, str]:
|
|
"""Read the cron schedule config from SystemSettings, defaults for any unset."""
|
|
stored = await settings_store.get_map(db, SCHEDULE_DEFAULTS)
|
|
return {key: (stored.get(key) or default) for key, default in SCHEDULE_DEFAULTS.items()}
|
|
|
|
|
|
def reschedule_jobs(schedule_config: dict[str, str]) -> dict[str, str]:
|
|
"""Re-apply cron triggers to the running scheduler after a settings change."""
|
|
tz = schedule_config.get("schedule_timezone") or SCHEDULE_DEFAULTS["schedule_timezone"]
|
|
applied: dict[str, str] = {}
|
|
for job_id, key in _CRON_JOBS.items():
|
|
if scheduler.get_job(job_id) is None:
|
|
continue
|
|
expr = schedule_config.get(key) or SCHEDULE_DEFAULTS[key]
|
|
scheduler.reschedule_job(job_id, trigger=_cron_trigger(expr, tz, key))
|
|
applied[job_id] = expr
|
|
_log_event(logging.INFO, "jobs_rescheduled", applied=applied, timezone=tz)
|
|
return applied
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scheduler setup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
|
|
"""Add all jobs to the scheduler.
|
|
|
|
Call this once before scheduler.start(). Removes any existing jobs first to
|
|
ensure idempotency. ``schedule_config`` supplies the cron strings + timezone
|
|
for the cron-driven jobs (daily/intraday pipelines, fundamentals); defaults
|
|
are used for anything missing.
|
|
"""
|
|
cfg = {**SCHEDULE_DEFAULTS, **(schedule_config or {})}
|
|
tz = cfg["schedule_timezone"]
|
|
scheduler.remove_all_jobs()
|
|
|
|
# Pipeline members: registered but PAUSED (next_run_time=None) so they never
|
|
# auto-fire on their own timer — the pipelines drive them in order. The long
|
|
# interval is just a backstop after a manual trigger (which re-arms an
|
|
# interval job). They stay manually triggerable from Admin → Jobs.
|
|
_members = [
|
|
(collect_ohlcv, "data_collector", "Data Collector (OHLCV)"),
|
|
(collect_sentiment, "sentiment_collector", "Sentiment Collector"),
|
|
(scan_rr, "rr_scanner", "R:R Scanner"),
|
|
(evaluate_outcomes, "outcome_evaluator", "Outcome Evaluator"),
|
|
(compute_market_regime, "market_regime", "Market Regime"),
|
|
(compute_regime_monitor, "regime_monitor", "Regime Monitor"),
|
|
]
|
|
for fn, job_id, job_name in _members:
|
|
scheduler.add_job(
|
|
fn, "interval", weeks=520, id=job_id, name=job_name,
|
|
replace_existing=True, next_run_time=None,
|
|
)
|
|
|
|
# Cron-driven jobs (admin-configurable times)
|
|
scheduler.add_job(
|
|
run_daily_pipeline,
|
|
_cron_trigger(cfg["schedule_daily_pipeline_cron"], tz, "schedule_daily_pipeline_cron"),
|
|
id="daily_pipeline", name="Daily Pipeline", replace_existing=True,
|
|
)
|
|
scheduler.add_job(
|
|
run_intraday_pipeline,
|
|
_cron_trigger(cfg["schedule_intraday_pipeline_cron"], tz, "schedule_intraday_pipeline_cron"),
|
|
id="intraday_pipeline", name="Intraday Pipeline", replace_existing=True,
|
|
)
|
|
# Fundamentals — quarterly-ish data; weekly by default (conserves API quota).
|
|
# Its own early cron so the slow, rate-limited fetch finishes before the day.
|
|
scheduler.add_job(
|
|
collect_fundamentals,
|
|
_cron_trigger(cfg["schedule_fundamentals_cron"], tz, "schedule_fundamentals_cron"),
|
|
id="fundamental_collector", name="Fundamental Collector", replace_existing=True,
|
|
)
|
|
|
|
# Independent interval jobs (own cadence, no ordering dependency)
|
|
scheduler.add_job(
|
|
sync_ticker_universe, "interval", hours=24,
|
|
id="ticker_universe_sync", name="Ticker Universe Sync", replace_existing=True,
|
|
)
|
|
alerts_interval = _parse_frequency(settings.alerts_frequency)
|
|
scheduler.add_job(
|
|
dispatch_alerts_job, "interval", **alerts_interval,
|
|
id="alerts", name="Alerts Dispatcher", replace_existing=True,
|
|
)
|
|
scheduler.add_job(
|
|
run_backtest_job, "interval", hours=168,
|
|
id="backtest", name="Backtest", replace_existing=True,
|
|
)
|
|
# Deep history backfill: manual only (never auto-fires); triggered from
|
|
# Admin → Jobs when histories need deepening.
|
|
scheduler.add_job(
|
|
backfill_ohlcv, "interval", weeks=520,
|
|
id="data_backfill", name="Data Backfill (deep history)",
|
|
replace_existing=True, next_run_time=None,
|
|
)
|
|
|
|
_log_event(logging.INFO, "scheduler_configured", timezone=tz, daily_pipeline={
|
|
"cron": cfg["schedule_daily_pipeline_cron"],
|
|
"steps": [name for name, _ in _DAILY_PIPELINE_STEPS],
|
|
}, intraday_pipeline={
|
|
"cron": cfg["schedule_intraday_pipeline_cron"],
|
|
"steps": [name for name, _ in _INTRADAY_PIPELINE_STEPS],
|
|
}, fundamental_collector={"cron": cfg["schedule_fundamentals_cron"]}, independent=["ticker_universe_sync", "alerts", "backtest"])
|