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//! First-run universe seed: load the curated starter list, then backfill deep
2//! daily history for the symbols that do not have it yet.
3//!
4//! Resumable and quota-friendly: symbols that already hold history are skipped,
5//! so re-running `make seed` after a partial run continues where it stopped.
6//! Every history request goes through the persistent `EndpointGuard` (the
7//! `yahoo` one), which paces the loop and stops it early if the circuit breaker
8//! is open or the hourly budget is spent, instead of grinding the list against
9//! a guarded endpoint.
10
11use std::path::Path;
12use std::time::Instant;
13
14use anyhow::{Context, Result};
15use sqlx::SqlitePool;
16
17use crate::db::{now_ms, set_meta};
18use crate::guard::{EndpointGuard, Permit};
19use crate::providers::{DailyBar, HistoryProvider};
20use crate::Config;
21
22struct SeedSymbol {
23 ticker: String,
24 name: String,
25 kind: String,
26 exchange: Option<String>,
27 /// Phase 28: the curated benchmark index a fund tracks (e.g. `^SPX`),
28 /// for the relative-performance overlay on the ETF symbol page. Only
29 /// the broad-market ETFs carry one in `starter.csv`; everything else
30 /// (including stocks, futures, indexes, sector / bond / commodity ETFs)
31 /// leaves it empty and the overlay is hidden.
32 benchmark: Option<String>,
33}
34
35fn parse_universe(path: &Path) -> Result<Vec<SeedSymbol>> {
36 let mut rdr = csv::Reader::from_path(path)
37 .with_context(|| format!("opening universe file {}", path.display()))?;
38 let mut out = Vec::new();
39 for rec in rdr.records() {
40 let rec = rec?;
41 let cell = |i: usize| rec.get(i).unwrap_or("").trim().to_string();
42 let ticker = cell(0).to_uppercase();
43 if ticker.is_empty() {
44 continue;
45 }
46 let opt = |s: String| if s.is_empty() { None } else { Some(s) };
47 out.push(SeedSymbol {
48 ticker,
49 name: cell(1),
50 kind: cell(2),
51 exchange: opt(cell(3)),
52 benchmark: opt(cell(4)),
53 });
54 }
55 Ok(out)
56}
57
58/// Outcome of a universe sync, for logging.
59pub struct SyncReport {
60 /// Symbols in the curated CSV (all upserted).
61 pub total: usize,
62 /// Seeded symbols deleted because they were dropped from the CSV.
63 pub pruned: u64,
64}
65
66/// Reconcile the `symbols` table to the curated CSV. Local only, no network:
67/// upsert every listed symbol and prune the curated ones that were dropped from
68/// the list. Idempotent, so it runs on every boot (see `scheduler::run_boot_seed`)
69/// and a CSV edit (added or removed symbols) takes effect on the next deploy
70/// without a manual re-seed. The history backfill for any newly-added symbol is
71/// handled separately: the first-run seed loop below, or — once the seed has
72/// completed — the incremental `run_history` job, which picks up any symbol with
73/// a `NULL history_synced_at`.
74pub async fn sync_universe(pool: &SqlitePool, config: &Config) -> Result<SyncReport> {
75 let path = config.root.join("universe/starter.csv");
76 let symbols = parse_universe(&path)?;
77
78 // Upsert every symbol. Phase 28: the curated benchmark column is set from
79 // the CSV on each pass, so a re-run picks up any newly-curated mapping. A
80 // `NULL` benchmark stays `NULL` (they all live in the CSV).
81 for s in &symbols {
82 let now = now_ms();
83 sqlx::query(
84 "INSERT INTO symbols \
85 (ticker, name, kind, exchange, benchmark, is_seeded, created_at, updated_at) \
86 VALUES (?, ?, ?, ?, ?, 1, ?, ?) \
87 ON CONFLICT(ticker) DO UPDATE SET \
88 name = excluded.name, kind = excluded.kind, \
89 exchange = excluded.exchange, benchmark = excluded.benchmark, \
90 is_seeded = 1, updated_at = excluded.updated_at",
91 )
92 .bind(&s.ticker)
93 .bind(&s.name)
94 .bind(&s.kind)
95 .bind(&s.exchange)
96 .bind(&s.benchmark)
97 .bind(now)
98 .bind(now)
99 .execute(pool)
100 .await?;
101 }
102
103 // Prune curated symbols dropped from the CSV. Only `is_seeded = 1` rows are
104 // eligible, so a user-added symbol (`is_seeded = 0`) is never touched. The
105 // delete cascades to every child table (all `REFERENCES symbols(ticker) ON
106 // DELETE CASCADE`, foreign keys enabled in db::init), so no orphan rows are
107 // left behind. The IN-list is bounded by the curated list size (~560), well
108 // under SQLite's bound-parameter limit.
109 let placeholders = vec!["?"; symbols.len()].join(",");
110 let sql = format!(
111 "DELETE FROM symbols WHERE is_seeded = 1 AND ticker NOT IN ({placeholders})"
112 );
113 let mut q = sqlx::query(&sql);
114 for s in &symbols {
115 q = q.bind(&s.ticker);
116 }
117 let pruned = q.execute(pool).await?.rows_affected();
118
119 Ok(SyncReport { total: symbols.len(), pruned })
120}
121
122/// Run the seed: reconcile the universe, then backfill daily history for any
123/// symbol that still lacks it.
124pub async fn run(pool: &SqlitePool, config: &Config, history: &dyn HistoryProvider) -> Result<()> {
125 let started = Instant::now();
126 let report = sync_universe(pool, config).await?;
127 tracing::info!(
128 "seed: {} symbols synced, {} pruned",
129 report.total,
130 report.pruned
131 );
132
133 // Symbols that still need a deep backfill. Two cases:
134 // - `history_last_date IS NULL`: never fetched (the normal first-run case).
135 // - `history_first_date = history_last_date`: only a single stored bar,
136 // which means the symbol was added after the initial seed and a
137 // `daily_close` snapshot stamped its `history_last_date` before any
138 // range=max backfill ran — so the incremental path, which only asks for
139 // the window since the last bar, can never reach its deep history. These
140 // get re-fetched with range=max (the loop always passes `None`). A symbol
141 // Yahoo genuinely has one bar for simply stays a one-bar re-fetch; cheap.
142 // This keeps re-runs cheap and lets a quota-limited run resume later.
143 // Futures are included now: Yahoo serves `=F` daily history, unlike the
144 // Stooq source this replaced.
145 let pending: Vec<String> = sqlx::query_scalar(
146 "SELECT ticker FROM symbols \
147 WHERE is_seeded = 1 \
148 AND (history_last_date IS NULL OR history_first_date = history_last_date) \
149 ORDER BY ticker",
150 )
151 .fetch_all(pool)
152 .await?;
153
154 if pending.is_empty() {
155 set_meta(pool, "seed_completed", "1").await?;
156 tracing::info!("seed: every symbol already has history, nothing to fetch");
157 return Ok(());
158 }
159 tracing::info!("seed: {} symbols need a history backfill", pending.len());
160
161 // Every history request passes through the persistent endpoint guard: it
162 // paces the loop and, once the breaker opens or the hourly budget runs out,
163 // refuses further requests so the seed stops cleanly rather than grinding
164 // the rest of the list. A stopped seed is resumable (see below). History
165 // shares the `yahoo` guard with live quotes, so it must carry the same
166 // budget rather than the 200-default `new` ceiling.
167 let guard = EndpointGuard::with_budget(
168 pool.clone(),
169 history.name(),
170 crate::scheduler::YAHOO_BUDGET,
171 );
172
173 let mut ok = 0usize;
174 let mut stopped: Option<String> = None;
175 for (i, ticker) in pending.iter().enumerate() {
176 match guard.acquire().await? {
177 Permit::Granted => {}
178 Permit::Denied(why) => {
179 stopped = Some(why);
180 break;
181 }
182 }
183 match history.daily(ticker, None).await {
184 Ok(bars) if !bars.is_empty() => {
185 guard.record_success().await?;
186 let n = bars.len();
187 store_daily(pool, ticker, &bars).await?;
188 ok += 1;
189 tracing::info!(
190 "seed: {ticker} <- {n} daily bars ({}/{})",
191 i + 1,
192 pending.len()
193 );
194 }
195 Ok(_) => {
196 // A valid but empty response: the request itself succeeded, so
197 // it counts as a success for the guard; the symbol simply has
198 // no history to store.
199 guard.record_success().await?;
200 tracing::warn!("seed: {ticker} returned no data");
201 }
202 Err(e) => {
203 guard.record_failure(&e).await?;
204 tracing::warn!("seed: {ticker} failed: {e:#}");
205 }
206 }
207 }
208 if let Some(why) = &stopped {
209 tracing::warn!(
210 "seed: stopped early — {why}; {ok} symbols backfilled and kept, \
211 re-run `make seed` (or restart the server) later to continue"
212 );
213 }
214
215 let remaining: i64 = sqlx::query_scalar(
216 "SELECT COUNT(*) FROM symbols \
217 WHERE is_seeded = 1 AND history_last_date IS NULL",
218 )
219 .fetch_one(pool)
220 .await?;
221 if remaining == 0 {
222 set_meta(pool, "seed_completed", "1").await?;
223 set_meta(pool, "seed_at", &now_ms().to_string()).await?;
224 }
225 tracing::info!(
226 "seed: {ok} backfilled, {remaining} still missing, {:.1}s",
227 started.elapsed().as_secs_f64()
228 );
229 Ok(())
230}
231
232/// Upsert one symbol's daily bars in a single transaction and refresh its
233/// `symbols` history-range columns.
234pub async fn store_daily(pool: &SqlitePool, ticker: &str, bars: &[DailyBar]) -> Result<()> {
235 if bars.is_empty() {
236 return Ok(());
237 }
238 let mut tx = pool.begin().await?;
239 for b in bars {
240 sqlx::query(
241 "INSERT INTO daily_prices (ticker, d, open, high, low, close, volume) \
242 VALUES (?, ?, ?, ?, ?, ?, ?) \
243 ON CONFLICT(ticker, d) DO UPDATE SET \
244 open = excluded.open, high = excluded.high, low = excluded.low, \
245 close = excluded.close, volume = excluded.volume",
246 )
247 .bind(ticker)
248 .bind(&b.d)
249 .bind(b.open)
250 .bind(b.high)
251 .bind(b.low)
252 .bind(b.close)
253 .bind(b.volume)
254 .execute(&mut *tx)
255 .await?;
256 }
257 // Recompute the history range from the stored rows (visible inside this
258 // transaction) rather than from just the bars passed in. An incremental
259 // window or a single daily-close append carries only recent bars, so
260 // taking min/max of the argument alone would clobber the true earliest
261 // date with the window start.
262 let now = now_ms();
263 sqlx::query(
264 "UPDATE symbols SET history_synced_at = ?, \
265 history_first_date = (SELECT MIN(d) FROM daily_prices WHERE ticker = ?), \
266 history_last_date = (SELECT MAX(d) FROM daily_prices WHERE ticker = ?), \
267 updated_at = ? WHERE ticker = ?",
268 )
269 .bind(now)
270 .bind(ticker)
271 .bind(ticker)
272 .bind(now)
273 .bind(ticker)
274 .execute(&mut *tx)
275 .await?;
276 tx.commit().await?;
277 Ok(())
278}