Single-binary self-hosted market watcher for stocks, ETFs, indexes, and futures: live charts, key stats, fundamentals, SEC filings, and SSE streaming.
axumdockerfinancerustself-hostedsqlitestocksvite
1//! Background job scheduler (demand-only).
2//!
3//! One long-lived tokio task wakes on a fixed tick. Since the demand-only
4//! refocus (2026-06-03) the home dashboard's instruments are the **only**
5//! timed network fetching (the active home sweep, a 2026-06-10 user call);
6//! everything else is demand-driven. Each tick it:
7//! - broadcasts a market-session change so open pages update their pill (and
8//! re-pushes the dashboard summary, a local DB read, on a session flip);
9//! - runs the demand-driven intraday quote poll — Yahoo quotes for just the
10//! symbols a browser is currently viewing (the stream hub's interest
11//! registry), so nothing is polled when nobody is watching;
12//! - runs the active home sweep when due (every 15 minutes): re-quotes the
13//! dashboard's overview instruments + all watchlist symbols, viewer or not,
14//! so the home page always opens fresh — session-aware and guard-routed;
15//! - prunes aged `intraday_bars` and `fetch_log` rows (~daily, local only).
16//!
17//! All the old timed sweeps (daily-close, SEC, dividends, fund metadata, NAV,
18//! earnings, asset profile, periodic history) were removed in Phase A. The data
19//! they fetched is now pulled **on demand** when a symbol's page is viewed and
20//! its stored copy is stale — see `backfill_symbol` (the synchronous per-symbol
21//! pull the add-symbol route and the on-demand refresh use) and Phase B.
22//!
23//! The boot seed only reconciles the universe rows from the curated CSV (local,
24//! no network); a symbol's history and deep data fill in the first time it is
25//! viewed. Outbound calls still pass through the persistent `EndpointGuard`
26//! (see `src/guard.rs`), which paces requests and trips on rate limits.
27
28use std::collections::{HashMap, HashSet};
29use std::sync::Arc;
30use std::time::{Duration, Instant};
31
32use sqlx::SqlitePool;
33use tokio::task::JoinHandle;
34
35use crate::db::now_ms;
36use crate::guard::{EndpointGuard, Permit};
37use crate::market;
38use crate::providers::sec::SecProvider;
39use crate::providers::yahoo::YahooProvider;
40use crate::providers::{
41 self, DividendEvent, Fact, FilingRecord, FundId, FundMetadata,
42 FundShape, FundamentalsProvider, HistoryProvider, IntradayBar, OwnershipPerson,
43 PortfolioData, Quote, QuoteProvider,
44};
45use crate::stream::{Hub, QuoteUpdate, StreamEvent};
46use crate::{seed, Config};
47
48/// How often the loop wakes to check whether a job is due. The jobs themselves
49/// run hours apart; a one-minute tick is plenty responsive and nearly free
50/// (two small SELECTs per wake).
51const TICK: Duration = Duration::from_secs(60);
52
53/// Minimum seconds between intraday polls of the same viewed symbol. With the
54/// 60s tick this yields a ~5-minute per-symbol cadence: the
55/// dashboard's watchlist and an open symbol page refresh about every five
56/// minutes while watched, not every minute. Just under 5 min so a symbol due at
57/// the 5-minute mark isn't skipped to the next tick.
58const INTRADAY_MIN_INTERVAL_SECS: i64 = 4 * 60 + 45;
59
60/// Active home-dashboard sweep cadence (a user call, 2026-06-10): the home
61/// page's instruments — the market overview, the VIX / volume reads, and every
62/// watchlist symbol across all sessions — are re-quoted every 15 minutes even
63/// with nobody on the site, so the dashboard always opens fresh instead of
64/// waiting on the on-open refresh's round trip to Yahoo. This is the one
65/// deliberate exception to the demand-only model; it stays session-aware
66/// (off-hours only the ~24h instruments are polled) and guard-routed, so the
67/// worst case is a few dozen requests per hour against the 1000/hr budget.
68const HOME_SWEEP_INTERVAL_SECS: i64 = 15 * 60;
69
70/// Prune cadence and the two retention windows it enforces.
71const PRUNE_INTERVAL_SECS: i64 = 24 * 3600;
72const INTRADAY_RETENTION_DAYS: i64 = 14;
73const FETCH_LOG_RETENTION_DAYS: i64 = 30;
74/// ~13 months: one month past the fin_sid cookie's 12-month lifetime, so only
75/// sids that can no longer return are pruned.
76const WATCHLIST_STALE_DAYS: i64 = 396;
77
78/// Per-hour request ceiling for the Yahoo endpoint guard. Higher than Stooq's
79/// default 200: an intraday tick sweeps every viewed symbol, and a daily-close
80/// run touches the whole ~144-symbol universe at once. Still a hard cap that
81/// stops a runaway loop well short of anything Yahoo would refuse.
82/// `pub(crate)` so the add-symbol route builds its Yahoo guard with the same
83/// ceiling (see `routes::symbols`).
84pub(crate) const YAHOO_BUDGET: i64 = 1000;
85
86/// Per-hour request ceiling for the SEC endpoint guard. A first-run full sweep
87/// is one bulk ticker-map call plus two calls per stock (~220 for the starter
88/// universe); 600 clears that in a single budget hour while still capping a
89/// runaway loop well short of anything SEC's fair-access policy would refuse.
90const SEC_BUDGET: i64 = 600;
91
92/// Company-leadership refresh (Phase 14). Leadership changes slowly, so the
93/// roster is rebuilt monthly rather than on the weekly SEC cadence above. Each
94/// sweep parses at most this many of a company's most recent ownership filings
95/// (one HTTP request each).
96///
97/// At 10 filings per sweep, a first-time backfill of one company captures the
98/// recently-filing officers and board (a Form 3/4/5 from each active director
99/// + a handful of officer trades), while the steady-state monthly refresh is
100/// tiny — only the filings since `leadership_synced_at` are pulled. A higher
101/// chunk eagerly grabs more history but, multiplied by the few-hundred-stock
102/// universe, churned through the SEC endpoint's hourly budget during the
103/// Phase 14 backfill (markets-closed weekend burn). Smaller chunks spread
104/// that initial fill across more sweeps without changing the eventual roster.
105const LEADERSHIP_MAX_FILINGS: usize = 10;
106
107/// Spawn the scheduler. The returned handle is normally dropped: dropping it
108/// detaches the task, which then runs for the lifetime of the process.
109pub fn spawn(pool: SqlitePool, config: Arc<Config>, hub: Arc<Hub>) -> JoinHandle<()> {
110 tokio::spawn(async move {
111 tracing::info!("[scheduler] started");
112
113 if let Err(e) = reset_states_on_boot(&pool).await {
114 tracing::warn!("[scheduler] reset states: {e}");
115 }
116 if let Err(e) = register_endpoints(&pool).await {
117 tracing::warn!("[scheduler] register endpoints: {e}");
118 }
119
120 // Reconcile the universe rows from the curated CSV (local, no network).
121 // No history is fetched here any more — a symbol's data fills in on
122 // demand the first time its page is viewed (Phase B).
123 if let Err(e) = run_boot_seed(&pool, &config).await {
124 tracing::warn!("[scheduler] boot seed: {e:#}");
125 }
126
127 // Prune's last-run time is loop-local: a restart simply re-prunes once,
128 // which is harmless (local-only DELETEs, no network).
129 let mut last_prune: Option<i64> = None;
130 // The home sweep's last-run time is loop-local too: a restart sweeps
131 // once right away (the per-symbol throttle inside `refresh_quotes`
132 // keeps a quick restart from re-hitting Yahoo for fresh quotes).
133 let mut last_home_sweep: Option<i64> = None;
134 // The session last broadcast, so a transition (e.g. into after-hours)
135 // is pushed to connected clients exactly once.
136 let mut last_session: Option<market::Session> = None;
137 loop {
138 // Broadcast a market-session change so open pages update their pill.
139 let session = market::session_at(chrono::Utc::now());
140 if last_session != Some(session) {
141 if let Some(prev) = last_session {
142 tracing::info!(
143 "[scheduler] market {} -> {}",
144 prev.as_str(),
145 session.as_str()
146 );
147 }
148 hub.publish(StreamEvent::Market {
149 session: session.as_str().to_string(),
150 });
151 last_session = Some(session);
152 }
153
154 // Intraday quotes: demand-driven (only symbols a browser is
155 // viewing). Inside a trading session every viewed symbol is
156 // polled; outside it, only viewed futures, which trade nearly
157 // around the clock. Does no network work when nobody is watching.
158 if let Err(e) = run_intraday(&pool, &config, &hub, session).await {
159 tracing::warn!("[scheduler] intraday: {e:#}");
160 }
161
162 // The active home-dashboard sweep: every 15 minutes, re-quote the
163 // home page's instruments whether or not anyone is watching, so
164 // the dashboard always opens fresh.
165 if let Err(e) =
166 run_home_sweep_if_due(&pool, &config, &hub, session, &mut last_home_sweep).await
167 {
168 tracing::warn!("[scheduler] home sweep: {e:#}");
169 }
170
171 if let Err(e) = run_prune_if_due(&pool, &mut last_prune, &hub).await {
172 tracing::warn!("[scheduler] prune: {e:#}");
173 }
174 tokio::time::sleep(TICK).await;
175 }
176 })
177}
178
179/// Register the known data endpoints at startup, so the data-health page lists
180/// each one — with its correct hourly budget — from the first boot, rather than
181/// only once that endpoint's first request lazily creates its guard row. The
182/// ids and budgets mirror how each job below constructs its `EndpointGuard`.
183async fn register_endpoints(pool: &SqlitePool) -> anyhow::Result<()> {
184 // Stooq was retired 2026-05-30 (Yahoo now serves history too). Drop its
185 // stale guard row so it no longer lingers on the data-health page.
186 sqlx::query("DELETE FROM endpoint_guard WHERE endpoint = 'stooq'")
187 .execute(pool)
188 .await?;
189 // The demand-only refocus (Phase A) removed every timed sweep. Drop their
190 // leftover `data_status` rows so `/health` lists only the jobs that still
191 // run (the demand-driven intraday poll, the active home sweep, and the
192 // local prune); a prod DB carries rows from the old jobs that would
193 // otherwise show as stale.
194 sqlx::query("DELETE FROM data_status WHERE job NOT IN ('intraday', 'home', 'prune')")
195 .execute(pool)
196 .await?;
197 EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET)
198 .ensure_registered()
199 .await?;
200 EndpointGuard::with_budget(pool.clone(), "sec", SEC_BUDGET)
201 .ensure_registered()
202 .await?;
203 Ok(())
204}
205
206/// Clear any `fetching` state left behind by a crash mid-job. The owning task
207/// did not survive the restart, so the row must not stay stuck `fetching`.
208async fn reset_states_on_boot(pool: &SqlitePool) -> sqlx::Result<()> {
209 sqlx::query("UPDATE data_status SET state = 'idle', updated_at = ? WHERE state = 'fetching'")
210 .bind(now_ms())
211 .execute(pool)
212 .await?;
213 Ok(())
214}
215
216/// Reconcile the universe rows from the curated CSV on every boot — local only,
217/// no network. Upserts every listed symbol and prunes curated rows dropped from
218/// the CSV, so a `starter.csv` edit takes effect on deploy without a manual
219/// re-seed (user-added `is_seeded = 0` rows are never touched).
220///
221/// Unlike before the demand-only refocus, this no longer backfills any history:
222/// a symbol's deep daily history and SEC data fill in the first time its page is
223/// viewed and found stale (Phase B), or via an explicit `make seed` run.
224async fn run_boot_seed(pool: &SqlitePool, config: &Config) -> anyhow::Result<()> {
225 match seed::sync_universe(pool, config).await {
226 Ok(r) => tracing::info!(
227 "[scheduler] universe sync: {} symbols, {} pruned",
228 r.total,
229 r.pruned
230 ),
231 Err(e) => tracing::warn!("[scheduler] universe sync: {e:#}"),
232 }
233 Ok(())
234}
235
236/// Demand-driven intraday quote refresh.
237///
238/// Polls Yahoo only for the symbols a browser is currently viewing (the stream
239/// hub's interest registry). With nobody watching, `hub.viewed()` is empty and
240/// this returns at once having done no network work: the user's hard rule is
241/// to poll only what is on screen.
242///
243/// Which viewed symbols are polled depends on the session. Inside any trading
244/// session (pre, regular, post) every viewed symbol is fair game. Outside it,
245/// only viewed symbols that trade ~around the clock are polled — index futures,
246/// commodities, and crypto (BTC) — while indexes, stocks and ETFs sit frozen
247/// until the next session, so polling them off-hours would only re-fetch a flat
248/// quote. This is what keeps the dashboard's overview futures/commodity/BTC
249/// lines live overnight.
250///
251/// A clean run is recorded only in `data_status` (plus each `quotes.fetched_at`
252/// row); a `fetch_log` row is written only for a notable run, an error or a
253/// guard stop, so the minute-cadence job does not bury the log.
254async fn run_intraday(
255 pool: &SqlitePool,
256 config: &Config,
257 hub: &Hub,
258 session: market::Session,
259) -> anyhow::Result<()> {
260 let viewed = hub.viewed();
261 if viewed.is_empty() {
262 return Ok(());
263 }
264 // Inside a session, poll every viewed symbol; outside it, only the futures.
265 let mut targets: Vec<String> = if session.is_open() {
266 viewed
267 } else {
268 // Off-hours, only poll symbols that trade ~24h: index futures (and
269 // commodities) plus crypto (BTC). Cash indexes / stocks / ETFs are
270 // frozen, so polling them outside the session just burns budget.
271 let around_clock: HashSet<String> =
272 sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind IN ('future', 'crypto')")
273 .fetch_all(pool)
274 .await?
275 .into_iter()
276 .collect();
277 viewed
278 .into_iter()
279 .filter(|t| around_clock.contains(t))
280 .collect()
281 };
282 if targets.is_empty() {
283 return Ok(());
284 }
285 // Throttle to a ~5-minute per-symbol cadence: the loop
286 // ticks every 60s, but a symbol quoted within the last few minutes is left
287 // alone, so a dashboard left open polls each watchlist symbol about once
288 // every five minutes rather than every minute — light on the budget, and
289 // plenty "real-time" for delayed data. A symbol never quoted is always
290 // eligible. The set is small (only viewed symbols), so the lookup is cheap.
291 let throttle_cutoff = now_ms() - INTRADAY_MIN_INTERVAL_SECS * 1000;
292 let recent: HashSet<String> = sqlx::query_scalar(
293 "SELECT ticker FROM symbols WHERE last_quote_at IS NOT NULL AND last_quote_at >= ?",
294 )
295 .bind(throttle_cutoff)
296 .fetch_all(pool)
297 .await?
298 .into_iter()
299 .collect();
300 targets.retain(|t| !recent.contains(t));
301 if targets.is_empty() {
302 return Ok(());
303 }
304 let started = now_ms();
305 mark_fetching(pool, "intraday").await?;
306 notify_health(hub);
307
308 let yahoo = YahooProvider::new(providers::http::build_client(config));
309 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
310 let t0 = Instant::now();
311
312 let mut ok = 0i64;
313 let mut errors = 0i64;
314 let mut stopped: Option<String> = None;
315
316 for ticker in &targets {
317 match guard.acquire().await? {
318 Permit::Granted => {}
319 Permit::Denied(why) => {
320 stopped = Some(why);
321 break;
322 }
323 }
324 match yahoo.quote(ticker).await {
325 Ok(data) => {
326 guard.record_success().await?;
327 store_quote(pool, ticker, &data.quote).await?;
328 if !data.bars.is_empty() {
329 store_intraday(pool, ticker, &data.bars).await?;
330 }
331 hub.publish(StreamEvent::Quote(QuoteUpdate::new(
332 ticker.clone(),
333 data.quote.price,
334 data.quote.prev_close,
335 data.quote.market_state.clone(),
336 )));
337 ok += 1;
338 }
339 Err(e) => {
340 guard.record_failure(&e).await?;
341 errors += 1;
342 tracing::warn!("[scheduler] intraday {ticker} failed: {e:#}");
343 }
344 }
345 }
346
347 let dur = t0.elapsed().as_millis() as i64;
348 if let Some(why) = stopped {
349 let detail = format!("stopped early ({why}); {ok} ok, {errors} errors");
350 tracing::warn!("[scheduler] intraday: {detail}");
351 log_fetch(pool, "intraday", "yahoo", "skipped", Some(&detail), Some(ok), dur, started)
352 .await?;
353 mark_ok(pool, "intraday", None).await?;
354 } else if ok == 0 && errors > 0 {
355 let detail = format!("all {errors} quotes failed");
356 log_fetch(pool, "intraday", "yahoo", "error", Some(&detail), Some(0), dur, started).await?;
357 mark_error(pool, "intraday", &detail, None).await?;
358 } else {
359 if errors > 0 {
360 let detail = format!("{ok} ok, {errors} errors");
361 log_fetch(pool, "intraday", "yahoo", "ok", Some(&detail), Some(ok), dur, started)
362 .await?;
363 }
364 mark_ok(pool, "intraday", None).await?;
365 }
366
367 notify_health(hub);
368 Ok(())
369}
370
371/// The active home-dashboard sweep (a user call, 2026-06-10; see
372/// `HOME_SWEEP_INTERVAL_SECS`). Every 15 minutes it re-quotes the full
373/// dashboard set — the market overview's cash + futures tickers, the VIX and
374/// volume-proxy reads, and the union of every session's watchlist — without
375/// requiring a viewer, so the home page always opens on current figures.
376///
377/// Session-aware like `run_intraday`: inside any trading session everything is
378/// polled; outside it, only the instruments that trade ~around the clock
379/// (futures, crypto), since a frozen stock/ETF/index quote would just burn
380/// budget. Each pull rides `refresh_quotes`, so it is guard-routed, throttled
381/// per symbol (a quote fresher than ~5 minutes is skipped), and published to
382/// the hub so any open page live-ticks too.
383async fn run_home_sweep_if_due(
384 pool: &SqlitePool,
385 config: &Config,
386 hub: &Hub,
387 session: market::Session,
388 last: &mut Option<i64>,
389) -> anyhow::Result<()> {
390 let now = now_ms();
391 if let Some(t) = *last {
392 if (now - t) / 1000 < HOME_SWEEP_INTERVAL_SECS {
393 return Ok(());
394 }
395 }
396 *last = Some(now);
397
398 let mut targets: Vec<String> = crate::routes::home::dashboard_tickers()
399 .into_iter()
400 .map(str::to_string)
401 .collect();
402 let watchlisted: Vec<String> =
403 sqlx::query_scalar("SELECT DISTINCT ticker FROM watchlist ORDER BY ticker")
404 .fetch_all(pool)
405 .await?;
406 for t in watchlisted {
407 if !targets.contains(&t) {
408 targets.push(t);
409 }
410 }
411 if !session.is_open() {
412 let around_clock: HashSet<String> =
413 sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind IN ('future', 'crypto')")
414 .fetch_all(pool)
415 .await?
416 .into_iter()
417 .collect();
418 targets.retain(|t| around_clock.contains(t));
419 }
420 if targets.is_empty() {
421 return Ok(());
422 }
423
424 mark_fetching(pool, "home").await?;
425 notify_health(hub);
426 let refreshed = refresh_quotes(pool, config, hub, &targets).await;
427 if refreshed > 0 {
428 tracing::info!(
429 "[scheduler] home sweep: {refreshed}/{} refreshed",
430 targets.len()
431 );
432 }
433 mark_ok(pool, "home", Some(now + HOME_SWEEP_INTERVAL_SECS * 1000)).await?;
434 notify_health(hub);
435 Ok(())
436}
437
438/// Replace one stock's dividend history with what Yahoo returned. Yahoo serves
439/// the canonical, corrected history each call, so a `DELETE` + `INSERT` keeps
440/// the table honest if a payout is later retracted or restated. `pub(crate)`:
441/// the add-symbol backfill reuses it.
442pub(crate) async fn store_dividends(
443 pool: &SqlitePool,
444 ticker: &str,
445 events: &[DividendEvent],
446) -> sqlx::Result<()> {
447 let mut tx = pool.begin().await?;
448 sqlx::query("DELETE FROM dividends WHERE ticker = ?")
449 .bind(ticker)
450 .execute(&mut *tx)
451 .await?;
452 for e in events {
453 sqlx::query(
454 "INSERT INTO dividends (ticker, ex_date, amount) VALUES (?, ?, ?) \
455 ON CONFLICT(ticker, ex_date) DO UPDATE SET amount = excluded.amount",
456 )
457 .bind(ticker)
458 .bind(&e.ex_date)
459 .bind(e.amount)
460 .execute(&mut *tx)
461 .await?;
462 }
463 tx.commit().await?;
464 Ok(())
465}
466
467/// Stamp a stock as freshly dividend-synced.
468async fn mark_dividends_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
469 let now = now_ms();
470 sqlx::query("UPDATE symbols SET dividends_synced_at = ?, updated_at = ? WHERE ticker = ?")
471 .bind(now)
472 .bind(now)
473 .bind(ticker)
474 .execute(pool)
475 .await?;
476 Ok(())
477}
478
479/// Upsert one ETF's Yahoo `quoteSummary` metadata. Yahoo serves the full
480/// current snapshot each call, so the row is replaced wholesale: a field
481/// Yahoo no longer carries on a refresh becomes `NULL` here, rather than
482/// keeping a stale value. `pub(crate)`: the add-symbol backfill reuses it.
483pub(crate) async fn store_fund_metadata(
484 pool: &SqlitePool,
485 ticker: &str,
486 m: &FundMetadata,
487) -> sqlx::Result<()> {
488 sqlx::query(
489 "INSERT INTO fund_metadata \
490 (ticker, expense_ratio, yield_pct, trailing_yield_pct, nav_price, \
491 inception_date, category, fund_family, strategy_summary, updated_at) \
492 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
493 ON CONFLICT(ticker) DO UPDATE SET \
494 expense_ratio = excluded.expense_ratio, \
495 yield_pct = excluded.yield_pct, \
496 trailing_yield_pct = excluded.trailing_yield_pct, \
497 nav_price = excluded.nav_price, \
498 inception_date = excluded.inception_date, \
499 category = excluded.category, \
500 fund_family = excluded.fund_family, \
501 strategy_summary = excluded.strategy_summary, \
502 updated_at = excluded.updated_at",
503 )
504 .bind(ticker)
505 .bind(m.expense_ratio)
506 .bind(m.yield_pct)
507 .bind(m.trailing_yield_pct)
508 .bind(m.nav_price)
509 .bind(&m.inception_date)
510 .bind(&m.category)
511 .bind(&m.fund_family)
512 .bind(&m.strategy_summary)
513 .bind(now_ms())
514 .execute(pool)
515 .await?;
516 Ok(())
517}
518
519/// Stamp an ETF as freshly fund-metadata-synced. ETFs only — the column is
520/// `NULL` forever on every non-ETF row.
521async fn mark_fund_metadata_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
522 let now = now_ms();
523 sqlx::query(
524 "UPDATE symbols SET fund_metadata_synced_at = ?, updated_at = ? WHERE ticker = ?",
525 )
526 .bind(now)
527 .bind(now)
528 .bind(ticker)
529 .execute(pool)
530 .await?;
531 Ok(())
532}
533
534/// Write one stock's next-earnings date and stamp it as freshly synced.
535/// `next` is `None` when Yahoo has no upcoming date (the stored value is
536/// cleared so the page falls back to a cadence estimate). `pub(crate)`:
537/// the add-symbol backfill reuses it.
538pub(crate) async fn store_earnings_next(
539 pool: &SqlitePool,
540 ticker: &str,
541 next: Option<i64>,
542) -> sqlx::Result<()> {
543 let now = now_ms();
544 sqlx::query(
545 "UPDATE symbols SET next_earnings_at = ?, earnings_synced_at = ?, \
546 updated_at = ? \
547 WHERE ticker = ?",
548 )
549 .bind(next)
550 .bind(now)
551 .bind(now)
552 .bind(ticker)
553 .execute(pool)
554 .await?;
555 Ok(())
556}
557
558/// Write one stock's sector / industry classification and stamp it freshly
559/// synced. Either field may be `None` when Yahoo's coverage is partial; an
560/// empty / whitespace value is dropped at parse time, not stored. `pub(crate)`:
561/// the add-symbol backfill reuses it.
562pub(crate) async fn store_asset_profile(
563 pool: &SqlitePool,
564 ticker: &str,
565 profile: &crate::providers::AssetProfile,
566) -> sqlx::Result<()> {
567 let now = now_ms();
568 sqlx::query(
569 "UPDATE symbols SET sector = ?, industry = ?, \
570 asset_profile_synced_at = ?, updated_at = ? \
571 WHERE ticker = ?",
572 )
573 .bind(&profile.sector)
574 .bind(&profile.industry)
575 .bind(now)
576 .bind(now)
577 .bind(ticker)
578 .execute(pool)
579 .await?;
580 Ok(())
581}
582
583/// Stamp a stock as freshly asset-profile-synced without overwriting its
584/// stored sector / industry (used on a clean-empty Yahoo response so the
585/// sweep does not re-fetch).
586async fn mark_asset_profile_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
587 let now = now_ms();
588 sqlx::query(
589 "UPDATE symbols SET asset_profile_synced_at = ?, updated_at = ? WHERE ticker = ?",
590 )
591 .bind(now)
592 .bind(now)
593 .bind(ticker)
594 .execute(pool)
595 .await?;
596 Ok(())
597}
598
599/// Fill in `symbols.cik` for any stock found in the bulk SEC ticker map.
600/// Returns how many were newly resolved.
601async fn resolve_ciks(pool: &SqlitePool, map: &HashMap<String, String>) -> sqlx::Result<i64> {
602 let stocks: Vec<String> =
603 sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind = 'stock' AND cik IS NULL")
604 .fetch_all(pool)
605 .await?;
606 let mut resolved = 0;
607 for ticker in stocks {
608 if let Some(cik) = map.get(&crate::providers::sec::normalize_ticker(&ticker)) {
609 let now = now_ms();
610 sqlx::query("UPDATE symbols SET cik = ?, updated_at = ? WHERE ticker = ?")
611 .bind(cik)
612 .bind(now)
613 .bind(&ticker)
614 .execute(pool)
615 .await?;
616 resolved += 1;
617 }
618 }
619 Ok(resolved)
620}
621
622/// Fill in `symbols.cik` and `symbols.series_id` for any ETF found in the bulk
623/// SEC mutual-fund ticker map. Returns how many were newly resolved.
624async fn resolve_fund_ciks(
625 pool: &SqlitePool,
626 map: &HashMap<String, FundId>,
627) -> sqlx::Result<i64> {
628 let etfs: Vec<String> =
629 sqlx::query_scalar("SELECT ticker FROM symbols WHERE kind = 'etf' AND cik IS NULL")
630 .fetch_all(pool)
631 .await?;
632 let mut resolved = 0;
633 for ticker in etfs {
634 if let Some(id) = map.get(&crate::providers::sec::normalize_ticker(&ticker)) {
635 let now = now_ms();
636 sqlx::query(
637 "UPDATE symbols SET cik = ?, series_id = ?, updated_at = ? WHERE ticker = ?",
638 )
639 .bind(&id.cik)
640 .bind(&id.series_id)
641 .bind(now)
642 .bind(&ticker)
643 .execute(pool)
644 .await?;
645 resolved += 1;
646 }
647 }
648 Ok(resolved)
649}
650
651/// Stamp one of a symbol's SEC sync timestamps to now. `column` is one of a
652/// few hardcoded literals (never user input), so interpolating it is safe.
653async fn mark_sec_synced(pool: &SqlitePool, ticker: &str, column: &str) -> sqlx::Result<()> {
654 let now = now_ms();
655 let sql = format!("UPDATE symbols SET {column} = ?, updated_at = ? WHERE ticker = ?");
656 sqlx::query(&sql)
657 .bind(now)
658 .bind(now)
659 .bind(ticker)
660 .execute(pool)
661 .await?;
662 Ok(())
663}
664
665/// Upsert one company's fundamental facts. Keyed on (ticker, metric, period),
666/// so a later filing's restated figure overwrites the prior one.
667async fn store_fundamentals(pool: &SqlitePool, ticker: &str, facts: &[Fact]) -> sqlx::Result<()> {
668 let mut tx = pool.begin().await?;
669 for f in facts {
670 sqlx::query(
671 "INSERT INTO fundamentals \
672 (ticker, metric, period, fiscal_year, fiscal_qtr, period_end, \
673 value, unit, form, filed_at) \
674 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
675 ON CONFLICT(ticker, metric, period) DO UPDATE SET \
676 fiscal_year = excluded.fiscal_year, fiscal_qtr = excluded.fiscal_qtr, \
677 period_end = excluded.period_end, value = excluded.value, \
678 unit = excluded.unit, form = excluded.form, filed_at = excluded.filed_at",
679 )
680 .bind(ticker)
681 .bind(&f.metric)
682 .bind(&f.period)
683 .bind(f.fiscal_year)
684 .bind(f.fiscal_qtr)
685 .bind(&f.period_end)
686 .bind(f.value)
687 .bind(&f.unit)
688 .bind(&f.form)
689 .bind(&f.filed_at)
690 .execute(&mut *tx)
691 .await?;
692 }
693 tx.commit().await?;
694 Ok(())
695}
696
697/// Upsert one company's filing history. Keyed on (ticker, accession).
698async fn store_filings(
699 pool: &SqlitePool,
700 ticker: &str,
701 filings: &[FilingRecord],
702) -> sqlx::Result<()> {
703 let mut tx = pool.begin().await?;
704 for f in filings {
705 sqlx::query(
706 "INSERT INTO filings \
707 (ticker, accession, form, filed_at, period_of_report, \
708 primary_doc, url, description, items) \
709 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) \
710 ON CONFLICT(ticker, accession) DO UPDATE SET \
711 form = excluded.form, filed_at = excluded.filed_at, \
712 period_of_report = excluded.period_of_report, \
713 primary_doc = excluded.primary_doc, url = excluded.url, \
714 description = excluded.description, items = excluded.items",
715 )
716 .bind(ticker)
717 .bind(&f.accession)
718 .bind(&f.form)
719 .bind(&f.filed_at)
720 .bind(&f.period_of_report)
721 .bind(&f.primary_doc)
722 .bind(&f.url)
723 .bind(&f.description)
724 .bind(&f.items)
725 .execute(&mut *tx)
726 .await?;
727 }
728 tx.commit().await?;
729 Ok(())
730}
731
732/// Upsert a company's leadership roster from parsed ownership filings (Phase
733/// 14). `roster` arrives newest-filing-first, so the first entry seen for a
734/// person is their most recent filing and any later duplicate is skipped. The
735/// upsert is guarded on `last_seen`, so a stale re-parse never overwrites a
736/// person's role with an older filing's; departed insiders simply stop being
737/// re-stamped and age out of the symbol page's recency window.
738async fn store_leadership(
739 pool: &SqlitePool,
740 ticker: &str,
741 roster: &[(OwnershipPerson, String)],
742) -> sqlx::Result<()> {
743 let mut seen: HashSet<&str> = HashSet::new();
744 let mut tx = pool.begin().await?;
745 for (person, filed_at) in roster {
746 if !seen.insert(person.name.as_str()) {
747 continue;
748 }
749 sqlx::query(
750 "INSERT INTO leadership \
751 (ticker, name, is_director, is_officer, officer_title, last_seen) \
752 VALUES (?, ?, ?, ?, ?, ?) \
753 ON CONFLICT(ticker, name) DO UPDATE SET \
754 is_director = excluded.is_director, is_officer = excluded.is_officer, \
755 officer_title = excluded.officer_title, last_seen = excluded.last_seen \
756 WHERE excluded.last_seen >= leadership.last_seen",
757 )
758 .bind(ticker)
759 .bind(&person.name)
760 .bind(person.is_director as i64)
761 .bind(person.is_officer as i64)
762 .bind(&person.officer_title)
763 .bind(filed_at)
764 .execute(&mut *tx)
765 .await?;
766 }
767 tx.commit().await?;
768 Ok(())
769}
770
771/// Upsert one ETF's N-PORT fund profile and replace its stored holdings. The
772/// kept holdings are a small top slice, so they are deleted and re-inserted
773/// wholesale on each refresh.
774async fn store_fund_portfolio(
775 pool: &SqlitePool,
776 ticker: &str,
777 p: &PortfolioData,
778) -> anyhow::Result<()> {
779 // Asset / sector / geography mixes are each variable-length, so they ride
780 // in JSON columns rather than their own tables:
781 // [["Equity", 99.8], ["Cash & equivalents", 0.2], ...].
782 let asset_mix = serde_json::to_string(&p.asset_mix)?;
783 let sector_mix = serde_json::to_string(&p.sector_mix)?;
784 let geography_mix = serde_json::to_string(&p.geography_mix)?;
785 let mut tx = pool.begin().await?;
786 sqlx::query(
787 "INSERT INTO fund_profiles \
788 (ticker, kind, net_assets, total_assets, holdings_count, report_date, \
789 asset_mix, sector_mix, geography_mix, updated_at) \
790 VALUES (?, 'portfolio', ?, ?, ?, ?, ?, ?, ?, ?) \
791 ON CONFLICT(ticker) DO UPDATE SET \
792 kind = excluded.kind, net_assets = excluded.net_assets, \
793 total_assets = excluded.total_assets, holdings_count = excluded.holdings_count, \
794 report_date = excluded.report_date, asset_mix = excluded.asset_mix, \
795 sector_mix = excluded.sector_mix, geography_mix = excluded.geography_mix, \
796 updated_at = excluded.updated_at",
797 )
798 .bind(ticker)
799 .bind(p.net_assets)
800 .bind(p.total_assets)
801 .bind(p.holdings_count)
802 .bind(&p.report_date)
803 .bind(&asset_mix)
804 .bind(§or_mix)
805 .bind(&geography_mix)
806 .bind(now_ms())
807 .execute(&mut *tx)
808 .await?;
809 sqlx::query("DELETE FROM fund_holdings WHERE ticker = ?")
810 .bind(ticker)
811 .execute(&mut *tx)
812 .await?;
813 for (i, h) in p.top_holdings.iter().enumerate() {
814 sqlx::query(
815 "INSERT INTO fund_holdings (ticker, rank, name, pct, value_usd) \
816 VALUES (?, ?, ?, ?, ?)",
817 )
818 .bind(ticker)
819 .bind(i as i64 + 1)
820 .bind(&h.name)
821 .bind(h.pct)
822 .bind(h.value_usd)
823 .execute(&mut *tx)
824 .await?;
825 }
826 tx.commit().await?;
827 Ok(())
828}
829
830/// Record a physical-commodity grantor trust's profile: just the AUM read from
831/// its 10-K. It holds bullion, not a securities portfolio, so there are no
832/// holdings and no asset mix.
833async fn store_fund_commodity(
834 pool: &SqlitePool,
835 ticker: &str,
836 aum: Option<f64>,
837) -> sqlx::Result<()> {
838 sqlx::query(
839 "INSERT INTO fund_profiles \
840 (ticker, kind, net_assets, total_assets, holdings_count, report_date, \
841 asset_mix, updated_at) \
842 VALUES (?, 'commodity_trust', ?, NULL, NULL, NULL, NULL, ?) \
843 ON CONFLICT(ticker) DO UPDATE SET \
844 kind = excluded.kind, net_assets = excluded.net_assets, \
845 updated_at = excluded.updated_at",
846 )
847 .bind(ticker)
848 .bind(aum)
849 .bind(now_ms())
850 .execute(pool)
851 .await?;
852 Ok(())
853}
854
855/// Stamp an ETF's fund profile as freshly synced.
856async fn mark_fund_synced(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
857 let now = now_ms();
858 sqlx::query("UPDATE symbols SET fund_synced_at = ?, updated_at = ? WHERE ticker = ?")
859 .bind(now)
860 .bind(now)
861 .bind(ticker)
862 .execute(pool)
863 .await?;
864 Ok(())
865}
866
867/// Upsert one symbol's live quote into `quotes` and refresh the denormalized
868/// snapshot columns on `symbols` that the dashboard and SSE seeding read.
869/// `pub(crate)`: the add-symbol route stores the quote its Yahoo lookup
870/// already returned, rather than spending a second request.
871pub(crate) async fn store_quote(pool: &SqlitePool, ticker: &str, q: &Quote) -> sqlx::Result<()> {
872 let now = now_ms();
873 sqlx::query(
874 "INSERT INTO quotes \
875 (ticker, price, prev_close, open, day_high, day_low, volume, \
876 market_state, source, source_time, fetched_at) \
877 VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'yahoo', ?, ?) \
878 ON CONFLICT(ticker) DO UPDATE SET \
879 price = excluded.price, prev_close = excluded.prev_close, \
880 open = excluded.open, day_high = excluded.day_high, \
881 day_low = excluded.day_low, volume = excluded.volume, \
882 market_state = excluded.market_state, source = excluded.source, \
883 source_time = excluded.source_time, fetched_at = excluded.fetched_at",
884 )
885 .bind(ticker)
886 .bind(q.price)
887 .bind(q.prev_close)
888 .bind(q.open)
889 .bind(q.day_high)
890 .bind(q.day_low)
891 .bind(q.volume)
892 .bind(&q.market_state)
893 .bind(q.source_time)
894 .bind(now)
895 .execute(pool)
896 .await?;
897
898 sqlx::query(
899 "UPDATE symbols SET last_price = ?, prev_close = ?, last_quote_at = ?, \
900 updated_at = ? WHERE ticker = ?",
901 )
902 .bind(q.price)
903 .bind(q.prev_close)
904 .bind(now)
905 .bind(now)
906 .bind(ticker)
907 .execute(pool)
908 .await?;
909 Ok(())
910}
911
912/// Upsert one symbol's intraday bars in a single transaction. The prune job
913/// trims `intraday_bars` to a rolling ~14-day window, so nothing here grows
914/// without bound. `pub(crate)`: also called by the add-symbol route.
915pub(crate) async fn store_intraday(
916 pool: &SqlitePool,
917 ticker: &str,
918 bars: &[IntradayBar],
919) -> sqlx::Result<()> {
920 let mut tx = pool.begin().await?;
921 for b in bars {
922 sqlx::query(
923 "INSERT INTO intraday_bars (ticker, ts, open, high, low, close, volume) \
924 VALUES (?, ?, ?, ?, ?, ?, ?) \
925 ON CONFLICT(ticker, ts) DO UPDATE SET \
926 open = excluded.open, high = excluded.high, low = excluded.low, \
927 close = excluded.close, volume = excluded.volume",
928 )
929 .bind(ticker)
930 .bind(b.ts)
931 .bind(b.open)
932 .bind(b.high)
933 .bind(b.low)
934 .bind(b.close)
935 .bind(b.volume)
936 .execute(&mut *tx)
937 .await?;
938 }
939 tx.commit().await?;
940 Ok(())
941}
942
943// ── synchronous backfill for a freshly-added symbol (Phase 21) ─────────────
944
945/// Run one guarded outbound call: acquire a permit, await `call`, and feed the
946/// outcome back to the guard. `None` when the guard denied the request (the
947/// breaker is open or the hourly budget is spent) or the permit could not be
948/// acquired; `Some` carries whatever the call itself returned.
949async fn guarded<T>(
950 guard: &EndpointGuard,
951 call: impl std::future::Future<Output = anyhow::Result<T>>,
952) -> Option<anyhow::Result<T>> {
953 match guard.acquire().await {
954 Ok(Permit::Granted) => {
955 let result = call.await;
956 match &result {
957 Ok(_) => {
958 let _ = guard.record_success().await;
959 }
960 Err(e) => {
961 let _ = guard.record_failure(e).await;
962 }
963 }
964 Some(result)
965 }
966 Ok(Permit::Denied(why)) => {
967 tracing::info!("[backfill] guard denied: {why}");
968 None
969 }
970 Err(e) => {
971 tracing::warn!("[backfill] guard error: {e:#}");
972 None
973 }
974 }
975}
976
977/// Synchronously backfill everything for one just-added symbol: its deep daily
978/// history from Stooq and, for a stock or ETF, its SEC data. The add-symbol
979/// route (`routes::symbols`) calls this so a user-added symbol's page is
980/// complete the moment the add returns, rather than filling in over later
981/// scheduler cycles.
982///
983/// Best-effort and guard-routed: every outbound call passes through the same
984/// `EndpointGuard` the background jobs use, and a guard denial or upstream
985/// error for any one piece is logged and skipped. The symbol is already added;
986/// the normal scheduler sweeps pick up whatever this run missed.
987pub(crate) async fn backfill_symbol(pool: &SqlitePool, config: &Config, ticker: &str, kind: &str) {
988 backfill_history(pool, config, ticker, kind).await;
989 // Phase 26 + Phase 28: dividends covers stock dividends and ETF
990 // distributions (same Yahoo event series). Rides the Yahoo guard, so it
991 // runs independently of the SEC contact-email gate below.
992 if kind == "stock" || kind == "etf" {
993 backfill_dividends(pool, config, ticker).await;
994 }
995 // Phase 28: ETFs get their Yahoo fund_metadata pulled too — expense
996 // ratio, yield, NAV, inception, category, family, strategy summary.
997 if kind == "etf" {
998 backfill_fund_metadata(pool, config, ticker).await;
999 }
1000 // Phase 25: stocks get their Yahoo earnings calendar pulled too — so a
1001 // user-added stock's symbol page carries the next-expected earnings
1002 // date the moment the add returns, rather than waiting on the next
1003 // scheduler cycle.
1004 if kind == "stock" {
1005 backfill_earnings_calendar(pool, config, ticker).await;
1006 }
1007 // Phase 15: stocks get their Yahoo assetProfile pulled too — so a
1008 // user-added stock immediately shows up under its sector and industry
1009 // on /industries and carries the symbol-page header tag, rather than
1010 // waiting on the next monthly scheduler sweep.
1011 if kind == "stock" {
1012 backfill_asset_profile(pool, config, ticker).await;
1013 }
1014
1015 // SEC data covers stocks and ETFs; indexes and futures do not file. The
1016 // whole SEC step is skipped with no contact email configured, as `run_sec`
1017 // skips itself.
1018 if config.sec_contact_email.is_empty() {
1019 return;
1020 }
1021 let sec = SecProvider::new(providers::http::build_sec_client(config));
1022 let guard = EndpointGuard::with_budget(pool.clone(), sec.name(), SEC_BUDGET);
1023 match kind {
1024 "stock" => backfill_stock_sec(pool, &sec, &guard, ticker).await,
1025 "etf" => backfill_etf_sec(pool, &sec, &guard, ticker).await,
1026 _ => {}
1027 }
1028}
1029
1030/// Pull and store a freshly-added ETF's Yahoo `quoteSummary` metadata (Phase
1031/// 28). Mirrors `backfill_dividends`: same `yahoo` guard, best-effort, no
1032/// failure propagated to the add-symbol response.
1033async fn backfill_fund_metadata(pool: &SqlitePool, config: &Config, ticker: &str) {
1034 let yahoo = YahooProvider::new(providers::http::build_client(config));
1035 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1036 match guarded(&guard, yahoo.fund_metadata(ticker)).await {
1037 Some(Ok(Some(meta))) => match store_fund_metadata(pool, ticker, &meta).await {
1038 Ok(()) => {
1039 let _ = mark_fund_metadata_synced(pool, ticker).await;
1040 tracing::info!("[backfill] {ticker} <- fund_metadata");
1041 }
1042 Err(e) => tracing::warn!("[backfill] store fund_metadata {ticker}: {e:#}"),
1043 },
1044 // Yahoo answered cleanly but had no fund modules for this ETF — stamp
1045 // it checked so the next sweep does not re-fetch the same empty.
1046 Some(Ok(None)) => {
1047 let _ = mark_fund_metadata_synced(pool, ticker).await;
1048 }
1049 Some(Err(e)) => tracing::warn!("[backfill] fund_metadata {ticker}: {e:#}"),
1050 None => {}
1051 }
1052}
1053
1054/// Pull and store a freshly-added stock's sector / industry classification
1055/// (Phase 15). Mirrors `backfill_earnings_calendar`: same `yahoo` guard,
1056/// best-effort, no failure propagated to the add-symbol response.
1057async fn backfill_asset_profile(pool: &SqlitePool, config: &Config, ticker: &str) {
1058 let yahoo = YahooProvider::new(providers::http::build_client(config));
1059 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1060 match guarded(&guard, yahoo.asset_profile(ticker)).await {
1061 Some(Ok(Some(profile))) => match store_asset_profile(pool, ticker, &profile).await {
1062 Ok(()) => {
1063 tracing::info!(
1064 "[backfill] {ticker} <- asset_profile ({} / {})",
1065 profile.sector.as_deref().unwrap_or("—"),
1066 profile.industry.as_deref().unwrap_or("—"),
1067 );
1068 }
1069 Err(e) => tracing::warn!("[backfill] store asset_profile {ticker}: {e:#}"),
1070 },
1071 // Yahoo answered cleanly but had no profile for this stock — stamp
1072 // it so the next sweep does not re-fetch the same empty.
1073 Some(Ok(None)) => {
1074 let _ = mark_asset_profile_synced(pool, ticker).await;
1075 }
1076 Some(Err(e)) => tracing::warn!("[backfill] asset_profile {ticker}: {e:#}"),
1077 None => {}
1078 }
1079}
1080
1081/// Pull and store a freshly-added stock's next-expected earnings date
1082/// (Phase 25). Mirrors `backfill_dividends`: same `yahoo` guard,
1083/// best-effort, no failure propagated to the add-symbol response.
1084async fn backfill_earnings_calendar(pool: &SqlitePool, config: &Config, ticker: &str) {
1085 let yahoo = YahooProvider::new(providers::http::build_client(config));
1086 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1087 match guarded(&guard, yahoo.earnings_calendar(ticker)).await {
1088 Some(Ok(next)) => match store_earnings_next(pool, ticker, next).await {
1089 Ok(()) => {
1090 tracing::info!(
1091 "[backfill] {ticker} <- earnings_calendar ({})",
1092 next.map(|_| "next set").unwrap_or("no upcoming date"),
1093 );
1094 }
1095 Err(e) => tracing::warn!("[backfill] store earnings {ticker}: {e:#}"),
1096 },
1097 Some(Err(e)) => tracing::warn!("[backfill] earnings_calendar {ticker}: {e:#}"),
1098 None => {}
1099 }
1100}
1101
1102/// Pull and store a freshly-added symbol's dividend / distribution history
1103/// (Phase 26 + Phase 28: stocks and ETFs both use this path). Routed through
1104/// the same `yahoo` guard the dividends sweep uses. Best-effort: a guard
1105/// denial or upstream error leaves the symbol for the next normal sweep.
1106async fn backfill_dividends(pool: &SqlitePool, config: &Config, ticker: &str) {
1107 let yahoo = YahooProvider::new(providers::http::build_client(config));
1108 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1109 match guarded(&guard, yahoo.dividends(ticker)).await {
1110 Some(Ok(events)) => match store_dividends(pool, ticker, &events).await {
1111 Ok(()) => {
1112 let _ = mark_dividends_synced(pool, ticker).await;
1113 tracing::info!("[backfill] {ticker} <- {} dividends", events.len());
1114 }
1115 Err(e) => tracing::warn!("[backfill] store dividends {ticker}: {e:#}"),
1116 },
1117 Some(Err(e)) => tracing::warn!("[backfill] dividends {ticker}: {e:#}"),
1118 None => {}
1119 }
1120}
1121
1122/// Pull and store one symbol's deep daily history from Yahoo (one
1123/// `interval=1d&range=max` call). Used by the add-symbol backfill; all kinds
1124/// are eligible since Yahoo serves `=F` futures history too.
1125async fn backfill_history(pool: &SqlitePool, config: &Config, ticker: &str, _kind: &str) {
1126 let yahoo = YahooProvider::new(providers::http::build_client(config));
1127 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1128 match guarded(&guard, yahoo.daily(ticker, None)).await {
1129 Some(Ok(bars)) if !bars.is_empty() => match seed::store_daily(pool, ticker, &bars).await {
1130 Ok(()) => tracing::info!("[backfill] {ticker} <- {} daily bars", bars.len()),
1131 Err(e) => tracing::warn!("[backfill] store history {ticker}: {e:#}"),
1132 },
1133 // A valid but empty response (a historyless symbol): stamp it checked
1134 // so the history job does not immediately re-fetch it.
1135 Some(Ok(_)) => {
1136 let _ = mark_history_checked(pool, ticker).await;
1137 }
1138 _ => {}
1139 }
1140}
1141
1142/// Backfill a stock's SEC data: resolve its CIK, then pull fundamentals,
1143/// filings, and the officer/board roster.
1144async fn backfill_stock_sec(
1145 pool: &SqlitePool,
1146 sec: &SecProvider,
1147 guard: &EndpointGuard,
1148 ticker: &str,
1149) {
1150 let cik = match resolve_one_cik(pool, sec, guard, ticker, false).await {
1151 CikResolution::Found(c) => c,
1152 CikResolution::Absent => {
1153 // Not in SEC's company map: a non-filer, a foreign issuer, or a
1154 // delisted/renamed ticker. Stamp the sections checked so the page
1155 // shows an honest "no data" rather than a perpetual pending note.
1156 let _ = mark_sec_synced(pool, ticker, "fundamentals_synced_at").await;
1157 let _ = mark_sec_synced(pool, ticker, "filings_synced_at").await;
1158 tracing::info!("[backfill] {ticker}: not in SEC company map, marked checked");
1159 return;
1160 }
1161 CikResolution::Unavailable => {
1162 tracing::info!("[backfill] {ticker}: SEC CIK map unavailable, leaving it for the sec job");
1163 return;
1164 }
1165 };
1166 if let Some(Ok(facts)) = guarded(guard, sec.facts(&cik)).await {
1167 match store_fundamentals(pool, ticker, &facts).await {
1168 Ok(()) => {
1169 let _ = mark_sec_synced(pool, ticker, "fundamentals_synced_at").await;
1170 }
1171 Err(e) => tracing::warn!("[backfill] store facts {ticker}: {e:#}"),
1172 }
1173 }
1174 if let Some(Ok(filings)) = guarded(guard, sec.filings(&cik)).await {
1175 match store_filings(pool, ticker, &filings).await {
1176 Ok(()) => {
1177 let _ = mark_sec_synced(pool, ticker, "filings_synced_at").await;
1178 }
1179 Err(e) => tracing::warn!("[backfill] store filings {ticker}: {e:#}"),
1180 }
1181 }
1182 backfill_leadership(pool, sec, guard, ticker, &cik).await;
1183}
1184
1185/// Backfill a stock's officer/board roster from a window of its most recent
1186/// Form 3/4/5 ownership filings, mirroring the leadership sweep in `run_sec`.
1187async fn backfill_leadership(
1188 pool: &SqlitePool,
1189 sec: &SecProvider,
1190 guard: &EndpointGuard,
1191 ticker: &str,
1192 cik: &str,
1193) {
1194 let Some(Ok(index)) = guarded(guard, sec.ownership_index(cik)).await else {
1195 return;
1196 };
1197 let to_parse: Vec<_> = index.into_iter().take(LEADERSHIP_MAX_FILINGS).collect();
1198
1199 let mut roster: Vec<(OwnershipPerson, String)> = Vec::new();
1200 let mut complete = true;
1201 for f in &to_parse {
1202 match guarded(guard, sec.ownership_doc(cik, &f.accession, &f.primary_doc)).await {
1203 Some(Ok(people)) => {
1204 for p in people {
1205 if p.is_director || p.is_officer {
1206 roster.push((p, f.filed_at.clone()));
1207 }
1208 }
1209 }
1210 // A parse or network error for one filing: skip it and build the
1211 // roster from the rest, exactly as `run_sec` does.
1212 Some(Err(e)) => tracing::warn!("[backfill] ownership_doc {ticker}: {e:#}"),
1213 // A guard denial leaves the roster only partial: leave it unsynced
1214 // so the next `sec` cycle finishes it.
1215 None => complete = false,
1216 }
1217 }
1218 let _ = store_leadership(pool, ticker, &roster).await;
1219 if complete {
1220 let _ = mark_sec_synced(pool, ticker, "leadership_synced_at").await;
1221 }
1222}
1223
1224/// Backfill an ETF's fund profile: resolve its fund CIK, pull the filing list,
1225/// then either the N-PORT portfolio or a commodity trust's AUM.
1226async fn backfill_etf_sec(pool: &SqlitePool, sec: &SecProvider, guard: &EndpointGuard, ticker: &str) {
1227 let cik = match resolve_one_cik(pool, sec, guard, ticker, true).await {
1228 CikResolution::Found(c) => c,
1229 CikResolution::Absent => {
1230 // Not in SEC's mutual-fund map: a delisted/renamed fund (e.g. SPCX,
1231 // which renamed to SPCK) or one that does not file N-PORT. Stamp it
1232 // checked so the page shows an honest "no fund profile available"
1233 // rather than a pending note Refresh can never clear.
1234 let _ = mark_fund_synced(pool, ticker).await;
1235 tracing::info!("[backfill] {ticker}: not in SEC fund map (delisted/renamed?), marked checked");
1236 return;
1237 }
1238 CikResolution::Unavailable => {
1239 tracing::info!("[backfill] {ticker}: SEC fund map unavailable, leaving it for the sec job");
1240 return;
1241 }
1242 };
1243 // `resolve_fund_ciks` stored the series id alongside the CIK.
1244 let series_id: Option<String> =
1245 sqlx::query_scalar("SELECT series_id FROM symbols WHERE ticker = ?")
1246 .bind(ticker)
1247 .fetch_one(pool)
1248 .await
1249 .ok()
1250 .flatten();
1251 let id = FundId {
1252 cik: cik.clone(),
1253 series_id,
1254 };
1255
1256 let Some(Ok(ff)) = guarded(guard, sec.fund_filings(&id)).await else {
1257 return;
1258 };
1259 let _ = store_filings(pool, ticker, &ff.filings).await;
1260 match ff.shape {
1261 FundShape::Portfolio { nport_href } => {
1262 if let Some(Ok(portfolio)) = guarded(guard, sec.fund_portfolio(&nport_href)).await {
1263 if store_fund_portfolio(pool, ticker, &portfolio).await.is_ok() {
1264 let _ = mark_fund_synced(pool, ticker).await;
1265 }
1266 }
1267 }
1268 FundShape::CommodityTrust => {
1269 if let Some(Ok(aum)) = guarded(guard, sec.fund_aum(&cik)).await {
1270 if store_fund_commodity(pool, ticker, aum).await.is_ok() {
1271 let _ = mark_fund_synced(pool, ticker).await;
1272 }
1273 }
1274 }
1275 FundShape::Unknown => {
1276 let _ = mark_fund_synced(pool, ticker).await;
1277 }
1278 }
1279}
1280
1281/// Outcome of resolving a symbol's SEC CIK from the bulk ticker map.
1282enum CikResolution {
1283 /// A CIK is on file for this symbol (resolved now or on a prior run).
1284 Found(String),
1285 /// The bulk map was fetched successfully but does not list this ticker:
1286 /// the company/fund genuinely has no SEC entry (delisted, renamed, a
1287 /// foreign issuer, or a non-filer). The caller stamps the affected section
1288 /// *checked* so the page shows an honest "no data available" instead of a
1289 /// perpetual "not synced yet, hit Refresh".
1290 Absent,
1291 /// The map could not be fetched (guard denied / network error): nothing was
1292 /// learned, so the caller leaves the section unsynced for a later retry.
1293 Unavailable,
1294}
1295
1296/// Resolve and store a freshly-added symbol's SEC CIK from the bulk ticker map.
1297/// `fund` selects the mutual-fund map (ETFs) over the operating-company map
1298/// (stocks). Distinguishes a genuinely-absent ticker from an unreachable map so
1299/// the caller can render an honest empty state (see [`CikResolution`]).
1300async fn resolve_one_cik(
1301 pool: &SqlitePool,
1302 sec: &SecProvider,
1303 guard: &EndpointGuard,
1304 ticker: &str,
1305 fund: bool,
1306) -> CikResolution {
1307 // Whether the bulk map was actually fetched this call (vs guard-denied /
1308 // errored). A symbol may already carry a CIK from a prior run regardless.
1309 let fetched = if fund {
1310 match guarded(guard, sec.fund_ticker_map()).await {
1311 Some(Ok(map)) => {
1312 let _ = resolve_fund_ciks(pool, &map).await;
1313 true
1314 }
1315 _ => false,
1316 }
1317 } else {
1318 match guarded(guard, sec.cik_map()).await {
1319 Some(Ok(map)) => {
1320 let _ = resolve_ciks(pool, &map).await;
1321 true
1322 }
1323 _ => false,
1324 }
1325 };
1326 let cik: Option<String> =
1327 sqlx::query_scalar::<_, Option<String>>("SELECT cik FROM symbols WHERE ticker = ?")
1328 .bind(ticker)
1329 .fetch_one(pool)
1330 .await
1331 .ok()
1332 .flatten();
1333 match (cik, fetched) {
1334 (Some(c), _) => CikResolution::Found(c),
1335 (None, true) => CikResolution::Absent,
1336 (None, false) => CikResolution::Unavailable,
1337 }
1338}
1339
1340/// Prune aged rows once per `PRUNE_INTERVAL_SECS`. `intraday_bars` keeps a
1341/// rolling ~14-day window; `fetch_log` keeps ~30 days. `daily_prices` is
1342/// permanent and never touched here.
1343async fn run_prune_if_due(
1344 pool: &SqlitePool,
1345 last: &mut Option<i64>,
1346 hub: &Hub,
1347) -> anyhow::Result<()> {
1348 let now = now_ms();
1349 if let Some(t) = *last {
1350 if (now - t) / 1000 < PRUNE_INTERVAL_SECS {
1351 return Ok(());
1352 }
1353 }
1354
1355 let t0 = Instant::now();
1356 let intraday_cutoff = now - INTRADAY_RETENTION_DAYS * 86_400 * 1000;
1357 let log_cutoff = now - FETCH_LOG_RETENTION_DAYS * 86_400 * 1000;
1358
1359 let bars = sqlx::query("DELETE FROM intraday_bars WHERE ts < ?")
1360 .bind(intraday_cutoff)
1361 .execute(pool)
1362 .await?
1363 .rows_affected();
1364 let logs = sqlx::query("DELETE FROM fetch_log WHERE started_at < ?")
1365 .bind(log_cutoff)
1366 .execute(pool)
1367 .await?
1368 .rows_affected();
1369 // Watchlist sids whose newest row is >13 months old belong to expired
1370 // cookies: fin_sid lives 12 months from mint and is never refreshed, and
1371 // every added_at falls within that lifetime, so 13 months of silence means
1372 // the sid can never come back. Without this, every visitor ever leaves 5
1373 // starter rows behind forever.
1374 let wl_cutoff = now - WATCHLIST_STALE_DAYS * 86_400 * 1000;
1375 let sids = sqlx::query(
1376 "DELETE FROM watchlist WHERE sid IN \
1377 (SELECT sid FROM watchlist GROUP BY sid HAVING MAX(added_at) < ?)",
1378 )
1379 .bind(wl_cutoff)
1380 .execute(pool)
1381 .await?
1382 .rows_affected();
1383
1384 let dur = t0.elapsed().as_millis() as i64;
1385 let detail =
1386 format!("{bars} intraday bars, {logs} fetch_log rows, {sids} stale watchlist rows");
1387 tracing::info!("[scheduler] prune: removed {detail}");
1388 log_fetch(pool, "prune", "-", "ok", Some(&detail), Some((bars + logs) as i64), dur, now).await?;
1389
1390 *last = Some(now);
1391 notify_health(hub);
1392 Ok(())
1393}
1394
1395// ── data_status / fetch_log helpers ───────────────────────────────────────
1396
1397/// Nudge any connected `/health` page to pull a fresh snapshot. Sent whenever a
1398/// job changes state or appends a `fetch_log` row, so the data-health page
1399/// tracks the worker in near real time. Carries no payload (see `StreamEvent`).
1400fn notify_health(hub: &Hub) {
1401 hub.publish(StreamEvent::Health);
1402}
1403
1404/// Move a job's `data_status` row to the `fetching` state.
1405async fn mark_fetching(pool: &SqlitePool, job: &str) -> sqlx::Result<()> {
1406 let now = now_ms();
1407 sqlx::query(
1408 "INSERT INTO data_status (job, state, updated_at) VALUES (?, 'fetching', ?) \
1409 ON CONFLICT(job) DO UPDATE SET state = 'fetching', updated_at = excluded.updated_at",
1410 )
1411 .bind(job)
1412 .bind(now)
1413 .execute(pool)
1414 .await?;
1415 Ok(())
1416}
1417
1418/// Mark a job finished-OK, recording when it next falls due (`None` for
1419/// one-shot jobs like the seed).
1420async fn mark_ok(pool: &SqlitePool, job: &str, next_run_at: Option<i64>) -> sqlx::Result<()> {
1421 let now = now_ms();
1422 sqlx::query(
1423 "INSERT INTO data_status (job, state, last_ok_at, next_run_at, updated_at) \
1424 VALUES (?, 'ok', ?, ?, ?) \
1425 ON CONFLICT(job) DO UPDATE SET \
1426 state = 'ok', last_ok_at = excluded.last_ok_at, \
1427 next_run_at = excluded.next_run_at, updated_at = excluded.updated_at",
1428 )
1429 .bind(job)
1430 .bind(now)
1431 .bind(next_run_at)
1432 .bind(now)
1433 .execute(pool)
1434 .await?;
1435 Ok(())
1436}
1437
1438/// Mark a job failed, recording the error and when it should be retried.
1439async fn mark_error(
1440 pool: &SqlitePool,
1441 job: &str,
1442 msg: &str,
1443 next_run_at: Option<i64>,
1444) -> sqlx::Result<()> {
1445 let now = now_ms();
1446 sqlx::query(
1447 "INSERT INTO data_status (job, state, last_error, last_error_at, next_run_at, updated_at) \
1448 VALUES (?, 'error', ?, ?, ?, ?) \
1449 ON CONFLICT(job) DO UPDATE SET \
1450 state = 'error', last_error = excluded.last_error, \
1451 last_error_at = excluded.last_error_at, next_run_at = excluded.next_run_at, \
1452 updated_at = excluded.updated_at",
1453 )
1454 .bind(job)
1455 .bind(msg)
1456 .bind(now)
1457 .bind(next_run_at)
1458 .bind(now)
1459 .execute(pool)
1460 .await?;
1461 Ok(())
1462}
1463
1464/// Stamp a symbol as history-checked without storing bars: used when the
1465/// upstream returned a valid response that simply held nothing new.
1466async fn mark_history_checked(pool: &SqlitePool, ticker: &str) -> sqlx::Result<()> {
1467 let now = now_ms();
1468 sqlx::query("UPDATE symbols SET history_synced_at = ?, updated_at = ? WHERE ticker = ?")
1469 .bind(now)
1470 .bind(now)
1471 .bind(ticker)
1472 .execute(pool)
1473 .await?;
1474 Ok(())
1475}
1476
1477/// Append one `fetch_log` row. `ticker` is left NULL: these are bulk jobs, so
1478/// a run logs once rather than once per symbol.
1479#[allow(clippy::too_many_arguments)]
1480async fn log_fetch(
1481 pool: &SqlitePool,
1482 job: &str,
1483 provider: &str,
1484 status: &str,
1485 detail: Option<&str>,
1486 rows: Option<i64>,
1487 duration_ms: i64,
1488 started_at: i64,
1489) -> sqlx::Result<()> {
1490 sqlx::query(
1491 "INSERT INTO fetch_log \
1492 (job, provider, ticker, status, detail, rows, duration_ms, started_at, finished_at) \
1493 VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?)",
1494 )
1495 .bind(job)
1496 .bind(provider)
1497 .bind(status)
1498 .bind(detail)
1499 .bind(rows)
1500 .bind(duration_ms)
1501 .bind(started_at)
1502 .bind(now_ms())
1503 .execute(pool)
1504 .await?;
1505 Ok(())
1506}
1507
1508// ───────────────────── on-demand refresh pipeline (Phase B) ─────────────────
1509//
1510// Since the demand-only refocus there are no timed sweeps: a viewed symbol's
1511// data is pulled here, on demand, when its page loads (the fast price steps
1512// always run; the slow SEC / metadata steps run only when their stored copy is
1513// stale) or when the user hits Refresh (`force`, which runs everything). The
1514// symbol-page SSE route (`routes::symbols::refresh_stream`) drives it: it asks
1515// `refresh_plan` which steps will run, then runs each via `refresh_step`,
1516// streaming progress to the page's loading bar. Each step reuses the same
1517// guarded `backfill_*` helpers the add-symbol flow already uses.
1518
1519/// On-demand staleness windows for the gated (slow) steps. The always-run price
1520/// steps (quote + history) carry no window — they run on every load.
1521const REFRESH_SEC_STALE_SECS: i64 = 7 * 24 * 3600;
1522const REFRESH_LEADERSHIP_STALE_SECS: i64 = 30 * 24 * 3600;
1523const REFRESH_META_STALE_SECS: i64 = 7 * 24 * 3600;
1524
1525/// One step in a symbol's refresh, shown on the page's loading bar.
1526pub(crate) struct RefreshStep {
1527 /// Stable key the route passes back to `refresh_step`.
1528 pub key: &'static str,
1529 /// Human label for the loading bar.
1530 pub label: &'static str,
1531 /// Whether this refreshes a server-rendered "deep" section — if any deep
1532 /// step ran, the page reloads to show it; a load that runs only the (live)
1533 /// price steps patches the price in place instead.
1534 pub deep: bool,
1535}
1536
1537const fn step(key: &'static str, label: &'static str, deep: bool) -> RefreshStep {
1538 RefreshStep { key, label, deep }
1539}
1540
1541/// Decide which steps a symbol's refresh will run. The two price steps always
1542/// run; the slow steps are included only when stale (or `force`). An index /
1543/// future / unknown kind gets just the price steps.
1544pub(crate) async fn refresh_plan(
1545 pool: &SqlitePool,
1546 config: &Config,
1547 ticker: &str,
1548 kind: &str,
1549 force: bool,
1550) -> Vec<RefreshStep> {
1551 let mut steps = vec![
1552 step("quote", "Live quote", false),
1553 step("history", "Daily history", false),
1554 ];
1555 let Some(s) =
1556 sqlx::query_as::<_, crate::models::SymbolRow>("SELECT * FROM symbols WHERE ticker = ?")
1557 .bind(ticker)
1558 .fetch_optional(pool)
1559 .await
1560 .ok()
1561 .flatten()
1562 else {
1563 return steps;
1564 };
1565 let now = now_ms();
1566 let stale = |at: Option<i64>, secs: i64| force || at.map_or(true, |t| now - t > secs * 1000);
1567 let sec_ok = !config.sec_contact_email.is_empty();
1568 match kind {
1569 "stock" => {
1570 if sec_ok
1571 && (stale(s.fundamentals_synced_at, REFRESH_SEC_STALE_SECS)
1572 || stale(s.filings_synced_at, REFRESH_SEC_STALE_SECS)
1573 || stale(s.leadership_synced_at, REFRESH_LEADERSHIP_STALE_SECS))
1574 {
1575 steps.push(step("sec", "Fundamentals, filings & leadership", true));
1576 }
1577 if stale(s.earnings_synced_at, REFRESH_META_STALE_SECS) {
1578 steps.push(step("earnings", "Earnings date", true));
1579 }
1580 if stale(s.asset_profile_synced_at, REFRESH_META_STALE_SECS) {
1581 steps.push(step("profile", "Sector & industry", true));
1582 }
1583 if stale(s.dividends_synced_at, REFRESH_META_STALE_SECS) {
1584 steps.push(step("dividends", "Dividends", true));
1585 }
1586 }
1587 "etf" => {
1588 if sec_ok && stale(s.fund_synced_at, REFRESH_SEC_STALE_SECS) {
1589 steps.push(step("fund_sec", "Holdings & filings", true));
1590 }
1591 if stale(s.fund_metadata_synced_at, REFRESH_META_STALE_SECS) {
1592 steps.push(step("fund_meta", "Fund details & NAV", true));
1593 }
1594 if stale(s.dividends_synced_at, REFRESH_META_STALE_SECS) {
1595 steps.push(step("dividends", "Distributions", true));
1596 }
1597 }
1598 _ => {}
1599 }
1600 steps
1601}
1602
1603/// Run one refresh step by key. Returns a short status for the loading bar:
1604/// "ok" when it ran, "skipped" when the guard denied it (breaker open / budget
1605/// spent). The backfill helpers are best-effort and swallow their own errors,
1606/// so the deep steps report "ok" once attempted; the price steps, which this
1607/// runs inline through the guard, distinguish a guard denial.
1608pub(crate) async fn refresh_step(
1609 pool: &SqlitePool,
1610 config: &Config,
1611 hub: &Hub,
1612 ticker: &str,
1613 kind: &str,
1614 key: &str,
1615) -> &'static str {
1616 let _ = kind;
1617 match key {
1618 "quote" => refresh_quote(pool, config, hub, ticker).await,
1619 "history" => refresh_history_incremental(pool, config, ticker).await,
1620 "sec" => refresh_sec(pool, config, ticker, false).await,
1621 "fund_sec" => refresh_sec(pool, config, ticker, true).await,
1622 "earnings" => {
1623 backfill_earnings_calendar(pool, config, ticker).await;
1624 "ok"
1625 }
1626 "profile" => {
1627 backfill_asset_profile(pool, config, ticker).await;
1628 "ok"
1629 }
1630 "dividends" => {
1631 backfill_dividends(pool, config, ticker).await;
1632 "ok"
1633 }
1634 "fund_meta" => refresh_fund_meta(pool, config, ticker).await,
1635 _ => "ok",
1636 }
1637}
1638
1639/// Pull fresh quotes for a set of dashboard symbols on demand — the dashboard's
1640/// on-open refresh. A symbol quoted within the last few
1641/// minutes is skipped, so a reload (or the add/remove reload) does not re-hit
1642/// Yahoo, and overnight the gate keeps a re-open from re-polling the same frozen
1643/// close. Runs regardless of session: opening the dashboard after the close
1644/// should still confirm the latest (closing) prices rather than show a stale
1645/// snapshot. Each fetch publishes to the hub so open cards live-tick. Returns how
1646/// many symbols were actually refreshed.
1647pub(crate) async fn refresh_quotes(
1648 pool: &SqlitePool,
1649 config: &Config,
1650 hub: &Hub,
1651 tickers: &[String],
1652) -> usize {
1653 let cutoff = now_ms() - INTRADAY_MIN_INTERVAL_SECS * 1000;
1654 let mut refreshed = 0;
1655 for t in tickers {
1656 // Skip a symbol with a quote younger than the throttle window.
1657 let fresh: Option<i64> = sqlx::query_scalar(
1658 "SELECT last_quote_at FROM symbols WHERE ticker = ? AND last_quote_at >= ?",
1659 )
1660 .bind(t)
1661 .bind(cutoff)
1662 .fetch_optional(pool)
1663 .await
1664 .ok()
1665 .flatten();
1666 if fresh.is_some() {
1667 continue;
1668 }
1669 if refresh_quote(pool, config, hub, t).await == "ok" {
1670 refreshed += 1;
1671 }
1672 }
1673 refreshed
1674}
1675
1676/// Pull one live quote + its intraday bars and publish it to the hub so an open
1677/// page patches its price in place (mirrors `run_intraday`'s per-symbol body).
1678async fn refresh_quote(pool: &SqlitePool, config: &Config, hub: &Hub, ticker: &str) -> &'static str {
1679 let yahoo = YahooProvider::new(providers::http::build_client(config));
1680 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1681 match guarded(&guard, yahoo.quote(ticker)).await {
1682 Some(Ok(data)) => {
1683 let _ = store_quote(pool, ticker, &data.quote).await;
1684 if !data.bars.is_empty() {
1685 let _ = store_intraday(pool, ticker, &data.bars).await;
1686 }
1687 hub.publish(StreamEvent::Quote(QuoteUpdate::new(
1688 ticker.to_string(),
1689 data.quote.price,
1690 data.quote.prev_close,
1691 data.quote.market_state.clone(),
1692 )));
1693 "ok"
1694 }
1695 Some(Err(e)) => {
1696 tracing::warn!("[refresh] quote {ticker}: {e:#}");
1697 "error"
1698 }
1699 None => "skipped",
1700 }
1701}
1702
1703/// Intraday range that covers a whole trading week (Mon–Fri) of 15-minute bars
1704/// in one request — enough for the end-of-week dashboard view.
1705const INTRADAY_WEEK_RANGE: &str = "5d";
1706
1707/// Backfill the whole trading week's 15-minute bars for `tickers` when the
1708/// stored bars don't already cover the early week. The routine intraday poll
1709/// only ever stores one day at a time (`range=1d`), so the end-of-week view is
1710/// missing any day the dashboard wasn't open. One guarded `range=5d` request per
1711/// still-incomplete symbol fills the gap; symbols whose stored bars already
1712/// reach the week's start are skipped, so a reload doesn't re-hit Yahoo.
1713pub(crate) async fn backfill_intraday_week(
1714 pool: &SqlitePool,
1715 config: &Config,
1716 tickers: &[String],
1717 week_start_ms: i64,
1718 week_end_ms: i64,
1719) -> usize {
1720 // "Already covered": the earliest in-window bar sits within ~36h of the
1721 // week's open. A normally-polled week starts at Monday's open; a
1722 // holiday-Monday week at Tuesday's — both inside this margin, so they're not
1723 // refetched. Only a week missing its first two days (the gap the user sees)
1724 // falls outside it.
1725 let covered_before = week_start_ms + 36 * 3_600 * 1000;
1726 let mut filled = 0;
1727 for t in tickers {
1728 let earliest: Option<i64> = sqlx::query_scalar(
1729 "SELECT MIN(ts) FROM intraday_bars WHERE ticker = ? AND ts >= ? AND ts <= ?",
1730 )
1731 .bind(t)
1732 .bind(week_start_ms)
1733 .bind(week_end_ms)
1734 .fetch_optional(pool)
1735 .await
1736 .ok()
1737 .flatten()
1738 .flatten();
1739 if matches!(earliest, Some(ms) if ms <= covered_before) {
1740 continue;
1741 }
1742 let yahoo = YahooProvider::new(providers::http::build_client(config));
1743 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1744 match guarded(&guard, yahoo.intraday_window(t, INTRADAY_WEEK_RANGE)).await {
1745 Some(Ok(data)) if !data.bars.is_empty() => {
1746 let _ = store_intraday(pool, t, &data.bars).await;
1747 filled += 1;
1748 }
1749 Some(Err(e)) => tracing::warn!("[week] intraday {t}: {e:#}"),
1750 _ => {}
1751 }
1752 }
1753 filled
1754}
1755
1756/// Pull the daily history a viewed symbol is missing: the window since its last
1757/// stored bar (incremental) when it already has history, else a full
1758/// `range=max` backfill. Cheaper than the deep re-fetch on a routine load.
1759async fn refresh_history_incremental(
1760 pool: &SqlitePool,
1761 config: &Config,
1762 ticker: &str,
1763) -> &'static str {
1764 let last: Option<String> =
1765 sqlx::query_scalar("SELECT history_last_date FROM symbols WHERE ticker = ?")
1766 .bind(ticker)
1767 .fetch_optional(pool)
1768 .await
1769 .ok()
1770 .flatten();
1771 let yahoo = YahooProvider::new(providers::http::build_client(config));
1772 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1773 match guarded(&guard, yahoo.daily(ticker, last.as_deref())).await {
1774 Some(Ok(bars)) if !bars.is_empty() => {
1775 let _ = seed::store_daily(pool, ticker, &bars).await;
1776 "ok"
1777 }
1778 Some(Ok(_)) => {
1779 let _ = mark_history_checked(pool, ticker).await;
1780 "ok"
1781 }
1782 Some(Err(e)) => {
1783 tracing::warn!("[refresh] history {ticker}: {e:#}");
1784 "error"
1785 }
1786 None => "skipped",
1787 }
1788}
1789
1790/// Backfill a viewed symbol's SEC data on demand (stock fundamentals/filings/
1791/// leadership, or an ETF's holdings/filings). Skipped cleanly with no contact
1792/// email configured, as the old sweep was.
1793async fn refresh_sec(pool: &SqlitePool, config: &Config, ticker: &str, fund: bool) -> &'static str {
1794 if config.sec_contact_email.is_empty() {
1795 return "skipped";
1796 }
1797 let sec = SecProvider::new(providers::http::build_sec_client(config));
1798 let guard = EndpointGuard::with_budget(pool.clone(), sec.name(), SEC_BUDGET);
1799 if fund {
1800 backfill_etf_sec(pool, &sec, &guard, ticker).await;
1801 } else {
1802 backfill_stock_sec(pool, &sec, &guard, ticker).await;
1803 }
1804 "ok"
1805}
1806
1807/// Refresh an ETF's Yahoo fund metadata and its NAV (the price-vs-NAV premium
1808/// behind the quality read's tracking factor needs a fresh NAV; see the
1809/// hard-won lesson). Two cheap `quoteSummary` calls through the
1810/// Yahoo guard.
1811async fn refresh_fund_meta(pool: &SqlitePool, config: &Config, ticker: &str) -> &'static str {
1812 backfill_fund_metadata(pool, config, ticker).await;
1813 let yahoo = YahooProvider::new(providers::http::build_client(config));
1814 let guard = EndpointGuard::with_budget(pool.clone(), "yahoo", YAHOO_BUDGET);
1815 if let Some(Ok(nav)) = guarded(&guard, yahoo.fund_nav(ticker)).await {
1816 let _ = store_fund_nav(pool, ticker, nav).await;
1817 }
1818 "ok"
1819}
1820
1821/// Upsert an ETF's freshly-fetched NAV + its sync stamp, touching only the NAV
1822/// columns so the static fields stay intact. A `None` nav clears any prior NAV
1823/// (honest: no fresh value to read a premium against). Re-added for the Phase-B
1824/// on-demand NAV pull after the daily `fund_nav` job was removed in Phase A.
1825async fn store_fund_nav(pool: &SqlitePool, ticker: &str, nav: Option<f64>) -> sqlx::Result<()> {
1826 let now = now_ms();
1827 sqlx::query(
1828 "INSERT INTO fund_metadata (ticker, nav_price, nav_synced_at, updated_at) \
1829 VALUES (?, ?, ?, ?) \
1830 ON CONFLICT(ticker) DO UPDATE SET \
1831 nav_price = excluded.nav_price, \
1832 nav_synced_at = excluded.nav_synced_at, \
1833 updated_at = excluded.updated_at",
1834 )
1835 .bind(ticker)
1836 .bind(nav)
1837 .bind(now)
1838 .bind(now)
1839 .execute(pool)
1840 .await?;
1841 Ok(())
1842}