diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index c7127ff..84c19ae 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -20,6 +20,7 @@ pub mod mempool; pub mod ohlcv; pub mod pool; pub mod poolvisitor; +pub mod priceseries; pub mod tokenlist; pub mod tokentoken; pub mod tx; diff --git a/src/db/cauldron/priceseries.rs b/src/db/cauldron/priceseries.rs new file mode 100644 index 0000000..0dfa407 --- /dev/null +++ b/src/db/cauldron/priceseries.rs @@ -0,0 +1,411 @@ +// Copyright (C) 2025-2026 Whiterun LLC +// +// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. +// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html + +//! Cached 7-day price curve for the token list. +//! +//! The series samples the *aggregate pool price* — summed reserves over every live +//! pool of the token — which is the same measurement `price_now`, `price_7d` and +//! therefore `change_7d_bp` are built from. Serving it with the list lets a client +//! draw a 7d curve whose ends are the very numbers it prints as percentages, +//! without a request per token. +//! +//! Candlesticks cannot fill that role: they price *executed trades* (a per-tx +//! VWAP, so they carry slippage and, for a token with many pools, only the pool +//! that happened to trade). On a quiet token their last close can sit percent +//! away from the pool price, which makes a curve drawn from them contradict the +//! percentage beside it. +//! +//! Cost per cycle is one snapshot query per token plus two window-wide queries; +//! everything inside the window is replayed in memory from `pool_history_entry`. + +use std::collections::HashMap; + +use anyhow::Result; +use bitcoincash::TokenID; +use riftenlabs_defi::chainutil::OutPointHash; +use rust_decimal::prelude::*; +use rust_decimal::Decimal; +use sqlx::{Row, SqlitePool}; + +use crate::db::blob::blob_to_display_hex; +use crate::db::cauldron::poolvisitor::{ + db_visit_pool_entries, OptionalFields, OptionalPoolFields, PoolFilters, PoolVisitor, +}; +use crate::db::cauldron::tokenlist::token_utils::{dec_round, dec_to_f64_bounded}; + +/// Points in a series: 7 days sampled every 6h, both ends included. The 6h step +/// puts the 24h boundary exactly on index 24, so the 1d and 7d moves can be read +/// straight off the array. +pub(crate) const SERIES_POINTS: usize = 29; + +/// Decimals kept per ratio. Five is ~1e-5 relative precision — far finer than a +/// hundred-pixel curve can show, and it keeps the array near 200 bytes. +const RATIO_SCALE: u32 = 5; + +/// Reserves of one pool: (sats, token base units). +type Reserves = (u64, u64); + +/// A pool state change inside the window. +/// +/// `reserves == None` marks a withdrawal, which has no `pool_history_entry` row +/// of its own — it only sets `pool.withdrawn_in_utxo` — so it has to be loaded +/// separately or the pool's reserves would linger in the sum forever. +#[derive(Clone)] +pub(crate) struct PoolEvent { + ts: i64, + sequence: i64, + pool_id: String, + reserves: Option, +} + +/// Every pool state change after a cutoff, grouped by token hex. +pub(crate) struct WindowEvents { + by_token: HashMap>, +} + +impl WindowEvents { + pub(crate) fn for_token(&self, token_id: &str) -> &[PoolEvent] { + self.by_token + .get(token_id) + .map(|v| v.as_slice()) + .unwrap_or(&[]) + } +} + +#[derive(Default)] +struct ReservesVisitor { + reserves: HashMap, +} + +impl PoolVisitor for ReservesVisitor { + fn optional_fields_wanted(&self) -> u64 { + OptionalPoolFields::PoolId as u64 + } + + fn visit(&mut self, sats: u64, tokens: u64, optional: OptionalFields) -> Result { + if let Some(pool_id) = optional.pool_id { + self.reserves.insert(pool_id, (sats, tokens)); + } + Ok(true) + } +} + +/// Per-pool reserves for `token_id` as of `ts`. +/// +/// Uses the same liveness rules as `price_at_or_before_2` (each pool's latest +/// entry at or before `ts`, pools withdrawn before `ts` excluded), so summing +/// this map reproduces that function's price exactly — which is what makes +/// entry 0 of the series equal `price_7d`. +pub(crate) async fn pool_reserves_at( + cauldron_pool: &SqlitePool, + ts: i64, + token_id: &TokenID, +) -> Result> { + let mut visitor = ReservesVisitor::default(); + db_visit_pool_entries( + cauldron_pool, + &mut visitor, + PoolFilters::new() + .lte(ts as u64) + .token_id(&token_id.to_string()), + ) + .await?; + Ok(visitor.reserves) +} + +/// Load every pool state change after `since`, for all tokens at once. +/// +/// One pass over the window beats a query per token: the whole 7d window is a +/// few thousand rows, and `idx_phe_token_id_ts` keeps it an index range scan. +pub(crate) async fn load_window_events( + cauldron_pool: &SqlitePool, + since: i64, +) -> Result { + let mut by_token: HashMap> = HashMap::new(); + + let entries = sqlx::query( + r#" + SELECT token_id, pool, sats, token_amount, effective_timestamp, sequence + FROM pool_history_entry + WHERE effective_timestamp > ?1 + "#, + ) + .bind(since) + .fetch_all(cauldron_pool) + .await?; + + for row in entries { + let token_blob: Vec = row.get(0); + let Ok(token_id) = blob_to_display_hex::(&token_blob) else { + continue; + }; + let pool_blob: Vec = row.get(1); + let Ok(pool_id) = blob_to_display_hex::(&pool_blob) else { + continue; + }; + let sats: i64 = row.get(2); + let tokens: i64 = row.get(3); + by_token.entry(token_id).or_default().push(PoolEvent { + ts: row.get(4), + sequence: row.get(5), + pool_id, + reserves: Some((sats.max(0) as u64, tokens.max(0) as u64)), + }); + } + + let withdrawals = sqlx::query( + r#" + SELECT p.token_id, p.creation_utxo, t.effective_timestamp + FROM pool p + JOIN utxo_spending us ON us.spent_utxo_hash = p.withdrawn_in_utxo + JOIN tx t ON us.txid = t.txid + WHERE t.effective_timestamp > ?1 + "#, + ) + .bind(since) + .fetch_all(cauldron_pool) + .await?; + + for row in withdrawals { + let token_blob: Vec = row.get(0); + let Ok(token_id) = blob_to_display_hex::(&token_blob) else { + continue; + }; + let pool_blob: Vec = row.get(1); + let Ok(pool_id) = blob_to_display_hex::(&pool_blob) else { + continue; + }; + let withdrawn_ts: i64 = row.get(2); + by_token.entry(token_id).or_default().push(PoolEvent { + // A pool whose withdrawal lands exactly on the sample timestamp is + // still counted there (`db_visit_pool_entries` keeps it while the + // withdrawal is `>=` the bound), so it may only drop out strictly + // after. Recording the event a second late lets the replay use one + // `<=` comparison for both kinds of event. + ts: withdrawn_ts.saturating_add(1), + // after any entry sharing the timestamp + sequence: i64::MAX, + pool_id, + reserves: None, + }); + } + + for events in by_token.values_mut() { + events.sort_by_key(|e| (e.ts, e.sequence)); + } + + Ok(WindowEvents { by_token }) +} + +/// Aggregate pool price, per smallest token unit. +/// +/// `None` when the token has no priceable reserves — the same situation in which +/// `price_at_or_before_2` reports no price rather than zero. +fn aggregate_price(reserves: &HashMap) -> Option { + let mut sats: u128 = 0; + let mut tokens: u128 = 0; + for (s, t) in reserves.values() { + sats = sats.saturating_add(*s as u128); + tokens = tokens.saturating_add(*t as u128); + } + if sats == 0 || tokens == 0 { + return None; + } + Decimal::from_u128(sats)?.checked_div(Decimal::from_u128(tokens)?) +} + +/// Sample the aggregate pool price across `[window_start, now]`. +/// +/// `reserves` is the state at `window_start` per [`pool_reserves_at`], `events` +/// that token's changes after it (sorted, as [`load_window_events`] returns +/// them). Entry `k` is the price at `window_start + k * span / (SERIES_POINTS-1)`, +/// so entry 0 is the window start and the last entry is `now`. `None` marks a +/// bucket where the token had no priceable pool yet — a token younger than the +/// window simply starts with a run of them. +pub(crate) fn build_price_series( + mut reserves: HashMap, + events: &[PoolEvent], + window_start: i64, + now: i64, +) -> Vec> { + let span = (now - window_start).max(0); + let last = (SERIES_POINTS - 1) as i64; + let mut out = Vec::with_capacity(SERIES_POINTS); + let mut next = 0usize; + + for k in 0..SERIES_POINTS as i64 { + // scaled from the span rather than a fixed step so the final sample lands + // exactly on `now` however the window is sized + let t = window_start + (span * k) / last; + while next < events.len() && events[next].ts <= t { + let event = &events[next]; + match event.reserves { + Some(r) => { + reserves.insert(event.pool_id.clone(), r); + } + None => { + reserves.remove(&event.pool_id); + } + } + next += 1; + } + out.push(aggregate_price(&reserves)); + } + + out +} + +/// Serialize a series as a JSON array of ratios to `base`, `null` where unknown. +/// +/// Ratios rather than prices: they are unit-free (no decimals to reconcile), +/// they compress to a few bytes each, and a client that wants absolute values +/// already has the exact endpoints in `price_7d`/`price_now`. Returns `None` when +/// there is nothing to anchor against or no price in the whole window, so the +/// column stays NULL instead of holding a meaningless array. +pub(crate) fn encode_price_series(series: &[Option], base: Decimal) -> Option { + if base <= Decimal::ZERO || series.iter().all(|p| p.is_none()) { + return None; + } + let ratios: Vec> = series + .iter() + .map(|p| { + p.and_then(|price| price.checked_div(base)) + .map(|ratio| dec_to_f64_bounded(dec_round(ratio, RATIO_SCALE))) + }) + .collect(); + serde_json::to_string(&ratios).ok() +} + +#[cfg(test)] +mod tests { + use super::*; + use rust_decimal_macros::dec; + + const DAY: i64 = 86_400; + const WEEK: i64 = 7 * DAY; + const STEP: i64 = 6 * 3600; + + fn reserves(pairs: &[(&str, u64, u64)]) -> HashMap { + pairs + .iter() + .map(|(id, s, t)| (id.to_string(), (*s, *t))) + .collect() + } + + fn entry(ts: i64, pool: &str, sats: u64, tokens: u64) -> PoolEvent { + PoolEvent { + ts, + sequence: ts, + pool_id: pool.to_string(), + reserves: Some((sats, tokens)), + } + } + + fn withdrawal(ts: i64, pool: &str) -> PoolEvent { + PoolEvent { + ts: ts + 1, + sequence: i64::MAX, + pool_id: pool.to_string(), + reserves: None, + } + } + + #[test] + fn samples_both_ends_of_the_window() { + let series = build_price_series(reserves(&[("a", 100, 200)]), &[], 0, WEEK); + assert_eq!(series.len(), SERIES_POINTS); + // nothing happened, so every sample is the starting price + assert!(series.iter().all(|p| *p == Some(dec!(0.5)))); + } + + #[test] + fn the_24h_boundary_lands_on_index_24() { + // a 6h step over 7 days puts now-24h exactly four samples from the end + let series = build_price_series( + reserves(&[("a", 100, 100)]), + &[entry(WEEK - DAY, "a", 200, 100)], + 0, + WEEK, + ); + assert_eq!(series[23], Some(dec!(1))); + assert_eq!(series[24], Some(dec!(2))); + assert_eq!(series[SERIES_POINTS - 1], Some(dec!(2))); + } + + #[test] + fn replays_trades_onto_the_grid() { + let series = build_price_series( + reserves(&[("a", 100, 100)]), + &[entry(STEP, "a", 150, 100), entry(3 * STEP, "a", 300, 100)], + 0, + WEEK, + ); + assert_eq!(series[0], Some(dec!(1))); + assert_eq!(series[1], Some(dec!(1.5))); + assert_eq!(series[2], Some(dec!(1.5))); + assert_eq!(series[3], Some(dec!(3))); + } + + #[test] + fn sums_every_live_pool() { + // the aggregate is summed reserves, not an average of per-pool prices + let series = + build_price_series(reserves(&[("a", 100, 100), ("b", 900, 100)]), &[], 0, WEEK); + assert_eq!(series[0], Some(dec!(5))); + } + + #[test] + fn drops_a_withdrawn_pool_only_after_its_timestamp() { + // matches db_visit_pool_entries, which still counts a pool whose + // withdrawal is exactly at the sample timestamp + let series = build_price_series( + reserves(&[("a", 100, 100), ("b", 900, 100)]), + &[withdrawal(2 * STEP, "b")], + 0, + WEEK, + ); + assert_eq!(series[2], Some(dec!(5))); + assert_eq!(series[3], Some(dec!(1))); + } + + #[test] + fn a_pool_created_inside_the_window_starts_the_curve() { + // a token younger than the window has no price until its first pool + let series = build_price_series(HashMap::new(), &[entry(3 * STEP, "a", 100, 100)], 0, WEEK); + assert_eq!(series[2], None); + assert_eq!(series[3], Some(dec!(1))); + } + + #[test] + fn no_price_at_all_stays_null() { + let series = build_price_series(HashMap::new(), &[], 0, WEEK); + assert!(series.iter().all(|p| p.is_none())); + assert_eq!(encode_price_series(&series, dec!(1)), None); + } + + #[test] + fn encodes_ratios_to_the_base_price() { + let series = vec![Some(dec!(0.5)), None, Some(dec!(1)), Some(dec!(2))]; + let json = encode_price_series(&series, dec!(1)).unwrap(); + assert_eq!(json, "[0.5,null,1.0,2.0]"); + } + + #[test] + fn encoding_needs_a_usable_base() { + let series = vec![Some(dec!(1))]; + assert_eq!(encode_price_series(&series, dec!(0)), None); + assert_eq!(encode_price_series(&series, dec!(-1)), None); + } + + #[test] + fn encoding_keeps_the_payload_small() { + // ~200 bytes per token is the whole point: it replaces a per-row request + let series: Vec> = (0..SERIES_POINTS) + .map(|i| Some(dec!(1) + Decimal::from(i) / dec!(100000))) + .collect(); + let json = encode_price_series(&series, dec!(1)).unwrap(); + assert!(json.len() < 300, "series encoded to {} bytes", json.len()); + } +} diff --git a/src/db/cauldron/tokenlist/db_utils.rs b/src/db/cauldron/tokenlist/db_utils.rs index 77b5364..ac3b721 100644 --- a/src/db/cauldron/tokenlist/db_utils.rs +++ b/src/db/cauldron/tokenlist/db_utils.rs @@ -212,6 +212,27 @@ CREATE TABLE IF NOT EXISTS cached_token_metrics ( add_column_if_missing(pool, "cached_token_metrics", "first_pool_ts", "INTEGER").await?; add_column_if_missing(pool, "cached_token_metrics", "bcmr_json", "TEXT").await?; add_column_if_missing(pool, "cached_token_metrics", "bcmr_well_known_json", "TEXT").await?; + // Timestamp each change_* was actually measured from. For a token younger than + // the window this is its first pool, not `now - window`, so consumers can label + // a since-launch move as such instead of passing it off as a full 24h/7d one. + add_column_if_missing( + pool, + "cached_token_metrics", + "change_24h_anchor_ts", + "INTEGER", + ) + .await?; + add_column_if_missing( + pool, + "cached_token_metrics", + "change_7d_anchor_ts", + "INTEGER", + ) + .await?; + // 7d aggregate-pool-price curve as a JSON array of ratios to price_now; see + // db::cauldron::priceseries. Lets the token list ship a sparkline instead of + // every client asking for candlesticks per row. + add_column_if_missing(pool, "cached_token_metrics", "price_series_7d", "TEXT").await?; Ok(()) } diff --git a/src/db/cauldron/tokenlist/list_cached.rs b/src/db/cauldron/tokenlist/list_cached.rs index 059fd07..fa76e18 100644 --- a/src/db/cauldron/tokenlist/list_cached.rs +++ b/src/db/cauldron/tokenlist/list_cached.rs @@ -34,8 +34,18 @@ pub struct TokenListItemCached { pub price_7d_usd: Option, pub change_24h_usd_bp: Option, pub change_7d_usd_bp: Option, + /// Timestamp the matching `change_*` was measured from. Equals `now - window` + /// for a token with enough history; for a younger one it is the token's first + /// pool, i.e. the change is a since-launch move over a shorter period. + pub change_24h_anchor_ts: Option, + pub change_7d_anchor_ts: Option, pub apy_30d_bp: Option, pub first_pool_ts: Option, + /// 7d aggregate-pool-price curve: 29 samples every 6h ending at now, each a + /// ratio to `price_now`, `null` before the token had a pool. Same measurement + /// as `change_7d_bp`, so a client can draw a sparkline that agrees with the + /// percentages without fetching candlesticks per token. + pub price_series_7d: Option>>, } const TOKEN_METRICS_COLUMNS: &str = r#" @@ -56,11 +66,14 @@ const TOKEN_METRICS_COLUMNS: &str = r#" price_7d_usd, change_24h_usd_bp, change_7d_usd_bp, + change_24h_anchor_ts, + change_7d_anchor_ts, apy_30d_bp, first_pool_ts, score_rank, bcmr_json, - bcmr_well_known_json + bcmr_well_known_json, + price_series_7d "#; fn parse_bcmr_columns( @@ -93,12 +106,17 @@ fn parse_row(row: &sqlx::sqlite::SqliteRow) -> Result { let price_7d_usd: Option = row.get("price_7d_usd"); let change_24h_usd_bp: Option = row.get("change_24h_usd_bp"); let change_7d_usd_bp: Option = row.get("change_7d_usd_bp"); + let change_24h_anchor_ts: Option = row.get("change_24h_anchor_ts"); + let change_7d_anchor_ts: Option = row.get("change_7d_anchor_ts"); let apy_30d_bp: Option = row.get("apy_30d_bp"); // 0 is the "not backfilled yet" sentinel — treat it as unknown. let first_pool_ts: Option = row .get::, _>("first_pool_ts") .filter(|&ts| ts != 0); let score_rank: i64 = row.get("score_rank"); + let price_series_7d: Option>> = row + .get::, _>("price_series_7d") + .and_then(|s| serde_json::from_str(&s).ok()); let bcmr_json_str: Option = row.get("bcmr_json"); let bcmr_wk_json_str: Option = row.get("bcmr_well_known_json"); let (bcmr, bcmr_well_known) = parse_bcmr_columns(bcmr_json_str, bcmr_wk_json_str); @@ -125,8 +143,11 @@ fn parse_row(row: &sqlx::sqlite::SqliteRow) -> Result { price_7d_usd, change_24h_usd_bp, change_7d_usd_bp, + change_24h_anchor_ts, + change_7d_anchor_ts, apy_30d_bp, first_pool_ts, + price_series_7d, }) } diff --git a/src/db/cauldron/tokenlist/metrics_cache.rs b/src/db/cauldron/tokenlist/metrics_cache.rs index c92930f..ee5dbf6 100644 --- a/src/db/cauldron/tokenlist/metrics_cache.rs +++ b/src/db/cauldron/tokenlist/metrics_cache.rs @@ -10,16 +10,18 @@ use sqlx::{Row, SqlitePool}; use crate::db::bcmr::{get_token_bcmr, get_well_known_bcmr}; use crate::db::blob::{blob_to_display_hex, display_hex_to_blob}; use crate::db::cauldron::poolvisitor::{db_visit_pool_entries, PoolFilters}; +use crate::db::cauldron::priceseries::{ + build_price_series, encode_price_series, load_window_events, pool_reserves_at, +}; use crate::db::cauldron::tokenlist::db_utils::{ cache_first_pool_ts_if_empty, db_first_pool_creation_row, }; use crate::db::cauldron::tokenlist::token_utils::{ apy_30d_bp_for_token, compute_score, dec_round, dec_to_f64_bounded, overflow_f64_fallback, - pct_change_bp_dec, pow10_dec, price_from_tvl, resolve_decimals, resolve_display_labels, - usd_per_bch_at_or_before, SATS_PER_BCH, + pct_change_bp_dec, pow10_dec, price_anchor_for_window, price_from_tvl, resolve_decimals, + resolve_display_labels, usd_per_bch_at_or_before, SATS_PER_BCH, }; use crate::db::DB; -use crate::rpc::price::price_at_or_before_2; use crate::rpc::tvl::TvlByTokenVisitor; use crate::signal::shutdown_requested; use crate::timeutil::time_now; @@ -241,6 +243,10 @@ pub async fn recompute_score_ranks(pool: &SqlitePool) -> anyhow::Result<()> { Ok(()) } +/// One row of the metrics read phase: (token_id blob, display hex, tvl_sats, +/// tvl_tokens, first_pool_ts). +type TokenReadRow = (Vec, String, u64, u64, Option); + /// Holds computed metrics for a single token, ready to be written to DB. struct TokenMetricsUpdate { token_blob: Vec, @@ -255,6 +261,10 @@ struct TokenMetricsUpdate { price_7d_usd: Option, change_24h_usd_bp: Option, change_7d_usd_bp: Option, + change_24h_anchor_ts: Option, + change_7d_anchor_ts: Option, + /// 7d curve as a JSON array of ratios to price_now, see `priceseries`. + price_series_7d: Option, display_name: Option, display_symbol: Option, bcmr_json: Option, @@ -286,6 +296,9 @@ async fn flush_metrics_update_batch( display_symbol = ?14, bcmr_json = ?15, bcmr_well_known_json = ?16, + change_24h_anchor_ts = ?17, + change_7d_anchor_ts = ?18, + price_series_7d = ?19, updated_at = CAST(strftime('%s','now') AS INTEGER) WHERE token_id = ?1 "#, @@ -306,6 +319,9 @@ async fn flush_metrics_update_batch( .bind(&u.display_symbol) .bind(&u.bcmr_json) .bind(&u.bcmr_well_known_json) + .bind(u.change_24h_anchor_ts) + .bind(u.change_7d_anchor_ts) + .bind(&u.price_series_7d) .execute(&mut *tx) .await?; } @@ -313,6 +329,27 @@ async fn flush_metrics_update_batch( Ok(()) } +/// USD-per-sat at `ts`, memoized. Young tokens anchor their change windows at their +/// own launch time, so without the cache the oracle is re-queried once per token. +/// A missing oracle point yields `ZERO`, the "unknown rate" sentinel callers already +/// test for before writing USD fields. +async fn usd_per_sat_at( + oracle_pool: &SqlitePool, + cache: &mut HashMap, + ts: i64, +) -> Decimal { + if let Some(v) = cache.get(&ts) { + return *v; + } + let sats_per_bch = Decimal::from_i64(SATS_PER_BCH).unwrap(); + let rate = usd_per_bch_at_or_before(oracle_pool, ts) + .await + .map(|usd_per_bch| usd_per_bch / sats_per_bch) + .unwrap_or(Decimal::ZERO); + cache.insert(ts, rate); + rate +} + // ---------- 5 min: price changes (24h/7d), score, volume + ranking ---------- pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result<()> { let now = time_now(); @@ -352,14 +389,20 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< let usd_per_sat_24h = usd_per_bch_at_or_before(&db.oracle_r, ts_24h).await? / sats_per_bch; let usd_per_sat_7d = usd_per_bch_at_or_before(&db.oracle_r, ts_7d).await? / sats_per_bch; + // Every pool state change in the 7d window, for all tokens in one pass — the + // per-token snapshots below only reach the window start, so the series needs + // these to walk forward from it. + let window_events = load_window_events(&db.cauldron_r, ts_7d).await?; + // ========== READ PHASE: collect all token data ========== let token_rows = sqlx::query( - "SELECT token_id, tvl_sats, tvl_tokens FROM cached_token_metrics WHERE tvl_sats > 0", + "SELECT token_id, tvl_sats, tvl_tokens, first_pool_ts + FROM cached_token_metrics WHERE tvl_sats > 0", ) .fetch_all(&db.cauldron_r) .await?; - let mut tokens_data: Vec<(Vec, String, u64, u64)> = Vec::new(); + let mut tokens_data: Vec = Vec::new(); for row in token_rows { let token_blob: Vec = row.get(0); let token_id = match blob_to_display_hex::(&token_blob) { @@ -368,19 +411,24 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< }; let tvl_sats: i64 = row.get(1); let tvl_tokens: i64 = row.get(2); + // 0 is the "not backfilled yet" sentinel — treat it as unknown. + let first_pool_ts: Option = row.get::, _>(3).filter(|&ts| ts != 0); tokens_data.push(( token_blob, token_id, tvl_sats.max(0) as u64, tvl_tokens.max(0) as u64, + first_pool_ts, )); } // ========== COMPUTE PHASE: compute all updates ========== let mut updates: Vec = Vec::with_capacity(tokens_data.len()); let mut dec_cache: HashMap = HashMap::new(); + let mut usd_rate_cache: HashMap = + HashMap::from([(ts_24h, usd_per_sat_24h), (ts_7d, usd_per_sat_7d)]); - for (token_blob, token_id, tvl_sats, tvl_tokens) in tokens_data { + for (token_blob, token_id, tvl_sats, tvl_tokens, first_pool_ts) in tokens_data { let decimals_u32 = if let Some(d) = dec_cache.get(&token_id) { *d } else { @@ -445,6 +493,9 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< price_7d_usd: None, change_24h_usd_bp: None, change_7d_usd_bp: None, + change_24h_anchor_ts: None, + change_7d_anchor_ts: None, + price_series_7d: None, display_name, display_symbol, bcmr_json, @@ -503,6 +554,9 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< price_7d_usd: None, change_24h_usd_bp: None, change_7d_usd_bp: None, + change_24h_anchor_ts: None, + change_7d_anchor_ts: None, + price_series_7d: None, display_name, display_symbol, bcmr_json, @@ -512,55 +566,86 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< } }; + // Historical anchors. For a token younger than the window these fall back to + // its first pool, so a new token reports a since-launch move instead of NULL. + let anchor_24h = price_anchor_for_window(&db.cauldron_r, ts_24h, first_pool_ts, &tid).await; + let anchor_7d = price_anchor_for_window(&db.cauldron_r, ts_7d, first_pool_ts, &tid).await; + // The anchor reports the pool entry the price was read from, which on a token + // that stopped trading can be years before the window. The change still spans + // the whole window — the price simply held — so clamp to the window start. + // That is both the period consumers should be told about and the BCH/USD rate + // the USD side must convert at: anchoring to a years-old entry priced the USD + // side at a years-old rate, or at none at all once the entry predated the + // oracle history, turning a correct 0.00% in BCH into N/A in USD. + let anchor_24h_ts = anchor_24h.as_ref().map(|a| a.ts.max(ts_24h)); + let anchor_7d_ts = anchor_7d.as_ref().map(|a| a.ts.max(ts_7d)); + + // 7d curve, sampled from the same aggregate pool price as the changes + // above so a client can draw it without a request per token. Always the + // nominal window, never the launch-time anchor: the grid has to mean the + // same thing on every row. + let price_series_7d = match pool_reserves_at(&db.cauldron_r, ts_7d, &tid).await { + Ok(reserves) => { + let series = + build_price_series(reserves, window_events.for_token(&token_id), ts_7d, now); + encode_price_series(&series, price_now_dec) + } + Err(e) => { + warn!("price series unavailable for {token_id}: {e:?}"); + None + } + }; + + // Convert at the rate of the snapshot we actually read, not of the nominal + // window start — otherwise a launch-time price gets a 7-days-ago BCH/USD rate. + let usd_per_sat_at_24h = match anchor_24h_ts { + Some(ts) => usd_per_sat_at(&db.oracle_r, &mut usd_rate_cache, ts).await, + None => Decimal::ZERO, + }; + let usd_per_sat_at_7d = match anchor_7d_ts { + Some(ts) => usd_per_sat_at(&db.oracle_r, &mut usd_rate_cache, ts).await, + None => Decimal::ZERO, + }; + // --------- FAST PATH (no trades in 30d) ---------- - // Only gate by *existence* of a historical point; if none → keep NULLs + // No trades means the token price is unchanged; only gate on the *existence* + // of an anchor, and if there is none → keep NULLs. if vol == 0 { - let p24_exists = matches!( - price_at_or_before_2(&db.cauldron_r, ts_24h, &tid).await, - Ok((_ts, _)) - ); - let p7d_exists = matches!( - price_at_or_before_2(&db.cauldron_r, ts_7d, &tid).await, - Ok((_ts, _)) - ); + let price_24h_human_f = anchor_24h + .as_ref() + .map(|_| dec_to_f64_bounded(dec_round(price_now_human_dec, 12))); + let price_7d_human_f = anchor_7d + .as_ref() + .map(|_| dec_to_f64_bounded(dec_round(price_now_human_dec, 12))); - let price_24h_human_f = if p24_exists { - Some(dec_to_f64_bounded(dec_round(price_now_human_dec, 12))) + let d24_bp_opt = anchor_24h.as_ref().map(|_| 0i64); + let d7d_bp_opt = anchor_7d.as_ref().map(|_| 0i64); + + // Keep the unrounded Decimals: `dec_round(_, 12)` is twelve *decimal + // places*, not significant digits, so a token worth ~1e-12 USD per unit + // survives it with a single digit. Taking the delta from the rounded + // value made a flat token report that rounding error as a price move + // (one at 9e-12 came out at +0.90%, another at +18.75%). + let price_24h_usd_dec = if price_24h_human_f.is_some() && !usd_per_sat_at_24h.is_zero() + { + price_now_human_dec.checked_mul(usd_per_sat_at_24h) } else { None }; - let price_7d_human_f = if p7d_exists { - Some(dec_to_f64_bounded(dec_round(price_now_human_dec, 12))) + let price_7d_usd_dec = if price_7d_human_f.is_some() && !usd_per_sat_at_7d.is_zero() { + price_now_human_dec.checked_mul(usd_per_sat_at_7d) } else { None }; - let d24_bp_opt = if p24_exists { Some(0) } else { None }; - let d7d_bp_opt = if p7d_exists { Some(0) } else { None }; + let price_24h_usd_f = price_24h_usd_dec.map(|d| dec_to_f64_bounded(dec_round(d, 12))); + let price_7d_usd_f = price_7d_usd_dec.map(|d| dec_to_f64_bounded(dec_round(d, 12))); - let price_24h_usd_f = if p24_exists && !usd_per_sat_24h.is_zero() { - price_now_human_dec - .checked_mul(usd_per_sat_24h) - .map(|d| dec_to_f64_bounded(dec_round(d, 12))) - } else { - None - }; - let price_7d_usd_f = if p7d_exists && !usd_per_sat_7d.is_zero() { - price_now_human_dec - .checked_mul(usd_per_sat_7d) - .map(|d| dec_to_f64_bounded(dec_round(d, 12))) - } else { - None - }; - - let old24_u_dec_opt = price_24h_usd_f.and_then(Decimal::from_f64); - let old7d_u_dec_opt = price_7d_usd_f.and_then(Decimal::from_f64); - - let d24_usd_bp_opt = match (price_now_usd_dec_opt, old24_u_dec_opt) { + let d24_usd_bp_opt = match (price_now_usd_dec_opt, price_24h_usd_dec) { (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), _ => None, }; - let d7d_usd_bp_opt = match (price_now_usd_dec_opt, old7d_u_dec_opt) { + let d7d_usd_bp_opt = match (price_now_usd_dec_opt, price_7d_usd_dec) { (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), _ => None, }; @@ -578,6 +663,9 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< price_7d_usd: price_7d_usd_f, change_24h_usd_bp: d24_usd_bp_opt, change_7d_usd_bp: d7d_usd_bp_opt, + change_24h_anchor_ts: anchor_24h_ts, + change_7d_anchor_ts: anchor_7d_ts, + price_series_7d, display_name, display_symbol, bcmr_json: bcmr_json.clone(), @@ -587,14 +675,8 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< } // --------- FULL PATH (vol > 0) ---------- - let p24_dec_opt = price_at_or_before_2(&db.cauldron_r, ts_24h, &tid) - .await - .ok() - .and_then(|(_, p)| Decimal::from_f64(p)); - let p7d_dec_opt = price_at_or_before_2(&db.cauldron_r, ts_7d, &tid) - .await - .ok() - .and_then(|(_, p)| Decimal::from_f64(p)); + let p24_dec_opt = anchor_24h.map(|a| a.price); + let p7d_dec_opt = anchor_7d.map(|a| a.price); // scale to "human" with overflow guard (nullable) let price_24h_human_opt = p24_dec_opt.and_then(|p| p.checked_mul(factor)); @@ -605,12 +687,12 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< let d7d_bp_opt = p7d_dec_opt.map(|p| pct_change_bp_dec(price_now_dec, p)); // USD historicals (nullable if oracle missing or overflow) - let price_24h_usd_opt = match (price_24h_human_opt, !usd_per_sat_24h.is_zero()) { - (Some(h), true) => h.checked_mul(usd_per_sat_24h), + let price_24h_usd_opt = match (price_24h_human_opt, !usd_per_sat_at_24h.is_zero()) { + (Some(h), true) => h.checked_mul(usd_per_sat_at_24h), _ => None, }; - let price_7d_usd_opt = match (price_7d_human_opt, !usd_per_sat_7d.is_zero()) { - (Some(h), true) => h.checked_mul(usd_per_sat_7d), + let price_7d_usd_opt = match (price_7d_human_opt, !usd_per_sat_at_7d.is_zero()) { + (Some(h), true) => h.checked_mul(usd_per_sat_at_7d), _ => None, }; @@ -618,9 +700,9 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< p24_dec_opt.map(|p24| to_human_f64_with_fallback(p24, decimals_u32)); let price_7d_human_f = p7d_dec_opt.map(|p7d| to_human_f64_with_fallback(p7d, decimals_u32)); let price_24h_usd_f = - to_usd_f64_with_fallback(price_24h_human_opt, usd_per_sat_24h, decimals_u32); + to_usd_f64_with_fallback(price_24h_human_opt, usd_per_sat_at_24h, decimals_u32); let price_7d_usd_f = - to_usd_f64_with_fallback(price_7d_human_opt, usd_per_sat_7d, decimals_u32); + to_usd_f64_with_fallback(price_7d_human_opt, usd_per_sat_at_7d, decimals_u32); let d24_usd_bp_opt = match (price_now_usd_dec_opt, price_24h_usd_opt) { (Some(now_u), Some(old_u)) => Some(pct_change_bp_dec(now_u, old_u)), @@ -644,6 +726,9 @@ pub async fn update_changes_score_volume_and_ranking(db: &DB) -> anyhow::Result< price_7d_usd: price_7d_usd_f, change_24h_usd_bp: d24_usd_bp_opt, change_7d_usd_bp: d7d_usd_bp_opt, + change_24h_anchor_ts: anchor_24h_ts, + change_7d_anchor_ts: anchor_7d_ts, + price_series_7d, display_name, display_symbol, bcmr_json, diff --git a/src/db/cauldron/tokenlist/test.rs b/src/db/cauldron/tokenlist/test.rs index 1c44197..6a41937 100644 --- a/src/db/cauldron/tokenlist/test.rs +++ b/src/db/cauldron/tokenlist/test.rs @@ -24,7 +24,7 @@ mod tests { }; use crate::db::cauldron::tokenlist::token_utils::{ apy_30d_bp_for_token, compute_score, overflow_f64_fallback, pct_change_bp_dec, pow10_dec, - price_from_tvl, resolve_decimals, resolve_display_labels, + price_anchor_for_window, price_from_tvl, resolve_decimals, resolve_display_labels, }; use crate::db::cauldron::tx::{insert_block_tx, insert_mempool_tx}; use crate::db::cauldron::utxo_funding::insert_utxo_funding; @@ -587,7 +587,7 @@ mod tests { } #[test] - fn young_token_under_24h_is_null() { + fn window_with_no_anchor_at_all_is_null() { let now = dec!(0.00000123); // in sats/base (arbitrary) let factor = dec!(100); // pretend 2 decimals let usd_now = dec!(0.000002); // usd per sat now @@ -731,36 +731,37 @@ mod tests { } #[tokio::test] - async fn core_updater_keeps_deltas_null_for_young_token() { + async fn core_updater_anchors_young_token_at_first_pool_instead_of_null() { let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; let pool = &mock.cauldron_w; - // Seed a YOUNG token with *volume* so we do the full path, but with no 24h anchor. - // Make both events within the last hour. + // A token launched 17h ago: younger than BOTH windows, so neither the 24h nor + // the 7d probe finds a snapshot at the window start. It used to report N/A. let now = time_now(); let token = TokenID::from_byte_array([0xCC; 32]); - let t0 = now - 3600; // 1h ago + let t0 = now - 17 * 3600; let t1 = now - 600; // 10m ago + // launch price = 1_000/500 = 2 sats per base unit seed_minimal_token_history(pool, token, t0, t1, 1_000, 500, 2_000, 1_000).await; - // Ensure there is a cached row picked up by the updater (tvl_sats>0 is enough). + // price_now = 3_000/1_000 = 3 sats → +50% since launch = +5000 bp sqlx::query( "INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, price_now, price_now_usd, updated_at) - VALUES (?, 0, 1, 1, 0, 1.0, 0.5, strftime('%s','now'))", + VALUES (?, 0, 3000, 1000, 0, 3.0, 1.5, strftime('%s','now'))", ) .bind(token.to_blob()) .execute(pool) .await .unwrap(); - // Run core updater; for a "young" token with no historicals, deltas remain NULL update_changes_score_volume_and_ranking(&mock) .await .unwrap(); let row = sqlx::query( - "SELECT change_24h_bp, change_7d_bp, change_24h_usd_bp, change_7d_usd_bp + "SELECT change_24h_bp, change_7d_bp, change_24h_usd_bp, change_7d_usd_bp, + change_24h_anchor_ts, change_7d_anchor_ts FROM cached_token_metrics WHERE token_id=?", ) .bind(token.to_blob()) @@ -772,11 +773,361 @@ mod tests { let c7d: Option = row.get(1); let c24u: Option = row.get(2); let c7du: Option = row.get(3); + let a24: Option = row.get(4); + let a7d: Option = row.get(5); - assert!(c24.is_none()); - assert!(c7d.is_none()); - assert!(c24u.is_none()); - assert!(c7du.is_none()); + assert_eq!(c24, Some(5000), "24h change should measure since launch"); + assert_eq!(c7d, Some(5000), "7d change should measure since launch"); + assert!(c24u.is_some()); + assert!(c7du.is_some()); + // Both windows collapsed onto the token's first pool, and say so. + assert_eq!(a24, Some(t0)); + assert_eq!(a7d, Some(t0)); + } + + #[tokio::test] + async fn core_updater_prices_a_long_dormant_token_at_window_rates() { + let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; + let pool = &mock.cauldron_w; + + // A token whose pool last moved long before the oracle history begins. Its + // BCH change is a genuine 0%, and the USD change is whatever BCH/USD did over + // the window — but anchoring the rate to that ancient pool entry found no + // oracle sample at all and wrote NULL, so the row showed 0.00% in BCH and + // N/A in USD. + let now = time_now(); + let token = TokenID::from_byte_array([0xC3; 32]); + let ancient = now - 800 * 86_400; + seed_minimal_token_history( + pool, + token, + ancient, + ancient + 60, + 1_000, + 1_000, + 2_000, + 2_000, + ) + .await; + + sqlx::query( + "INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, + price_now, price_now_usd, updated_at) + VALUES (?, 0, 3000, 3000, 0, 1.0, 0.5, strftime('%s','now'))", + ) + .bind(token.to_blob()) + .execute(pool) + .await + .unwrap(); + + update_changes_score_volume_and_ranking(&mock) + .await + .unwrap(); + + let row = sqlx::query( + "SELECT change_24h_bp, change_7d_bp, change_24h_usd_bp, change_7d_usd_bp, + change_24h_anchor_ts, change_7d_anchor_ts, price_7d_usd + FROM cached_token_metrics WHERE token_id=?", + ) + .bind(token.to_blob()) + .fetch_one(pool) + .await + .unwrap(); + + assert_eq!( + row.get::, _>(0), + Some(0), + "flat in BCH over 24h" + ); + assert_eq!(row.get::, _>(1), Some(0), "flat in BCH over 7d"); + assert!( + row.get::, _>(2).is_some(), + "24h USD change must not be N/A when the pool simply held its price" + ); + assert!( + row.get::, _>(3).is_some(), + "7d USD change must not be N/A when the pool simply held its price" + ); + assert!( + row.get::, _>(6).is_some(), + "7d USD price present" + ); + // The change spans the window, so that is what the anchors report — not the + // 800-day-old pool entry the price happens to come from. + assert_eq!(row.get::, _>(4), Some(now - 86_400)); + assert_eq!(row.get::, _>(5), Some(now - 7 * 86_400)); + } + + #[tokio::test] + async fn core_updater_keeps_a_micro_priced_dormant_token_flat_in_usd() { + let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; + let pool = &mock.cauldron_w; + + // Sub-1e-9 USD per unit: rounding the historical price to twelve decimal + // places leaves one significant digit, and taking the delta from that made a + // dormant token report the rounding error as a price move. + let now = time_now(); + let token = TokenID::from_byte_array([0xC4; 32]); + seed_minimal_token_history( + pool, + token, + now - 300 * 86_400, + now - 299 * 86_400, + 3, + 3_000_000, + 7, + 9_000_000, + ) + .await; + + // 7/9_000_000 sats per unit at the fixture's $500/BCH is 3.888…e-12 USD, which + // twelve decimal places cannot hold — it lands on 4e-12. + sqlx::query( + "INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, + price_now, price_now_usd, updated_at) + VALUES (?, 0, 7, 9000000, 0, 0.000000777777, 0.000000000004, strftime('%s','now'))", + ) + .bind(token.to_blob()) + .execute(pool) + .await + .unwrap(); + + update_changes_score_volume_and_ranking(&mock) + .await + .unwrap(); + + let row = sqlx::query( + "SELECT change_24h_usd_bp, change_7d_usd_bp FROM cached_token_metrics WHERE token_id=?", + ) + .bind(token.to_blob()) + .fetch_one(pool) + .await + .unwrap(); + + // The oracle fixture holds one rate, so with the price held the USD move is + // exactly zero — any non-zero value here is quantization leaking through. + assert_eq!(row.get::, _>(0), Some(0)); + assert_eq!(row.get::, _>(1), Some(0)); + } + + /// The cached 7d curve, as a client reads it: ratios to price_now. + async fn read_price_series(pool: &SqlitePool, token: TokenID) -> Option>> { + let row = sqlx::query("SELECT price_series_7d FROM cached_token_metrics WHERE token_id=?") + .bind(token.to_blob()) + .fetch_one(pool) + .await + .unwrap(); + row.get::, _>(0) + .map(|s| serde_json::from_str(&s).unwrap()) + } + + #[tokio::test] + async fn core_updater_series_agrees_with_the_change_columns() { + let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; + let pool = &mock.cauldron_w; + + // Aggregate pool price doubles two days ago: 1000/1000 = 1 sat per base unit + // from 9 days ago, then a second pool joins it at 3000/1000 so the summed + // reserves become 4000/2000 = 2. + let now = time_now(); + let token = TokenID::from_byte_array([0xC1; 32]); + seed_minimal_token_history( + pool, + token, + now - 9 * 86_400, + now - 2 * 86_400, + 1_000, + 1_000, + 3_000, + 1_000, + ) + .await; + + sqlx::query( + "INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, + price_now, price_now_usd, updated_at) + VALUES (?, 0, 4000, 2000, 0, 2.0, 1.0, strftime('%s','now'))", + ) + .bind(token.to_blob()) + .execute(pool) + .await + .unwrap(); + + update_changes_score_volume_and_ranking(&mock) + .await + .unwrap(); + + let series = read_price_series(pool, token) + .await + .expect("a token with a pool at the window start gets a series"); + assert_eq!(series.len(), 29, "7 days sampled every 6h, both ends"); + + let row = sqlx::query( + "SELECT price_7d, price_now, change_7d_bp, change_24h_bp + FROM cached_token_metrics WHERE token_id=?", + ) + .bind(token.to_blob()) + .fetch_one(pool) + .await + .unwrap(); + let price_7d: Option = row.get(0); + let price_now: f64 = row.get(1); + let change_7d_bp: Option = row.get(2); + let change_24h_bp: Option = row.get(3); + + // The whole point: the curve's ends ARE the numbers the columns print, so a + // client can draw it without the shape contradicting the percentage. + let first = series[0].expect("price known at the window start"); + let last = series[28].expect("price known now"); + assert_eq!(first, 0.5, "1 sat against a price_now of 2"); + assert_eq!(last, 1.0, "the last sample is price_now itself"); + assert_eq!(price_7d, Some(first * price_now)); + assert_eq!(change_7d_bp, Some(10_000), "price doubled over the week"); + assert_eq!( + ((1.0 / first - 1.0) * 10_000.0).round() as i64, + change_7d_bp.unwrap(), + "change_7d_bp must be recoverable from the first ratio" + ); + + // A 6h step puts now-24h exactly on index 24, and nothing moved in the last + // two days, so that sample and the 1d column agree on "flat". + assert_eq!(series[24], Some(1.0)); + assert_eq!(change_24h_bp, Some(0)); + + // The doubling lands on the sample covering now-2d and not the one before it. + assert_eq!(series[19], Some(0.5)); + assert_eq!(series[20], Some(1.0)); + } + + #[tokio::test] + async fn core_updater_series_is_null_padded_for_a_young_token() { + let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; + let pool = &mock.cauldron_w; + + // Launched three days ago, so the first half of the window predates the token. + let now = time_now(); + let token = TokenID::from_byte_array([0xC2; 32]); + seed_minimal_token_history( + pool, + token, + now - 3 * 86_400, + now - 86_400, + 1_000, + 1_000, + 3_000, + 1_000, + ) + .await; + + sqlx::query( + "INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, + price_now, price_now_usd, updated_at) + VALUES (?, 0, 4000, 2000, 0, 2.0, 1.0, strftime('%s','now'))", + ) + .bind(token.to_blob()) + .execute(pool) + .await + .unwrap(); + + update_changes_score_volume_and_ranking(&mock) + .await + .unwrap(); + + let series = read_price_series(pool, token) + .await + .expect("series present"); + // now-3d is sample 16; everything before it is unknown rather than invented. + assert!( + series[..16].iter().all(|v| v.is_none()), + "buckets before the first pool must stay null: {series:?}" + ); + assert_eq!(series[16], Some(0.5)); + assert_eq!(series[28], Some(1.0)); + } + + #[tokio::test] + async fn core_updater_keeps_deltas_null_when_token_has_no_history() { + let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; + let pool = &mock.cauldron_w; + + // Cached row with liquidity but no pool/tx rows at all: there is no launch to + // fall back to, so the deltas must stay NULL rather than be invented. + let token = TokenID::from_byte_array([0xCD; 32]); + sqlx::query( + "INSERT INTO cached_token_metrics(token_id, trade_volume, tvl_sats, tvl_tokens, score, + price_now, price_now_usd, updated_at) + VALUES (?, 0, 1, 1, 0, 1.0, 0.5, strftime('%s','now'))", + ) + .bind(token.to_blob()) + .execute(pool) + .await + .unwrap(); + + update_changes_score_volume_and_ranking(&mock) + .await + .unwrap(); + + let row = sqlx::query( + "SELECT change_24h_bp, change_7d_bp, change_24h_usd_bp, change_7d_usd_bp, + change_24h_anchor_ts, change_7d_anchor_ts + FROM cached_token_metrics WHERE token_id=?", + ) + .bind(token.to_blob()) + .fetch_one(pool) + .await + .unwrap(); + + for i in 0..6 { + let v: Option = row.get(i); + assert!(v.is_none(), "column {i} should be NULL without any history"); + } + } + + #[tokio::test] + async fn price_anchor_falls_back_to_launch_only_for_young_tokens() { + let mock = mock_db_pool(|pool| async move { setup_basic_schemas(pool).await }).await; + let pool = &mock.cauldron_r; + let now = time_now(); + let window_start = now - 7 * 86_400; + + // Young: launched inside the window, so the window start has no snapshot. + let young = TokenID::from_byte_array([0xC1; 32]); + let young_t0 = now - 2 * 86_400; + seed_minimal_token_history(pool, young, young_t0, now - 3600, 1_000, 500, 2_000, 1_000) + .await; + + // Not knowing first_pool_ts must give the same answer as knowing it — new + // tokens are exactly the ones the first_pool_ts backfill hasn't reached yet. + let uncached = price_anchor_for_window(pool, window_start, None, &young) + .await + .expect("young token should anchor at its launch"); + let cached = price_anchor_for_window(pool, window_start, Some(young_t0), &young) + .await + .expect("young token should anchor at its launch"); + assert_eq!(uncached.ts, young_t0); + assert_eq!(cached.ts, young_t0); + assert_eq!(uncached.price, dec!(2)); // 1_000 sats / 500 base units + + // Old: history reaches past the window start, so the window start wins and the + // launch fallback must not kick in. + let old = TokenID::from_byte_array([0xC2; 32]); + let old_t0 = now - 30 * 86_400; + let old_t1 = now - 20 * 86_400; + seed_minimal_token_history(pool, old, old_t0, old_t1, 1_000, 500, 2_000, 1_000).await; + + let anchor = price_anchor_for_window(pool, window_start, None, &old) + .await + .expect("old token has a snapshot at the window start"); + assert_eq!( + anchor.ts, old_t1, + "should read the latest snapshot at or before the window start, not the launch" + ); + + // No pools at all → no anchor to invent. + let unknown = TokenID::from_byte_array([0xC3; 32]); + assert!(price_anchor_for_window(pool, window_start, None, &unknown) + .await + .is_none()); } #[tokio::test] diff --git a/src/db/cauldron/tokenlist/token_utils.rs b/src/db/cauldron/tokenlist/token_utils.rs index 08ec43a..4e5f12e 100644 --- a/src/db/cauldron/tokenlist/token_utils.rs +++ b/src/db/cauldron/tokenlist/token_utils.rs @@ -10,9 +10,11 @@ use sqlx::SqlitePool; use crate::db::blob::display_hex_to_blob; use crate::db::cauldron::pool::{get_injections_between, get_pool_period_snapshot}; +use crate::db::cauldron::tokenlist::db_utils::db_first_pool_creation_row; use crate::db::oracle::get_closest; use crate::rpc::apy::apyaggregator::APYAggregator; use crate::rpc::apy::poolperiod::split_at_injections; +use crate::rpc::price::price_at_or_before_2; use malachite::base::num::arithmetic::traits::FloorSqrt; use malachite::Integer; @@ -70,6 +72,70 @@ pub async fn usd_per_bch_at_or_before( Ok(price / Decimal::from_i64(ORACLE_SCALE).unwrap()) } +/// The historical end of a price-change window. +/// +/// `ts` is the timestamp of the snapshot the price was actually read from, which is +/// *not* always the window start: for a token younger than the window it is the +/// token's first pool instead, and for a dormant one it can be far older than the +/// window. Callers persist it so consumers can tell a real 7d move from a +/// since-launch one — clamping it to their window first, since a price that merely +/// held does not shorten the period being measured. +#[derive(Debug, Clone, Copy)] +pub struct PriceAnchor { + pub ts: i64, + pub price: Decimal, +} + +async fn price_snapshot_at( + cauldron_pool: &SqlitePool, + ts: i64, + token_id: &TokenID, +) -> Option { + let (snapshot_ts, price) = price_at_or_before_2(cauldron_pool, ts, token_id) + .await + .ok()?; + Decimal::from_f64(price).map(|price| PriceAnchor { + ts: snapshot_ts, + price, + }) +} + +/// Price at `window_start`, falling back to the token's first pool when the token is +/// younger than the window. +/// +/// Without the fallback every token under 7 days old has a NULL `change_7d_bp` and +/// the UI can only show "N/A"; with it a young token reports its change since launch. +/// `first_pool_ts` is the cached value when known — it is looked up on demand +/// otherwise, since brand-new tokens are exactly the ones the backfill hasn't reached. +pub async fn price_anchor_for_window( + cauldron_pool: &SqlitePool, + window_start: i64, + first_pool_ts: Option, + token_id: &TokenID, +) -> Option { + if let Some(anchor) = price_snapshot_at(cauldron_pool, window_start, token_id).await { + return Some(anchor); + } + + let launch_ts = match first_pool_ts.filter(|&ts| ts > 0) { + Some(ts) => ts, + None => db_first_pool_creation_row(cauldron_pool, &token_id.to_string()) + .await + .ok() + .flatten() + .map(|(_creation_utxo, _txid, ts, _height)| ts)?, + }; + + // History already reaches past the window start, so the miss above is a gap in + // the data (e.g. every pool was withdrawn), not a young token. Don't invent an + // anchor that would report an older move as if it were a 7d one. + if launch_ts <= window_start { + return None; + } + + price_snapshot_at(cauldron_pool, launch_ts, token_id).await +} + #[inline] pub fn pct_change_bp_dec(current: Decimal, past: Decimal) -> i64 { if past.is_zero() { diff --git a/src/db/search/mod.rs b/src/db/search/mod.rs index 271f9d1..59244d6 100644 --- a/src/db/search/mod.rs +++ b/src/db/search/mod.rs @@ -235,7 +235,9 @@ pub async fn db_search_tokens_cached( price_now, price_24h, price_7d, change_24h_bp, change_7d_bp, display_name, display_symbol, price_now_usd, price_24h_usd, price_7d_usd, change_24h_usd_bp, change_7d_usd_bp, - apy_30d_bp, first_pool_ts, score_rank, bcmr_json, bcmr_well_known_json + change_24h_anchor_ts, change_7d_anchor_ts, + apy_30d_bp, first_pool_ts, score_rank, bcmr_json, bcmr_well_known_json, + price_series_7d FROM cached_token_metrics {where_sql} {order_sql} @@ -291,11 +293,16 @@ pub async fn db_search_tokens_cached( let price_7d_usd: Option = row.get("price_7d_usd"); let change_24h_usd_bp: Option = row.get("change_24h_usd_bp"); let change_7d_usd_bp: Option = row.get("change_7d_usd_bp"); + let change_24h_anchor_ts: Option = row.get("change_24h_anchor_ts"); + let change_7d_anchor_ts: Option = row.get("change_7d_anchor_ts"); let apy_30d_bp: Option = row.get("apy_30d_bp"); // 0 is the "not backfilled yet" sentinel — treat it as unknown. let first_pool_ts: Option = row .get::, _>("first_pool_ts") .filter(|&ts| ts != 0); + let price_series_7d: Option>> = row + .get::, _>("price_series_7d") + .and_then(|s| serde_json::from_str(&s).ok()); let bcmr_json_str: Option = row.get("bcmr_json"); let bcmr_wk_json_str: Option = row.get("bcmr_well_known_json"); let bcmr: Option = bcmr_json_str.and_then(|s| serde_json::from_str(&s).ok()); @@ -325,8 +332,11 @@ pub async fn db_search_tokens_cached( price_7d_usd, change_24h_usd_bp, change_7d_usd_bp, + change_24h_anchor_ts, + change_7d_anchor_ts, apy_30d_bp, first_pool_ts, + price_series_7d, }); }