1894 lines
79 KiB
Python
1894 lines
79 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 and_, 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,
|
||
pipeline_run,
|
||
sentiment_service,
|
||
settings_store,
|
||
shadow_book_service,
|
||
fundamentals_parity_service,
|
||
fundamental_data_refresh_service,
|
||
)
|
||
from app.services.data_import import (
|
||
STATUS_DEFERRED,
|
||
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,
|
||
DEFAULT_BACKTEST_CADENCE,
|
||
PRODUCTION_GTL_TARGET_MODEL,
|
||
run_and_store as run_backtest_and_store,
|
||
validate_backtest_cadence,
|
||
validate_backtest_target_model,
|
||
)
|
||
from app.services.benchmark_service import refresh_benchmark_prices
|
||
from app.services.market_regime_service import update_market_regime
|
||
from app.services.regime_monitor_service import update_regime_monitor
|
||
from app.services.event_study_service import run_and_store as run_event_study_and_store
|
||
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",
|
||
"dolt_earnings_import",
|
||
"sec_fundamentals_import",
|
||
"fundamentals_parity_report",
|
||
"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",
|
||
]
|
||
|
||
|
||
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}
|
||
_next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
|
||
_next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def queue_backtest_options(
|
||
target_model: str | None,
|
||
cadence: str | None,
|
||
) -> tuple[str, str]:
|
||
"""Select model and cadence for the next manual backtest run only.
|
||
|
||
Scheduled and subsequent manual runs return to production GTL at the
|
||
resource-safe weekly cadence.
|
||
"""
|
||
global _next_backtest_target_model, _next_backtest_cadence
|
||
selected_model = validate_backtest_target_model(
|
||
target_model or PRODUCTION_GTL_TARGET_MODEL
|
||
)
|
||
selected_cadence = validate_backtest_cadence(
|
||
cadence or DEFAULT_BACKTEST_CADENCE
|
||
)
|
||
_next_backtest_target_model = selected_model
|
||
_next_backtest_cadence = selected_cadence
|
||
return selected_model, selected_cadence
|
||
|
||
|
||
def queue_backtest_target_model(target_model: str | None) -> str:
|
||
"""Compatibility wrapper for callers selecting only the target model."""
|
||
selected, _ = queue_backtest_options(target_model, DEFAULT_BACKTEST_CADENCE)
|
||
return selected
|
||
|
||
|
||
def _consume_backtest_options() -> tuple[str, str]:
|
||
global _next_backtest_target_model, _next_backtest_cadence
|
||
selected = (_next_backtest_target_model, _next_backtest_cadence)
|
||
_next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
|
||
_next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
|
||
return selected
|
||
|
||
|
||
def _consume_backtest_target_model() -> str:
|
||
"""Compatibility wrapper consuming all queued one-run options."""
|
||
selected, _ = _consume_backtest_options()
|
||
return selected
|
||
|
||
|
||
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),
|
||
)
|
||
|
||
|
||
async def _record_system_event(
|
||
*,
|
||
severity: str,
|
||
source: str,
|
||
code: str,
|
||
message: str,
|
||
symbol: str | None = None,
|
||
dedup_key: str | None = None,
|
||
) -> None:
|
||
"""Best-effort durable event for Admin → Jobs and the top-nav badge."""
|
||
from app.services.system_event_service import log_event_standalone
|
||
|
||
await log_event_standalone(
|
||
severity=severity,
|
||
source=source,
|
||
code=code,
|
||
message=message,
|
||
symbol=symbol,
|
||
dedup_key=dedup_key,
|
||
)
|
||
|
||
|
||
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
|
||
# Durable event for error / rate-limit finishes (badge + Admin → Jobs panel).
|
||
if status in ("error", "rate_limited"):
|
||
severity = "error" if status == "error" else "warning"
|
||
try:
|
||
loop = asyncio.get_running_loop()
|
||
loop.create_task(
|
||
_record_system_event(
|
||
severity=severity,
|
||
source=job_name,
|
||
code=f"job_{status}",
|
||
message=message or f"Job {job_name} finished with status {status}",
|
||
dedup_key=f"job:{job_name}:{status}:{(message or '')[:80]}"[:200],
|
||
)
|
||
)
|
||
except RuntimeError:
|
||
pass
|
||
|
||
|
||
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_top_pick_feeder_ids(db: AsyncSession) -> set[int]:
|
||
"""Ticker ids whose latest LONG setup makes them a top-pick feeder.
|
||
|
||
A dashboard 'top pick' is the highest residual-momentum *qualified* setup.
|
||
Sentiment can never move a ticker's activation percentile (the gate's core
|
||
axis) — only its confidence and EV ranking. So the only tickers that are, or
|
||
could become with positive sentiment, a top pick are residual-momentum leaders
|
||
that already have a tradeable long setup clearing the R:R floor. That set is exactly:
|
||
|
||
latest long setup with momentum_percentile >= gate AND rr_ratio >= floor.
|
||
|
||
It contains both the currently-qualified setups and the near-miss ones held
|
||
back only by a neutral/missing sentiment — the cases the user saw surface as
|
||
top picks with no sentiment. Only meaningful with the momentum gate on
|
||
(min_momentum_percentile > 0); off, there is no leader axis to anchor on and we
|
||
defer to the filler set. Best-effort: a config failure must not stop collection.
|
||
"""
|
||
from app.models.trade_setup import TradeSetup
|
||
|
||
try:
|
||
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))
|
||
min_rr = float(activation.get("min_rr", 0.0))
|
||
except Exception:
|
||
logger.exception("Sentiment top-pick scoping failed; using filler set only")
|
||
return set()
|
||
|
||
if min_pct <= 0:
|
||
return set()
|
||
|
||
# Latest long setup per ticker, then keep those clearing the gate's momentum
|
||
# percentile and R:R floor. (Sentiment runs before the day's scan, so this
|
||
# reads the previous scan's setups — momentum is a slow, cross-sectional signal,
|
||
# so yesterday's leaders are the right anchor.)
|
||
latest_long = (
|
||
select(TradeSetup.ticker_id, func.max(TradeSetup.detected_at).label("md"))
|
||
.where(TradeSetup.direction == "long")
|
||
.group_by(TradeSetup.ticker_id)
|
||
.subquery()
|
||
)
|
||
rows = await db.execute(
|
||
select(TradeSetup.ticker_id)
|
||
.join(
|
||
latest_long,
|
||
and_(
|
||
TradeSetup.ticker_id == latest_long.c.ticker_id,
|
||
TradeSetup.detected_at == latest_long.c.md,
|
||
),
|
||
)
|
||
.where(
|
||
TradeSetup.direction == "long",
|
||
TradeSetup.rr_ratio >= min_rr,
|
||
TradeSetup.momentum_percentile.is_not(None),
|
||
TradeSetup.momentum_percentile >= min_pct,
|
||
)
|
||
)
|
||
return {r[0] for r in rows.all()}
|
||
|
||
|
||
async def _stale_sentiment_symbols(
|
||
db: AsyncSession, ticker_ids: set[int], cutoff: datetime
|
||
) -> list[str]:
|
||
"""Symbols among ``ticker_ids`` whose newest sentiment is missing or older than
|
||
``cutoff``, ordered missing-first → oldest → alphabetical."""
|
||
if not ticker_ids:
|
||
return []
|
||
latest_ts = func.max(SentimentScore.timestamp)
|
||
missing_first = case((latest_ts.is_(None), 0), else_=1)
|
||
stmt = (
|
||
select(Ticker.symbol)
|
||
.outerjoin(SentimentScore, SentimentScore.ticker_id == Ticker.id)
|
||
.where(Ticker.id.in_(ticker_ids))
|
||
.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())
|
||
)
|
||
result = await db.execute(stmt)
|
||
return list(result.scalars().all())
|
||
|
||
|
||
async def _get_sentiment_priority_tickers(db: AsyncSession) -> list[str]:
|
||
"""Symbols to fetch sentiment for, skipping anything refreshed within
|
||
``sentiment_fresh_hours``.
|
||
|
||
No per-run cap: the relevant set is naturally bounded (curated watchlist <= 20,
|
||
a handful of open trades and top-pick feeders, top-N composite), so refreshing
|
||
all of it stays well inside the free search tier — and everything that matters
|
||
is always fully covered. The two tiers only affect ORDER, so a mid-run provider
|
||
rate limit still lands the names we care about first:
|
||
|
||
Priority: top-pick feeders (residual-momentum leaders with a tradeable long setup, see
|
||
``_get_top_pick_feeder_ids``) + the curated watchlist + open paper trades —
|
||
the set we never want shown without sentiment.
|
||
Filler: top-N by composite — a cheap discovery net for names not yet covered.
|
||
|
||
Once the 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
|
||
|
||
cutoff = datetime.now(timezone.utc) - timedelta(hours=settings.sentiment_fresh_hours)
|
||
|
||
# Priority: the set we always want fresh — top-pick feeders, the curated
|
||
# watchlist, and open positions.
|
||
priority_ids = await _get_top_pick_feeder_ids(db)
|
||
wl = await db.execute(
|
||
select(WatchlistEntry.ticker_id)
|
||
.where(WatchlistEntry.entry_type != "dismissed")
|
||
.distinct()
|
||
)
|
||
priority_ids.update(r[0] for r in wl.all())
|
||
pt = await db.execute(
|
||
select(PaperTrade.ticker_id).where(PaperTrade.status == "open").distinct()
|
||
)
|
||
priority_ids.update(r[0] for r in pt.all())
|
||
|
||
# Filler: top-N by composite, a discovery net for names not already covered.
|
||
top = await db.execute(
|
||
select(CompositeScore.ticker_id)
|
||
.order_by(CompositeScore.score.desc())
|
||
.limit(settings.sentiment_top_composite)
|
||
)
|
||
filler_ids = {r[0] for r in top.all()} - priority_ids
|
||
|
||
if not priority_ids and not filler_ids:
|
||
return []
|
||
|
||
# No cap — fetch every stale name. Priority first so a rate limit mid-run still
|
||
# covers the curated/at-risk set before the discovery net.
|
||
priority_syms = await _stale_sentiment_symbols(db, priority_ids, cutoff)
|
||
filler_syms = await _stale_sentiment_symbols(db, filler_ids, cutoff)
|
||
return priority_syms + filler_syms
|
||
|
||
|
||
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",
|
||
*,
|
||
refetch_days: int = 0,
|
||
) -> 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: overlap last_ingested_date so partial bars refresh
|
||
- 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.
|
||
|
||
``refetch_days`` re-pulls the last N days regardless of ingestion progress —
|
||
the after-close run uses it to overwrite the day's partial intraday bar, which
|
||
resume logic would otherwise skip as "already up to date".
|
||
"""
|
||
_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()
|
||
# An explicit start_date makes fetch_and_ingest re-pull that window instead
|
||
# of resuming from the last stored bar (upsert overwrites, so this is safe).
|
||
if full_backfill:
|
||
backfill_start = end_date - timedelta(days=settings.ohlcv_history_days)
|
||
elif refetch_days:
|
||
backfill_start = end_date - timedelta(days=refetch_days)
|
||
else:
|
||
backfill_start = 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 == "stale":
|
||
await _record_system_event(
|
||
severity="warning",
|
||
source=job_name,
|
||
code="ohlcv_stale",
|
||
message=result.message or f"No new OHLCV bars for {symbol}",
|
||
symbol=symbol,
|
||
dedup_key=f"ohlcv_stale:{symbol}",
|
||
)
|
||
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)
|
||
await _record_system_event(
|
||
severity="error",
|
||
source=job_name,
|
||
code="job_ticker_error",
|
||
message=f"{type(exc).__name__}: {exc}",
|
||
symbol=symbol,
|
||
dedup_key=f"job_ticker_error:{job_name}:{symbol}:{type(exc).__name__}",
|
||
)
|
||
|
||
# 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")
|
||
|
||
|
||
async def run_shadow_book() -> None:
|
||
"""Open the strategy's own positions from the latest qualifying scan.
|
||
|
||
The shadow book is the faithful live twin of the backtest: top-ranked
|
||
qualified setups, up to capacity, 1% risk, no human input. It runs straight
|
||
after the near-close scan so its entries are marked at the same near-close
|
||
prices the discretionary book sees, leaving *selection* as the only
|
||
difference between the two books.
|
||
|
||
When run as a pipeline step it acts only on the scan that stamped *this
|
||
pipeline's* run id (``expected_run_id``): if the pipeline's own scan was
|
||
disabled or failed, the stored run id is some other scan's — including a
|
||
manual scan that overlapped and finished last — and shadow refuses.
|
||
Triggered directly from Admin (no pipeline context) it falls back to the
|
||
scan-freshness window — an explicit operator action.
|
||
|
||
Opt-in (``shadow_book_enabled``) because it writes live trades.
|
||
"""
|
||
job_name = "shadow_book"
|
||
expected_run_id = pipeline_run.current()
|
||
_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 False
|
||
if not await shadow_book_service.is_enabled(db):
|
||
_log_event(logging.INFO, "job_skipped", job=job_name, reason="not enabled in settings")
|
||
_runtime_finish(job_name, "skipped", processed=0, total=1, message="Not enabled")
|
||
return
|
||
|
||
from app.services.admin_service import get_activation_config
|
||
|
||
activation_config = await get_activation_config(db)
|
||
summary = await shadow_book_service.open_shadow_positions(
|
||
db,
|
||
activation_config=activation_config,
|
||
expected_run_id=expected_run_id,
|
||
)
|
||
symbols = await shadow_book_service.symbols_for(db, summary["symbols"])
|
||
|
||
_runtime_progress(job_name, processed=1, total=1)
|
||
_runtime_finish(
|
||
job_name, "completed", processed=1, total=1,
|
||
message=(
|
||
f"Opened {summary['opened']} ({', '.join(symbols) if symbols else 'none'}); "
|
||
f"held {summary['skipped_held']}, gate-locked {summary['skipped_locked']}"
|
||
),
|
||
)
|
||
_log_event(logging.INFO, "job_complete", job=job_name, opened=summary["opened"], symbols=symbols)
|
||
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))
|
||
|
||
|
||
async def collect_ohlcv_final() -> None:
|
||
"""After-close OHLCV refresh that replaces the day's partial bar.
|
||
|
||
Intraday runs store today's bar while the session is still open, so ingestion
|
||
progress already reads "today" and incremental resume would skip the day
|
||
entirely — leaving a partial bar as the permanent record. ``refetch_days``
|
||
forces the last few sessions to be re-pulled so outcome evaluation and
|
||
fill-quality checks grade against the real close.
|
||
"""
|
||
await collect_ohlcv(refetch_days=_FINAL_REFETCH_DAYS)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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
|
||
if await fundamental_data_refresh_service.is_enabled(db):
|
||
message = "SEC + Dolt fundamentals cutover is active"
|
||
_log_event(
|
||
logging.INFO,
|
||
"job_skipped",
|
||
job=job_name,
|
||
reason="sec_dolt_cutover_active",
|
||
)
|
||
_runtime_finish(
|
||
job_name,
|
||
"skipped",
|
||
processed=0,
|
||
total=0,
|
||
message=message,
|
||
)
|
||
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))
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Jobs: shadow fundamentals sources
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool:
|
||
"""Run an importer and return whether its scheduled job was enabled.
|
||
|
||
The SEC wrapper uses the return value to run its activated local cache step
|
||
after deferred, failed, no-op, promoted, or source-locked attempts while honoring
|
||
the job-level disable switch.
|
||
"""
|
||
_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 True
|
||
|
||
revision = f" · {run.revision[:12]}" if run.revision else ""
|
||
message = f"{run.status}{revision}"
|
||
if run.status == STATUS_DEFERRED:
|
||
message = run.error_details or message
|
||
_log_event(logging.INFO, "job_deferred", job=job_name, message=message)
|
||
_runtime_finish(job_name, "deferred", processed=0, total=1, message=message)
|
||
return True
|
||
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 True
|
||
|
||
_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)
|
||
return True
|
||
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))
|
||
return True
|
||
|
||
|
||
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 SEC facts, then run the activated local compat-cache refresh.
|
||
|
||
The refresh is deliberately separate from the network import result. Once
|
||
activated it therefore still runs from stored snapshots/earnings/prices when
|
||
SEC is unavailable, unchanged, or another SEC import owns the source lock.
|
||
"""
|
||
job_name = "sec_fundamentals_import"
|
||
job_enabled = await _run_shadow_import(job_name, SecFundamentalsImporter())
|
||
if not job_enabled:
|
||
return
|
||
|
||
try:
|
||
async with async_session_factory() as db:
|
||
summary = await fundamental_data_refresh_service.refresh_if_enabled(db)
|
||
except asyncio.CancelledError:
|
||
_runtime_finish(
|
||
job_name, "error", processed=0, total=1, message="Cancelled"
|
||
)
|
||
raise
|
||
except Exception as exc:
|
||
message = f"Local fundamental_data refresh failed: {exc}"
|
||
_log_event(
|
||
logging.ERROR,
|
||
"fundamental_data_refresh_error",
|
||
job=job_name,
|
||
error_type=type(exc).__name__,
|
||
message=str(exc),
|
||
)
|
||
_runtime_finish(job_name, "error", processed=0, total=1, message=message)
|
||
return
|
||
|
||
if not summary["enabled"]:
|
||
_log_event(
|
||
logging.INFO,
|
||
"fundamental_data_refresh_skipped",
|
||
job=job_name,
|
||
reason="cutover_disabled",
|
||
setting=fundamental_data_refresh_service.ACTIVATION_KEY,
|
||
)
|
||
return
|
||
|
||
_log_event(
|
||
logging.INFO,
|
||
"fundamental_data_refresh_complete",
|
||
job=job_name,
|
||
**summary,
|
||
)
|
||
runtime = get_job_runtime_snapshot(job_name)
|
||
if runtime.get("status") == "completed":
|
||
import_message = runtime.get("message") or "import completed"
|
||
cache_message = (
|
||
f"cache {summary['refreshed']} · "
|
||
f"{summary['score_inputs_changed']} score inputs changed"
|
||
)
|
||
_runtime_finish(
|
||
job_name,
|
||
"completed",
|
||
processed=1,
|
||
total=1,
|
||
message=f"{import_message} · {cache_message}",
|
||
)
|
||
|
||
|
||
async def run_fundamentals_parity_report() -> None:
|
||
"""Generate the A5 comparison bundle without mutating live fundamentals/scores."""
|
||
job_name = "fundamentals_parity_report"
|
||
_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):
|
||
_runtime_finish(
|
||
job_name, "skipped", processed=0, total=1, message="Disabled"
|
||
)
|
||
return
|
||
report, artifacts = await fundamentals_parity_service.generate_and_store(
|
||
db, settings.fundamentals_parity_report_dir
|
||
)
|
||
summary = report["summary"]
|
||
message = (
|
||
f"{summary['universe_count']} tickers · "
|
||
f"{summary['fundamental_score_material_changes']} material score changes"
|
||
)
|
||
_runtime_finish(job_name, "completed", processed=1, total=1, message=message)
|
||
_log_event(
|
||
logging.INFO,
|
||
"job_complete",
|
||
job=job_name,
|
||
generated_at=report["generated_at"],
|
||
json_path=artifacts["json"],
|
||
csv_path=artifacts["csv"],
|
||
)
|
||
except asyncio.CancelledError:
|
||
_runtime_finish(job_name, "error", processed=0, total=1, message="Cancelled")
|
||
raise
|
||
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: 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: Benchmark Collector (SPY closes for paper-trade alpha)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def collect_benchmark() -> None:
|
||
"""Refresh the stored benchmark (SPY) daily closes used for paper-trade alpha."""
|
||
job_name = "benchmark_collector"
|
||
_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
|
||
|
||
written = await refresh_benchmark_prices(db)
|
||
|
||
_runtime_progress(job_name, processed=1, total=1)
|
||
_runtime_finish(job_name, "completed", processed=1, total=1, message=f"{written} rows")
|
||
_log_event(logging.INFO, "job_complete", job=job_name, rows=written)
|
||
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)
|
||
|
||
state = result.get("state") or {}
|
||
warning = result.get("warning") or {}
|
||
_runtime_progress(job_name, processed=1, total=1)
|
||
_runtime_finish(
|
||
job_name, "completed", processed=1, total=1,
|
||
message=f"State: {state.get('score')} · Warning: {warning.get('score')}",
|
||
)
|
||
_log_event(
|
||
logging.INFO,
|
||
"job_complete",
|
||
job=job_name,
|
||
state=state.get("score"),
|
||
warning=warning.get("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"
|
||
target_model, cadence = _consume_backtest_options()
|
||
_log_event(
|
||
logging.INFO,
|
||
"job_start",
|
||
job=job_name,
|
||
target_model=target_model,
|
||
cadence=cadence,
|
||
)
|
||
_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,
|
||
target_model=target_model,
|
||
cadence=cadence,
|
||
)
|
||
|
||
_runtime_finish(
|
||
job_name, "completed",
|
||
processed=report.get("tickers", 0), total=report.get("tickers", 0),
|
||
message=(
|
||
f"{BACKTEST_TARGET_MODELS[target_model]}: "
|
||
f"{cadence} cadence, "
|
||
f"{report.get('candidates', 0)} setups, "
|
||
f"{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: Event Study (manual)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def run_event_study_job() -> None:
|
||
"""Measure indicator lead time vs. historical drawdowns and cache the report.
|
||
|
||
Manual only (never auto-fires) — it does a universe-wide OHLCV scan. Triggered
|
||
from Admin → Jobs when you want to re-run the early-warning measurement.
|
||
"""
|
||
job_name = "event_study"
|
||
_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
|
||
|
||
report = await run_event_study_and_store(db)
|
||
|
||
_runtime_progress(job_name, processed=1, total=1)
|
||
if report.get("available"):
|
||
metrics = report.get("metrics") or {}
|
||
msg = (
|
||
f"{metrics.get('events_warned', 0)}/{metrics.get('events', 0)} warned, "
|
||
f"{metrics.get('false_alarms_per_year', 0)} false alarms/year"
|
||
)
|
||
else:
|
||
msg = report.get("reason", "no data")
|
||
_runtime_finish(job_name, "completed", processed=1, total=1, message=msg)
|
||
_log_event(logging.INFO, "job_complete", job=job_name, events=len(report.get("events", [])))
|
||
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: 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.
|
||
# Morning (America/New_York ~02:00): refresh data + display context. No R:R scan
|
||
# — the qualifying full-universe scan runs once near the US close so post-stop
|
||
# gate-reset sees one observation per trading day (plus the trade_policy
|
||
# distinct-day guard for manual re-scans).
|
||
# Sessions re-pulled by the after-close fetch so the consolidated bar overwrites
|
||
# 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"),
|
||
("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"),
|
||
]
|
||
|
||
# Warn if near-close fetch+scan+alert drifts past this — entries leave the close
|
||
# and the stale_close floor quietly becomes the ceiling.
|
||
_NEAR_CLOSE_DURATION_WARN_SECONDS = 600
|
||
|
||
|
||
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.
|
||
|
||
A unique run id is bound for the invocation and visible to every step via the
|
||
shared task context: the scan step stamps it into its completion markers and
|
||
the shadow step requires an exact match, so only a scan that ran inside this
|
||
pipeline can drive the shadow book.
|
||
"""
|
||
_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
|
||
token = pipeline_run.bind(pipeline_run.new_run_id())
|
||
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))
|
||
finally:
|
||
pipeline_run.release(token)
|
||
|
||
|
||
async def run_daily_pipeline() -> None:
|
||
"""Morning flow: OHLCV → benchmark → sentiment → market regime (no scan)."""
|
||
await _run_pipeline("daily_pipeline", _DAILY_PIPELINE_STEPS)
|
||
|
||
|
||
async def run_near_close_pipeline() -> None:
|
||
"""Near-close flow: OHLCV fetch → R:R scan → Telegram alerts.
|
||
|
||
Logs wall duration; warn if past 10 minutes so operators notice close drift.
|
||
"""
|
||
import time
|
||
|
||
started = time.monotonic()
|
||
await _run_pipeline("near_close_pipeline", _NEAR_CLOSE_PIPELINE_STEPS)
|
||
elapsed = time.monotonic() - started
|
||
payload = {
|
||
"job": "near_close_pipeline",
|
||
"duration_seconds": round(elapsed, 1),
|
||
}
|
||
if elapsed > _NEAR_CLOSE_DURATION_WARN_SECONDS:
|
||
_log_event(
|
||
logging.WARNING,
|
||
"near_close_pipeline_slow",
|
||
**payload,
|
||
threshold_seconds=_NEAR_CLOSE_DURATION_WARN_SECONDS,
|
||
message=(
|
||
"Near-close pipeline exceeded 10 minutes — entries may drift from "
|
||
"the close toward the stale_close research floor"
|
||
),
|
||
)
|
||
else:
|
||
_log_event(logging.INFO, "near_close_pipeline_duration", **payload)
|
||
|
||
|
||
async def run_after_close_pipeline() -> None:
|
||
"""After-close flow: OHLCV fetch (final bar) → outcome eval (+paper close)."""
|
||
await _run_pipeline("after_close_pipeline", _AFTER_CLOSE_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.
|
||
|
||
# All wall times are America/New_York after the near-close execution cutover.
|
||
# Stored SystemSetting values shadow these defaults — deploy migration 023
|
||
# rewrites schedule_* keys so prod does not keep scanning at 07:00 Berlin.
|
||
# DAY-OF-WEEK MUST BE NAMES, NEVER NUMBERS. APScheduler's from_crontab() passes
|
||
# field 5 straight to its own day_of_week, where 0=Monday — so "1-5" resolves to
|
||
# Tue–Sat, silently skipping every Monday and scanning on Saturdays. Names are
|
||
# unambiguous in both dialects.
|
||
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 * * *",
|
||
# Bulk source imports. The SEC job writes the legacy compat cache only after
|
||
# the explicit, default-off A5 cutover setting is enabled.
|
||
"schedule_dolt_earnings_cron": "30 2 * * *",
|
||
"schedule_sec_fundamentals_cron": "0 4 * * *",
|
||
"schedule_fundamentals_parity_cron": "30 5 * * *",
|
||
# 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).
|
||
"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",
|
||
# Weekly fundamentals early Monday NY.
|
||
"schedule_fundamentals_cron": "0 1 * * mon",
|
||
}
|
||
|
||
# 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",
|
||
"fundamentals_parity_report": "schedule_fundamentals_parity_cron",
|
||
"near_close_pipeline": "schedule_near_close_pipeline_cron",
|
||
"after_close_pipeline": "schedule_after_close_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_benchmark, "benchmark_collector", "Benchmark Collector"),
|
||
(collect_sentiment, "sentiment_collector", "Sentiment Collector"),
|
||
(scan_rr, "rr_scanner", "R:R Scanner"),
|
||
(run_shadow_book, "shadow_book", "Shadow Book (auto-traded strategy)"),
|
||
(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="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",
|
||
replace_existing=True,
|
||
)
|
||
scheduler.add_job(
|
||
run_fundamentals_parity_report,
|
||
_cron_trigger(
|
||
cfg["schedule_fundamentals_parity_cron"],
|
||
tz,
|
||
"schedule_fundamentals_parity_cron",
|
||
),
|
||
id="fundamentals_parity_report",
|
||
name="Fundamentals Parity Report (read-only)",
|
||
replace_existing=True,
|
||
)
|
||
scheduler.add_job(
|
||
run_near_close_pipeline,
|
||
_cron_trigger(
|
||
cfg["schedule_near_close_pipeline_cron"],
|
||
tz,
|
||
"schedule_near_close_pipeline_cron",
|
||
),
|
||
id="near_close_pipeline",
|
||
name="Near-Close Pipeline (scan+alert)",
|
||
replace_existing=True,
|
||
)
|
||
scheduler.add_job(
|
||
run_after_close_pipeline,
|
||
_cron_trigger(
|
||
cfg["schedule_after_close_pipeline_cron"],
|
||
tz,
|
||
"schedule_after_close_pipeline_cron",
|
||
),
|
||
id="after_close_pipeline",
|
||
name="After-Close Pipeline (outcome)",
|
||
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 auto-fire only via near_close_pipeline (scan → alert before MOC).
|
||
# Keep the job registered for Admin manual trigger; no independent interval.
|
||
scheduler.add_job(
|
||
dispatch_alerts_job, "interval", weeks=520,
|
||
id="alerts", name="Alerts Dispatcher",
|
||
replace_existing=True, next_run_time=None,
|
||
)
|
||
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,
|
||
)
|
||
# Event study: manual only (universe-wide scan); triggered from Admin → Jobs.
|
||
scheduler.add_job(
|
||
run_event_study_job, "interval", weeks=520,
|
||
id="event_study", name="Event Study",
|
||
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],
|
||
},
|
||
dolt_earnings_import={"cron": cfg["schedule_dolt_earnings_cron"]},
|
||
sec_fundamentals_import={"cron": cfg["schedule_sec_fundamentals_cron"]},
|
||
fundamentals_parity_report={
|
||
"cron": cfg["schedule_fundamentals_parity_cron"]
|
||
},
|
||
near_close_pipeline={
|
||
"cron": cfg["schedule_near_close_pipeline_cron"],
|
||
"steps": [name for name, _ in _NEAR_CLOSE_PIPELINE_STEPS],
|
||
},
|
||
after_close_pipeline={
|
||
"cron": cfg["schedule_after_close_pipeline_cron"],
|
||
"steps": [name for name, _ in _AFTER_CLOSE_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", "backtest"],
|
||
manual_only=["alerts", "data_backfill", "event_study"],
|
||
)
|