71.2 KB
raw
//! Background job scheduler (demand-only).
//!
//! One long-lived tokio task wakes on a fixed tick. Since the demand-only
//! refocus (2026-06-03) the home dashboard's instruments are the **only**
//! timed network fetching (the active home sweep, a 2026-06-10 user call);
//! everything else is demand-driven. Each tick it:
//! - broadcasts a market-session change so open pages update their pill (and
//! re-pushes the dashboard summary, a local DB read, on a session flip);
//! - runs the demand-driven intraday quote poll — Yahoo quotes for just the
//! symbols a browser is currently viewing (the stream hub's interest
//! registry), so nothing is polled when nobody is watching;
//! - runs the active home sweep when due (every 15 minutes): re-quotes the
//! dashboard's overview instruments + all watchlist symbols, viewer or not,
//! so the home page always opens fresh — session-aware and guard-routed;
//! - prunes aged `intraday_bars` and `fetch_log` rows (~daily, local only).
//!
//! All the old timed sweeps (daily-close, SEC, dividends, fund metadata, NAV,
//! earnings, asset profile, periodic history) were removed in Phase A. The data
//! they fetched is now pulled **on demand** when a symbol's page is viewed and
//! its stored copy is stale — see `backfill_symbol` (the synchronous per-symbol
//! pull the add-symbol route and the on-demand refresh use) and Phase B.
//!
//! The boot seed only reconciles the universe rows from the curated CSV (local,
//! no network); a symbol's history and deep data fill in the first time it is
//! viewed. Outbound calls still pass through the persistent `EndpointGuard`
//! (see `src/guard.rs`), which paces requests and trips on rate limits.
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
use sqlx::SqlitePool;
use tokio::task::JoinHandle;
use crate::db::now_ms;
use crate::guard::{EndpointGuard, Permit};
use crate::market;
use crate::providers::sec::SecProvider;
use crate::providers::yahoo::YahooProvider;
use crate::providers::{
self, DividendEvent, Fact, FilingRecord, FundId, FundMetadata,
FundShape, FundamentalsProvider, HistoryProvider, IntradayBar, OwnershipPerson,
PortfolioData, Quote, QuoteProvider,
};
use crate::stream::{Hub, QuoteUpdate, StreamEvent};
use crate::{seed, Config};
/// How often the loop wakes to check whether a job is due. The jobs themselves
/// run hours apart; a one-minute tick is plenty responsive and nearly free
/// (two small SELECTs per wake).
const TICK: Duration = Duration::from_secs(60);
/// Minimum seconds between intraday polls of the same viewed symbol. With the
/// 60s tick this yields a ~5-minute per-symbol cadence: the
/// dashboard's watchlist and an open symbol page refresh about every five
/// minutes while watched, not every minute. Just under 5 min so a symbol due at
/// the 5-minute mark isn't skipped to the next tick.
const INTRADAY_MIN_INTERVAL_SECS: i64 = 4 * 60 + 45;
/// Active home-dashboard sweep cadence (a user call, 2026-06-10): the home
/// page's instruments — the market overview, the VIX / volume reads, and every
/// watchlist symbol across all sessions — are re-quoted every 15 minutes even
/// with nobody on the site, so the dashboard always opens fresh instead of
/// waiting on the on-open refresh's round trip to Yahoo. This is the one
/// deliberate exception to the demand-only model; it stays session-aware
/// (off-hours only the ~24h instruments are polled) and guard-routed, so the
/// worst case is a few dozen requests per hour against the 1000/hr budget.
const HOME_SWEEP_INTERVAL_SECS: i64 = 15 * 60;
/// Prune cadence and the two retention windows it enforces.
const PRUNE_INTERVAL_SECS: i64 = 24 * 3600;
const INTRADAY_RETENTION_DAYS: i64 = 14;
const FETCH_LOG_RETENTION_DAYS: i64 = 30;
/// Per-hour request ceiling for the Yahoo endpoint guard. Higher than Stooq's
/// default 200: an intraday tick sweeps every viewed symbol, and a daily-close
/// run touches the whole ~144-symbol universe at once. Still a hard cap that
/// stops a runaway loop well short of anything Yahoo would refuse.
/// `pub(crate)` so the add-symbol route builds its Yahoo guard with the same
/// ceiling (see `routes::symbols`).
pub(crate) const YAHOO_BUDGET: i64 = 1000;
/// Per-hour request ceiling for the SEC endpoint guard. A first-run full sweep
/// is one bulk ticker-map call plus two calls per stock (~220 for the starter
/// universe); 600 clears that in a single budget hour while still capping a
/// runaway loop well short of anything SEC's fair-access policy would refuse.
const SEC_BUDGET: i64 = 600;
/// Company-leadership refresh (Phase 14). Leadership changes slowly, so the
/// roster is rebuilt monthly rather than on the weekly SEC cadence above. Each
/// sweep parses at most this many of a company's most recent ownership filings
/// (one HTTP request each).
///
/// At 10 filings per sweep, a first-time backfill of one company captures the
/// recently-filing officers and board (a Form 3/4/5 from each active director
/// + a handful of officer trades), while the steady-state monthly refresh is
/// tiny — only the filings since `leadership_synced_at` are pulled. A higher
/// chunk eagerly grabs more history but, multiplied by the few-hundred-stock
/// universe, churned through the SEC endpoint's hourly budget during the
/// Phase 14 backfill (markets-closed weekend burn). Smaller chunks spread
/// that initial fill across more sweeps without changing the eventual roster.
const LEADERSHIP_MAX_FILINGS: usize = 10;
/// Spawn the scheduler. The returned handle is normally dropped: dropping it
/// detaches the task, which then runs for the lifetime of the process.
pub fn spawn(pool: SqlitePool, config: Arc<Config>, hub: Arc<Hub>) -> JoinHandle<()> {
tokio::spawn(async move {
tracing::info!("[scheduler] started");
if let Err(e) = reset_states_on_boot(&pool).await {
tracing::warn!("[scheduler] reset states: {e}");
}
if let Err(e) = register_endpoints(&pool).await {
tracing::warn!("[scheduler] register endpoints: {e}");
}
// Reconcile the universe rows from the curated CSV (local, no network).
// No history is fetched here any more — a symbol's data fills in on
// demand the first time its page is viewed (Phase B).
if let Err(e) = run_boot_seed(&pool, &config).await {
tracing::warn!("[scheduler] boot seed: {e:#}");
}
// Prune's last-run time is loop-local: a restart simply re-prunes once,
// which is harmless (local-only DELETEs, no network).
let mut last_prune: Option<i64> = None;
// The home sweep's last-run time is loop-local too: a restart sweeps
// once right away (the per-symbol throttle inside `refresh_quotes`
// keeps a quick restart from re-hitting Yahoo for fresh quotes).
let mut last_home_sweep: Option<i64> = None;
// The session last broadcast, so a transition (e.g. into after-hours)
// is pushed to connected clients exactly once.
let mut last_session: Option<market::Session> = None;
loop {
// Broadcast a market-session change so open pages update their pill.
let session = market::session_at(chrono::Utc::now());
if last_session != Some(session) {
if let Some(prev) = last_session {
tracing::info!(
"[scheduler] market {} -> {}",
prev.as_str(),
session.as_str()
);
}
hub.publish(StreamEvent::Market {
session: session.as_str().to_string(),
});
last_session = Some(session);
}
// Intraday quotes: demand-driven (only symbols a browser is
// viewing). Inside a trading session every viewed symbol is
// polled; outside it, only viewed futures, which trade nearly
// around the clock. Does no network work when nobody is watching.
if let Err(e) = run_intraday(&pool, &config, &hub, session).await {
tracing::warn!("[scheduler] intraday: {e:#}");
}
// The active home-dashboard sweep: every 15 minutes, re-quote the
// home page's instruments whether or not anyone is watching, so
// the dashboard always opens fresh.
if let Err(e) =
run_home_sweep_if_due(&pool, &config, &hub, session, &mut last_home_sweep).await
{
tracing::warn!("[scheduler] home sweep: {e:#}");
}
if let Err(e) = run_prune_if_due(&pool, &mut last_prune, &hub).await {
tracing::warn!("[scheduler] prune: {e:#}");
}
tokio::time::sleep(TICK).await;
}
})
}
/// Register the known data endpoints at startup, so the data-health page lists
/// each one — with its correct hourly budget — from the first boot, rather than
/// only once that endpoint's first request lazily creates its guard row. The
/// ids and budgets mirror how each job below constructs its `EndpointGuard`.
async fn register_endpoints(pool: &SqlitePool) -> anyhow::Result<()> {
// Stooq was retired 2026-05-30 (Yahoo now serves history too). Drop its
// stale guard row so it no longer lingers on the data-health page.
sqlx::query("DELETE FROM endpoint_guard WHERE endpoint = 'stooq'")
.execute(pool)
.await?;
// The demand-only refocus (Phase A) removed every timed sweep. Drop their
// leftover `data_status` rows so `/health` lists only the jobs that still
// run (the demand-driven intraday poll, the active home sweep, and the
// local prune); a prod DB carries rows from the old jobs that would
// otherwise show as stale.
sqlx::query("DELETE FROM data_status WHERE job NOT IN ('intraday', 'home', 'prune')")
.execute(pool)
.await?;
EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET)
.ensure_registered()
.await?;
EndpointGuard::with_budget(pool.clone(), "sec", SEC_BUDGET)
.ensure_registered()
.await?;
Ok(())
}
/// Clear any `fetching` state left behind by a crash mid-job. The owning task
/// did not survive the restart, so the row must not stay stuck `fetching`.
async fn reset_states_on_boot(pool: &SqlitePool) -> sqlx::Result<()> {
sqlx::query("UPDATE data_status SET state = 'idle', updated_at = ? WHERE state = 'fetching'")
.bind(now_ms())
.execute(pool)
.await?;
Ok(())
}
/// Reconcile the universe rows from the curated CSV on every boot — local only,
/// no network. Upserts every listed symbol and prunes curated rows dropped from
/// the CSV, so a `starter.csv` edit takes effect on deploy without a manual
/// re-seed (user-added `is_seeded = 0` rows are never touched).
///
/// Unlike before the demand-only refocus, this no longer backfills any history:
/// a symbol's deep daily history and SEC data fill in the first time its page is
/// viewed and found stale (Phase B), or via an explicit `make seed` run.
async fn run_boot_seed(pool: &SqlitePool, config: &Config) -> anyhow::Result<()> {
match seed::sync_universe(pool, config).await {
Ok(r) => tracing::info!(
"[scheduler] universe sync: {} symbols, {} pruned",
r.total,
r.pruned
),
Err(e) => tracing::warn!("[scheduler] universe sync: {e:#}"),
}
Ok(())
}
/// Demand-driven intraday quote refresh.
///
/// Polls Yahoo only for the symbols a browser is currently viewing (the stream
/// hub's interest registry). With nobody watching, `hub.viewed()` is empty and
/// this returns at once having done no network work: the user's hard rule is
/// to poll only what is on screen.
///
/// Which viewed symbols are polled depends on the session. Inside any trading
/// session (pre, regular, post) every viewed symbol is fair game. Outside it,
/// only viewed symbols that trade ~around the clock are polled — index futures,
/// commodities, and crypto (BTC) — while indexes, stocks and ETFs sit frozen
/// until the next session, so polling them off-hours would only re-fetch a flat
/// quote. This is what keeps the dashboard's overview futures/commodity/BTC
/// lines live overnight.
///
/// A clean run is recorded only in `data_status` (plus each `quotes.fetched_at`
/// row); a `fetch_log` row is written only for a notable run, an error or a
/// guard stop, so the minute-cadence job does not bury the log.
async fn run_intraday(
pool: &SqlitePool,
config: &Config,
hub: &Hub,
session: market::Session,
) -> anyhow::Result<()> {
let viewed = hub.viewed();
if viewed.is_empty() {
return Ok(());
}
// Inside a session, poll every viewed symbol; outside it, only the futures.
let mut targets: Vec<String> = if session.is_open() {
viewed
} else {
// Off-hours, only poll symbols that trade ~24h: index futures (and
// commodities) plus crypto (BTC). Cash indexes / stocks / ETFs are
// frozen, so polling them outside the session just burns budget.
let around_clock: HashSet<String> =
sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind IN ('future', 'crypto')")
.fetch_all(pool)
.await?
.into_iter()
.collect();
viewed
.into_iter()
.filter(|t| around_clock.contains(t))
.collect()
};
if targets.is_empty() {
return Ok(());
}
// Throttle to a ~5-minute per-symbol cadence: the loop
// ticks every 60s, but a symbol quoted within the last few minutes is left
// alone, so a dashboard left open polls each watchlist symbol about once
// every five minutes rather than every minute — light on the budget, and
// plenty "real-time" for delayed data. A symbol never quoted is always
// eligible. The set is small (only viewed symbols), so the lookup is cheap.
let throttle_cutoff = now_ms() - INTRADAY_MIN_INTERVAL_SECS * 1000;
let recent: HashSet<String> = sqlx::query_scalar(
"SELECT ticker FROM symbols WHERE last_quote_at IS NOT NULL AND last_quote_at >= ?",
)
.bind(throttle_cutoff)
.fetch_all(pool)
.await?
.into_iter()
.collect();
targets.retain(|t| !recent.contains(t));
if targets.is_empty() {
return Ok(());
}
let started = now_ms();
mark_fetching(pool, "intraday").await?;
notify_health(hub);
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
let t0 = Instant::now();
let mut ok = 0i64;
let mut errors = 0i64;
let mut stopped: Option<String> = None;
for ticker in &targets {
match guard.acquire().await? {
Permit::Granted => {}
Permit::Denied(why) => {
stopped = Some(why);
break;
}
}
match yahoo.quote(ticker).await {
Ok(data) => {
guard.record_success().await?;
store_quote(pool, ticker, &data.quote).await?;
if !data.bars.is_empty() {
store_intraday(pool, ticker, &data.bars).await?;
}
hub.publish(StreamEvent::Quote(QuoteUpdate::new(
ticker.clone(),
data.quote.price,
data.quote.prev_close,
data.quote.market_state.clone(),
)));
ok += 1;
}
Err(e) => {
guard.record_failure(&e).await?;
errors += 1;
tracing::warn!("[scheduler] intraday {ticker} failed: {e:#}");
}
}
}
let dur = t0.elapsed().as_millis() as i64;
if let Some(why) = stopped {
let detail = format!("stopped early ({why}); {ok} ok, {errors} errors");
tracing::warn!("[scheduler] intraday: {detail}");
log_fetch(pool, "intraday", "yahoo", "skipped", Some(&detail), Some(ok), dur, started)
.await?;
mark_ok(pool, "intraday", None).await?;
} else if ok == 0 && errors > 0 {
let detail = format!("all {errors} quotes failed");
log_fetch(pool, "intraday", "yahoo", "error", Some(&detail), Some(0), dur, started).await?;
mark_error(pool, "intraday", &detail, None).await?;
} else {
if errors > 0 {
let detail = format!("{ok} ok, {errors} errors");
log_fetch(pool, "intraday", "yahoo", "ok", Some(&detail), Some(ok), dur, started)
.await?;
}
mark_ok(pool, "intraday", None).await?;
}
notify_health(hub);
Ok(())
}
/// The active home-dashboard sweep (a user call, 2026-06-10; see
/// `HOME_SWEEP_INTERVAL_SECS`). Every 15 minutes it re-quotes the full
/// dashboard set — the market overview's cash + futures tickers, the VIX and
/// volume-proxy reads, and the union of every session's watchlist — without
/// requiring a viewer, so the home page always opens on current figures.
///
/// Session-aware like `run_intraday`: inside any trading session everything is
/// polled; outside it, only the instruments that trade ~around the clock
/// (futures, crypto), since a frozen stock/ETF/index quote would just burn
/// budget. Each pull rides `refresh_quotes`, so it is guard-routed, throttled
/// per symbol (a quote fresher than ~5 minutes is skipped), and published to
/// the hub so any open page live-ticks too.
async fn run_home_sweep_if_due(
pool: &SqlitePool,
config: &Config,
hub: &Hub,
session: market::Session,
last: &mut Option<i64>,
) -> anyhow::Result<()> {
let now = now_ms();
if let Some(t) = *last {
if (now - t) / 1000 < HOME_SWEEP_INTERVAL_SECS {
return Ok(());
}
}
*last = Some(now);
let mut targets: Vec<String> = crate::routes::home::dashboard_tickers()
.into_iter()
.map(str::to_string)
.collect();
let watchlisted: Vec<String> =
sqlx::query_scalar("SELECT DISTINCT ticker FROM watchlist ORDER BY ticker")
.fetch_all(pool)
.await?;
for t in watchlisted {
if !targets.contains(&t) {
targets.push(t);
}
}
if !session.is_open() {
let around_clock: HashSet<String> =
sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind IN ('future', 'crypto')")
.fetch_all(pool)
.await?
.into_iter()
.collect();
targets.retain(|t| around_clock.contains(t));
}
if targets.is_empty() {
return Ok(());
}
mark_fetching(pool, "home").await?;
notify_health(hub);
let refreshed = refresh_quotes(pool, config, hub, &targets).await;
if refreshed > 0 {
tracing::info!(
"[scheduler] home sweep: {refreshed}/{} refreshed",
targets.len()
);
}
mark_ok(pool, "home", Some(now + HOME_SWEEP_INTERVAL_SECS * 1000)).await?;
notify_health(hub);
Ok(())
}
/// Replace one stock's dividend history with what Yahoo returned. Yahoo serves
/// the canonical, corrected history each call, so a `DELETE` + `INSERT` keeps
/// the table honest if a payout is later retracted or restated. `pub(crate)`:
/// the add-symbol backfill reuses it.
pub(crate) async fn store_dividends(
pool: &SqlitePool,
ticker: &str,
events: &[DividendEvent],
) -> sqlx::Result<()> {
let mut tx = pool.begin().await?;
sqlx::query("DELETE FROM dividends WHERE ticker = ?")
.bind(ticker)
.execute(&mut *tx)
.await?;
for e in events {
sqlx::query(
"INSERT INTO dividends (ticker, ex_date, amount) VALUES (?, ?, ?) \
ON CONFLICT(ticker, ex_date) DO UPDATE SET amount = excluded.amount",
)
.bind(ticker)
.bind(&e.ex_date)
.bind(e.amount)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
/// Stamp a stock as freshly dividend-synced.
async fn mark_dividends_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query("UPDATE symbols SET dividends_synced_at = ?, updated_at = ? WHERE ticker = ?")
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Upsert one ETF's Yahoo `quoteSummary` metadata. Yahoo serves the full
/// current snapshot each call, so the row is replaced wholesale: a field
/// Yahoo no longer carries on a refresh becomes `NULL` here, rather than
/// keeping a stale value. `pub(crate)`: the add-symbol backfill reuses it.
pub(crate) async fn store_fund_metadata(
pool: &SqlitePool,
ticker: &str,
m: &FundMetadata,
) -> sqlx::Result<()> {
sqlx::query(
"INSERT INTO fund_metadata \
(ticker, expense_ratio, yield_pct, trailing_yield_pct, nav_price, \
inception_date, category, fund_family, strategy_summary, updated_at) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(ticker) DO UPDATE SET \
expense_ratio = excluded.expense_ratio, \
yield_pct = excluded.yield_pct, \
trailing_yield_pct = excluded.trailing_yield_pct, \
nav_price = excluded.nav_price, \
inception_date = excluded.inception_date, \
category = excluded.category, \
fund_family = excluded.fund_family, \
strategy_summary = excluded.strategy_summary, \
updated_at = excluded.updated_at",
)
.bind(ticker)
.bind(m.expense_ratio)
.bind(m.yield_pct)
.bind(m.trailing_yield_pct)
.bind(m.nav_price)
.bind(&m.inception_date)
.bind(&m.category)
.bind(&m.fund_family)
.bind(&m.strategy_summary)
.bind(now_ms())
.execute(pool)
.await?;
Ok(())
}
/// Stamp an ETF as freshly fund-metadata-synced. ETFs only — the column is
/// `NULL` forever on every non-ETF row.
async fn mark_fund_metadata_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"UPDATE symbols SET fund_metadata_synced_at = ?, updated_at = ? WHERE ticker = ?",
)
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Write one stock's next-earnings date and stamp it as freshly synced.
/// `next` is `None` when Yahoo has no upcoming date (the stored value is
/// cleared so the page falls back to a cadence estimate). `pub(crate)`:
/// the add-symbol backfill reuses it.
pub(crate) async fn store_earnings_next(
pool: &SqlitePool,
ticker: &str,
next: Option<i64>,
) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"UPDATE symbols SET next_earnings_at = ?, earnings_synced_at = ?, \
updated_at = ? \
WHERE ticker = ?",
)
.bind(next)
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Write one stock's sector / industry classification and stamp it freshly
/// synced. Either field may be `None` when Yahoo's coverage is partial; an
/// empty / whitespace value is dropped at parse time, not stored. `pub(crate)`:
/// the add-symbol backfill reuses it.
pub(crate) async fn store_asset_profile(
pool: &SqlitePool,
ticker: &str,
profile: &crate::providers::AssetProfile,
) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"UPDATE symbols SET sector = ?, industry = ?, \
asset_profile_synced_at = ?, updated_at = ? \
WHERE ticker = ?",
)
.bind(&profile.sector)
.bind(&profile.industry)
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Stamp a stock as freshly asset-profile-synced without overwriting its
/// stored sector / industry (used on a clean-empty Yahoo response so the
/// sweep does not re-fetch).
async fn mark_asset_profile_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"UPDATE symbols SET asset_profile_synced_at = ?, updated_at = ? WHERE ticker = ?",
)
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Fill in `symbols.cik` for any stock found in the bulk SEC ticker map.
/// Returns how many were newly resolved.
async fn resolve_ciks(pool: &SqlitePool, map: &HashMap<String, String>) -> sqlx::Result<i64> {
let stocks: Vec<String> =
sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind = 'stock' AND cik IS NULL")
.fetch_all(pool)
.await?;
let mut resolved = 0;
for ticker in stocks {
if let Some(cik) = map.get(&crate::providers::sec::normalize_ticker(&ticker)) {
let now = now_ms();
sqlx::query("UPDATE symbols SET cik = ?, updated_at = ? WHERE ticker = ?")
.bind(cik)
.bind(now)
.bind(&ticker)
.execute(pool)
.await?;
resolved += 1;
}
}
Ok(resolved)
}
/// Fill in `symbols.cik` and `symbols.series_id` for any ETF found in the bulk
/// SEC mutual-fund ticker map. Returns how many were newly resolved.
async fn resolve_fund_ciks(
pool: &SqlitePool,
map: &HashMap<String, FundId>,
) -> sqlx::Result<i64> {
let etfs: Vec<String> =
sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind = 'etf' AND cik IS NULL")
.fetch_all(pool)
.await?;
let mut resolved = 0;
for ticker in etfs {
if let Some(id) = map.get(&crate::providers::sec::normalize_ticker(&ticker)) {
let now = now_ms();
sqlx::query(
"UPDATE symbols SET cik = ?, series_id = ?, updated_at = ? WHERE ticker = ?",
)
.bind(&id.cik)
.bind(&id.series_id)
.bind(now)
.bind(&ticker)
.execute(pool)
.await?;
resolved += 1;
}
}
Ok(resolved)
}
/// Stamp one of a symbol's SEC sync timestamps to now. `column` is one of a
/// few hardcoded literals (never user input), so interpolating it is safe.
async fn mark_sec_synced(pool: &SqlitePool, ticker: &str, column: &str) -> sqlx::Result<()> {
let now = now_ms();
let sql = format!("UPDATE symbols SET {column} = ?, updated_at = ? WHERE ticker = ?");
sqlx::query(&sql)
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Upsert one company's fundamental facts. Keyed on (ticker, metric, period),
/// so a later filing's restated figure overwrites the prior one.
async fn store_fundamentals(pool: &SqlitePool, ticker: &str, facts: &[Fact]) -> sqlx::Result<()> {
let mut tx = pool.begin().await?;
for f in facts {
sqlx::query(
"INSERT INTO fundamentals \
(ticker, metric, period, fiscal_year, fiscal_qtr, period_end, \
value, unit, form, filed_at) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(ticker, metric, period) DO UPDATE SET \
fiscal_year = excluded.fiscal_year, fiscal_qtr = excluded.fiscal_qtr, \
period_end = excluded.period_end, value = excluded.value, \
unit = excluded.unit, form = excluded.form, filed_at = excluded.filed_at",
)
.bind(ticker)
.bind(&f.metric)
.bind(&f.period)
.bind(f.fiscal_year)
.bind(f.fiscal_qtr)
.bind(&f.period_end)
.bind(f.value)
.bind(&f.unit)
.bind(&f.form)
.bind(&f.filed_at)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
/// Upsert one company's filing history. Keyed on (ticker, accession).
async fn store_filings(
pool: &SqlitePool,
ticker: &str,
filings: &[FilingRecord],
) -> sqlx::Result<()> {
let mut tx = pool.begin().await?;
for f in filings {
sqlx::query(
"INSERT INTO filings \
(ticker, accession, form, filed_at, period_of_report, \
primary_doc, url, description, items) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(ticker, accession) DO UPDATE SET \
form = excluded.form, filed_at = excluded.filed_at, \
period_of_report = excluded.period_of_report, \
primary_doc = excluded.primary_doc, url = excluded.url, \
description = excluded.description, items = excluded.items",
)
.bind(ticker)
.bind(&f.accession)
.bind(&f.form)
.bind(&f.filed_at)
.bind(&f.period_of_report)
.bind(&f.primary_doc)
.bind(&f.url)
.bind(&f.description)
.bind(&f.items)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
/// Upsert a company's leadership roster from parsed ownership filings (Phase
/// 14). `roster` arrives newest-filing-first, so the first entry seen for a
/// person is their most recent filing and any later duplicate is skipped. The
/// upsert is guarded on `last_seen`, so a stale re-parse never overwrites a
/// person's role with an older filing's; departed insiders simply stop being
/// re-stamped and age out of the symbol page's recency window.
async fn store_leadership(
pool: &SqlitePool,
ticker: &str,
roster: &[(OwnershipPerson, String)],
) -> sqlx::Result<()> {
let mut seen: HashSet<&str> = HashSet::new();
let mut tx = pool.begin().await?;
for (person, filed_at) in roster {
if !seen.insert(person.name.as_str()) {
continue;
}
sqlx::query(
"INSERT INTO leadership \
(ticker, name, is_director, is_officer, officer_title, last_seen) \
VALUES (?, ?, ?, ?, ?, ?) \
ON CONFLICT(ticker, name) DO UPDATE SET \
is_director = excluded.is_director, is_officer = excluded.is_officer, \
officer_title = excluded.officer_title, last_seen = excluded.last_seen \
WHERE excluded.last_seen >= leadership.last_seen",
)
.bind(ticker)
.bind(&person.name)
.bind(person.is_director as i64)
.bind(person.is_officer as i64)
.bind(&person.officer_title)
.bind(filed_at)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
/// Upsert one ETF's N-PORT fund profile and replace its stored holdings. The
/// kept holdings are a small top slice, so they are deleted and re-inserted
/// wholesale on each refresh.
async fn store_fund_portfolio(
pool: &SqlitePool,
ticker: &str,
p: &PortfolioData,
) -> anyhow::Result<()> {
// Asset / sector / geography mixes are each variable-length, so they ride
// in JSON columns rather than their own tables:
// [["Equity", 99.8], ["Cash & equivalents", 0.2], ...].
let asset_mix = serde_json::to_string(&p.asset_mix)?;
let sector_mix = serde_json::to_string(&p.sector_mix)?;
let geography_mix = serde_json::to_string(&p.geography_mix)?;
let mut tx = pool.begin().await?;
sqlx::query(
"INSERT INTO fund_profiles \
(ticker, kind, net_assets, total_assets, holdings_count, report_date, \
asset_mix, sector_mix, geography_mix, updated_at) \
VALUES (?, 'portfolio', ?, ?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(ticker) DO UPDATE SET \
kind = excluded.kind, net_assets = excluded.net_assets, \
total_assets = excluded.total_assets, holdings_count = excluded.holdings_count, \
report_date = excluded.report_date, asset_mix = excluded.asset_mix, \
sector_mix = excluded.sector_mix, geography_mix = excluded.geography_mix, \
updated_at = excluded.updated_at",
)
.bind(ticker)
.bind(p.net_assets)
.bind(p.total_assets)
.bind(p.holdings_count)
.bind(&p.report_date)
.bind(&asset_mix)
.bind(§or_mix)
.bind(&geography_mix)
.bind(now_ms())
.execute(&mut *tx)
.await?;
sqlx::query("DELETE FROM fund_holdings WHERE ticker = ?")
.bind(ticker)
.execute(&mut *tx)
.await?;
for (i, h) in p.top_holdings.iter().enumerate() {
sqlx::query(
"INSERT INTO fund_holdings (ticker, rank, name, pct, value_usd) \
VALUES (?, ?, ?, ?, ?)",
)
.bind(ticker)
.bind(i as i64 + 1)
.bind(&h.name)
.bind(h.pct)
.bind(h.value_usd)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
/// Record a physical-commodity grantor trust's profile: just the AUM read from
/// its 10-K. It holds bullion, not a securities portfolio, so there are no
/// holdings and no asset mix.
async fn store_fund_commodity(
pool: &SqlitePool,
ticker: &str,
aum: Option<f64>,
) -> sqlx::Result<()> {
sqlx::query(
"INSERT INTO fund_profiles \
(ticker, kind, net_assets, total_assets, holdings_count, report_date, \
asset_mix, updated_at) \
VALUES (?, 'commodity_trust', ?, NULL, NULL, NULL, NULL, ?) \
ON CONFLICT(ticker) DO UPDATE SET \
kind = excluded.kind, net_assets = excluded.net_assets, \
updated_at = excluded.updated_at",
)
.bind(ticker)
.bind(aum)
.bind(now_ms())
.execute(pool)
.await?;
Ok(())
}
/// Stamp an ETF's fund profile as freshly synced.
async fn mark_fund_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query("UPDATE symbols SET fund_synced_at = ?, updated_at = ? WHERE ticker = ?")
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Upsert one symbol's live quote into `quotes` and refresh the denormalized
/// snapshot columns on `symbols` that the dashboard and SSE seeding read.
/// `pub(crate)`: the add-symbol route stores the quote its Yahoo lookup
/// already returned, rather than spending a second request.
pub(crate) async fn store_quote(pool: &SqlitePool, ticker: &str, q: &Quote) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"INSERT INTO quotes \
(ticker, price, prev_close, open, day_high, day_low, volume, \
market_state, source, source_time, fetched_at) \
VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'yahoo', ?, ?) \
ON CONFLICT(ticker) DO UPDATE SET \
price = excluded.price, prev_close = excluded.prev_close, \
open = excluded.open, day_high = excluded.day_high, \
day_low = excluded.day_low, volume = excluded.volume, \
market_state = excluded.market_state, source = excluded.source, \
source_time = excluded.source_time, fetched_at = excluded.fetched_at",
)
.bind(ticker)
.bind(q.price)
.bind(q.prev_close)
.bind(q.open)
.bind(q.day_high)
.bind(q.day_low)
.bind(q.volume)
.bind(&q.market_state)
.bind(q.source_time)
.bind(now)
.execute(pool)
.await?;
sqlx::query(
"UPDATE symbols SET last_price = ?, prev_close = ?, last_quote_at = ?, \
updated_at = ? WHERE ticker = ?",
)
.bind(q.price)
.bind(q.prev_close)
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Upsert one symbol's intraday bars in a single transaction. The prune job
/// trims `intraday_bars` to a rolling ~14-day window, so nothing here grows
/// without bound. `pub(crate)`: also called by the add-symbol route.
pub(crate) async fn store_intraday(
pool: &SqlitePool,
ticker: &str,
bars: &[IntradayBar],
) -> sqlx::Result<()> {
let mut tx = pool.begin().await?;
for b in bars {
sqlx::query(
"INSERT INTO intraday_bars (ticker, ts, open, high, low, close, volume) \
VALUES (?, ?, ?, ?, ?, ?, ?) \
ON CONFLICT(ticker, ts) DO UPDATE SET \
open = excluded.open, high = excluded.high, low = excluded.low, \
close = excluded.close, volume = excluded.volume",
)
.bind(ticker)
.bind(b.ts)
.bind(b.open)
.bind(b.high)
.bind(b.low)
.bind(b.close)
.bind(b.volume)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
// ── synchronous backfill for a freshly-added symbol (Phase 21) ─────────────
/// Run one guarded outbound call: acquire a permit, await `call`, and feed the
/// outcome back to the guard. `None` when the guard denied the request (the
/// breaker is open or the hourly budget is spent) or the permit could not be
/// acquired; `Some` carries whatever the call itself returned.
async fn guarded<T>(
guard: &EndpointGuard,
call: impl std::future::Future<Output = anyhow::Result<T>>,
) -> Option<anyhow::Result<T>> {
match guard.acquire().await {
Ok(Permit::Granted) => {
let result = call.await;
match &result {
Ok(_) => {
let _ = guard.record_success().await;
}
Err(e) => {
let _ = guard.record_failure(e).await;
}
}
Some(result)
}
Ok(Permit::Denied(why)) => {
tracing::info!("[backfill] guard denied: {why}");
None
}
Err(e) => {
tracing::warn!("[backfill] guard error: {e:#}");
None
}
}
}
/// Synchronously backfill everything for one just-added symbol: its deep daily
/// history from Stooq and, for a stock or ETF, its SEC data. The add-symbol
/// route (`routes::symbols`) calls this so a user-added symbol's page is
/// complete the moment the add returns, rather than filling in over later
/// scheduler cycles.
///
/// Best-effort and guard-routed: every outbound call passes through the same
/// `EndpointGuard` the background jobs use, and a guard denial or upstream
/// error for any one piece is logged and skipped. The symbol is already added;
/// the normal scheduler sweeps pick up whatever this run missed.
pub(crate) async fn backfill_symbol(pool: &SqlitePool, config: &Config, ticker: &str, kind: &str) {
backfill_history(pool, config, ticker, kind).await;
// Phase 26 + Phase 28: dividends covers stock dividends and ETF
// distributions (same Yahoo event series). Rides the Yahoo guard, so it
// runs independently of the SEC contact-email gate below.
if kind == "stock" || kind == "etf" {
backfill_dividends(pool, config, ticker).await;
}
// Phase 28: ETFs get their Yahoo fund_metadata pulled too — expense
// ratio, yield, NAV, inception, category, family, strategy summary.
if kind == "etf" {
backfill_fund_metadata(pool, config, ticker).await;
}
// Phase 25: stocks get their Yahoo earnings calendar pulled too — so a
// user-added stock's symbol page carries the next-expected earnings
// date the moment the add returns, rather than waiting on the next
// scheduler cycle.
if kind == "stock" {
backfill_earnings_calendar(pool, config, ticker).await;
}
// Phase 15: stocks get their Yahoo assetProfile pulled too — so a
// user-added stock immediately shows up under its sector and industry
// on /industries and carries the symbol-page header tag, rather than
// waiting on the next monthly scheduler sweep.
if kind == "stock" {
backfill_asset_profile(pool, config, ticker).await;
}
// SEC data covers stocks and ETFs; indexes and futures do not file. The
// whole SEC step is skipped with no contact email configured, as `run_sec`
// skips itself.
if config.sec_contact_email.is_empty() {
return;
}
let sec = SecProvider::new(providers::http::build_sec_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), sec.name(), SEC_BUDGET);
match kind {
"stock" => backfill_stock_sec(pool, &sec, &guard, ticker).await,
"etf" => backfill_etf_sec(pool, &sec, &guard, ticker).await,
_ => {}
}
}
/// Pull and store a freshly-added ETF's Yahoo `quoteSummary` metadata (Phase
/// 28). Mirrors `backfill_dividends`: same `yahoo` guard, best-effort, no
/// failure propagated to the add-symbol response.
async fn backfill_fund_metadata(pool: &SqlitePool, config: &Config, ticker: &str) {
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.fund_metadata(ticker)).await {
Some(Ok(Some(meta))) => match store_fund_metadata(pool, ticker, &meta).await {
Ok(()) => {
let _ = mark_fund_metadata_synced(pool, ticker).await;
tracing::info!("[backfill] {ticker} <- fund_metadata");
}
Err(e) => tracing::warn!("[backfill] store fund_metadata {ticker}: {e:#}"),
},
// Yahoo answered cleanly but had no fund modules for this ETF — stamp
// it checked so the next sweep does not re-fetch the same empty.
Some(Ok(None)) => {
let _ = mark_fund_metadata_synced(pool, ticker).await;
}
Some(Err(e)) => tracing::warn!("[backfill] fund_metadata {ticker}: {e:#}"),
None => {}
}
}
/// Pull and store a freshly-added stock's sector / industry classification
/// (Phase 15). Mirrors `backfill_earnings_calendar`: same `yahoo` guard,
/// best-effort, no failure propagated to the add-symbol response.
async fn backfill_asset_profile(pool: &SqlitePool, config: &Config, ticker: &str) {
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.asset_profile(ticker)).await {
Some(Ok(Some(profile))) => match store_asset_profile(pool, ticker, &profile).await {
Ok(()) => {
tracing::info!(
"[backfill] {ticker} <- asset_profile ({} / {})",
profile.sector.as_deref().unwrap_or("—"),
profile.industry.as_deref().unwrap_or("—"),
);
}
Err(e) => tracing::warn!("[backfill] store asset_profile {ticker}: {e:#}"),
},
// Yahoo answered cleanly but had no profile for this stock — stamp
// it so the next sweep does not re-fetch the same empty.
Some(Ok(None)) => {
let _ = mark_asset_profile_synced(pool, ticker).await;
}
Some(Err(e)) => tracing::warn!("[backfill] asset_profile {ticker}: {e:#}"),
None => {}
}
}
/// Pull and store a freshly-added stock's next-expected earnings date
/// (Phase 25). Mirrors `backfill_dividends`: same `yahoo` guard,
/// best-effort, no failure propagated to the add-symbol response.
async fn backfill_earnings_calendar(pool: &SqlitePool, config: &Config, ticker: &str) {
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.earnings_calendar(ticker)).await {
Some(Ok(next)) => match store_earnings_next(pool, ticker, next).await {
Ok(()) => {
tracing::info!(
"[backfill] {ticker} <- earnings_calendar ({})",
next.map(|_| "next set").unwrap_or("no upcoming date"),
);
}
Err(e) => tracing::warn!("[backfill] store earnings {ticker}: {e:#}"),
},
Some(Err(e)) => tracing::warn!("[backfill] earnings_calendar {ticker}: {e:#}"),
None => {}
}
}
/// Pull and store a freshly-added symbol's dividend / distribution history
/// (Phase 26 + Phase 28: stocks and ETFs both use this path). Routed through
/// the same `yahoo` guard the dividends sweep uses. Best-effort: a guard
/// denial or upstream error leaves the symbol for the next normal sweep.
async fn backfill_dividends(pool: &SqlitePool, config: &Config, ticker: &str) {
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.dividends(ticker)).await {
Some(Ok(events)) => match store_dividends(pool, ticker, &events).await {
Ok(()) => {
let _ = mark_dividends_synced(pool, ticker).await;
tracing::info!("[backfill] {ticker} <- {} dividends", events.len());
}
Err(e) => tracing::warn!("[backfill] store dividends {ticker}: {e:#}"),
},
Some(Err(e)) => tracing::warn!("[backfill] dividends {ticker}: {e:#}"),
None => {}
}
}
/// Pull and store one symbol's deep daily history from Yahoo (one
/// `interval=1d&range=max` call). Used by the add-symbol backfill; all kinds
/// are eligible since Yahoo serves `=F` futures history too.
async fn backfill_history(pool: &SqlitePool, config: &Config, ticker: &str, _kind: &str) {
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.daily(ticker, None)).await {
Some(Ok(bars)) if !bars.is_empty() => match seed::store_daily(pool, ticker, &bars).await {
Ok(()) => tracing::info!("[backfill] {ticker} <- {} daily bars", bars.len()),
Err(e) => tracing::warn!("[backfill] store history {ticker}: {e:#}"),
},
// A valid but empty response (a historyless symbol): stamp it checked
// so the history job does not immediately re-fetch it.
Some(Ok(_)) => {
let _ = mark_history_checked(pool, ticker).await;
}
_ => {}
}
}
/// Backfill a stock's SEC data: resolve its CIK, then pull fundamentals,
/// filings, and the officer/board roster.
async fn backfill_stock_sec(
pool: &SqlitePool,
sec: &SecProvider,
guard: &EndpointGuard,
ticker: &str,
) {
let cik = match resolve_one_cik(pool, sec, guard, ticker, false).await {
CikResolution::Found(c) => c,
CikResolution::Absent => {
// Not in SEC's company map: a non-filer, a foreign issuer, or a
// delisted/renamed ticker. Stamp the sections checked so the page
// shows an honest "no data" rather than a perpetual pending note.
let _ = mark_sec_synced(pool, ticker, "fundamentals_synced_at").await;
let _ = mark_sec_synced(pool, ticker, "filings_synced_at").await;
tracing::info!("[backfill] {ticker}: not in SEC company map, marked checked");
return;
}
CikResolution::Unavailable => {
tracing::info!("[backfill] {ticker}: SEC CIK map unavailable, leaving it for the sec job");
return;
}
};
if let Some(Ok(facts)) = guarded(guard, sec.facts(&cik)).await {
match store_fundamentals(pool, ticker, &facts).await {
Ok(()) => {
let _ = mark_sec_synced(pool, ticker, "fundamentals_synced_at").await;
}
Err(e) => tracing::warn!("[backfill] store facts {ticker}: {e:#}"),
}
}
if let Some(Ok(filings)) = guarded(guard, sec.filings(&cik)).await {
match store_filings(pool, ticker, &filings).await {
Ok(()) => {
let _ = mark_sec_synced(pool, ticker, "filings_synced_at").await;
}
Err(e) => tracing::warn!("[backfill] store filings {ticker}: {e:#}"),
}
}
backfill_leadership(pool, sec, guard, ticker, &cik).await;
}
/// Backfill a stock's officer/board roster from a window of its most recent
/// Form 3/4/5 ownership filings, mirroring the leadership sweep in `run_sec`.
async fn backfill_leadership(
pool: &SqlitePool,
sec: &SecProvider,
guard: &EndpointGuard,
ticker: &str,
cik: &str,
) {
let Some(Ok(index)) = guarded(guard, sec.ownership_index(cik)).await else {
return;
};
let to_parse: Vec<_> = index.into_iter().take(LEADERSHIP_MAX_FILINGS).collect();
let mut roster: Vec<(OwnershipPerson, String)> = Vec::new();
let mut complete = true;
for f in &to_parse {
match guarded(guard, sec.ownership_doc(cik, &f.accession, &f.primary_doc)).await {
Some(Ok(people)) => {
for p in people {
if p.is_director || p.is_officer {
roster.push((p, f.filed_at.clone()));
}
}
}
// A parse or network error for one filing: skip it and build the
// roster from the rest, exactly as `run_sec` does.
Some(Err(e)) => tracing::warn!("[backfill] ownership_doc {ticker}: {e:#}"),
// A guard denial leaves the roster only partial: leave it unsynced
// so the next `sec` cycle finishes it.
None => complete = false,
}
}
let _ = store_leadership(pool, ticker, &roster).await;
if complete {
let _ = mark_sec_synced(pool, ticker, "leadership_synced_at").await;
}
}
/// Backfill an ETF's fund profile: resolve its fund CIK, pull the filing list,
/// then either the N-PORT portfolio or a commodity trust's AUM.
async fn backfill_etf_sec(pool: &SqlitePool, sec: &SecProvider, guard: &EndpointGuard, ticker: &str) {
let cik = match resolve_one_cik(pool, sec, guard, ticker, true).await {
CikResolution::Found(c) => c,
CikResolution::Absent => {
// Not in SEC's mutual-fund map: a delisted/renamed fund (e.g. SPCX,
// which renamed to SPCK) or one that does not file N-PORT. Stamp it
// checked so the page shows an honest "no fund profile available"
// rather than a pending note Refresh can never clear.
let _ = mark_fund_synced(pool, ticker).await;
tracing::info!("[backfill] {ticker}: not in SEC fund map (delisted/renamed?), marked checked");
return;
}
CikResolution::Unavailable => {
tracing::info!("[backfill] {ticker}: SEC fund map unavailable, leaving it for the sec job");
return;
}
};
// `resolve_fund_ciks` stored the series id alongside the CIK.
let series_id: Option<String> =
sqlx::query_scalar("SELECT series_id FROM symbols WHERE ticker = ?")
.bind(ticker)
.fetch_one(pool)
.await
.ok()
.flatten();
let id = FundId {
cik: cik.clone(),
series_id,
};
let Some(Ok(ff)) = guarded(guard, sec.fund_filings(&id)).await else {
return;
};
let _ = store_filings(pool, ticker, &ff.filings).await;
match ff.shape {
FundShape::Portfolio { nport_href } => {
if let Some(Ok(portfolio)) = guarded(guard, sec.fund_portfolio(&nport_href)).await {
if store_fund_portfolio(pool, ticker, &portfolio).await.is_ok() {
let _ = mark_fund_synced(pool, ticker).await;
}
}
}
FundShape::CommodityTrust => {
if let Some(Ok(aum)) = guarded(guard, sec.fund_aum(&cik)).await {
if store_fund_commodity(pool, ticker, aum).await.is_ok() {
let _ = mark_fund_synced(pool, ticker).await;
}
}
}
FundShape::Unknown => {
let _ = mark_fund_synced(pool, ticker).await;
}
}
}
/// Outcome of resolving a symbol's SEC CIK from the bulk ticker map.
enum CikResolution {
/// A CIK is on file for this symbol (resolved now or on a prior run).
Found(String),
/// The bulk map was fetched successfully but does not list this ticker:
/// the company/fund genuinely has no SEC entry (delisted, renamed, a
/// foreign issuer, or a non-filer). The caller stamps the affected section
/// *checked* so the page shows an honest "no data available" instead of a
/// perpetual "not synced yet, hit Refresh".
Absent,
/// The map could not be fetched (guard denied / network error): nothing was
/// learned, so the caller leaves the section unsynced for a later retry.
Unavailable,
}
/// Resolve and store a freshly-added symbol's SEC CIK from the bulk ticker map.
/// `fund` selects the mutual-fund map (ETFs) over the operating-company map
/// (stocks). Distinguishes a genuinely-absent ticker from an unreachable map so
/// the caller can render an honest empty state (see [`CikResolution`]).
async fn resolve_one_cik(
pool: &SqlitePool,
sec: &SecProvider,
guard: &EndpointGuard,
ticker: &str,
fund: bool,
) -> CikResolution {
// Whether the bulk map was actually fetched this call (vs guard-denied /
// errored). A symbol may already carry a CIK from a prior run regardless.
let fetched = if fund {
match guarded(guard, sec.fund_ticker_map()).await {
Some(Ok(map)) => {
let _ = resolve_fund_ciks(pool, &map).await;
true
}
_ => false,
}
} else {
match guarded(guard, sec.cik_map()).await {
Some(Ok(map)) => {
let _ = resolve_ciks(pool, &map).await;
true
}
_ => false,
}
};
let cik: Option<String> =
sqlx::query_scalar::<_, Option<String>>("SELECT cik FROM symbols WHERE ticker = ?")
.bind(ticker)
.fetch_one(pool)
.await
.ok()
.flatten();
match (cik, fetched) {
(Some(c), _) => CikResolution::Found(c),
(None, true) => CikResolution::Absent,
(None, false) => CikResolution::Unavailable,
}
}
/// Prune aged rows once per `PRUNE_INTERVAL_SECS`. `intraday_bars` keeps a
/// rolling ~14-day window; `fetch_log` keeps ~30 days. `daily_prices` is
/// permanent and never touched here.
async fn run_prune_if_due(
pool: &SqlitePool,
last: &mut Option<i64>,
hub: &Hub,
) -> anyhow::Result<()> {
let now = now_ms();
if let Some(t) = *last {
if (now - t) / 1000 < PRUNE_INTERVAL_SECS {
return Ok(());
}
}
let t0 = Instant::now();
let intraday_cutoff = now - INTRADAY_RETENTION_DAYS * 86_400 * 1000;
let log_cutoff = now - FETCH_LOG_RETENTION_DAYS * 86_400 * 1000;
let bars = sqlx::query("DELETE FROM intraday_bars WHERE ts < ?")
.bind(intraday_cutoff)
.execute(pool)
.await?
.rows_affected();
let logs = sqlx::query("DELETE FROM fetch_log WHERE started_at < ?")
.bind(log_cutoff)
.execute(pool)
.await?
.rows_affected();
let dur = t0.elapsed().as_millis() as i64;
let detail = format!("{bars} intraday bars, {logs} fetch_log rows");
tracing::info!("[scheduler] prune: removed {detail}");
log_fetch(pool, "prune", "-", "ok", Some(&detail), Some((bars + logs) as i64), dur, now).await?;
*last = Some(now);
notify_health(hub);
Ok(())
}
// ── data_status / fetch_log helpers ───────────────────────────────────────
/// Nudge any connected `/health` page to pull a fresh snapshot. Sent whenever a
/// job changes state or appends a `fetch_log` row, so the data-health page
/// tracks the worker in near real time. Carries no payload (see `StreamEvent`).
fn notify_health(hub: &Hub) {
hub.publish(StreamEvent::Health);
}
/// Move a job's `data_status` row to the `fetching` state.
async fn mark_fetching(pool: &SqlitePool, job: &str) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"INSERT INTO data_status (job, state, updated_at) VALUES (?, 'fetching', ?) \
ON CONFLICT(job) DO UPDATE SET state = 'fetching', updated_at = excluded.updated_at",
)
.bind(job)
.bind(now)
.execute(pool)
.await?;
Ok(())
}
/// Mark a job finished-OK, recording when it next falls due (`None` for
/// one-shot jobs like the seed).
async fn mark_ok(pool: &SqlitePool, job: &str, next_run_at: Option<i64>) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"INSERT INTO data_status (job, state, last_ok_at, next_run_at, updated_at) \
VALUES (?, 'ok', ?, ?, ?) \
ON CONFLICT(job) DO UPDATE SET \
state = 'ok', last_ok_at = excluded.last_ok_at, \
next_run_at = excluded.next_run_at, updated_at = excluded.updated_at",
)
.bind(job)
.bind(now)
.bind(next_run_at)
.bind(now)
.execute(pool)
.await?;
Ok(())
}
/// Mark a job failed, recording the error and when it should be retried.
async fn mark_error(
pool: &SqlitePool,
job: &str,
msg: &str,
next_run_at: Option<i64>,
) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"INSERT INTO data_status (job, state, last_error, last_error_at, next_run_at, updated_at) \
VALUES (?, 'error', ?, ?, ?, ?) \
ON CONFLICT(job) DO UPDATE SET \
state = 'error', last_error = excluded.last_error, \
last_error_at = excluded.last_error_at, next_run_at = excluded.next_run_at, \
updated_at = excluded.updated_at",
)
.bind(job)
.bind(msg)
.bind(now)
.bind(next_run_at)
.bind(now)
.execute(pool)
.await?;
Ok(())
}
/// Stamp a symbol as history-checked without storing bars: used when the
/// upstream returned a valid response that simply held nothing new.
async fn mark_history_checked(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query("UPDATE symbols SET history_synced_at = ?, updated_at = ? WHERE ticker = ?")
.bind(now)
.bind(now)
.bind(ticker)
.execute(pool)
.await?;
Ok(())
}
/// Append one `fetch_log` row. `ticker` is left NULL: these are bulk jobs, so
/// a run logs once rather than once per symbol.
#[allow(clippy::too_many_arguments)]
async fn log_fetch(
pool: &SqlitePool,
job: &str,
provider: &str,
status: &str,
detail: Option<&str>,
rows: Option<i64>,
duration_ms: i64,
started_at: i64,
) -> sqlx::Result<()> {
sqlx::query(
"INSERT INTO fetch_log \
(job, provider, ticker, status, detail, rows, duration_ms, started_at, finished_at) \
VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?)",
)
.bind(job)
.bind(provider)
.bind(status)
.bind(detail)
.bind(rows)
.bind(duration_ms)
.bind(started_at)
.bind(now_ms())
.execute(pool)
.await?;
Ok(())
}
// ───────────────────── on-demand refresh pipeline (Phase B) ─────────────────
//
// Since the demand-only refocus there are no timed sweeps: a viewed symbol's
// data is pulled here, on demand, when its page loads (the fast price steps
// always run; the slow SEC / metadata steps run only when their stored copy is
// stale) or when the user hits Refresh (`force`, which runs everything). The
// symbol-page SSE route (`routes::symbols::refresh_stream`) drives it: it asks
// `refresh_plan` which steps will run, then runs each via `refresh_step`,
// streaming progress to the page's loading bar. Each step reuses the same
// guarded `backfill_*` helpers the add-symbol flow already uses.
/// On-demand staleness windows for the gated (slow) steps. The always-run price
/// steps (quote + history) carry no window — they run on every load.
const REFRESH_SEC_STALE_SECS: i64 = 7 * 24 * 3600;
const REFRESH_LEADERSHIP_STALE_SECS: i64 = 30 * 24 * 3600;
const REFRESH_META_STALE_SECS: i64 = 7 * 24 * 3600;
/// One step in a symbol's refresh, shown on the page's loading bar.
pub(crate) struct RefreshStep {
/// Stable key the route passes back to `refresh_step`.
pub key: &'static str,
/// Human label for the loading bar.
pub label: &'static str,
/// Whether this refreshes a server-rendered "deep" section — if any deep
/// step ran, the page reloads to show it; a load that runs only the (live)
/// price steps patches the price in place instead.
pub deep: bool,
}
const fn step(key: &'static str, label: &'static str, deep: bool) -> RefreshStep {
RefreshStep { key, label, deep }
}
/// Decide which steps a symbol's refresh will run. The two price steps always
/// run; the slow steps are included only when stale (or `force`). An index /
/// future / unknown kind gets just the price steps.
pub(crate) async fn refresh_plan(
pool: &SqlitePool,
config: &Config,
ticker: &str,
kind: &str,
force: bool,
) -> Vec<RefreshStep> {
let mut steps = vec![
step("quote", "Live quote", false),
step("history", "Daily history", false),
];
let Some(s) =
sqlx::query_as::<_, crate::models::SymbolRow>("SELECT * FROM symbols WHERE ticker = ?")
.bind(ticker)
.fetch_optional(pool)
.await
.ok()
.flatten()
else {
return steps;
};
let now = now_ms();
let stale = |at: Option<i64>, secs: i64| force || at.map_or(true, |t| now - t > secs * 1000);
let sec_ok = !config.sec_contact_email.is_empty();
match kind {
"stock" => {
if sec_ok
&& (stale(s.fundamentals_synced_at, REFRESH_SEC_STALE_SECS)
|| stale(s.filings_synced_at, REFRESH_SEC_STALE_SECS)
|| stale(s.leadership_synced_at, REFRESH_LEADERSHIP_STALE_SECS))
{
steps.push(step("sec", "Fundamentals, filings & leadership", true));
}
if stale(s.earnings_synced_at, REFRESH_META_STALE_SECS) {
steps.push(step("earnings", "Earnings date", true));
}
if stale(s.asset_profile_synced_at, REFRESH_META_STALE_SECS) {
steps.push(step("profile", "Sector & industry", true));
}
if stale(s.dividends_synced_at, REFRESH_META_STALE_SECS) {
steps.push(step("dividends", "Dividends", true));
}
}
"etf" => {
if sec_ok && stale(s.fund_synced_at, REFRESH_SEC_STALE_SECS) {
steps.push(step("fund_sec", "Holdings & filings", true));
}
if stale(s.fund_metadata_synced_at, REFRESH_META_STALE_SECS) {
steps.push(step("fund_meta", "Fund details & NAV", true));
}
if stale(s.dividends_synced_at, REFRESH_META_STALE_SECS) {
steps.push(step("dividends", "Distributions", true));
}
}
_ => {}
}
steps
}
/// Run one refresh step by key. Returns a short status for the loading bar:
/// "ok" when it ran, "skipped" when the guard denied it (breaker open / budget
/// spent). The backfill helpers are best-effort and swallow their own errors,
/// so the deep steps report "ok" once attempted; the price steps, which this
/// runs inline through the guard, distinguish a guard denial.
pub(crate) async fn refresh_step(
pool: &SqlitePool,
config: &Config,
hub: &Hub,
ticker: &str,
kind: &str,
key: &str,
) -> &'static str {
let _ = kind;
match key {
"quote" => refresh_quote(pool, config, hub, ticker).await,
"history" => refresh_history_incremental(pool, config, ticker).await,
"sec" => refresh_sec(pool, config, ticker, false).await,
"fund_sec" => refresh_sec(pool, config, ticker, true).await,
"earnings" => {
backfill_earnings_calendar(pool, config, ticker).await;
"ok"
}
"profile" => {
backfill_asset_profile(pool, config, ticker).await;
"ok"
}
"dividends" => {
backfill_dividends(pool, config, ticker).await;
"ok"
}
"fund_meta" => refresh_fund_meta(pool, config, ticker).await,
_ => "ok",
}
}
/// Pull fresh quotes for a set of dashboard symbols on demand — the dashboard's
/// on-open refresh. A symbol quoted within the last few
/// minutes is skipped, so a reload (or the add/remove reload) does not re-hit
/// Yahoo, and overnight the gate keeps a re-open from re-polling the same frozen
/// close. Runs regardless of session: opening the dashboard after the close
/// should still confirm the latest (closing) prices rather than show a stale
/// snapshot. Each fetch publishes to the hub so open cards live-tick. Returns how
/// many symbols were actually refreshed.
pub(crate) async fn refresh_quotes(
pool: &SqlitePool,
config: &Config,
hub: &Hub,
tickers: &[String],
) -> usize {
let cutoff = now_ms() - INTRADAY_MIN_INTERVAL_SECS * 1000;
let mut refreshed = 0;
for t in tickers {
// Skip a symbol with a quote younger than the throttle window.
let fresh: Option<i64> = sqlx::query_scalar(
"SELECT last_quote_at FROM symbols WHERE ticker = ? AND last_quote_at >= ?",
)
.bind(t)
.bind(cutoff)
.fetch_optional(pool)
.await
.ok()
.flatten();
if fresh.is_some() {
continue;
}
if refresh_quote(pool, config, hub, t).await == "ok" {
refreshed += 1;
}
}
refreshed
}
/// Pull one live quote + its intraday bars and publish it to the hub so an open
/// page patches its price in place (mirrors `run_intraday`'s per-symbol body).
async fn refresh_quote(pool: &SqlitePool, config: &Config, hub: &Hub, ticker: &str) -> &'static str {
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.quote(ticker)).await {
Some(Ok(data)) => {
let _ = store_quote(pool, ticker, &data.quote).await;
if !data.bars.is_empty() {
let _ = store_intraday(pool, ticker, &data.bars).await;
}
hub.publish(StreamEvent::Quote(QuoteUpdate::new(
ticker.to_string(),
data.quote.price,
data.quote.prev_close,
data.quote.market_state.clone(),
)));
"ok"
}
Some(Err(e)) => {
tracing::warn!("[refresh] quote {ticker}: {e:#}");
"error"
}
None => "skipped",
}
}
/// Intraday range that covers a whole trading week (Mon–Fri) of 15-minute bars
/// in one request — enough for the end-of-week dashboard view.
const INTRADAY_WEEK_RANGE: &str = "5d";
/// Backfill the whole trading week's 15-minute bars for `tickers` when the
/// stored bars don't already cover the early week. The routine intraday poll
/// only ever stores one day at a time (`range=1d`), so the end-of-week view is
/// missing any day the dashboard wasn't open. One guarded `range=5d` request per
/// still-incomplete symbol fills the gap; symbols whose stored bars already
/// reach the week's start are skipped, so a reload doesn't re-hit Yahoo.
pub(crate) async fn backfill_intraday_week(
pool: &SqlitePool,
config: &Config,
tickers: &[String],
week_start_ms: i64,
week_end_ms: i64,
) -> usize {
// "Already covered": the earliest in-window bar sits within ~36h of the
// week's open. A normally-polled week starts at Monday's open; a
// holiday-Monday week at Tuesday's — both inside this margin, so they're not
// refetched. Only a week missing its first two days (the gap the user sees)
// falls outside it.
let covered_before = week_start_ms + 36 * 3_600 * 1000;
let mut filled = 0;
for t in tickers {
let earliest: Option<i64> = sqlx::query_scalar(
"SELECT MIN(ts) FROM intraday_bars WHERE ticker = ? AND ts >= ? AND ts <= ?",
)
.bind(t)
.bind(week_start_ms)
.bind(week_end_ms)
.fetch_optional(pool)
.await
.ok()
.flatten()
.flatten();
if matches!(earliest, Some(ms) if ms <= covered_before) {
continue;
}
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.intraday_window(t, INTRADAY_WEEK_RANGE)).await {
Some(Ok(data)) if !data.bars.is_empty() => {
let _ = store_intraday(pool, t, &data.bars).await;
filled += 1;
}
Some(Err(e)) => tracing::warn!("[week] intraday {t}: {e:#}"),
_ => {}
}
}
filled
}
/// Pull the daily history a viewed symbol is missing: the window since its last
/// stored bar (incremental) when it already has history, else a full
/// `range=max` backfill. Cheaper than the deep re-fetch on a routine load.
async fn refresh_history_incremental(
pool: &SqlitePool,
config: &Config,
ticker: &str,
) -> &'static str {
let last: Option<String> =
sqlx::query_scalar("SELECT history_last_date FROM symbols WHERE ticker = ?")
.bind(ticker)
.fetch_optional(pool)
.await
.ok()
.flatten();
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
match guarded(&guard, yahoo.daily(ticker, last.as_deref())).await {
Some(Ok(bars)) if !bars.is_empty() => {
let _ = seed::store_daily(pool, ticker, &bars).await;
"ok"
}
Some(Ok(_)) => {
let _ = mark_history_checked(pool, ticker).await;
"ok"
}
Some(Err(e)) => {
tracing::warn!("[refresh] history {ticker}: {e:#}");
"error"
}
None => "skipped",
}
}
/// Backfill a viewed symbol's SEC data on demand (stock fundamentals/filings/
/// leadership, or an ETF's holdings/filings). Skipped cleanly with no contact
/// email configured, as the old sweep was.
async fn refresh_sec(pool: &SqlitePool, config: &Config, ticker: &str, fund: bool) -> &'static str {
if config.sec_contact_email.is_empty() {
return "skipped";
}
let sec = SecProvider::new(providers::http::build_sec_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), sec.name(), SEC_BUDGET);
if fund {
backfill_etf_sec(pool, &sec, &guard, ticker).await;
} else {
backfill_stock_sec(pool, &sec, &guard, ticker).await;
}
"ok"
}
/// Refresh an ETF's Yahoo fund metadata and its NAV (the price-vs-NAV premium
/// behind the quality read's tracking factor needs a fresh NAV; see the
/// hard-won lesson). Two cheap `quoteSummary` calls through the
/// Yahoo guard.
async fn refresh_fund_meta(pool: &SqlitePool, config: &Config, ticker: &str) -> &'static str {
backfill_fund_metadata(pool, config, ticker).await;
let yahoo = YahooProvider::new(providers::http::build_client(config));
let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
if let Some(Ok(nav)) = guarded(&guard, yahoo.fund_nav(ticker)).await {
let _ = store_fund_nav(pool, ticker, nav).await;
}
"ok"
}
/// Upsert an ETF's freshly-fetched NAV + its sync stamp, touching only the NAV
/// columns so the static fields stay intact. A `None` nav clears any prior NAV
/// (honest: no fresh value to read a premium against). Re-added for the Phase-B
/// on-demand NAV pull after the daily `fund_nav` job was removed in Phase A.
async fn store_fund_nav(pool: &SqlitePool, ticker: &str, nav: Option<f64>) -> sqlx::Result<()> {
let now = now_ms();
sqlx::query(
"INSERT INTO fund_metadata (ticker, nav_price, nav_synced_at, updated_at) \
VALUES (?, ?, ?, ?) \
ON CONFLICT(ticker) DO UPDATE SET \
nav_price = excluded.nav_price, \
nav_synced_at = excluded.nav_synced_at, \
updated_at = excluded.updated_at",
)
.bind(ticker)
.bind(nav)
.bind(now)
.bind(now)
.execute(pool)
.await?;
Ok(())
}