diff --git a/src/db/cauldron/candlestick/mod.rs b/src/db/cauldron/candlestick/mod.rs new file mode 100644 index 0000000..5e8745a --- /dev/null +++ b/src/db/cauldron/candlestick/mod.rs @@ -0,0 +1,415 @@ +// Copyright (C) 2024-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 + +use crate::db::blob::display_hex_to_blob; +use crate::db::cauldron::ohlcv; +use anyhow::{bail, Result}; +use bitcoincash::TokenID; +use serde::Serialize; +use sqlx::{Row, SqlitePool}; + +#[derive(Debug, Serialize)] +pub struct CandlestickData { + pub time: i64, // start of the interval + pub open: f64, + pub close: f64, + pub high: f64, + pub low: f64, + pub volume_sats: i64, + pub volume_tokens: i64, + 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, 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, signed_sats, signed_tokens, vol_sats, vol_tokens) = all_trades[trade_index]; + if ts < interval_start { + trade_index += 1; + continue; + } + if ts >= interval_end { + break; + } + + if signed_tokens != 0 { + let price = (signed_sats as f64 / signed_tokens as f64).abs(); + 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) +} + +fn fill_ohlcv_candles( + rows: Vec, + start: i64, + end: i64, + mut found_first_trade: bool, + mut last_close: Option, +) -> (Vec, bool, Option) { + let mut result = Vec::new(); + let mut row_iter = rows.into_iter().peekable(); + 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, + open: r.open, + close: r.close, + high: r.high, + low: r.low, + volume_sats: r.volume_sats, + 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, + }); + } + } + bucket += 3600; + } + + (result, found_first_trade, last_close) +} + +async fn fetch_raw_trades( + pool: &SqlitePool, + token_blob: &[u8], + timestamp_start: i64, + timestamp_end: i64, +) -> Result> { + let sql = r#" +WITH per_pool_tx_raw AS ( + SELECT + phe.txid, + phe.effective_timestamp, + phe.utxo, + phe.sats_delta, + phe.token_delta, + phe.sequence + FROM pool_history_entry AS phe + WHERE phe.token_id = ? + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? +), +per_pool_tx AS ( + SELECT + txid, + effective_timestamp, + utxo, + MIN(sequence) AS min_sequence, + SUM(sats_delta) AS signed_sats_pool, + SUM(token_delta) AS signed_tokens_pool, + SUM(ABS(sats_delta)) AS volume_sats_pool, + SUM(ABS(token_delta)) AS volume_tokens_pool + FROM per_pool_tx_raw + GROUP BY txid, effective_timestamp, utxo +), +tx_trades AS ( + SELECT + txid, + effective_timestamp, + MIN(min_sequence) AS min_sequence, + SUM(signed_sats_pool) AS signed_sats, + SUM(signed_tokens_pool) AS signed_tokens, + SUM(volume_sats_pool) AS volume_sats, + SUM(volume_tokens_pool) AS volume_tokens + FROM per_pool_tx + GROUP BY txid, effective_timestamp +) +SELECT + effective_timestamp, + signed_sats, + signed_tokens, + volume_sats, + volume_tokens +FROM tx_trades +ORDER BY effective_timestamp ASC, min_sequence ASC; +"#; + let rows = sqlx::query(sql) + .bind(token_blob) + .bind(timestamp_start) + .bind(timestamp_end) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4))) + .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( + pool: &SqlitePool, + token_blob: &[u8], + timestamp_end: i64, +) -> Result> { + let sql = r#" +WITH per_pool_tx_raw AS ( + SELECT + phe.txid, + phe.effective_timestamp, + phe.utxo, + phe.sats_delta, + phe.token_delta, + phe.sequence + FROM pool_history_entry AS phe + WHERE phe.token_id = ? + AND phe.effective_timestamp < ? +), +per_pool_tx AS ( + SELECT + txid, + effective_timestamp, + utxo, + MIN(sequence) AS min_sequence, + SUM(sats_delta) AS signed_sats_pool, + SUM(token_delta) AS signed_tokens_pool + FROM per_pool_tx_raw + GROUP BY txid, effective_timestamp, utxo +), +tx_trades AS ( + SELECT + txid, + effective_timestamp, + MIN(min_sequence) AS min_sequence, + SUM(signed_sats_pool) AS signed_sats, + SUM(signed_tokens_pool) AS signed_tokens + FROM per_pool_tx + GROUP BY txid, effective_timestamp +) +SELECT ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS close_price +FROM tx_trades +WHERE signed_tokens != 0 +ORDER BY effective_timestamp DESC, min_sequence DESC +LIMIT 1 +"#; + let row = sqlx::query(sql) + .bind(token_blob) + .bind(timestamp_end) + .fetch_optional(pool) + .await?; + + Ok(row.map(|r| r.get::(0))) +} + +/// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`. +/// Pass 0 to always use the raw CTE path. +pub async fn candlesticks( + pool: &SqlitePool, + timestamp_start: i64, + timestamp_end: i64, + step_size: i64, + token_id: &str, + ohlcv_materialized_end: i64, +) -> Result> { + if timestamp_start > timestamp_end { + bail!("Start cannot be higher than end"); + } + + 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 + // start would produce candles whose boundaries disagree with the raw path. + if step_size == 3600 && ohlcv_materialized_end > timestamp_start && timestamp_start % 3600 == 0 + { + let ohlcv_end = ohlcv_materialized_end.min(timestamp_end); + + // 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, + ); + + 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); + } + + 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) +} + +#[cfg(test)] +mod tests; diff --git a/src/db/cauldron/candlestick/tests.rs b/src/db/cauldron/candlestick/tests.rs new file mode 100644 index 0000000..7166028 --- /dev/null +++ b/src/db/cauldron/candlestick/tests.rs @@ -0,0 +1,135 @@ +// Copyright (C) 2024-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 + +use crate::db::blob::ToBlob; +use crate::db::cauldron::{ + ohlcv, + pool::{self, dummy_init_seq}, + tx::{self, insert_block_tx, insert_mempool_tx}, + utxo_funding::{self, insert_utxo_funding}, +}; +use crate::utiltest::mock_db_pool; +use bitcoin_hashes::Hash; +use bitcoincash::{BlockHash, PubkeyHash, TokenID, Txid}; +use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash}; + +fn dummy_cauldron( + txid: &Txid, + utxo: &OutPointHash, + token: &TokenID, + sats: u64, + tokens: i64, + pkh: &PubkeyHash, +) -> ParsedContract { + ParsedContract { + pkh: *pkh, + is_withdrawn: false, + spent_utxo_hash: OutPointHash::all_zeros(), + new_utxo_hash: Some(*utxo), + new_utxo_txid: Some(*txid), + new_utxo_n: Some(0), + token_id: Some(*token), + sats: Some(sats), + token_amount: Some(tokens), + } +} + +async fn setup_db(pool: sqlx::SqlitePool) { + utxo_funding::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_inner([txid_byte; 32]); + let utxo = OutPointHash::from_inner([txid_byte; 32]); + let pool_hash = OutPointHash::from_inner([txid_byte.wrapping_add(0x80); 32]); + let block = BlockHash::all_zeros(); + + let cauldron = dummy_cauldron( + &txid, + &utxo, + token, + sats_delta.unsigned_abs(), + token_delta, + &PubkeyHash::all_zeros(), + ); + + 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, + ) + .await + .unwrap(); +} + +#[tokio::test] +async fn test_fetch_last_close_before_no_trades() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_inner([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_inner([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_inner([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); +} diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index 253fe3c..c7127ff 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -13,6 +13,7 @@ use crate::db::cauldron::tokenlist::db_utils::{ create_cached_token_metrics_table, }; +pub mod candlestick; pub mod config; pub mod header; pub mod mempool; diff --git a/src/rpc/candlesticks/mod.rs b/src/rpc/candlesticks/mod.rs index 1cbd5e4..ed37e84 100644 --- a/src/rpc/candlesticks/mod.rs +++ b/src/rpc/candlesticks/mod.rs @@ -3,431 +3,18 @@ // 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 crate::db::blob::display_hex_to_blob; -use crate::db::cauldron::ohlcv; +use crate::db::cauldron::candlestick::candlesticks; use crate::db::DB; use crate::rpc::err::{bad_request, ApiErrorCode, CachedApiResult}; use crate::rpc::response::{cached_ok, CACHE_IMMUTABLE, CACHE_NONE}; use crate::timeutil::time_now; use crate::OhlcvState; -use anyhow::{bail, Result}; -use bitcoincash::TokenID; use rocket::{get, State}; -use serde::Serialize; use serde_json::json; use serde_json::Value; -use sqlx::{Row, SqlitePool}; use std::sync::atomic::Ordering; use std::sync::Arc; -#[derive(Debug, Serialize)] -pub struct CandlestickData { - pub time: i64, // start of the interval - pub open: f64, - pub close: f64, - pub high: f64, - pub low: f64, - pub volume_sats: i64, - pub volume_tokens: i64, - 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 { - pub 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, - } - } - - pub 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, - }) - } - - pub fn end(&self) -> i64 { - self.start + self.step - } -} - -/// Aggregate a sorted list of raw trades into candlestick intervals. -/// Returns the filled candles and the last close price seen (for continuing into a tail query). -fn aggregate_raw_trades( - all_trades: &[(i64, i64, 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, signed_sats, signed_tokens, vol_sats, vol_tokens) = all_trades[trade_index]; - if ts < interval_start { - trade_index += 1; - continue; - } - if ts >= interval_end { - break; - } - - if signed_tokens != 0 { - let price = (signed_sats as f64 / signed_tokens as f64).abs(); - 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) -} - -/// Fill in 1-hour candles from pre-materialised `ohlcv_1h` rows, adding flat gap-filler -/// candles between active buckets just like the raw path does. -/// Returns the filled candles and the last close price for continuing into a raw tail. -fn fill_ohlcv_candles( - rows: Vec, - start: i64, - end: i64, - mut found_first_trade: bool, - mut last_close: Option, -) -> (Vec, bool, Option) { - let mut result = Vec::new(); - let mut row_iter = rows.into_iter().peekable(); - 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, - open: r.open, - close: r.close, - high: r.high, - low: r.low, - volume_sats: r.volume_sats, - 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, - }); - } - } - bucket += 3600; - } - - (result, found_first_trade, last_close) -} - -async fn fetch_raw_trades( - pool: &SqlitePool, - token_blob: &[u8], - timestamp_start: i64, - timestamp_end: i64, -) -> Result> { - let sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - phe.txid, - phe.effective_timestamp, - phe.utxo, - phe.sats_delta, - phe.token_delta, - phe.sequence - FROM pool_history_entry AS phe - WHERE phe.token_id = ? - AND phe.effective_timestamp >= ? - AND phe.effective_timestamp < ? -), -per_pool_tx AS ( - SELECT - txid, - effective_timestamp, - utxo, - MIN(sequence) AS min_sequence, - SUM(sats_delta) AS signed_sats_pool, - SUM(token_delta) AS signed_tokens_pool, - SUM(ABS(sats_delta)) AS volume_sats_pool, - SUM(ABS(token_delta)) AS volume_tokens_pool - FROM per_pool_tx_raw - GROUP BY txid, effective_timestamp, utxo -), -tx_trades AS ( - SELECT - txid, - effective_timestamp, - MIN(min_sequence) AS min_sequence, - SUM(signed_sats_pool) AS signed_sats, - SUM(signed_tokens_pool) AS signed_tokens, - SUM(volume_sats_pool) AS volume_sats, - SUM(volume_tokens_pool) AS volume_tokens - FROM per_pool_tx - GROUP BY txid, effective_timestamp -) -SELECT - effective_timestamp, - signed_sats, - signed_tokens, - volume_sats, - volume_tokens -FROM tx_trades -ORDER BY effective_timestamp ASC, min_sequence ASC; -"#; - let rows = sqlx::query(sql) - .bind(token_blob) - .bind(timestamp_start) - .bind(timestamp_end) - .fetch_all(pool) - .await?; - - Ok(rows - .into_iter() - .map(|r| (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4))) - .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( - pool: &SqlitePool, - token_blob: &[u8], - timestamp_end: i64, -) -> Result> { - let sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - phe.txid, - phe.effective_timestamp, - phe.utxo, - phe.sats_delta, - phe.token_delta, - phe.sequence - FROM pool_history_entry AS phe - WHERE phe.token_id = ? - AND phe.effective_timestamp < ? -), -per_pool_tx AS ( - SELECT - txid, - effective_timestamp, - utxo, - MIN(sequence) AS min_sequence, - SUM(sats_delta) AS signed_sats_pool, - SUM(token_delta) AS signed_tokens_pool - FROM per_pool_tx_raw - GROUP BY txid, effective_timestamp, utxo -), -tx_trades AS ( - SELECT - txid, - effective_timestamp, - MIN(min_sequence) AS min_sequence, - SUM(signed_sats_pool) AS signed_sats, - SUM(signed_tokens_pool) AS signed_tokens - FROM per_pool_tx - GROUP BY txid, effective_timestamp -) -SELECT ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS close_price -FROM tx_trades -WHERE signed_tokens != 0 -ORDER BY effective_timestamp DESC, min_sequence DESC -LIMIT 1 -"#; - let row = sqlx::query(sql) - .bind(token_blob) - .bind(timestamp_end) - .fetch_optional(pool) - .await?; - - Ok(row.map(|r| r.get::(0))) -} - -/// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`. -/// Pass 0 to always use the raw CTE path. -pub async fn candlesticks( - pool: &SqlitePool, - timestamp_start: i64, - timestamp_end: i64, - step_size: i64, - token_id: &str, - ohlcv_materialized_end: i64, -) -> Result> { - if timestamp_start > timestamp_end { - bail!("Start cannot be higher than end"); - } - - 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 - // start would produce candles whose boundaries disagree with the raw path. - if step_size == 3600 && ohlcv_materialized_end > timestamp_start && timestamp_start % 3600 == 0 - { - let ohlcv_end = ohlcv_materialized_end.min(timestamp_end); - - // 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, - ); - - 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); - } - - 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) -} - /// Fetch candlesticks in BCH satoshis for a token. /// /// If an interval has no trades, it will be omitted from the result. diff --git a/src/rpc/candlesticks/tests.rs b/src/rpc/candlesticks/tests.rs index 520faea..2fab3a5 100644 --- a/src/rpc/candlesticks/tests.rs +++ b/src/rpc/candlesticks/tests.rs @@ -1,4 +1,8 @@ -use crate::db::blob::ToBlob; +// Copyright (C) 2024-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 + use crate::db::cauldron::{ ohlcv, pool::{self, dummy_init_seq, insert_new_pool}, @@ -38,7 +42,6 @@ const TIME_2: u64 = 1727963600; const TIME_3: u64 = 1727963900; const TIME_4: u64 = 1727964200; -/// Helper to build a ParsedContract fn dummy_cauldron( txid: &Txid, utxo: &OutPointHash, @@ -61,7 +64,6 @@ fn dummy_cauldron( } async fn setup_mock_db(pool: sqlx::SqlitePool) { - // Create tables utxo_funding::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; @@ -72,7 +74,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) { let token_zero = TokenID::all_zeros(); let pkh_zero = PubkeyHash::all_zeros(); - // We'll make 4 trades with distinct times let txid1 = Txid::from_inner([0xf1; 32]); let txid2 = Txid::from_inner([0xf2; 32]); let txid3 = Txid::from_inner([0xf3; 32]); @@ -100,29 +101,24 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) { .await .unwrap(); - // Insert tx rows in realistic order: mempool first, then confirmed. let block_zero = BlockHash::all_zeros(); insert_mempool_tx(&mut *conn, &txid1, TIME_1).await.unwrap(); insert_block_tx(&mut *conn, &txid1, &block_zero, TIME_1 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid2, TIME_2).await.unwrap(); insert_block_tx(&mut *conn, &txid2, &block_zero, TIME_2 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid3, TIME_3).await.unwrap(); insert_block_tx(&mut *conn, &txid3, &block_zero, TIME_3 as i64) .await .unwrap(); - insert_mempool_tx(&mut *conn, &txid4, TIME_4).await.unwrap(); insert_block_tx(&mut *conn, &txid4, &block_zero, TIME_4 as i64) .await .unwrap(); - // Insert pool_history_entry let pool1 = OutPointHash::from_inner([0x0a; 32]); let pool2 = OutPointHash::from_inner([0x0b; 32]); let pool3 = OutPointHash::from_inner([0x0c; 32]); @@ -173,7 +169,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) { .await .unwrap(); - // Insert pools let token1 = TokenID::from_inner([0xda; 32]); let pkh1 = PubkeyHash::from_inner([0xca; 20]); insert_new_pool( @@ -202,6 +197,60 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) { .unwrap(); } +// ── Seeded gap-fill helpers ─────────────────────────────────────────────── + +async fn setup_seed_db(pool: sqlx::SqlitePool) { + utxo_funding::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_inner([txid_byte; 32]); + let utxo = OutPointHash::from_inner([txid_byte; 32]); + let pool_hash = OutPointHash::from_inner([txid_byte.wrapping_add(0x80); 32]); + let block = BlockHash::all_zeros(); + + let cauldron = dummy_cauldron( + &txid, + &utxo, + token, + sats_delta.unsigned_abs(), + token_delta, + &PubkeyHash::all_zeros(), + ); + + 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, + ) + .await + .unwrap(); +} + +// ── Endpoint tests ──────────────────────────────────────────────────────── + #[rocket::async_test] async fn test_future_end_timestamp() { let mock_db = mock_db_pool(setup_mock_db).await; @@ -213,7 +262,6 @@ async fn test_future_end_timestamp() { .await .expect("valid rocket instance"); - // Use a far-future timestamp for 'end' let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; let future_end = time_now() + 1_000_000; @@ -299,7 +347,6 @@ async fn test_multiple_candlesticks_endpoint() { // ----- Candle #1 ----- let cndl1 = &cndl_array[0]; - println!("First candlestick: {cndl1:?}"); // 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 @@ -314,7 +361,6 @@ async fn test_multiple_candlesticks_endpoint() { // ----- Candle #2 ----- let cndl2 = &cndl_array[1]; - println!("Second candlestick: {cndl2:?}"); // 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 @@ -323,7 +369,6 @@ async fn test_multiple_candlesticks_endpoint() { 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 assert_eq!(cndl2["volume_sats"].as_i64().unwrap(), 160_000 + 200_000); assert_eq!(cndl2["volume_tokens"].as_i64().unwrap(), 4_000); @@ -332,9 +377,7 @@ async fn test_multiple_candlesticks_endpoint() { #[rocket::async_test] async fn test_single_swap_multiple_pools() { - // Set up a fresh mock DB let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { - // Create required tables utxo_funding::create_table(&pool).await; tx::create_table(&pool).await; pool::create_table(&pool).await; @@ -344,12 +387,9 @@ async fn test_single_swap_multiple_pools() { let token_zero = TokenID::all_zeros(); let pkh_zero = PubkeyHash::all_zeros(); - - // Create a single transaction that will be used for multiple pool trades let txid_multi = Txid::from_inner([0xaa; 32]); let block_zero = BlockHash::all_zeros(); - // Create multiple pool and pool_history_entry records for the same txid let mut pools = Vec::new(); let times = [TIME_1, TIME_2]; for (i, &time) in times.iter().enumerate() { @@ -368,7 +408,6 @@ async fn test_single_swap_multiple_pools() { insert_utxo_funding(&mut *conn, &vec![cauldron.clone()], &txid_multi) .await .unwrap(); - insert_block_tx(&mut *conn, &txid_multi, &block_zero, time as i64) .await .unwrap(); @@ -376,23 +415,19 @@ async fn test_single_swap_multiple_pools() { .await .unwrap(); - let sats_delta = (100_000 * (i + 1)) as i64; - let token_delta = 2_000i64; - pool::insert_pool_history_entry( &mut *conn, &pool_hash, &cauldron, Some(time), Some(time), - sats_delta, - token_delta, + (100_000 * (i + 1)) as i64, + 2_000i64, ) .await .unwrap(); } - // Insert pools into the pool table let token1 = TokenID::from_inner([0xda; 32]); let pkh1 = PubkeyHash::from_inner([0xca; 20]); for pool_hash in &pools { @@ -406,7 +441,6 @@ async fn test_single_swap_multiple_pools() { }) .await; - // Build Rocket instance with our endpoint let rocket = rocket::build() .manage(mock_db) .manage(no_ohlcv()) @@ -417,7 +451,6 @@ async fn test_single_swap_multiple_pools() { let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - // Query a time range that includes our test transactions let response = client .get(format!( "/api/price/{}/candlesticks?start={}&end={}&stepsize=600", @@ -434,17 +467,11 @@ async fn test_single_swap_multiple_pools() { let candles = json["candlesticks"].as_array().unwrap(); let first_candle = &candles[0]; - let expected_volume_sats = 100_000 + 200_000; - let expected_volume_tokens = 2000 + 2000; - assert_eq!( first_candle["volume_sats"].as_i64().unwrap(), - expected_volume_sats - ); - assert_eq!( - first_candle["volume_tokens"].as_i64().unwrap(), - expected_volume_tokens + 100_000 + 200_000 ); + assert_eq!(first_candle["volume_tokens"].as_i64().unwrap(), 2000 + 2000); } #[rocket::async_test] @@ -572,13 +599,11 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { insert_utxo_funding(&mut *conn, &vec![ca_b1.clone()], &txid_net0) .await .unwrap(); - insert_block_tx(&mut *conn, &txid_net0, &block_zero, t1 as i64) .await .unwrap(); insert_mempool_tx(&mut *conn, &txid_net0, t1).await.unwrap(); - // Opposite deltas within the same tx pool::insert_pool_history_entry( &mut *conn, &pool_a, @@ -590,6 +615,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { ) .await .unwrap(); + // Opposite deltas within the same tx pool::insert_pool_history_entry( &mut *conn, &pool_b, @@ -604,7 +630,6 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { }) .await; - // Build Rocket and call endpoint across the two intervals let rocket = rocket::build() .manage(mock_db) .manage(no_ohlcv()) @@ -656,7 +681,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() { assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1); } -// ── ohlcv_1h fast-path tests ───────────────────────────────────────────── +// ── ohlcv_1h fast-path tests ────────────────────────────────────────────── // // Test data from setup_mock_db — four trades at: // TIME_1=1727963400, TIME_2=1727963600, TIME_3=1727963900 → hour bucket 1727960400 @@ -841,115 +866,10 @@ async fn test_ohlcv_skipped_for_non_aligned_start() { assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 0); } -// ── fetch_last_close_before + gap-fill seeding tests ─────────────────── +// ── Seeded gap-fill endpoint tests ──────────────────────────────────────── -/// Insert a single confirmed trade for a token. -/// `txid_byte` is used to derive unique txid/utxo hashes. -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_inner([txid_byte; 32]); - let utxo = OutPointHash::from_inner([txid_byte; 32]); - let pool_hash = OutPointHash::from_inner([txid_byte.wrapping_add(0x80); 32]); - let block = BlockHash::all_zeros(); - - let cauldron = dummy_cauldron( - &txid, - &utxo, - token, - sats_delta.unsigned_abs(), - token_delta, - &PubkeyHash::all_zeros(), - ); - - 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, - ) - .await - .unwrap(); -} - -async fn setup_seed_db(pool: sqlx::SqlitePool) { - utxo_funding::create_table(&pool).await; - tx::create_table(&pool).await; - pool::create_table(&pool).await; - ohlcv::create_table(&pool).await; - dummy_init_seq(); -} - -/// fetch_last_close_before returns None when there are no trades at all. -#[tokio::test] -async fn test_fetch_last_close_before_no_trades() { - let db = mock_db_pool(setup_seed_db).await; - let token = TokenID::from_inner([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()); -} - -/// fetch_last_close_before returns None when all trades are after timestamp_end. -#[tokio::test] -async fn test_fetch_last_close_before_only_future_trades() { - let db = mock_db_pool(setup_seed_db).await; - let token = TokenID::from_inner([0xAB; 32]); - let token_blob = token.to_blob(); - - let mut conn = db.cauldron_w.acquire().await.unwrap(); - // Trade at ts=2000, query for ts < 1000 → None - 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()); -} - -/// fetch_last_close_before returns the most recent trade before the cutoff, -/// not earlier ones, and computes the correct price. -#[tokio::test] -async fn test_fetch_last_close_before_returns_most_recent() { - let db = mock_db_pool(setup_seed_db).await; - let token = TokenID::from_inner([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); -} - -/// Raw path: when the window has no trades but there is a prior trade, -/// all candles in the window are flat gap-fill at the prior close price. +/// When the window has no trades but there is a prior trade, +/// all candles are flat gap-fill at the prior close price. #[rocket::async_test] async fn test_raw_path_seeded_gap_fill_no_in_window_trades() { let token = TokenID::from_inner([0xBA; 32]); @@ -958,7 +878,6 @@ 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(); - // price = 100_000 / 2_000 = 50 — BEFORE the query window insert_trade_at(&mut conn, &token_copy, 0x10, 1_000, 100_000, 2_000).await; }) .await; @@ -969,7 +888,6 @@ async fn test_raw_path_seeded_gap_fill_no_in_window_trades() { .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket).await.unwrap(); - // Window: [2000, 3000), step=500 — no in-window trades let resp = client .get(format!( "/api/price/{}/candlesticks?start=2000&end=3000&stepsize=500", @@ -994,7 +912,7 @@ async fn test_raw_path_seeded_gap_fill_no_in_window_trades() { } } -/// Raw path: pre-window seed fills the gap before the first in-window trade, +/// Pre-window seed fills the gap before the first in-window trade, /// then real candle and post-trade gap-fill use the in-window close. #[rocket::async_test] async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { @@ -1004,9 +922,7 @@ 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(); - // Pre-window trade: price = 50 insert_trade_at(&mut conn, &token_copy, 0x20, 1_000, 100_000, 2_000).await; - // In-window trade at ts=2500: price = 150_000 / 2_000 = 75 insert_trade_at(&mut conn, &token_copy, 0x21, 2_500, 150_000, 2_000).await; }) .await; @@ -1017,7 +933,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket).await.unwrap(); - // Window [2000, 3500), step=500 → intervals: [2000,2500), [2500,3000), [3000,3500) let resp = client .get(format!( "/api/price/{}/candlesticks?start=2000&end=3500&stepsize=500", @@ -1033,7 +948,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { assert_eq!(candles.len(), 3); - // First interval: gap-fill seeded from pre-window close (50) assert_eq!(candles[0]["time"].as_i64().unwrap(), 2000); assert_eq!(candles[0]["transaction_count"].as_i64().unwrap(), 0); assert!( @@ -1042,7 +956,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { candles[0]["close"] ); - // Second interval: real in-window trade (75) assert_eq!(candles[1]["time"].as_i64().unwrap(), 2500); assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 1); assert!( @@ -1051,7 +964,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { candles[1]["close"] ); - // Third interval: gap-fill from in-window close (75), not from seed (50) assert_eq!(candles[2]["time"].as_i64().unwrap(), 3000); assert_eq!(candles[2]["transaction_count"].as_i64().unwrap(), 0); assert!( @@ -1061,8 +973,7 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() { ); } -/// Without a pre-window trade, no candles appear before the first in-window trade. -/// This confirms the seed is only active when a prior trade actually exists. +/// Without a pre-window trade no candles appear before the first in-window trade. #[rocket::async_test] async fn test_raw_path_no_seed_no_prefill() { let token = TokenID::from_inner([0xBC; 32]); @@ -1071,7 +982,6 @@ 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(); - // Only an in-window trade at ts=2500: price=75 — no pre-window trade insert_trade_at(&mut conn, &token_copy, 0x30, 2_500, 150_000, 2_000).await; }) .await; @@ -1095,7 +1005,6 @@ async fn test_raw_path_no_seed_no_prefill() { let json: serde_json::Value = serde_json::from_str(&body).unwrap(); let candles = json["candlesticks"].as_array().unwrap(); - // [2000,2500) must be absent — no seed, no prior close to gap-fill from assert_eq!( candles.len(), 2,