dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: decommission FMP, Finnhub and Alpha Vantage (A6)

Browse Source 
The A5 cutover has been on and observed in production, so SEC Company Facts +
DoltHub earnings are already the live source for `fundamental_data`. This
removes everything the legacy path still occupied.

Gone: the three providers and their config/env keys; the weekly
`fundamental_collector` job; the cutover toggle (SEC + Dolt is now the
unconditional path, so `off` can no longer silently freeze scoring inputs); the
A5 parity report, whose deltas became structurally zero once the candidate
builder started writing the table it compared against; and the FMP tier of
universe bootstrap.

Two behavioral notes:

- Disabling **SEC Fundamentals Import** now stops the SEC network fetch only.
  The local cache refresh moved outside the job-enable check, because candidates
  also derive from daily closes and earnings events — freezing those on an
  ingestion pause would stale scoring with no fallback left to recover from.
- `/ingestion/fetch?sources=fundamentals` still accepts the key and reports
  `skipped`; there is no per-ticker fetch any more.

Migration 029 does not blanket-delete the leftover settings rows. Migrations run
before the service restart, and pre-A6 code reads an absent `job_*_enabled` row
as *enabled* — so the two behavior-bearing keys become tombstones pinned to safe
values (hidden in Admin) and only the inert three are deleted. Removing the
provider keys from the production `.env` is the matching rollout step.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
dennisthiessen
2026-08-07 11:19:28 +02:00
co-authored by Claude Opus 5
parent f5d4b516ab
commit 3e83d63b05
51 changed files with 368 additions and 3787 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/scheduler.py
+32 -260
View File
@@ -1,9 +1,9 @@
"""APScheduler job definitions and FastAPI lifespan integration.
Defines four scheduled jobs:
Defines the scheduled jobs, among them:
- Data Collector (OHLCV fetch for all tickers)
- Sentiment Collector (sentiment for all tickers)
- Fundamental Collector (fundamentals for all tickers)
- Dolt Earnings / SEC Fundamentals imports (bulk fundamentals sources)
- R:R Scanner (trade setup scan for all tickers)
Each job processes tickers independently, logs errors as structured JSON,
@@ -25,22 +25,18 @@ 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 (
@@ -93,7 +89,6 @@ _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
@@ -102,10 +97,8 @@ _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",
@@ -466,23 +459,6 @@ async def _get_sentiment_priority_tickers(db: AsyncSession) -> list[str]:
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).
@@ -815,139 +791,6 @@ async def collect_sentiment() -> None:
_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
# ---------------------------------------------------------------------------
@@ -956,9 +799,9 @@ async def collect_fundamentals() -> None:
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.
The SEC wrapper uses the return value only to word its runtime message: its
local cache step runs after deferred, failed, no-op, promoted, source-locked
and disabled attempts alike.
"""
_log_event(logging.INFO, "job_start", job=job_name)
_runtime_start(job_name, total=1)
@@ -968,7 +811,7 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool:
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
return False
run = await run_import(importer)
if run is None:
@@ -1015,25 +858,26 @@ async def _run_shadow_import(job_name: str, importer: SourceImporter) -> bool:
async def run_dolt_earnings_import() -> None:
"""Pull and import the Dolt earnings calendar/results feed in shadow."""
"""Pull and import the Dolt earnings calendar/results feed."""
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.
"""Import SEC facts, then refresh the local compat cache.
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.
The refresh is deliberately independent of the network import: it reads only
stored snapshots, earnings events and closes, so it runs identically when SEC
is unavailable, unchanged, or owned by another import — and also when the
job's ingestion is switched off in Admin → Jobs. Disabling the job stops
SEC network access, not the cache; prices and earnings move daily even when
no filing does, and `fundamental_data` feeds scoring.
"""
job_name = "sec_fundamentals_import"
job_enabled = await _run_shadow_import(job_name, SecFundamentalsImporter())
if not job_enabled:
return
import_ran = await _run_shadow_import(job_name, SecFundamentalsImporter())
try:
async with async_session_factory() as db:
summary = await fundamental_data_refresh_service.refresh_if_enabled(db)
summary = await fundamental_data_refresh_service.refresh(db)
except asyncio.CancelledError:
_runtime_finish(
job_name, "error", processed=0, total=1, message="Cancelled"
@@ -1051,29 +895,27 @@ async def run_sec_fundamentals_import() -> None:
_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,
)
cache_message = (
f"cache {summary['refreshed']} · "
f"{summary['score_inputs_changed']} score inputs changed"
)
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"
if not import_ran:
_runtime_finish(
job_name,
"completed",
processed=1,
total=1,
message=f"Import disabled · {cache_message}",
)
elif runtime.get("status") == "completed":
import_message = runtime.get("message") or "import completed"
_runtime_finish(
job_name,
"completed",
@@ -1083,49 +925,6 @@ async def run_sec_fundamentals_import() -> None:
)
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
# ---------------------------------------------------------------------------
@@ -1666,19 +1465,16 @@ 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.
# Bulk source imports. The SEC job also refreshes the fundamental_data compat
# cache that scoring reads — locally, from stored snapshots/earnings/closes.
"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
@@ -1686,11 +1482,9 @@ _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",
}
@@ -1779,7 +1573,7 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
"schedule_dolt_earnings_cron",
),
id="dolt_earnings_import",
name="Dolt Earnings Import (shadow)",
name="Dolt Earnings Import",
replace_existing=True,
)
scheduler.add_job(
@@ -1793,17 +1587,6 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
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(
@@ -1831,13 +1614,6 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
_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(
@@ -1879,9 +1655,6 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
},
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],
@@ -1894,7 +1667,6 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
"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"],
)