From bff99822a343e80366744c43c1b9a32a8fc642b8 Mon Sep 17 00:00:00 2001 From: jakobsn Date: Tue, 11 Aug 2026 15:03:50 +0200 Subject: [PATCH] Nice spot --- src/db/cauldron/candlestick/mod.rs | 335 ++++-------- src/db/cauldron/candlestick/tests.rs | 733 ++++++++++++++++++--------- src/db/cauldron/mod.rs | 1 + src/db/cauldron/ohlcv.rs | 570 +++++++++++++-------- src/db/cauldron/spot.rs | 513 +++++++++++++++++++ src/rpc/candlesticks/tests.rs | 320 ++++++------ 6 files changed, 1628 insertions(+), 844 deletions(-) create mode 100644 src/db/cauldron/spot.rs diff --git a/src/db/cauldron/candlestick/mod.rs b/src/db/cauldron/candlestick/mod.rs index bd40d46..10057b1 100644 --- a/src/db/cauldron/candlestick/mod.rs +++ b/src/db/cauldron/candlestick/mod.rs @@ -3,8 +3,11 @@ // 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 +use std::collections::HashMap; + use crate::db::blob::display_hex_to_blob; use crate::db::cauldron::ohlcv; +use crate::db::cauldron::spot::{self, build_spot_ohlc, Confirmed, SpotState}; use anyhow::{bail, Result}; use bitcoincash::TokenID; use serde::Serialize; @@ -22,159 +25,28 @@ pub struct CandlestickData { pub transaction_count: i64, } -struct PriceInterval { - start: i64, - step: i64, - low: f64, - high: f64, - open: Option, - close: Option, - volume_sats: i64, - volume_tokens: i64, - transaction_count: i64, -} - -impl PriceInterval { - fn new(start: i64, step: i64) -> Self { - Self { - start, - step, - low: f64::MAX, - high: f64::MIN, - open: None, - close: None, - volume_sats: 0, - volume_tokens: 0, - transaction_count: 0, - } - } - - fn to_candlestick_data(&self) -> Option { - if self.transaction_count == 0 { - return None; - } - Some(CandlestickData { - time: self.start, - open: self.open?, - close: self.close?, - high: self.high, - low: self.low, - volume_sats: self.volume_sats, - volume_tokens: self.volume_tokens, - transaction_count: self.transaction_count, - }) - } - - fn end(&self) -> i64 { - self.start + self.step - } -} - -fn aggregate_raw_trades( - all_trades: &[(i64, i64, i64)], - intervals: Vec, - step_size: i64, - mut found_first_trade: bool, - mut last_close_price: Option, -) -> (Vec, bool, Option) { - let mut result = Vec::with_capacity(intervals.len()); - let mut trade_index = 0; - - for interval in intervals { - let interval_start = interval.start; - let interval_end = interval.end(); - let mut pi = PriceInterval::new(interval_start, step_size); - let mut first_trade_in_interval = true; - - while trade_index < all_trades.len() { - let (ts, vol_sats, vol_tokens) = all_trades[trade_index]; - if ts < interval_start { - trade_index += 1; - continue; - } - if ts >= interval_end { - break; - } - - if vol_tokens != 0 { - let price = vol_sats as f64 / vol_tokens as f64; - if first_trade_in_interval { - pi.open = Some(price); - pi.high = price; - pi.low = price; - first_trade_in_interval = false; - } - pi.close = Some(price); - if price.is_finite() { - pi.high = pi.high.max(price); - pi.low = pi.low.min(price); - } - } - - pi.volume_sats += vol_sats; - pi.volume_tokens += vol_tokens; - pi.transaction_count += 1; - trade_index += 1; - } - - // Carry forward last close when volume exists but net tokens are zero. - if pi.transaction_count > 0 && (pi.open.is_none() || pi.close.is_none()) { - if let Some(prev) = last_close_price { - if pi.open.is_none() { - pi.open = Some(prev); - } - if pi.close.is_none() { - pi.close = Some(prev); - } - if pi.high == f64::MIN { - pi.high = prev; - } - if pi.low == f64::MAX { - pi.low = prev; - } - } - } - - if let Some(candle) = pi.to_candlestick_data() { - found_first_trade = true; - if let Some(close_price) = pi.close { - last_close_price = Some(close_price); - } - result.push(candle); - } else if found_first_trade { - if let Some(prev_close) = last_close_price { - result.push(CandlestickData { - time: interval_start, - open: prev_close, - close: prev_close, - high: prev_close, - low: prev_close, - volume_sats: 0, - volume_tokens: 0, - transaction_count: 0, - }); - } - } - } - - (result, found_first_trade, last_close_price) -} - +/// Fills the buckets `ohlcv_1h` does not store. +/// +/// Only buckets containing a pool change are materialised; a bucket with no events +/// repeats the previous close exactly, so it is reconstructed here. `seed` is the +/// price entering the window — without it, a window that opens on a quiet stretch +/// would start blank and disagree with the same period viewed at another timeframe. +/// +/// Returns the candles and the close carried out of the window. fn fill_ohlcv_candles( rows: Vec, start: i64, end: i64, - mut found_first_trade: bool, - mut last_close: Option, -) -> (Vec, bool, Option) { + seed: Option, +) -> (Vec, Option) { let mut result = Vec::new(); let mut row_iter = rows.into_iter().peekable(); + let mut last_close = seed; let mut bucket = start; while bucket < end { if row_iter.peek().map(|r| r.bucket_ts) == Some(bucket) { let r = row_iter.next().unwrap(); - found_first_trade = true; last_close = Some(r.close); result.push(CandlestickData { time: r.bucket_ts, @@ -186,36 +58,34 @@ fn fill_ohlcv_candles( volume_tokens: r.volume_tokens, transaction_count: r.tx_count, }); - } else if found_first_trade { - if let Some(prev) = last_close { - result.push(CandlestickData { - time: bucket, - open: prev, - close: prev, - high: prev, - low: prev, - volume_sats: 0, - volume_tokens: 0, - transaction_count: 0, - }); - } + } else if let Some(prev) = last_close { + result.push(CandlestickData { + time: bucket, + open: prev, + close: prev, + high: prev, + low: prev, + volume_sats: 0, + volume_tokens: 0, + transaction_count: 0, + }); } bucket += 3600; } - (result, found_first_trade, last_close) + (result, last_close) } /// Returns one row per transaction: `(effective_timestamp, volume_sats, volume_tokens)`. /// -/// Volumes are gross sums of the absolute per-leg deltas, so the derived price -/// `volume_sats / volume_tokens` is the volume-weighted average of the prices actually -/// executed by that transaction's legs, and is therefore always bounded by the cheapest -/// and dearest leg. Summing the *signed* deltas instead lets a multi-pool arbitrage -/// transaction — which buys from one pool and sells into others — cancel almost all of -/// its token movement and divide real satoshis by a near-zero remainder, fabricating a -/// price no leg ever traded at. -async fn fetch_raw_trades( +/// Volumes are gross sums of the absolute per-leg deltas. Summing the *signed* deltas +/// instead lets a multi-pool arbitrage transaction — which buys from one pool and +/// sells into others — cancel almost all of its token movement and report a fraction +/// of the volume it really moved. +/// +/// Unlike `ohlcv_1h`, this path includes mempool transactions, so a trade shows up in +/// the newest candle as soon as it is seen. +async fn fetch_tx_volumes( pool: &SqlitePool, token_blob: &[u8], timestamp_start: i64, @@ -247,36 +117,59 @@ ORDER BY phe.effective_timestamp ASC, min_sequence ASC; .collect()) } -/// Returns the close price of the most recent priceable trade strictly before -/// `timestamp_end`, using the same per-tx aggregation as `fetch_raw_trades`. -/// Returns `None` when no prior trade exists (new token, no history). -async fn fetch_last_close_before( +/// Candles for `[start, end)` built straight from `pool_history_entry`. +/// +/// Prices come from replaying the aggregate pool spot price; volumes from the gross +/// per-transaction sums. The two are independent on purpose: a withdrawal moves the +/// price with no volume, and a trade that nets to zero tokens still moved satoshis. +async fn raw_candles( pool: &SqlitePool, token_blob: &[u8], - timestamp_end: i64, -) -> Result> { - let sql = r#" -SELECT - CAST(SUM(ABS(phe.sats_delta)) AS REAL) / CAST(SUM(ABS(phe.token_delta)) AS REAL) AS close_price -FROM pool_history_entry AS phe -WHERE phe.token_id = ? - AND phe.effective_timestamp < ? -GROUP BY phe.txid, phe.effective_timestamp -HAVING SUM(ABS(phe.token_delta)) != 0 -ORDER BY phe.effective_timestamp DESC, MIN(phe.sequence) DESC -LIMIT 1 -"#; - let row = sqlx::query(sql) - .bind(token_blob) - .bind(timestamp_end) - .fetch_optional(pool) - .await?; + start: i64, + end: i64, + step_size: i64, +) -> Result> { + let reserves = spot::load_snapshot(pool, Some(token_blob), start, Confirmed::OrMempool) + .await? + .remove(token_blob) + .unwrap_or_default(); + let events = spot::load_events(pool, Some(token_blob), start, end, Confirmed::OrMempool) + .await? + .remove(token_blob) + .unwrap_or_default(); - Ok(row.map(|r| r.get::(0))) + let mut volumes: HashMap = HashMap::new(); + for (ts, vol_sats, vol_tokens) in fetch_tx_volumes(pool, token_blob, start, end).await? { + let bucket = start + ((ts - start) / step_size) * step_size; + let entry = volumes.entry(bucket).or_insert((0, 0, 0)); + entry.0 += vol_sats; + entry.1 += vol_tokens; + entry.2 += 1; + } + + Ok( + build_spot_ohlc(SpotState::new(reserves), &events, start, end, step_size) + .into_iter() + .map(|c| { + let (volume_sats, volume_tokens, transaction_count) = + volumes.get(&c.time).copied().unwrap_or((0, 0, 0)); + CandlestickData { + time: c.time, + open: c.open, + close: c.close, + high: c.high, + low: c.low, + volume_sats, + volume_tokens, + transaction_count, + } + }) + .collect(), + ) } /// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`. -/// Pass 0 to always use the raw CTE path. +/// Pass 0 to always use the raw path. pub async fn candlesticks( pool: &SqlitePool, timestamp_start: i64, @@ -291,12 +184,6 @@ pub async fn candlesticks( let token_blob = display_hex_to_blob::(token_id)?; - // Seed gap-fill with the last known close price before this window so that - // switching between timeframes (e.g. 1W vs 1M) produces consistent prices - // for any overlapping period. - let seed_close = fetch_last_close_before(pool, &token_blob, timestamp_start).await?; - let seed_found = seed_close.is_some(); - // Fast path: use pre-materialised ohlcv_1h when step_size is exactly 1 hour, // ohlcv covers at least part of the range, AND the start is hour-aligned. // ohlcv_1h buckets are always aligned to multiples of 3600, so a non-aligned @@ -305,55 +192,39 @@ pub async fn candlesticks( { let ohlcv_end = ohlcv_materialized_end.min(timestamp_end); + // Seed gap-fill with the pool price entering the window, so switching between + // timeframes (e.g. 1W vs 1M) produces the same prices for the overlap. + let seed = SpotState::new( + spot::load_snapshot( + pool, + Some(&token_blob), + timestamp_start, + Confirmed::OrMempool, + ) + .await? + .remove(&token_blob) + .unwrap_or_default(), + ) + .price(); + // timestamp_start is guaranteed hour-aligned by the entry condition above. let ohlcv_rows = ohlcv::get_active_candles(pool, &token_blob, timestamp_start, ohlcv_end).await?; - let (mut result, found_first, last_close) = fill_ohlcv_candles( - ohlcv_rows, - timestamp_start, - ohlcv_end, - seed_found, - seed_close, - ); + let (mut result, _) = fill_ohlcv_candles(ohlcv_rows, timestamp_start, ohlcv_end, seed); if ohlcv_end < timestamp_end { - // Tail: query raw for [ohlcv_end, timestamp_end) and append. - let raw_trades = fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?; - - let mut tail_intervals = Vec::new(); - let mut t = ohlcv_end; - while t < timestamp_end { - tail_intervals.push(PriceInterval::new(t, step_size)); - t += step_size; - } - - let (tail, _, _) = aggregate_raw_trades( - &raw_trades, - tail_intervals, - step_size, - found_first, - last_close, - ); - result.extend(tail); + // Tail: replay raw for [ohlcv_end, timestamp_end) and append. It takes its + // own snapshot at `ohlcv_end`, which is the same aggregate price the last + // materialised bucket closed at, so the join is seamless. + result + .extend(raw_candles(pool, &token_blob, ohlcv_end, timestamp_end, step_size).await?); } return Ok(result); } - // Raw path: full CTE scan (all non-3600 step sizes, or when ohlcv is not ready). - let mut intervals = Vec::new(); - let mut current_start = timestamp_start; - while current_start < timestamp_end { - intervals.push(PriceInterval::new(current_start, step_size)); - current_start += step_size; - } - - let all_trades = fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?; - - let (result, _, _) = - aggregate_raw_trades(&all_trades, intervals, step_size, seed_found, seed_close); - Ok(result) + raw_candles(pool, &token_blob, timestamp_start, timestamp_end, step_size).await } #[cfg(test)] diff --git a/src/db/cauldron/candlestick/tests.rs b/src/db/cauldron/candlestick/tests.rs index 7ddeb4f..6a5c36b 100644 --- a/src/db/cauldron/candlestick/tests.rs +++ b/src/db/cauldron/candlestick/tests.rs @@ -4,16 +4,21 @@ // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html use crate::db::blob::ToBlob; +use crate::db::cauldron::spot::{self, Confirmed, SpotState}; use crate::db::cauldron::{ ohlcv, pool::{self, dummy_init_seq}, tx::{self, insert_block_tx, insert_mempool_tx}, utxo_funding::{self, insert_utxo_funding}, + utxo_spending, }; use crate::utiltest::mock_db_pool; use bitcoin_hashes::Hash; use bitcoincash::{BlockHash, PubkeyHash, TokenID, Txid}; use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash}; +use sqlx::SqlitePool; + +const HOUR: u64 = 3600; fn dummy_cauldron( txid: &Txid, @@ -38,320 +43,554 @@ fn dummy_cauldron( async fn setup_db(pool: sqlx::SqlitePool) { utxo_funding::create_table(&pool).await; + utxo_spending::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; ohlcv::create_table(&pool).await; dummy_init_seq(); } -async fn insert_trade_at( - conn: &mut sqlx::pool::PoolConnection, - token: &TokenID, - txid_byte: u8, - ts: u64, - sats_delta: i64, - token_delta: i64, -) { - let txid = Txid::from_byte_array([txid_byte; 32]); - let utxo = OutPointHash::from_byte_array([txid_byte; 32]); - let pool_hash = OutPointHash::from_byte_array([txid_byte.wrapping_add(0x80); 32]); - let block = BlockHash::all_zeros(); +/// One pool touched by a transaction: the reserves it is left holding, and the +/// deltas that got it there. +struct Leg { + pool: u8, + reserves: (u64, i64), + deltas: (i64, i64), +} - let cauldron = dummy_cauldron( - &txid, - &utxo, - token, - sats_delta.unsigned_abs(), - token_delta, - &PubkeyHash::all_zeros(), - ); +fn pool_hash(pool: u8) -> OutPointHash { + OutPointHash::from_byte_array([pool; 32]) +} - insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid) - .await - .unwrap(); - insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap(); - insert_block_tx(&mut **conn, &txid, &block, ts as i64) - .await - .unwrap(); - pool::insert_pool_history_entry( - &mut **conn, - &pool_hash, - &cauldron, - Some(ts), - Some(ts), - sats_delta, - token_delta, +/// Register a pool so the `pool` join in [`spot::load_snapshot`] can see it. +/// Production writes this row via `insert_new_pool` when the pool is summoned. +async fn create_pool(conn: &mut sqlx::SqliteConnection, pool: u8, token: &TokenID) { + sqlx::query( + "INSERT OR IGNORE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) + VALUES (?, ?, ?, NULL)", ) + .bind(pool_hash(pool).to_blob()) + .bind(PubkeyHash::all_zeros().to_blob()) + .bind(token.to_blob()) + .execute(&mut *conn) .await .unwrap(); } -/// Insert one transaction that touches several pools, each leg with its own -/// `(sats_delta, token_delta)`. Models a router/arbitrage transaction. -async fn insert_multileg_trade_at( +/// Insert one transaction touching every pool in `legs`. +async fn insert_tx( conn: &mut sqlx::pool::PoolConnection, token: &TokenID, txid_byte: u8, ts: u64, - legs: &[(i64, i64)], + legs: &[Leg], + confirmed: bool, ) { let txid = Txid::from_byte_array([txid_byte; 32]); - let block = BlockHash::all_zeros(); insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap(); - insert_block_tx(&mut **conn, &txid, &block, ts as i64) - .await - .unwrap(); + if confirmed { + insert_block_tx(&mut **conn, &txid, &BlockHash::all_zeros(), ts as i64) + .await + .unwrap(); + } - for (i, (sats_delta, token_delta)) in legs.iter().enumerate() { + for leg in legs { + create_pool(&mut **conn, leg.pool, token).await; + + // The entry's own utxo must be unique per (transaction, pool). let mut utxo_bytes = [txid_byte; 32]; - utxo_bytes[0] = i as u8; + utxo_bytes[0] = leg.pool; let utxo = OutPointHash::from_byte_array(utxo_bytes); - let mut pool_bytes = [txid_byte.wrapping_add(0x80); 32]; - pool_bytes[0] = i as u8; - let pool_hash = OutPointHash::from_byte_array(pool_bytes); - - // Post-trade reserves large enough to look like a real pool. let cauldron = dummy_cauldron( &txid, &utxo, token, - sats_delta.unsigned_abs() * 10, - token_delta.abs() * 10, + leg.reserves.0, + leg.reserves.1, &PubkeyHash::all_zeros(), ); - insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid) .await .unwrap(); pool::insert_pool_history_entry( &mut **conn, - &pool_hash, + &pool_hash(leg.pool), &cauldron, Some(ts), Some(ts), - *sats_delta, - *token_delta, + leg.deltas.0, + leg.deltas.1, ) .await .unwrap(); } } -/// Regression for the netting artifact: a multi-pool arbitrage transaction that buys -/// from one pool and sells into another nets its token movement to almost nothing. -/// Dividing the signed sums produced a price no leg traded at — on mainnet token NWB -/// (tx 1E84F4E9…1916, 27 legs) that printed 30,792,599.5 sats/unit against legs that -/// actually executed between 0.288 and 0.335. -#[tokio::test] -async fn test_multipool_arb_priced_by_gross_volume_not_net() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xC1; 32]); - let token_blob = token.to_blob(); - - // Real leg totals from the NWB transaction, collapsed to two legs. - let sell = (-446_491_239i64, 1_334_527_069i64); // executes at 0.334569 - let buy = (384_906_040i64, -1_334_527_067i64); // executes at 0.288421 - let mut conn = db.cauldron_w.acquire().await.unwrap(); - insert_multileg_trade_at(&mut conn, &token, 0x21, 1000, &[sell, buy]).await; - - let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000) - .await - .unwrap() - .expect("arb transaction must price"); - - let low_leg = 384_906_040.0 / 1_334_527_067.0; - let high_leg = 446_491_239.0 / 1_334_527_069.0; - assert!( - price >= low_leg && price <= high_leg, - "price {price} must lie within the executed leg range [{low_leg}, {high_leg}]" - ); - - // Net-ratio pricing would divide 61,585,199 sats by 2 token units. - let net_ratio = 61_585_199.0 / 2.0; - assert!( - price < net_ratio / 1000.0, - "price {price} must not resemble the netting artifact {net_ratio}" - ); - - let expected = 831_397_279.0 / 2_669_054_136.0; - assert!((price - expected).abs() < 1e-9, "expected {expected}"); +/// A single-pool trade: reserves left behind, and the deltas that made them. +fn leg(pool: u8, reserves: (u64, i64), deltas: (i64, i64)) -> Leg { + Leg { + pool, + reserves, + deltas, + } } -/// Every leg pointing the same way is the ordinary case: gross and net agree exactly, -/// so 99.76% of mainnet prints — including the OLA supply-shock crash — are untouched. -#[tokio::test] -async fn test_single_direction_multileg_price_matches_net_ratio() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xC2; 32]); - let token_blob = token.to_blob(); - - let legs = [(150_000i64, -3_000i64), (50_000, -1_000), (99_000, -2_000)]; - let mut conn = db.cauldron_w.acquire().await.unwrap(); - insert_multileg_trade_at(&mut conn, &token, 0x22, 1000, &legs).await; - - let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000) +/// Withdraw `pool` in a transaction at `ts`, the way `flag_as_withdrawn` does. +async fn withdraw_pool( + conn: &mut sqlx::pool::PoolConnection, + pool: u8, + txid_byte: u8, + ts: u64, +) { + let txid = Txid::from_byte_array([txid_byte; 32]); + insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap(); + insert_block_tx(&mut **conn, &txid, &BlockHash::all_zeros(), ts as i64) .await - .unwrap() - .expect("router transaction must price"); - - let signed_sats: i64 = legs.iter().map(|l| l.0).sum(); - let signed_tokens: i64 = legs.iter().map(|l| l.1).sum(); - let net_ratio = (signed_sats as f64 / signed_tokens as f64).abs(); - assert!( - (price - net_ratio).abs() < f64::EPSILON, - "single-direction transactions must be unaffected: {price} vs {net_ratio}" - ); + .unwrap(); + sqlx::query("INSERT OR REPLACE INTO utxo_spending (spent_utxo_hash, txid) VALUES (?, ?)") + .bind(pool_hash(pool).to_blob()) + .bind(txid.to_blob()) + .execute(&mut **conn) + .await + .unwrap(); + sqlx::query("UPDATE pool SET withdrawn_in_utxo = ? WHERE creation_utxo = ?") + .bind(pool_hash(pool).to_blob()) + .bind(pool_hash(pool).to_blob()) + .execute(&mut **conn) + .await + .unwrap(); } -/// A transaction whose legs cancel exactly used to print nothing at all (the chart -/// carried the previous close). Its legs are real executions and now price normally. -#[tokio::test] -async fn test_exactly_cancelling_legs_still_price() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xC3; 32]); +/// The aggregate pool price a window would open at — what seeds gap-fill. +async fn seed_price(conn: &SqlitePool, token: &TokenID, ts: i64) -> Option { let token_blob = token.to_blob(); + SpotState::new( + spot::load_snapshot(conn, Some(&token_blob), ts, Confirmed::OrMempool) + .await + .unwrap() + .remove(&token_blob) + .unwrap_or_default(), + ) + .price() +} +/// The bug this pricing rule exists for. +/// +/// Reserves and deltas are mainnet GIRL's (token 63664918…f455), the buy at +/// 2026-08-10 15:45:54 and the sell at 2026-08-11 08:52:47. Priced by execution +/// average those print 0.0784 then 0.0836 — the chart stepping *up* on a sell, in +/// a pool whose price had just fallen from 0.0926 to 0.0760. +#[tokio::test] +async fn test_sell_after_buy_closes_lower() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xC0; 32]); let mut conn = db.cauldron_w.acquire().await.unwrap(); - insert_multileg_trade_at( + + insert_tx( &mut conn, &token, - 0x23, - 1000, - &[(-9_000, 1_000), (11_000, -1_000)], + 0x01, + HOUR, + &[leg( + 1, + (540_052, 818_802_757_370_920), + (100_000, -185_511_337_091_708), + )], + true, + ) + .await; + insert_tx( + &mut conn, + &token, + 0x02, + 2 * HOUR, + &[leg( + 1, + (640_052, 691_199_193_943_404), + (100_000, -127_603_563_427_516), + )], + true, + ) + .await; + insert_tx( + &mut conn, + &token, + 0x03, + 3 * HOUR, + &[leg( + 1, + (580_053, 762_931_584_125_945), + (-59_999, 71_732_390_182_541), + )], + true, ) .await; - let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000) - .await - .unwrap() - .expect("zero-net transaction must still price from its legs"); + let candles = super::candlesticks( + &db.cauldron_r, + HOUR as i64, + 4 * HOUR as i64, + HOUR as i64, + &token.to_string(), + 0, + ) + .await + .unwrap(); - // 20_000 gross sats over 2_000 gross tokens, between the 9 and 11 leg prices. - assert!((price - 10.0).abs() < f64::EPSILON, "got {price}"); -} - -/// Legs that move no tokens cannot produce a price (division by zero volume). -#[tokio::test] -async fn test_token_less_legs_do_not_price() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xC4; 32]); - let token_blob = token.to_blob(); - - let mut conn = db.cauldron_w.acquire().await.unwrap(); - insert_multileg_trade_at(&mut conn, &token, 0x24, 1000, &[(5_000, 0), (7_000, 0)]).await; - - let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000) - .await - .unwrap(); - assert!(price.is_none(), "no token movement means no price"); -} - -#[tokio::test] -async fn test_fetch_last_close_before_no_trades() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xAA; 32]); - let token_blob = token.to_blob(); - - let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2_000_000_000) - .await - .unwrap(); - assert!(result.is_none()); -} - -#[tokio::test] -async fn test_fetch_last_close_before_only_future_trades() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xAB; 32]); - let token_blob = token.to_blob(); - - let mut conn = db.cauldron_w.acquire().await.unwrap(); - insert_trade_at(&mut conn, &token, 0x01, 2000, 100_000, 2_000).await; - - let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 1000) - .await - .unwrap(); - assert!(result.is_none()); -} - -#[tokio::test] -async fn test_fetch_last_close_before_returns_most_recent() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xAC; 32]); - let token_blob = token.to_blob(); - - let mut conn = db.cauldron_w.acquire().await.unwrap(); - // ts=1000: price = 40_000/2_000 = 20 - insert_trade_at(&mut conn, &token, 0x01, 1000, 40_000, 2_000).await; - // ts=2000: price = 100_000/2_000 = 50 ← most recent before 3000 - insert_trade_at(&mut conn, &token, 0x02, 2000, 100_000, 2_000).await; - // ts=4000: after cutoff, must be excluded - insert_trade_at(&mut conn, &token, 0x03, 4000, 200_000, 2_000).await; - - let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 3000) - .await - .unwrap(); - assert!(result.is_some()); - assert!((result.unwrap() - 50.0).abs() < f64::EPSILON); -} - -/// Trade exactly AT timestamp_end must be excluded — the query uses strict `<`. -#[tokio::test] -async fn test_fetch_last_close_before_boundary_excluded() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xAD; 32]); - let token_blob = token.to_blob(); - - let mut conn = db.cauldron_w.acquire().await.unwrap(); - insert_trade_at(&mut conn, &token, 0x04, 1000, 100_000, 2_000).await; - - // Query exactly at ts=1000: that trade must NOT be included (strict <). - let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 1000) - .await - .unwrap(); - assert!(result.is_none(), "trade at cutoff must be excluded"); -} - -/// Trades with net-zero token delta (signed_tokens == 0) are invisible to pricing. -/// Only the last priceable trade before the cutoff should be returned. -#[tokio::test] -async fn test_fetch_last_close_before_skips_net_zero_token_trades() { - let db = mock_db_pool(setup_db).await; - let token = TokenID::from_byte_array([0xAE; 32]); - let token_blob = token.to_blob(); - - let mut conn = db.cauldron_w.acquire().await.unwrap(); - // ts=500: priceable trade, price = 50_000/1_000 = 50 - insert_trade_at(&mut conn, &token, 0x05, 500, 50_000, 1_000).await; - // ts=800: net-zero token trade — should be invisible to pricing - insert_trade_at(&mut conn, &token, 0x06, 800, 10_000, 0).await; - - let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 1000) - .await - .unwrap(); - assert!(result.is_some()); - // Must return the priceable trade's close (50), not be confused by the net-zero one. + assert_eq!(candles.len(), 3); + let buy = candles[1].close; + let sell = candles[2].close; assert!( - (result.unwrap() - 50.0).abs() < f64::EPSILON, - "net-zero trade must not affect close price" + (buy * 1e8 - 0.0926).abs() < 1e-4, + "buy must close at the price it created, got {}", + buy * 1e8 + ); + assert!( + (sell * 1e8 - 0.0760).abs() < 1e-4, + "sell must close at the price it created, got {}", + sell * 1e8 + ); + assert!( + sell < buy, + "a sell closed at {} above the buy before it at {} — the execution-average artifact", + sell * 1e8, + buy * 1e8 ); } -/// Trades for a different token must not bleed into results for the queried token. +/// A multi-pool arbitrage transaction buys from one pool and sells into another, so +/// its legs nearly cancel. Dividing the signed deltas printed a price no leg traded +/// at — on mainnet token NWB (tx 1E84F4E9…1916, 27 legs) that was 30,792,599.5 +/// sats/unit against legs executing between 0.288 and 0.335. Pricing by reserves +/// cannot express the artifact at all: reserves are never negative and never cancel. #[tokio::test] -async fn test_fetch_last_close_before_token_isolation() { +async fn test_multipool_arb_prices_at_its_reserves() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xC1; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + // Real leg totals from the NWB transaction, collapsed to two pools. + insert_tx( + &mut conn, + &token, + 0x21, + HOUR, + &[ + leg(1, (1_000_000, 3_000_000), (-446_491_239, 1_334_527_069)), + leg(2, (2_000_000, 6_000_000), (384_906_040, -1_334_527_067)), + ], + true, + ) + .await; + + let price = seed_price(&db.cauldron_r, &token, 2 * HOUR as i64) + .await + .expect("arb transaction must price"); + + let expected = 3_000_000.0 / 9_000_000.0; + assert!((price - expected).abs() < 1e-12, "got {price}"); + + // Net-ratio pricing would divide 61,585,199 sats by 2 token units. + assert!( + price < 61_585_199.0 / 2.0 / 1000.0, + "price {price} must not resemble the netting artifact" + ); +} + +/// Price comes from reserves, but volume still comes from the gross per-leg sums — +/// an arbitrage transaction moved every satoshi and token its legs moved. +#[tokio::test] +async fn test_multipool_arb_keeps_gross_volume() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xC2; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x22, + HOUR, + &[ + leg(1, (1_000_000, 3_000_000), (-446_491_239, 1_334_527_069)), + leg(2, (2_000_000, 6_000_000), (384_906_040, -1_334_527_067)), + ], + true, + ) + .await; + + let candles = super::candlesticks( + &db.cauldron_r, + HOUR as i64, + 2 * HOUR as i64, + HOUR as i64, + &token.to_string(), + 0, + ) + .await + .unwrap(); + + assert_eq!(candles[0].volume_sats, 446_491_239 + 384_906_040); + assert_eq!(candles[0].volume_tokens, 1_334_527_069 + 1_334_527_067); + assert_eq!(candles[0].transaction_count, 1, "one transaction, two legs"); +} + +/// A withdrawal has no `pool_history_entry` row of its own; if the replay misses it +/// the drained pool's reserves stay in the sum forever. +#[tokio::test] +async fn test_withdrawal_moves_price_without_volume() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xC3; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x31, + HOUR, + &[leg(1, (100, 100), (10, -10)), leg(2, (900, 100), (10, -10))], + true, + ) + .await; + withdraw_pool(&mut conn, 2, 0x32, 2 * HOUR).await; + + let candles = super::candlesticks( + &db.cauldron_r, + HOUR as i64, + 4 * HOUR as i64, + HOUR as i64, + &token.to_string(), + 0, + ) + .await + .unwrap(); + + assert_eq!(candles[0].close, 5.0, "both pools live"); + // recorded a second late, so it lands in the bucket after the withdrawal + assert_eq!(candles[1].close, 1.0, "withdrawn pool must leave the sum"); + assert_eq!(candles[1].volume_sats, 0, "a withdrawal is not volume"); + assert_eq!(candles[2].close, 1.0, "and must stay out"); +} + +#[tokio::test] +async fn test_seed_price_without_pools_is_none() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xAA; 32]); + assert!(seed_price(&db.cauldron_r, &token, 2_000_000_000) + .await + .is_none()); +} + +#[tokio::test] +async fn test_seed_price_ignores_later_trades() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xAB; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x01, + 2000, + &[leg(1, (100_000, 2_000), (100_000, 2_000))], + true, + ) + .await; + + assert!(seed_price(&db.cauldron_r, &token, 1000).await.is_none()); +} + +/// Each pool contributes its latest state before the cutoff, not its first. +#[tokio::test] +async fn test_seed_price_takes_the_latest_state_per_pool() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xAC; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x01, + 1000, + &[leg(1, (40_000, 2_000), (40_000, 2_000))], + true, + ) + .await; + insert_tx( + &mut conn, + &token, + 0x02, + 2000, + &[leg(1, (100_000, 2_000), (60_000, 0))], + true, + ) + .await; + insert_tx( + &mut conn, + &token, + 0x03, + 4000, + &[leg(1, (200_000, 2_000), (100_000, 0))], + true, + ) + .await; + + let price = seed_price(&db.cauldron_r, &token, 3000).await.unwrap(); + assert!((price - 50.0).abs() < f64::EPSILON, "got {price}"); +} + +/// An entry exactly at the cutoff belongs to the window, not to its seed. +#[tokio::test] +async fn test_seed_price_excludes_the_boundary() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xAD; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x04, + 1000, + &[leg(1, (100_000, 2_000), (100_000, 2_000))], + true, + ) + .await; + + assert!( + seed_price(&db.cauldron_r, &token, 1000).await.is_none(), + "entry at the cutoff must be left to the window" + ); +} + +/// The price is summed reserves over every live pool, not an average of per-pool +/// prices — a deep pool must dominate a shallow one. +#[tokio::test] +async fn test_seed_price_sums_live_pools() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xAE; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x05, + 500, + &[ + leg(1, (100, 100), (100, 100)), + leg(2, (9_900, 900), (9_900, 900)), + ], + true, + ) + .await; + + let price = seed_price(&db.cauldron_r, &token, 1000).await.unwrap(); + // 10_000 sats over 1_000 tokens; the per-pool average would be 6. + assert!((price - 10.0).abs() < f64::EPSILON, "got {price}"); +} + +#[tokio::test] +async fn test_seed_price_is_token_isolated() { let db = mock_db_pool(setup_db).await; let token_a = TokenID::from_byte_array([0xAF; 32]); let token_b = TokenID::from_byte_array([0xBF; 32]); - let token_a_blob = token_a.to_blob(); - let mut conn = db.cauldron_w.acquire().await.unwrap(); - // Only insert a trade for token_b; token_a has nothing. - insert_trade_at(&mut conn, &token_b, 0x07, 500, 100_000, 2_000).await; - let result = super::fetch_last_close_before(&db.cauldron_r, &token_a_blob, 1000) - .await - .unwrap(); - assert!(result.is_none(), "other token's trade must not appear"); + insert_tx( + &mut conn, + &token_b, + 0x07, + 500, + &[leg(1, (100_000, 2_000), (100_000, 2_000))], + true, + ) + .await; + + assert!( + seed_price(&db.cauldron_r, &token_a, 1000).await.is_none(), + "another token's pool must not appear" + ); +} + +/// The live path shows a trade as soon as it is seen; `ohlcv_1h` waits for a block. +#[tokio::test] +async fn test_raw_path_prices_mempool_trades() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xB1; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x41, + HOUR, + &[leg(1, (300, 100), (300, 100))], + false, + ) + .await; + + let candles = super::candlesticks( + &db.cauldron_r, + HOUR as i64, + 2 * HOUR as i64, + HOUR as i64, + &token.to_string(), + 0, + ) + .await + .unwrap(); + assert_eq!(candles[0].close, 3.0); + + let confirmed = spot::load_events( + &db.cauldron_r, + Some(&token.to_blob()), + HOUR as i64, + 2 * HOUR as i64, + Confirmed::Only, + ) + .await + .unwrap(); + assert!( + confirmed.is_empty(), + "an unconfirmed trade must stay out of the materialised table" + ); +} + +/// A quiet stretch repeats the last close rather than dropping out of the series, +/// so the same period looks identical at every timeframe. +#[tokio::test] +async fn test_quiet_buckets_carry_the_last_close() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xB2; 32]); + let mut conn = db.cauldron_w.acquire().await.unwrap(); + + insert_tx( + &mut conn, + &token, + 0x51, + HOUR, + &[leg(1, (700, 100), (700, 100))], + true, + ) + .await; + + let candles = super::candlesticks( + &db.cauldron_r, + HOUR as i64, + 5 * HOUR as i64, + HOUR as i64, + &token.to_string(), + 0, + ) + .await + .unwrap(); + + assert_eq!(candles.len(), 4); + for candle in &candles { + assert_eq!(candle.close, 7.0); + } + assert_eq!( + candles[3].transaction_count, 0, + "quiet buckets have no trades" + ); } diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index 84c19ae..644a5a9 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -21,6 +21,7 @@ pub mod ohlcv; pub mod pool; pub mod poolvisitor; pub mod priceseries; +pub mod spot; pub mod tokenlist; pub mod tokentoken; pub mod tx; diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index fb168c0..8932a5b 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -3,7 +3,10 @@ // 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 +use std::collections::HashMap; + use crate::db::cauldron::config::{config_get, config_set}; +use crate::db::cauldron::spot::{self, build_spot_ohlc, Confirmed, SpotState, TokenKey}; use anyhow::Result; use sqlx::{Row, SqlitePool}; @@ -12,7 +15,9 @@ use sqlx::{Row, SqlitePool}; /// `INSERT OR IGNORE`, so existing rows are never corrected in place. /// /// 2: price switched from the signed net ratio to the gross volume ratio. -pub const OHLCV_VERSION: u32 = 2; +/// 3: price switched from the trades' execution average to the reserves they left +/// behind, so a sell can no longer print above the buy before it (see `spot`). +pub const OHLCV_VERSION: u32 = 3; const OHLCV_VERSION_KEY: &str = "ohlcv_version"; pub async fn create_table(pool: &SqlitePool) { @@ -81,11 +86,63 @@ pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result> { Ok(row.and_then(|r| r.0)) } -/// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls -/// in `[since_ts, until_ts)`. +/// Gross traded volume per (token, bucket): `(volume_sats, volume_tokens, tx_count)`. /// -/// Two-phase approach: the slow aggregation SELECT runs against `read_pool` (no write lock), -/// then the pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly). +/// Volumes are gross sums of the absolute per-leg deltas. Summing the *signed* +/// deltas instead lets a multi-pool arbitrage transaction — which buys from one pool +/// and sells into others — cancel almost all of its token movement and report a +/// fraction of the volume it actually moved. +async fn load_bucket_volumes( + read_pool: &SqlitePool, + since_ts: i64, + until_ts: i64, +) -> Result> { + let sql = r#" +WITH tx_trades AS ( + SELECT + phe.token_id AS token_id, + (phe.effective_timestamp / 3600) * 3600 AS bucket_ts, + SUM(ABS(phe.sats_delta)) AS vol_sats, + SUM(ABS(phe.token_delta)) AS vol_tokens + FROM pool_history_entry AS phe + JOIN tx ON tx.txid = phe.txid + WHERE tx.blockhash IS NOT NULL + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? + GROUP BY phe.token_id, phe.txid, phe.effective_timestamp +) +SELECT token_id, bucket_ts, SUM(vol_sats), SUM(vol_tokens), COUNT(*) +FROM tx_trades +GROUP BY token_id, bucket_ts +"#; + + let rows = sqlx::query(sql) + .bind(since_ts) + .bind(until_ts) + .fetch_all(read_pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| { + let token_id: Vec = r.get(0); + let bucket_ts: i64 = r.get(1); + ((token_id, bucket_ts), (r.get(2), r.get(3), r.get(4))) + }) + .collect()) +} + +/// Materialise all 1-hour OHLCV buckets whose effective timestamp falls in +/// `[since_ts, until_ts)`, for confirmed transactions only. +/// +/// Price is the aggregate pool spot price replayed across the range (see [`spot`]), +/// so it tracks what the pools were actually quoting rather than what the trades +/// averaged. Only buckets containing a pool change are stored: a bucket with no +/// events repeats the previous close exactly, and the read path reconstructs it by +/// carrying that close forward. +/// +/// Two-phase approach: the reads run against `read_pool` (no write lock), then the +/// pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly). /// Uses INSERT OR IGNORE so existing rows are never overwritten. /// Returns the number of rows inserted. pub async fn rebuild_range( @@ -98,132 +155,77 @@ pub async fn rebuild_range( return Ok(0); } - // Phase 1: aggregate using the read pool — no write lock held during the slow CTE. - let select_sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - phe.token_id, - phe.txid, - phe.effective_timestamp AS ts, - phe.utxo, - phe.sats_delta, - phe.token_delta, - phe.sequence - FROM pool_history_entry AS phe - JOIN tx ON tx.txid = phe.txid - WHERE tx.blockhash IS NOT NULL - AND phe.effective_timestamp >= ? - AND phe.effective_timestamp < ? -), -per_pool_tx AS ( - SELECT - token_id, - txid, - ts, - (ts / 3600) * 3600 AS bucket_ts, - utxo, - MIN(sequence) AS min_sequence, - SUM(ABS(sats_delta)) AS vol_sats, - SUM(ABS(token_delta)) AS vol_tokens - FROM per_pool_tx_raw - GROUP BY token_id, txid, ts, utxo -), -tx_trades AS ( - SELECT - token_id, - txid, - ts, - bucket_ts, - MIN(min_sequence) AS min_sequence, - SUM(vol_sats) AS vol_sats, - SUM(vol_tokens) AS vol_tokens - FROM per_pool_tx - GROUP BY token_id, txid, ts -), -priceable AS ( - SELECT - token_id, - bucket_ts, - CAST(vol_sats AS REAL) / CAST(vol_tokens AS REAL) AS price, - ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, min_sequence ASC) AS rn_asc, - ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, min_sequence DESC) AS rn_desc - FROM tx_trades - WHERE vol_tokens != 0 -), -ohlc AS ( - SELECT - token_id, - bucket_ts, - MAX(CASE WHEN rn_asc = 1 THEN price END) AS open, - MAX(CASE WHEN rn_desc = 1 THEN price END) AS close, - MAX(price) AS high, - MIN(price) AS low - FROM priceable - GROUP BY token_id, bucket_ts -), -vol AS ( - SELECT - token_id, - bucket_ts, - SUM(vol_sats) AS volume_sats, - SUM(vol_tokens) AS volume_tokens, - COUNT(*) AS tx_count - FROM tx_trades - GROUP BY token_id, bucket_ts -) -SELECT - ohlc.token_id, - ohlc.bucket_ts, - ohlc.open, - ohlc.high, - ohlc.low, - ohlc.close, - vol.volume_sats, - vol.volume_tokens, - vol.tx_count -FROM ohlc -JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts -"#; + // Phase 1: read and replay using the read pool — no write lock held throughout. + let mut events = + spot::load_events(read_pool, None, since_ts, until_ts, Confirmed::Only).await?; + if events.is_empty() { + return Ok(0); + } + let mut snapshots = spot::load_snapshot(read_pool, None, since_ts, Confirmed::Only).await?; + let volumes = load_bucket_volumes(read_pool, since_ts, until_ts).await?; - let rows = sqlx::query(select_sql) - .bind(since_ts) - .bind(until_ts) - .fetch_all(read_pool) - .await?; + struct Materialised { + token_id: TokenKey, + bucket_ts: i64, + open: f64, + high: f64, + low: f64, + close: f64, + volume_sats: i64, + volume_tokens: i64, + tx_count: i64, + } - if rows.is_empty() { + let mut pending: Vec = Vec::new(); + let token_ids: Vec = events.keys().cloned().collect(); + for token_id in token_ids { + let token_events = events.remove(&token_id).unwrap_or_default(); + let state = SpotState::new(snapshots.remove(&token_id).unwrap_or_default()); + for candle in build_spot_ohlc(state, &token_events, since_ts, until_ts, 3600) { + if !candle.has_event { + continue; + } + let (volume_sats, volume_tokens, tx_count) = volumes + .get(&(token_id.clone(), candle.time)) + .copied() + .unwrap_or((0, 0, 0)); + pending.push(Materialised { + token_id: token_id.clone(), + bucket_ts: candle.time, + open: candle.open, + high: candle.high, + low: candle.low, + close: candle.close, + volume_sats, + volume_tokens, + tx_count, + }); + } + } + + if pending.is_empty() { return Ok(0); } // Phase 2: insert pre-computed rows inside a single transaction. - // The write lock is held only for these fast INSERTs, not during aggregation. + // The write lock is held only for these fast INSERTs, not during the replay. let mut tx = write_pool.begin().await?; let mut inserted = 0u64; - for row in &rows { - let token_id: Vec = row.get(0); - let bucket_ts: i64 = row.get(1); - let open: f64 = row.get(2); - let high: f64 = row.get(3); - let low: f64 = row.get(4); - let close: f64 = row.get(5); - let volume_sats: i64 = row.get(6); - let volume_tokens: i64 = row.get(7); - let tx_count: i64 = row.get(8); - + for row in pending { inserted += sqlx::query( "INSERT OR IGNORE INTO ohlcv_1h (token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ) - .bind(token_id) - .bind(bucket_ts) - .bind(open) - .bind(high) - .bind(low) - .bind(close) - .bind(volume_sats) - .bind(volume_tokens) - .bind(tx_count) + .bind(row.token_id) + .bind(row.bucket_ts) + .bind(row.open) + .bind(row.high) + .bind(row.low) + .bind(row.close) + .bind(row.volume_sats) + .bind(row.volume_tokens) + .bind(row.tx_count) .execute(&mut *tx) .await? .rows_affected(); @@ -282,12 +284,14 @@ pub async fn get_active_candles( #[cfg(test)] mod tests { use super::*; - use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding}; + use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding, utxo_spending}; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use std::sync::atomic::{AtomicU64, Ordering}; static OHLCV_TEST_COUNTER: AtomicU64 = AtomicU64::new(0); + const HOUR: i64 = 3600; + async fn test_pool() -> SqlitePool { let id = OHLCV_TEST_COUNTER.fetch_add(1, Ordering::SeqCst); let uri = format!("file:ohlcv_test_{}?mode=memory&cache=shared", id); @@ -300,108 +304,155 @@ mod tests { async fn setup_db(pool: &SqlitePool) { tx::create_table(pool).await; utxo_funding::create_table(pool).await; + utxo_spending::create_table(pool).await; cauldron_pool::create_table(pool).await; create_table(pool).await; // ohlcv_1h + idx_phe_txid } - /// Insert a single confirmed trade via raw SQL (FK disabled in tests). - async fn insert_confirmed_trade( - pool: &SqlitePool, + /// One trade leg, inserted via raw SQL (FK disabled in tests). + /// + /// `reserves` is the state the leg leaves its pool holding — what the price is + /// now read from — and `deltas` what it moved, which is what volume is read from. + #[allow(clippy::too_many_arguments)] + async fn insert_leg( + conn: &SqlitePool, txid: [u8; 32], utxo: [u8; 32], + pool_hash: [u8; 32], token_id: [u8; 32], - mtp_ts: i64, - sats_delta: i64, - token_delta: i64, + ts: i64, + confirmed: bool, + reserves: (i64, i64), + deltas: (i64, i64), ) { - let blockhash = [0xAA_u8; 32]; - sqlx::query("INSERT OR IGNORE INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)") + if confirmed { + sqlx::query( + "INSERT OR IGNORE INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)", + ) .bind(txid.as_slice()) - .bind(blockhash.as_slice()) - .bind(mtp_ts) - .execute(pool) + .bind([0xAA_u8; 32].as_slice()) + .bind(ts) + .execute(conn) .await .unwrap(); + } else { + sqlx::query("INSERT OR IGNORE INTO tx (txid, first_seen_timestamp) VALUES (?, ?)") + .bind(txid.as_slice()) + .bind(ts) + .execute(conn) + .await + .unwrap(); + } + + sqlx::query( + "INSERT OR IGNORE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) + VALUES (?, ?, ?, NULL)", + ) + .bind(pool_hash.as_slice()) + .bind([0u8; 20].as_slice()) + .bind(token_id.as_slice()) + .execute(conn) + .await + .unwrap(); + sqlx::query( "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) - VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + VALUES (?, ?, ?, ?, 0, ?, ?, ?)", ) .bind(utxo.as_slice()) .bind(txid.as_slice()) .bind([0u8; 32].as_slice()) .bind(txid.as_slice()) + .bind(reserves.0) + .bind(reserves.1) .bind(token_id.as_slice()) - .execute(pool) + .execute(conn) .await .unwrap(); + let seq: i64 = sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") - .fetch_one(pool) + .fetch_one(conn) .await .unwrap(); - sqlx::query( + let ts_column = if confirmed { + "mtp_timestamp" + } else { + "first_seen_timestamp" + }; + sqlx::query(&format!( "INSERT INTO pool_history_entry - (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, sequence, sats, token_amount, sats_delta, token_delta) - VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, ?, ?)", - ) + (utxo, pool, token_id, txid, tx_pos, {ts_column}, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, ?, ?, ?, ?)" + )) .bind(utxo.as_slice()) - .bind([0xBB_u8; 32].as_slice()) // dummy pool hash (FK disabled) + .bind(pool_hash.as_slice()) .bind(token_id.as_slice()) .bind(txid.as_slice()) - .bind(mtp_ts) + .bind(ts) .bind(seq) - .bind(sats_delta) - .bind(token_delta) - .execute(pool) + .bind(reserves.0) + .bind(reserves.1) + .bind(deltas.0) + .bind(deltas.1) + .execute(conn) .await .unwrap(); } - /// Insert a mempool-only trade (no blockhash on the tx row). - async fn insert_mempool_trade( - pool: &SqlitePool, + /// A confirmed single-pool trade whose reserves are irrelevant to the assertion. + async fn insert_confirmed_trade( + conn: &SqlitePool, txid: [u8; 32], utxo: [u8; 32], token_id: [u8; 32], - first_seen_ts: i64, + ts: i64, + sats_delta: i64, + token_delta: i64, ) { - sqlx::query("INSERT INTO tx (txid, first_seen_timestamp) VALUES (?, ?)") - .bind(txid.as_slice()) - .bind(first_seen_ts) - .execute(pool) - .await - .unwrap(); - sqlx::query( - "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) - VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + insert_leg( + conn, + txid, + utxo, + [0xBB; 32], + token_id, + ts, + true, + (1000, 1000), + (sats_delta, token_delta), ) - .bind(utxo.as_slice()) - .bind(txid.as_slice()) - .bind([0u8; 32].as_slice()) - .bind(txid.as_slice()) - .bind(token_id.as_slice()) - .execute(pool) - .await - .unwrap(); - let seq: i64 = - sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") - .fetch_one(pool) - .await - .unwrap(); - sqlx::query( - "INSERT INTO pool_history_entry - (utxo, pool, token_id, txid, tx_pos, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta) - VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, -1000, 25)", + .await; + } + + async fn insert_mempool_trade( + conn: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + ts: i64, + ) { + insert_leg( + conn, + txid, + utxo, + [0xBB; 32], + token_id, + ts, + false, + (1000, 1000), + (-1000, 25), + ) + .await; + } + + async fn closes(conn: &SqlitePool, token_id: [u8; 32]) -> Vec<(i64, f64)> { + sqlx::query_as( + "SELECT bucket_ts, close FROM ohlcv_1h WHERE token_id = ? ORDER BY bucket_ts", ) - .bind(utxo.as_slice()) - .bind([0xBB_u8; 32].as_slice()) .bind(token_id.as_slice()) - .bind(txid.as_slice()) - .bind(first_seen_ts) - .bind(seq) - .execute(pool) + .fetch_all(conn) .await - .unwrap(); + .unwrap() } /// `get_min_trade_bucket_ts` should floor a mid-hour timestamp to the hour boundary. @@ -476,51 +527,168 @@ mod tests { assert_eq!(n, 0, "mempool trades must not be materialised"); } - /// A multi-pool arbitrage transaction whose legs nearly cancel must materialise the - /// price its legs executed at, not the signed-net ratio. + /// The reason for `OHLCV_VERSION` 3. + /// + /// Reserves and deltas are mainnet GIRL's (token 63664918…f455) buy at + /// 2026-08-10 15:45:54 followed by its sell at 2026-08-11 08:52:47. Priced by + /// what the trades averaged, the sell materialises *above* the buy — 0.0836 + /// against 0.0784 per token — while the pool it traded against had just fallen + /// from 0.0926 to 0.0760. #[tokio::test] - async fn test_rebuild_range_prices_arb_by_gross_volume() { + async fn test_rebuild_range_prices_by_reserves_not_execution_average() { let pool = test_pool().await; setup_db(&pool).await; + let token = [0x03_u8; 32]; + insert_leg( + &pool, + [0x01; 32], + [0x11; 32], + [0xB1; 32], + token, + HOUR, + true, + (540_052, 818_802_757_370_920), + (100_000, -185_511_337_091_708), + ) + .await; + insert_leg( + &pool, + [0x02; 32], + [0x12; 32], + [0xB1; 32], + token, + 2 * HOUR, + true, + (640_052, 691_199_193_943_404), + (100_000, -127_603_563_427_516), + ) + .await; + insert_leg( + &pool, + [0x03; 32], + [0x13; 32], + [0xB1; 32], + token, + 3 * HOUR, + true, + (580_053, 762_931_584_125_945), + (-59_999, 71_732_390_182_541), + ) + .await; + + rebuild_range(&pool, &pool, 0, 4 * HOUR).await.unwrap(); + + let rows = closes(&pool, token).await; + assert_eq!(rows.len(), 3); + let buy = rows[1].1; + let sell = rows[2].1; + assert!( + (buy * 1e8 - 0.0926).abs() < 1e-4, + "buy must materialise at the price it created, got {}", + buy * 1e8 + ); + assert!( + (sell * 1e8 - 0.0760).abs() < 1e-4, + "sell must materialise at the price it created, got {}", + sell * 1e8 + ); + assert!( + sell < buy, + "a sell materialised at {} above the buy before it at {}", + sell * 1e8, + buy * 1e8 + ); + } + + /// A multi-pool arbitrage transaction prices at the reserves its legs left, and + /// still reports every satoshi and token those legs moved. + #[tokio::test] + async fn test_rebuild_range_arb_prices_by_reserves_and_keeps_gross_volume() { + let pool = test_pool().await; + setup_db(&pool).await; let token = [0x03_u8; 32]; let txid = [0x01_u8; 32]; + // Same transaction, two pools, opposite directions netting to +2 token units. - insert_confirmed_trade( + insert_leg( &pool, txid, [0x02; 32], + [0xB1; 32], token, - 1727963400, - -446_491_239, - 1_334_527_069, + HOUR, + true, + (1_000_000, 3_000_000), + (-446_491_239, 1_334_527_069), ) .await; - insert_confirmed_trade( + insert_leg( &pool, txid, [0x04; 32], + [0xB2; 32], token, - 1727963400, - 384_906_040, - -1_334_527_067, + HOUR, + true, + (2_000_000, 6_000_000), + (384_906_040, -1_334_527_067), ) .await; - rebuild_range(&pool, &pool, 1727960400, 1727964000) - .await - .unwrap(); + rebuild_range(&pool, &pool, 0, 2 * HOUR).await.unwrap(); - let close: f64 = sqlx::query_scalar("SELECT close FROM ohlcv_1h WHERE token_id = ?") - .bind(token.as_slice()) - .fetch_one(&pool) - .await - .unwrap(); + let (close, volume_sats, volume_tokens, tx_count): (f64, i64, i64, i64) = sqlx::query_as( + "SELECT close, volume_sats, volume_tokens, tx_count FROM ohlcv_1h WHERE token_id = ?", + ) + .bind(token.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); - let expected = 831_397_279.0 / 2_669_054_136.0; assert!( - (close - expected).abs() < 1e-9, - "materialised close {close} should be the gross ratio {expected}" + (close - 3_000_000.0 / 9_000_000.0).abs() < 1e-12, + "close {close} must be the summed reserves the legs left" + ); + // Net-ratio pricing would divide 61,585,199 sats by 2 token units. + assert!( + close < 61_585_199.0 / 2.0 / 1000.0, + "netting artifact: {close}" + ); + assert_eq!(volume_sats, 446_491_239 + 384_906_040); + assert_eq!(volume_tokens, 1_334_527_069 + 1_334_527_067); + assert_eq!(tx_count, 1, "one transaction, two legs"); + } + + /// A bucket with no pool change repeats the previous close exactly, so it is + /// reconstructed on read rather than stored — otherwise every token would need a + /// row for every hour it has ever existed. + #[tokio::test] + async fn test_rebuild_range_stores_only_buckets_with_events() { + let pool = test_pool().await; + setup_db(&pool).await; + let token = [0x03_u8; 32]; + + insert_leg( + &pool, + [0x01; 32], + [0x11; 32], + [0xB1; 32], + token, + HOUR, + true, + (700, 100), + (700, 100), + ) + .await; + + rebuild_range(&pool, &pool, 0, 10 * HOUR).await.unwrap(); + + let rows = closes(&pool, token).await; + assert_eq!( + rows, + vec![(HOUR, 7.0)], + "only the bucket that moved is stored" ); } diff --git a/src/db/cauldron/spot.rs b/src/db/cauldron/spot.rs new file mode 100644 index 0000000..238d2b1 --- /dev/null +++ b/src/db/cauldron/spot.rs @@ -0,0 +1,513 @@ +// 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 + +//! Aggregate pool spot price, replayed over a time range. +//! +//! A trade's *execution* price — the satoshis it moved divided by the tokens it +//! moved — is the volume-weighted average along the bonding curve, so it always +//! lands strictly between the pool's price before the trade and its price after. +//! A buy therefore prints below the price it created, a sell prints above it, and +//! two consecutive execution prices need not move in the same direction the pool +//! did: a modest sell after a large buy prints *up*, because the sell's average +//! (taken over the range it walked down) still sits above the buy's average +//! (taken over the range it walked up). +//! +//! Pricing by the reserves the trades left behind removes that whole class of +//! artifact — the price is a function of state, so it moves down exactly when the +//! pool moved down — and it makes candles agree with `/price` and the token list, +//! which already quote summed reserves over every live pool. + +use std::collections::HashMap; + +use anyhow::Result; +use sqlx::{Row, SqlitePool}; + +/// Reserves left by a pool's latest entry: (sats, token base units). +pub type Reserves = (i64, i64); + +/// A pool's identity — its `creation_utxo` / `pool_history_entry.pool` blob. +pub type PoolKey = Vec; + +/// A token's identity — its `token_id` blob, stored byte-reversed. +pub type TokenKey = Vec; + +/// A change to one pool's contribution to its token's price. +/// +/// `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, Debug)] +pub struct SpotEvent { + pub ts: i64, + pub sequence: i64, + pub pool: PoolKey, + pub reserves: Option, +} + +/// Restricts a load to confirmed transactions. +/// +/// `ohlcv_1h` is written once and never corrected, so it must not bake in a +/// mempool transaction that may never confirm; the live path has no such +/// constraint and shows unconfirmed trades as soon as they are seen. +#[derive(Copy, Clone, PartialEq, Eq)] +pub enum Confirmed { + Only, + OrMempool, +} + +impl Confirmed { + /// Predicate over a `pool_history_entry` alias' `txid`, or an empty string. + fn phe_clause(self, alias: &str) -> String { + match self { + Confirmed::Only => format!( + "AND EXISTS (SELECT 1 FROM tx WHERE tx.txid = {alias}.txid AND tx.blockhash IS NOT NULL)" + ), + Confirmed::OrMempool => String::new(), + } + } + + /// Predicate over an already-joined `tx` alias, or an empty string. + fn tx_clause(self, alias: &str) -> &'static str { + match self { + Confirmed::Only => { + debug_assert_eq!(alias, "t"); + "AND t.blockhash IS NOT NULL" + } + Confirmed::OrMempool => "", + } + } +} + +/// Running summed-reserve price for one token's live pools. +/// +/// The sums are maintained incrementally because a busy token can have hundreds of +/// live pools and thousands of events in a window; re-summing the map per event +/// would make the replay quadratic in pool count. +pub struct SpotState { + reserves: HashMap, + sats: i128, + tokens: i128, +} + +impl SpotState { + pub fn new(reserves: HashMap) -> Self { + let mut sats: i128 = 0; + let mut tokens: i128 = 0; + for (s, t) in reserves.values() { + sats += *s as i128; + tokens += *t as i128; + } + Self { + reserves, + sats, + tokens, + } + } + + /// Price per smallest token unit, in satoshis. + /// + /// `None` when the token has no priceable reserves — the same "no price" + /// `/price` reports rather than quoting zero. + pub fn price(&self) -> Option { + if self.sats <= 0 || self.tokens <= 0 { + return None; + } + let price = self.sats as f64 / self.tokens as f64; + price.is_finite().then_some(price) + } + + pub fn apply(&mut self, event: &SpotEvent) { + if let Some((sats, tokens)) = self.reserves.remove(&event.pool) { + self.sats -= sats as i128; + self.tokens -= tokens as i128; + } + if let Some((sats, tokens)) = event.reserves { + self.reserves.insert(event.pool.clone(), (sats, tokens)); + self.sats += sats as i128; + self.tokens += tokens as i128; + } + } +} + +/// Open/high/low/close of the aggregate spot price across one bucket. +pub struct SpotOhlc { + pub time: i64, + pub open: f64, + pub high: f64, + pub low: f64, + pub close: f64, + /// Whether any pool changed inside this bucket. Buckets without one repeat the + /// previous close exactly, so they can be reconstructed instead of stored. + pub has_event: bool, +} + +/// Walk `[start, end)` in `step` buckets, replaying `events` over `state`. +/// +/// `events` must be sorted by `(ts, sequence)` and hold nothing before `start`. +/// A bucket opens at the price it inherits, so consecutive candles never gap, and +/// closes at the price left by its last event; a bucket with no events is a flat +/// candle at the carried price, which is what the pool price genuinely did. Leading +/// buckets from before the token had any reserves are skipped rather than reported +/// as zero. +pub fn build_spot_ohlc( + mut state: SpotState, + events: &[SpotEvent], + start: i64, + end: i64, + step: i64, +) -> Vec { + let mut out = Vec::new(); + let mut next = 0usize; + let mut bucket = start; + + while bucket < end { + let bucket_end = bucket.saturating_add(step); + let mut open = state.price(); + let mut high = open; + let mut low = open; + let mut close = open; + let mut has_event = false; + + while next < events.len() && events[next].ts < bucket_end { + state.apply(&events[next]); + next += 1; + has_event = true; + if let Some(price) = state.price() { + // A token whose first pool is created mid-bucket opens there. + open.get_or_insert(price); + high = Some(high.map_or(price, |h| h.max(price))); + low = Some(low.map_or(price, |l| l.min(price))); + close = Some(price); + } + } + + if let (Some(open), Some(high), Some(low), Some(close)) = (open, high, low, close) { + out.push(SpotOhlc { + time: bucket, + open, + high, + low, + close, + has_event, + }); + } + + bucket = bucket_end; + } + + out +} + +/// Per-pool reserves, grouped by token, as of just before `ts`. +/// +/// "Just before" is deliberate: entries are taken with `effective_timestamp < ts` +/// and a pool whose withdrawal lands at or after `ts` is kept, so a caller can +/// replay [`load_events`] from `ts` on top without applying anything twice. The +/// liveness rules otherwise match `db_visit_pool_entries`, which is what makes the +/// replayed price equal the one `/price` reports. +pub async fn load_snapshot( + conn: &SqlitePool, + token: Option<&[u8]>, + ts: i64, + confirmed: Confirmed, +) -> Result>> { + let token_filter = if token.is_some() { + "AND p.token_id = ?" + } else { + "" + }; + let confirmed_filter = confirmed.phe_clause("inner_phe"); + + // Correlated subquery rather than a window function over the whole table: this + // is one index seek per pool, and the backfill runs it once per batch. + let sql = format!( + "SELECT p.token_id, p.creation_utxo, phe.sats, phe.token_amount + FROM pool p + JOIN pool_history_entry phe ON p.creation_utxo = phe.pool + AND phe.sequence = ( + SELECT MAX(inner_phe.sequence) FROM pool_history_entry AS inner_phe + WHERE inner_phe.pool = p.creation_utxo + AND inner_phe.effective_timestamp < ? + {confirmed_filter} + ) + WHERE (p.withdrawn_in_utxo IS NULL OR ( + SELECT t.effective_timestamp + FROM utxo_spending us JOIN tx t ON us.txid = t.txid + WHERE us.spent_utxo_hash = p.withdrawn_in_utxo + ) >= ?) + {token_filter}" + ); + + let mut query = sqlx::query(&sql).bind(ts).bind(ts); + if let Some(token) = token { + query = query.bind(token.to_vec()); + } + let rows = query.fetch_all(conn).await?; + + let mut out: HashMap> = HashMap::new(); + for row in rows { + let token_id: Vec = row.get(0); + let pool: Vec = row.get(1); + let sats: i64 = row.get(2); + let tokens: i64 = row.get(3); + out.entry(token_id) + .or_default() + .insert(pool, (sats.max(0), tokens.max(0))); + } + Ok(out) +} + +/// Every pool state change in `[start, end)`, grouped by token and sorted. +pub async fn load_events( + conn: &SqlitePool, + token: Option<&[u8]>, + start: i64, + end: i64, + confirmed: Confirmed, +) -> Result>> { + let mut by_token: HashMap> = HashMap::new(); + + let entry_token_filter = if token.is_some() { + "AND phe.token_id = ?" + } else { + "" + }; + let entries_sql = format!( + "SELECT phe.token_id, phe.pool, phe.sats, phe.token_amount, + phe.effective_timestamp, phe.sequence + FROM pool_history_entry AS phe + WHERE phe.effective_timestamp >= ? AND phe.effective_timestamp < ? + {entry_token_filter} + {}", + confirmed.phe_clause("phe") + ); + let mut query = sqlx::query(&entries_sql).bind(start).bind(end); + if let Some(token) = token { + query = query.bind(token.to_vec()); + } + for row in query.fetch_all(conn).await? { + let token_id: Vec = row.get(0); + let sats: i64 = row.get(2); + let tokens: i64 = row.get(3); + by_token.entry(token_id).or_default().push(SpotEvent { + ts: row.get(4), + sequence: row.get(5), + pool: row.get(1), + reserves: Some((sats.max(0), tokens.max(0))), + }); + } + + // A withdrawal that lands exactly on `start - 1` becomes an event at `start`, + // because `load_snapshot` still counts a pool whose withdrawal is at or after + // its bound — so the pool has to be dropped by the replay rather than by the + // snapshot. Recording every withdrawal a second late lets both kinds of event + // share one comparison. + let withdrawal_token_filter = if token.is_some() { + "AND p.token_id = ?" + } else { + "" + }; + let withdrawals_sql = format!( + "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 >= ? AND t.effective_timestamp < ? + {withdrawal_token_filter} + {}", + confirmed.tx_clause("t") + ); + let mut query = sqlx::query(&withdrawals_sql) + .bind(start.saturating_sub(1)) + .bind(end); + if let Some(token) = token { + query = query.bind(token.to_vec()); + } + for row in query.fetch_all(conn).await? { + let token_id: Vec = row.get(0); + let withdrawn_ts: i64 = row.get(2); + by_token.entry(token_id).or_default().push(SpotEvent { + ts: withdrawn_ts.saturating_add(1), + // after every entry sharing the timestamp + sequence: i64::MAX, + pool: row.get(1), + reserves: None, + }); + } + + for events in by_token.values_mut() { + events.sort_by_key(|e| (e.ts, e.sequence)); + } + Ok(by_token) +} + +#[cfg(test)] +mod tests { + use super::*; + + const HOUR: i64 = 3600; + + fn pool_key(n: u8) -> PoolKey { + vec![n; 32] + } + + fn state(pairs: &[(u8, i64, i64)]) -> SpotState { + SpotState::new( + pairs + .iter() + .map(|(p, s, t)| (pool_key(*p), (*s, *t))) + .collect(), + ) + } + + fn entry(ts: i64, pool: u8, sats: i64, tokens: i64) -> SpotEvent { + SpotEvent { + ts, + sequence: ts, + pool: pool_key(pool), + reserves: Some((sats, tokens)), + } + } + + fn withdrawal(ts: i64, pool: u8) -> SpotEvent { + SpotEvent { + ts: ts + 1, + sequence: i64::MAX, + pool: pool_key(pool), + reserves: None, + } + } + + /// The bug this module exists for: a sell must never raise the price. + /// + /// Reserves are GIRL's own, from the trades at 2026-08-10 15:45 and + /// 2026-08-11 08:52. Priced by execution average the pair prints 0.0784 then + /// 0.0836 — up, on a sell. Priced by reserves it prints the move the pool + /// actually made. + #[test] + fn a_sell_lowers_the_close() { + let candles = build_spot_ohlc( + state(&[(1, 540_052, 818_802_757_370_920)]), + &[ + entry(0, 1, 640_052, 691_199_193_943_404), + entry(HOUR, 1, 580_053, 762_931_584_125_945), + ], + 0, + 2 * HOUR, + HOUR, + ); + + let buy_close = candles[0].close; + let sell_close = candles[1].close; + assert!( + (buy_close * 1e8 - 0.0926).abs() < 1e-4, + "buy should close at the pool price it created, got {}", + buy_close * 1e8 + ); + assert!( + sell_close < buy_close, + "sell closed at {} after a buy closed at {} — a sell must not raise the price", + sell_close * 1e8, + buy_close * 1e8 + ); + assert!((sell_close * 1e8 - 0.0760).abs() < 1e-4); + } + + #[test] + fn a_bucket_opens_where_the_last_one_closed() { + let candles = build_spot_ohlc( + state(&[(1, 100, 100)]), + &[entry(10, 1, 400, 100), entry(HOUR + 10, 1, 200, 100)], + 0, + 2 * HOUR, + HOUR, + ); + assert_eq!(candles[0].open, 1.0); + assert_eq!(candles[0].close, 4.0); + // no gap: the second bucket starts from the first one's close + assert_eq!(candles[1].open, 4.0); + assert_eq!(candles[1].close, 2.0); + assert_eq!(candles[1].high, 4.0); + assert_eq!(candles[1].low, 2.0); + } + + #[test] + fn high_and_low_span_the_whole_bucket() { + let candles = build_spot_ohlc( + state(&[(1, 100, 100)]), + &[entry(10, 1, 500, 100), entry(20, 1, 50, 100)], + 0, + HOUR, + HOUR, + ); + assert_eq!(candles[0].open, 1.0); + assert_eq!(candles[0].high, 5.0); + assert_eq!(candles[0].low, 0.5); + assert_eq!(candles[0].close, 0.5); + } + + #[test] + fn a_quiet_bucket_is_flat_at_the_carried_price() { + let candles = build_spot_ohlc(state(&[(1, 300, 100)]), &[], 0, 3 * HOUR, HOUR); + assert_eq!(candles.len(), 3); + for candle in &candles { + assert_eq!( + (candle.open, candle.high, candle.low, candle.close), + (3.0, 3.0, 3.0, 3.0) + ); + assert!(!candle.has_event, "a quiet bucket need not be stored"); + } + } + + #[test] + fn the_price_sums_every_live_pool() { + // aggregate of summed reserves, not an average of per-pool prices + let candles = build_spot_ohlc(state(&[(1, 100, 100), (2, 900, 100)]), &[], 0, HOUR, HOUR); + assert_eq!(candles[0].close, 5.0); + } + + #[test] + fn a_withdrawal_moves_the_price_without_volume() { + let candles = build_spot_ohlc( + state(&[(1, 100, 100), (2, 900, 100)]), + &[withdrawal(HOUR, 2)], + 0, + 3 * HOUR, + HOUR, + ); + assert_eq!(candles[0].close, 5.0); + // withdrawal recorded a second late, so it lands in the second bucket + assert_eq!(candles[1].close, 1.0); + assert!(candles[1].has_event, "a withdrawal must be materialised"); + assert_eq!(candles[2].close, 1.0); + } + + #[test] + fn buckets_before_the_first_pool_are_skipped() { + let candles = build_spot_ohlc( + SpotState::new(HashMap::new()), + &[entry(2 * HOUR, 1, 100, 100)], + 0, + 4 * HOUR, + HOUR, + ); + assert_eq!(candles.len(), 2, "no price before the token had a pool"); + assert_eq!(candles[0].time, 2 * HOUR); + assert_eq!(candles[0].open, 1.0); + } + + #[test] + fn a_token_with_no_reserves_has_no_candles() { + let candles = build_spot_ohlc(SpotState::new(HashMap::new()), &[], 0, 3 * HOUR, HOUR); + assert!(candles.is_empty()); + } + + #[test] + fn an_empty_pool_is_not_priced_as_zero() { + // a pool drained to zero tokens has no price rather than an infinite one + let candles = build_spot_ohlc(state(&[(1, 100, 0)]), &[], 0, HOUR, HOUR); + assert!(candles.is_empty()); + } +} diff --git a/src/rpc/candlesticks/tests.rs b/src/rpc/candlesticks/tests.rs index d7fbc1f..21d6728 100644 --- a/src/rpc/candlesticks/tests.rs +++ b/src/rpc/candlesticks/tests.rs @@ -8,6 +8,7 @@ use crate::db::cauldron::{ pool::{self, dummy_init_seq, insert_new_pool}, tx::{self, insert_block_tx, insert_mempool_tx}, utxo_funding::{self, insert_utxo_funding}, + utxo_spending, }; use crate::utiltest::mock_db_pool; use crate::OhlcvState; @@ -64,6 +65,7 @@ fn dummy_cauldron( async fn setup_mock_db(pool: sqlx::SqlitePool) { utxo_funding::create_table(&pool).await; + utxo_spending::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; dummy_init_seq(); @@ -168,66 +170,68 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) { .await .unwrap(); - let token1 = TokenID::from_byte_array([0xda; 32]); + // Registered under the same token as their history entries: pricing reads the + // pool's reserves through this row, so a pool filed under another token would + // silently drop out of its own token's price. let pkh1 = PubkeyHash::from_byte_array([0xca; 20]); - insert_new_pool( - &mut conn, - &dummy_cauldron(&Txid::all_zeros(), &pool1, &token1, 0, 0, &pkh1), - ) - .await - .unwrap(); - insert_new_pool( - &mut conn, - &dummy_cauldron(&Txid::all_zeros(), &pool2, &token1, 0, 0, &pkh1), - ) - .await - .unwrap(); - insert_new_pool( - &mut conn, - &dummy_cauldron(&Txid::all_zeros(), &pool3, &token1, 0, 0, &pkh1), - ) - .await - .unwrap(); - insert_new_pool( - &mut conn, - &dummy_cauldron(&Txid::all_zeros(), &pool4, &token1, 0, 0, &pkh1), - ) - .await - .unwrap(); + for pool_hash in [&pool1, &pool2, &pool3, &pool4] { + insert_new_pool( + &mut conn, + &dummy_cauldron(&Txid::all_zeros(), pool_hash, &token_zero, 0, 0, &pkh1), + ) + .await + .unwrap(); + } } // ── Seeded gap-fill helpers ─────────────────────────────────────────────── async fn setup_seed_db(pool: sqlx::SqlitePool) { utxo_funding::create_table(&pool).await; + utxo_spending::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; ohlcv::create_table(&pool).await; dummy_init_seq(); } +/// A confirmed trade against `pool_byte`, leaving it holding `reserves`. async fn insert_trade_at( conn: &mut sqlx::pool::PoolConnection, token: &TokenID, txid_byte: u8, + pool_byte: u8, ts: u64, - sats_delta: i64, - token_delta: i64, + reserves: (u64, i64), + deltas: (i64, i64), ) { let txid = Txid::from_byte_array([txid_byte; 32]); let utxo = OutPointHash::from_byte_array([txid_byte; 32]); - let pool_hash = OutPointHash::from_byte_array([txid_byte.wrapping_add(0x80); 32]); + let pool_hash = OutPointHash::from_byte_array([pool_byte; 32]); let block = BlockHash::all_zeros(); let cauldron = dummy_cauldron( &txid, &utxo, token, - sats_delta.unsigned_abs(), - token_delta, + reserves.0, + reserves.1, &PubkeyHash::all_zeros(), ); + insert_new_pool( + &mut **conn, + &dummy_cauldron( + &Txid::all_zeros(), + &pool_hash, + token, + 0, + 0, + &PubkeyHash::all_zeros(), + ), + ) + .await + .unwrap(); insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid) .await .unwrap(); @@ -241,8 +245,8 @@ async fn insert_trade_at( &cauldron, Some(ts), Some(ts), - sats_delta, - token_delta, + deltas.0, + deltas.1, ) .await .unwrap(); @@ -344,31 +348,30 @@ async fn test_multiple_candlesticks_endpoint() { let cndl_array = json["candlesticks"].as_array().unwrap(); assert_eq!(cndl_array.len(), 2, "Should produce exactly two candles"); - // ----- Candle #1 ----- + // Each trade seeds a *different* pool, so the price is the running sum of all + // pools live at that moment, not the last one to trade: + // after T1 80k/2k = 40 after T2 200k/4k = 50 + // after T3 360k/6k = 60 after T4 560k/8k = 70 + + // ----- Candle #1: [1727963300, 1727963900) holds T1 and T2 ----- let cndl1 = &cndl_array[0]; - // Candle #1 => time=1727963300 - // trades at 1727963300 => ratio=40, 1727963600 => ratio=60 - // open=40, close=60, low=40, high=60, volume_sats=200k, volume_tokens=4k, transaction_count=2 assert_eq!(cndl1["time"], 1727963300); assert!((cndl1["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); - assert!((cndl1["close"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); + assert!((cndl1["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert!((cndl1["low"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); - assert!((cndl1["high"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); + assert!((cndl1["high"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert_eq!(cndl1["volume_sats"].as_i64().unwrap(), 80_000 + 120_000); assert_eq!(cndl1["volume_tokens"].as_i64().unwrap(), 2_000 + 2_000); assert_eq!(cndl1["transaction_count"].as_i64().unwrap(), 2); - // ----- Candle #2 ----- + // ----- Candle #2: [1727963900, 1727964500) holds T3 and T4 ----- let cndl2 = &cndl_array[1]; - // Candle #2 => time=1727963900 - // trades at 1727963900 => ratio=80, 1727964200 => ratio=100 - // open=80, close=100, low=80, high=100, volume_sats=360k, volume_tokens=4k, transaction_count=2 assert_eq!(cndl2["time"], 1727963900); - assert!((cndl2["open"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); - assert!((cndl2["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); - assert!((cndl2["low"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); - assert!((cndl2["high"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); - // volume_sats=160k+200k=360k, volume_tokens=2k+2k=4k, transaction_count=2 + // opens where candle #1 closed — the series never gaps + assert!((cndl2["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); + assert!((cndl2["close"].as_f64().unwrap() - 70.0).abs() < f64::EPSILON); + assert!((cndl2["low"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); + assert!((cndl2["high"].as_f64().unwrap() - 70.0).abs() < f64::EPSILON); assert_eq!(cndl2["volume_sats"].as_i64().unwrap(), 160_000 + 200_000); assert_eq!(cndl2["volume_tokens"].as_i64().unwrap(), 4_000); assert_eq!(cndl2["transaction_count"].as_i64().unwrap(), 2); @@ -378,6 +381,7 @@ async fn test_multiple_candlesticks_endpoint() { async fn test_single_swap_multiple_pools() { let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { utxo_funding::create_table(&pool).await; + utxo_spending::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; dummy_init_seq(); @@ -427,12 +431,11 @@ async fn test_single_swap_multiple_pools() { .unwrap(); } - let token1 = TokenID::from_byte_array([0xda; 32]); let pkh1 = PubkeyHash::from_byte_array([0xca; 20]); for pool_hash in &pools { insert_new_pool( &mut conn, - &dummy_cauldron(&Txid::all_zeros(), pool_hash, &token1, 0, 0, &pkh1), + &dummy_cauldron(&Txid::all_zeros(), pool_hash, &token_zero, 0, 0, &pkh1), ) .await .unwrap(); @@ -473,12 +476,18 @@ async fn test_single_swap_multiple_pools() { assert_eq!(first_candle["volume_tokens"].as_i64().unwrap(), 2000 + 2000); } +/// A transaction whose legs cancel — buying from one pool and selling into another +/// in equal size — nets to zero tokens moved. It still moved every satoshi and token +/// its legs moved, and it still leaves both pools holding reserves to price from. #[rocket::async_test] -async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { - // Fresh mock DB for this scenario - let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { - // --- boilerplate setup --- +async fn test_candle_prices_and_counts_volume_when_legs_cancel() { + let step: u64 = 600; // 10 minutes + let t0: u64 = 1_700_000_000; + let t1: u64 = t0 + step; + + let mock_db = mock_db_pool(move |pool: sqlx::SqlitePool| async move { utxo_funding::create_table(&pool).await; + utxo_spending::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; dummy_init_seq(); @@ -489,65 +498,21 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { let pkh_zero = PubkeyHash::all_zeros(); let block_zero = BlockHash::all_zeros(); - // Two pools (identities) let pool_a = OutPointHash::from_byte_array([0x1a; 32]); let pool_b = OutPointHash::from_byte_array([0x1b; 32]); + for pool_hash in [&pool_a, &pool_b] { + insert_new_pool( + &mut conn, + &dummy_cauldron(&Txid::all_zeros(), pool_hash, &token_zero, 0, 0, &pkh_zero), + ) + .await + .unwrap(); + } - // Register pools in `pool` table - let token1 = TokenID::from_byte_array([0xda; 32]); - let pkh1 = PubkeyHash::from_byte_array([0xca; 20]); - insert_new_pool( - &mut conn, - &ParsedContract { - pkh: pkh1, - is_withdrawn: false, - spent_utxo_hash: OutPointHash::all_zeros(), - new_utxo_hash: Some(pool_a), - new_utxo_txid: Some(Txid::all_zeros()), - new_utxo_n: Some(0), - token_id: Some(token1), - sats: Some(0), - token_amount: Some(0), - }, - ) - .await - .unwrap(); - insert_new_pool( - &mut conn, - &ParsedContract { - pkh: pkh1, - is_withdrawn: false, - spent_utxo_hash: OutPointHash::all_zeros(), - new_utxo_hash: Some(pool_b), - new_utxo_txid: Some(Txid::all_zeros()), - new_utxo_n: Some(0), - token_id: Some(token1), - sats: Some(0), - token_amount: Some(0), - }, - ) - .await - .unwrap(); - - // Times and step - let step: u64 = 600; // 10 minutes - let t0: u64 = 1_700_000_000; - let t1: u64 = t0 + step; - - // -------- Candle #1 (normal priceable trade) -------- + // -------- Candle #1: an ordinary trade, leaving pool A at 10_000/200 = 50 -------- let txid_price = Txid::from_byte_array([0x90; 32]); let utxo_p = OutPointHash::from_byte_array([0x21; 32]); - let cauldron_p = ParsedContract { - pkh: pkh_zero, - is_withdrawn: false, - spent_utxo_hash: OutPointHash::all_zeros(), - new_utxo_hash: Some(utxo_p), - new_utxo_txid: Some(txid_price), - new_utxo_n: Some(0), - token_id: Some(token_zero), - sats: Some(0), - token_amount: Some(0), - }; + let cauldron_p = dummy_cauldron(&txid_price, &utxo_p, &token_zero, 10_000, 200, &pkh_zero); insert_utxo_funding(&mut conn, &vec![cauldron_p.clone()], &txid_price) .await .unwrap(); @@ -555,7 +520,6 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { .await .unwrap(); insert_mempool_tx(&mut conn, &txid_price, t0).await.unwrap(); - // sats_delta=+10_000, token_delta=+200 -> price = 10000/200 = 50 pool::insert_pool_history_entry( &mut conn, &pool_a, @@ -568,27 +532,25 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { .await .unwrap(); - // -------- Candle #2 (net zero tokens but non-zero volume) -------- + // -------- Candle #2: two legs that cancel to zero net tokens -------- + // A ends at 20_000/400, B at 30_000/600 — both still 50, so the aggregate holds. let txid_net0 = Txid::from_byte_array([0x91; 32]); - let utxo_a1 = OutPointHash::from_byte_array([0x22; 32]); - let utxo_b1 = OutPointHash::from_byte_array([0x23; 32]); - - let ca_a1 = ParsedContract { - pkh: pkh_zero, - is_withdrawn: false, - spent_utxo_hash: OutPointHash::all_zeros(), - new_utxo_hash: Some(utxo_a1), - new_utxo_txid: Some(txid_net0), - new_utxo_n: Some(0), - token_id: Some(token_zero), - sats: Some(0), - token_amount: Some(0), - }; - let ca_b1 = ParsedContract { - new_utxo_hash: Some(utxo_b1), - new_utxo_txid: Some(txid_net0), - ..ca_a1 - }; + let ca_a1 = dummy_cauldron( + &txid_net0, + &OutPointHash::from_byte_array([0x22; 32]), + &token_zero, + 20_000, + 400, + &pkh_zero, + ); + let ca_b1 = dummy_cauldron( + &txid_net0, + &OutPointHash::from_byte_array([0x23; 32]), + &token_zero, + 30_000, + 600, + &pkh_zero, + ); insert_utxo_funding(&mut conn, &vec![ca_a1.clone()], &txid_net0) .await @@ -612,7 +574,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { ) .await .unwrap(); - // Opposite deltas within the same tx + // Opposite deltas within the same transaction pool::insert_pool_history_entry( &mut conn, &pool_b, @@ -636,11 +598,10 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { .expect("valid rocket instance"); let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - let start = 1_700_000_000u64; - let end = start + 2 * 600; + let end = t0 + 2 * step; let response = client .get(format!( - "/api/price/{token_id_zero}/candlesticks?start={start}&end={end}&stepsize=600" + "/api/price/{token_id_zero}/candlesticks?start={t0}&end={end}&stepsize=600" )) .dispatch() .await; @@ -652,29 +613,18 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { assert_eq!(candles.len(), 2, "Expected 2 candles (two intervals)"); - // --- Candle #1 (priceable) --- let c1 = &candles[0]; - assert_eq!(c1["time"].as_i64().unwrap(), start as i64); - // price = 10000 / 200 = 50 - assert!((c1["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); + assert_eq!(c1["time"].as_i64().unwrap(), t0 as i64); assert!((c1["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); - assert!((c1["low"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); - assert!((c1["high"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert_eq!(c1["volume_sats"].as_i64().unwrap(), 10_000); assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 200); assert_eq!(c1["transaction_count"].as_i64().unwrap(), 1); - // --- Candle #2 (legs cancel to zero net tokens) --- let c2 = &candles[1]; - assert_eq!(c2["time"].as_i64().unwrap(), (start + 600) as i64); - // Both legs executed at 50, so the gross ratio 20_000/400 prices the tx at 50 - // directly. (Before gross pricing this candle carried the previous close because - // the net token delta was zero; the value coincides, the derivation does not.) - assert!((c2["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); + assert_eq!(c2["time"].as_i64().unwrap(), (t0 + step) as i64); + // 50_000 sats over 1_000 tokens across both pools assert!((c2["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); - assert!((c2["low"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); - assert!((c2["high"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); - // Volume sums absolute per-pool deltas within the tx + // Volume sums the absolute per-pool deltas: the signed sum would be zero. assert_eq!(c2["volume_sats"].as_i64().unwrap(), 20_000); assert_eq!(c2["volume_tokens"].as_i64().unwrap(), 400); assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); @@ -686,9 +636,12 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { // TIME_1=1727963400, TIME_2=1727963600, TIME_3=1727963900 → hour bucket 1727960400 // TIME_4=1727964200 → hour bucket 1727964000 // -// Expected OHLCV (bucket 1727960400): open=40, close=80, high=80, low=40 +// Each trade seeds its own pool, so the price is the running sum over all live +// pools: 40 after T1, 50 after T2, 60 after T3, 70 after T4. +// +// Expected OHLCV (bucket 1727960400): open=40, close=60, high=60, low=40 // vol_sats=360_000 (80k+120k+160k), vol_tokens=6_000, tx_count=3 -// Expected OHLCV (bucket 1727964000): open=close=high=low=100 +// Expected OHLCV (bucket 1727964000): open=60, close=high=70, low=60 // vol_sats=200_000, vol_tokens=2_000, tx_count=1 /// Full range served from ohlcv_1h (materialized_end covers everything). @@ -734,8 +687,8 @@ async fn test_ohlcv_fast_path_full_range() { let c1 = &candles[0]; assert_eq!(c1["time"].as_i64().unwrap(), 1727960400); assert!((c1["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); - assert!((c1["close"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); - assert!((c1["high"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); + assert!((c1["close"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); + assert!((c1["high"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); assert!((c1["low"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); assert_eq!(c1["volume_sats"].as_i64().unwrap(), 360_000); assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 6_000); @@ -743,8 +696,8 @@ async fn test_ohlcv_fast_path_full_range() { let c2 = &candles[1]; assert_eq!(c2["time"].as_i64().unwrap(), 1727964000); - assert!((c2["open"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); - assert!((c2["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert!((c2["open"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); + assert!((c2["close"].as_f64().unwrap() - 70.0).abs() < f64::EPSILON); assert_eq!(c2["volume_sats"].as_i64().unwrap(), 200_000); assert_eq!(c2["volume_tokens"].as_i64().unwrap(), 2_000); assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); @@ -796,16 +749,18 @@ async fn test_ohlcv_fast_path_with_raw_tail() { let c1 = &candles[0]; assert_eq!(c1["time"].as_i64().unwrap(), 1727960400); assert!((c1["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); - assert!((c1["close"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON); + assert!((c1["close"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); assert_eq!(c1["volume_sats"].as_i64().unwrap(), 360_000); assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 6_000); assert_eq!(c1["transaction_count"].as_i64().unwrap(), 3); - // Candle 2 came from the raw CTE tail. + // Candle 2 came from the raw tail. Its open must be the price carried out of + // the materialised bucket — the tail takes its own snapshot at the seam, so a + // snapshot that missed the three pools T1–T3 seeded would open at 100 here. let c2 = &candles[1]; assert_eq!(c2["time"].as_i64().unwrap(), 1727964000); - assert!((c2["open"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); - assert!((c2["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert!((c2["open"].as_f64().unwrap() - 60.0).abs() < f64::EPSILON); + assert!((c2["close"].as_f64().unwrap() - 70.0).abs() < f64::EPSILON); assert_eq!(c2["volume_sats"].as_i64().unwrap(), 200_000); assert_eq!(c2["volume_tokens"].as_i64().unwrap(), 2_000); assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); @@ -838,8 +793,8 @@ async fn test_ohlcv_skipped_for_non_aligned_start() { // Start is NOT hour-aligned (1727963300 % 3600 != 0) — must fall back to raw. // With stepsize=3600 the first interval is [1727963300, 1727966900). // All four trades (T1–T4) fall within this single interval: - // candle 1 at 1727963300: open=40 (T1 first), close=100 (T4 last), tx_count=4 - // candle 2 at 1727966900: flat carry-forward at 100 (no trades) + // candle 1 at 1727963300: open=40 (T1 first), close=70 (all four pools), tx_count=4 + // candle 2 at 1727966900: flat carry-forward at 70 (no trades) let response = client .get(format!( "/api/price/{token_id_zero}/candlesticks\ @@ -858,7 +813,7 @@ async fn test_ohlcv_skipped_for_non_aligned_start() { assert_eq!(candles.len(), 2); assert_eq!(candles[0]["time"].as_i64().unwrap(), 1727963300); assert!((candles[0]["open"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); - assert!((candles[0]["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON); + assert!((candles[0]["close"].as_f64().unwrap() - 70.0).abs() < f64::EPSILON); assert_eq!(candles[0]["transaction_count"].as_i64().unwrap(), 4); // Flat carry-forward candle (no trades in second interval). assert_eq!(candles[1]["time"].as_i64().unwrap(), 1727966900); @@ -877,7 +832,16 @@ async fn test_raw_path_seeded_gap_fill_no_in_window_trades() { let db = mock_db_pool(move |pool: sqlx::SqlitePool| async move { setup_seed_db(pool.clone()).await; let mut conn = pool.acquire().await.unwrap(); - insert_trade_at(&mut conn, &token_copy, 0x10, 1_000, 100_000, 2_000).await; + insert_trade_at( + &mut conn, + &token_copy, + 0x10, + 0x90, + 1_000, + (100_000, 2_000), + (100_000, 2_000), + ) + .await; }) .await; @@ -921,8 +885,27 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { let db = mock_db_pool(move |pool: sqlx::SqlitePool| async move { setup_seed_db(pool.clone()).await; let mut conn = pool.acquire().await.unwrap(); - insert_trade_at(&mut conn, &token_copy, 0x20, 1_000, 100_000, 2_000).await; - insert_trade_at(&mut conn, &token_copy, 0x21, 2_500, 150_000, 2_000).await; + // Both trades hit the same pool, so the window's price is that pool's. + insert_trade_at( + &mut conn, + &token_copy, + 0x20, + 0x91, + 1_000, + (100_000, 2_000), + (100_000, 2_000), + ) + .await; + insert_trade_at( + &mut conn, + &token_copy, + 0x21, + 0x91, + 2_500, + (150_000, 2_000), + (50_000, 0), + ) + .await; }) .await; @@ -981,7 +964,16 @@ async fn test_raw_path_no_seed_no_prefill() { let db = mock_db_pool(move |pool: sqlx::SqlitePool| async move { setup_seed_db(pool.clone()).await; let mut conn = pool.acquire().await.unwrap(); - insert_trade_at(&mut conn, &token_copy, 0x30, 2_500, 150_000, 2_000).await; + insert_trade_at( + &mut conn, + &token_copy, + 0x30, + 0x92, + 2_500, + (150_000, 2_000), + (150_000, 2_000), + ) + .await; }) .await;