From b36c0b71a6fd7a41031e304621ead042ea930478 Mon Sep 17 00:00:00 2001 From: jakobsn Date: Tue, 23 Jun 2026 14:16:19 +0000 Subject: [PATCH] Use last trade as the basis for the price outside chart instead of the last token/sats amount --- src/db/cauldron/candlestick/mod.rs | 415 ++++++++ src/db/cauldron/candlestick/tests.rs | 195 ++++ src/db/cauldron/mod.rs | 1 + src/rpc/candlesticks.rs | 1318 -------------------------- src/rpc/candlesticks/mod.rs | 136 +++ src/rpc/candlesticks/tests.rs | 1016 ++++++++++++++++++++ 6 files changed, 1763 insertions(+), 1318 deletions(-) create mode 100644 src/db/cauldron/candlestick/mod.rs create mode 100644 src/db/cauldron/candlestick/tests.rs delete mode 100644 src/rpc/candlesticks.rs create mode 100644 src/rpc/candlesticks/mod.rs create mode 100644 src/rpc/candlesticks/tests.rs 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..f6e1409 --- /dev/null +++ b/src/db/cauldron/candlestick/tests.rs @@ -0,0 +1,195 @@ +// 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_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(); + + 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_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!( + (result.unwrap() - 50.0).abs() < f64::EPSILON, + "net-zero trade must not affect close price" + ); +} + +/// Trades for a different token must not bleed into results for the queried token. +#[tokio::test] +async fn test_fetch_last_close_before_token_isolation() { + 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"); +} 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.rs b/src/rpc/candlesticks.rs deleted file mode 100644 index e8dcaaa..0000000 --- a/src/rpc/candlesticks.rs +++ /dev/null @@ -1,1318 +0,0 @@ -// 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 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()) -} - -/// `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)?; - - // 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, false, None); - - 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, false, None); - Ok(result) -} - -/// Fetch candlesticks in BCH satoshis for a token. -/// -/// If an interval has no trades, it will be omitted from the result. -/// -/// -/// - start: unix timestamp for period start (default 30 days) -/// - end: unix timestamp for period end (default NOW) -/// - stepsize: seconds per interval (default: 3600 seconds) -/// -/// **Response Example:** -/// -/// ```json -/// { -/// "candlesticks": [ -/// {"close":64654136.35714286, -/// "high":87959043.0, -/// "low":58755326.0, -/// "open":87959043.0, -/// "time":1752522150, -/// "transaction_count":4, -/// "volume_sats":3170247594, -/// "volume_tokens":43}, -/// ] -/// } -/// ``` -/// -#[get("/price//candlesticks?&&")] -pub async fn price_candlesticks( - token: &str, - start: Option, - end: Option, - stepsize: Option, - conn: &State, - ohlcv: &State>, -) -> CachedApiResult { - let current_timestamp = time_now(); - - if let Some(end_ts) = end { - if end_ts > current_timestamp { - return Err(bad_request( - ApiErrorCode::FutureTimestamp, - "End timestamp cannot be in the future", - )); - } - } - - let effective_end = end.unwrap_or(current_timestamp); - let effective_start = start.unwrap_or(current_timestamp - 30 * 24 * 3600); - let effective_stepsize = stepsize.unwrap_or(3600); - - if effective_stepsize <= 0 { - return Err(bad_request( - ApiErrorCode::InvalidStepsize, - "stepsize must be > 0", - )); - } - - if effective_start >= effective_end { - return Err(bad_request( - ApiErrorCode::InvalidTimeRange, - "Start timestamp must be before end timestamp", - )); - } - - const MAX_INTERVALS: i64 = 10000; - let total_intervals = (effective_end - effective_start) / effective_stepsize; - if total_intervals > MAX_INTERVALS { - return Err(bad_request( - ApiErrorCode::TooManyIntervals, - &format!("Too many intervals ({total_intervals} > {MAX_INTERVALS})"), - )); - } - - // Windows fully in the past get immutable HTTP cache headers. - let is_historical = end.is_some() && effective_end < current_timestamp - effective_stepsize; - - let ohlcv_end = ohlcv.materialized_end.load(Ordering::Relaxed); - - let candlestick_data = candlesticks( - &conn.cauldron_r, - effective_start, - effective_end, - effective_stepsize, - token, - ohlcv_end, - ) - .await - .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?; - - let candlesticks_json: Vec = candlestick_data - .into_iter() - .map(|c| { - json!({ - "time": c.time, - "high": c.high, - "low": c.low, - "open": c.open, - "close": c.close, - "volume_sats": c.volume_sats, - "volume_tokens": c.volume_tokens, - "transaction_count": c.transaction_count - }) - }) - .collect(); - - let cache_duration = if is_historical { - CACHE_IMMUTABLE - } else { - CACHE_NONE - }; - - Ok(cached_ok( - json!({ "candlesticks": candlesticks_json }), - cache_duration, - )) -} - -#[cfg(test)] -mod tests { - use crate::db::cauldron::{ - ohlcv, - pool::{self, dummy_init_seq, insert_new_pool}, - tx::{self, insert_block_tx, insert_mempool_tx}, - utxo_funding::{self, insert_utxo_funding}, - }; - use crate::utiltest::mock_db_pool; - use crate::OhlcvState; - - use crate::timeutil::time_now; - use bitcoin_hashes::Hash; - use bitcoincash::{BlockHash, PubkeyHash, TokenID, Txid}; - use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash}; - use rocket::http::Status; - use rocket::local::asynchronous::Client; - use rocket::routes; - use std::sync::atomic::AtomicI64; - use std::sync::Arc; - - fn no_ohlcv() -> Arc { - Arc::new(OhlcvState { - materialized_end: AtomicI64::new(0), - }) - } - - /// Four trades, each 600-second bin is 10 minutes. - /// We'll have two bins: - /// Bin #1: [1727963300..1727963900) - /// T1=1727963300 => ratio=40.0 - /// T2=1727963600 => ratio=60.0 - /// Bin #2: [1727963900..1727964500) - /// T3=1727963900 => ratio=80.0 - /// T4=1727964200 => ratio=100.0 - const TIME_1: u64 = 1727963400; - 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, - 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_mock_db(pool: sqlx::SqlitePool) { - // Create tables - utxo_funding::create_table(&pool).await; - tx::create_table(&pool).await; - pool::create_table(&pool).await; - dummy_init_seq(); - - let mut conn = pool.acquire().await.unwrap(); - - let token_zero = TokenID::all_zeros(); - let pkh_zero = PubkeyHash::all_zeros(); - - // We'll make 4 trades with distinct times - let txid1 = Txid::from_byte_array([0xf1; 32]); - let txid2 = Txid::from_byte_array([0xf2; 32]); - let txid3 = Txid::from_byte_array([0xf3; 32]); - let txid4 = Txid::from_byte_array([0xf4; 32]); - let utxo1 = OutPointHash::from_byte_array([0xe1; 32]); - let utxo2 = OutPointHash::from_byte_array([0xe2; 32]); - let utxo3 = OutPointHash::from_byte_array([0xe3; 32]); - let utxo4 = OutPointHash::from_byte_array([0xe4; 32]); - - let cauldron1 = dummy_cauldron(&txid1, &utxo1, &token_zero, 80_000, 2_000, &pkh_zero); - let cauldron2 = dummy_cauldron(&txid2, &utxo2, &token_zero, 120_000, 2_000, &pkh_zero); - let cauldron3 = dummy_cauldron(&txid3, &utxo3, &token_zero, 160_000, 2_000, &pkh_zero); - let cauldron4 = dummy_cauldron(&txid4, &utxo4, &token_zero, 200_000, 2_000, &pkh_zero); - - insert_utxo_funding(&mut conn, &vec![cauldron1.clone()], &txid1) - .await - .unwrap(); - insert_utxo_funding(&mut conn, &vec![cauldron2.clone()], &txid2) - .await - .unwrap(); - insert_utxo_funding(&mut conn, &vec![cauldron3.clone()], &txid3) - .await - .unwrap(); - insert_utxo_funding(&mut conn, &vec![cauldron4.clone()], &txid4) - .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_byte_array([0x0a; 32]); - let pool2 = OutPointHash::from_byte_array([0x0b; 32]); - let pool3 = OutPointHash::from_byte_array([0x0c; 32]); - let pool4 = OutPointHash::from_byte_array([0x0d; 32]); - - pool::insert_pool_history_entry( - &mut conn, - &pool1, - &cauldron1, - Some(TIME_1), - Some(TIME_1), - 80_000, - 2_000, - ) - .await - .unwrap(); - pool::insert_pool_history_entry( - &mut conn, - &pool2, - &cauldron2, - Some(TIME_2), - Some(TIME_2), - 120_000, - 2_000, - ) - .await - .unwrap(); - pool::insert_pool_history_entry( - &mut conn, - &pool3, - &cauldron3, - Some(TIME_3), - Some(TIME_3), - 160_000, - 2_000, - ) - .await - .unwrap(); - pool::insert_pool_history_entry( - &mut conn, - &pool4, - &cauldron4, - Some(TIME_4), - Some(TIME_4), - 200_000, - 2_000, - ) - .await - .unwrap(); - - // Insert pools - let token1 = TokenID::from_byte_array([0xda; 32]); - 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(); - } - - #[rocket::async_test] - async fn test_future_end_timestamp() { - let mock_db = mock_db_pool(setup_mock_db).await; - let rocket = rocket::build() - .manage(mock_db) - .manage(no_ohlcv()) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket) - .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; - - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks?end={future_end}" - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::BadRequest); - let body = response.into_string().await.unwrap_or_default(); - let json: serde_json::Value = serde_json::from_str(&body).expect("Response should be JSON"); - assert_eq!(json["error"]["code"], "FUTURE_TIMESTAMP"); - assert!(json["error"]["message"] - .as_str() - .unwrap() - .contains("future")); - } - - #[rocket::async_test] - async fn test_start_after_end_timestamp() { - let mock_db = mock_db_pool(setup_mock_db).await; - let rocket = rocket::build() - .manage(mock_db) - .manage(no_ohlcv()) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket) - .await - .expect("valid rocket instance"); - - let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - let start_after_end = "2000&end=1000"; - - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks?start={start_after_end}" - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::BadRequest); - let body = response.into_string().await.unwrap_or_default(); - let json: serde_json::Value = serde_json::from_str(&body).expect("Response should be JSON"); - assert_eq!(json["error"]["code"], "INVALID_TIME_RANGE"); - assert!(json["error"]["message"] - .as_str() - .unwrap() - .contains("before end")); - } - - #[rocket::async_test] - async fn test_multiple_candlesticks_endpoint() { - let mock_db = mock_db_pool(setup_mock_db).await; - - let rocket = rocket::build() - .manage(mock_db) - .manage(no_ohlcv()) - .mount("/api", routes![super::price_candlesticks]); - - let client = Client::tracked(rocket) - .await - .expect("valid rocket instance"); - - let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - - // We'll query from 1727963300..1727964500 with step=600 => 10 minutes - // This yields 2 candles: one in [3300..3900), another in [3900..4500). - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks?start=1727963300&end=1727964500&stepsize=600" - )) - .dispatch().await; - - assert_eq!(response.status(), Status::Ok); - - let body = response.into_string().await.expect("No response body"); - let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); - - let cndl_array = json["candlesticks"].as_array().unwrap(); - assert_eq!(cndl_array.len(), 2, "Should produce exactly two candles"); - - // ----- 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 - 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["low"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); - assert!((cndl1["high"].as_f64().unwrap() - 60.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 ----- - 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 - 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 - 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); - } - - #[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; - dummy_init_seq(); - - let mut conn = pool.acquire().await.unwrap(); - - 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_byte_array([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() { - let utxo = OutPointHash::from_byte_array([0xe1 + i as u8; 32]); - let pool_hash = OutPointHash::from_byte_array([0x0a + i as u8; 32]); - let cauldron = dummy_cauldron( - &txid_multi, - &utxo, - &token_zero, - 100_000 * (i + 1) as u64, - 2_000, - &pkh_zero, - ); - pools.push(pool_hash); - - 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(); - insert_mempool_tx(&mut conn, &txid_multi, time) - .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, - ) - .await - .unwrap(); - } - - // Insert pools into the pool table - 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), - ) - .await - .unwrap(); - } - }) - .await; - - // Build Rocket instance with our endpoint - let rocket = rocket::build() - .manage(mock_db) - .manage(no_ohlcv()) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket) - .await - .expect("valid rocket instance"); - - 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", - token_id_zero, - TIME_1 - 100, - TIME_4 + 100 - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::Ok); - let body = response.into_string().await.expect("No response body"); - let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); - 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 - ); - } - - #[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 --- - utxo_funding::create_table(&pool).await; - tx::create_table(&pool).await; - pool::create_table(&pool).await; - dummy_init_seq(); - - let mut conn = pool.acquire().await.unwrap(); - - let token_zero = TokenID::all_zeros(); - 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]); - - // 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) -------- - 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), - }; - insert_utxo_funding(&mut conn, &vec![cauldron_p.clone()], &txid_price) - .await - .unwrap(); - insert_block_tx(&mut conn, &txid_price, &block_zero, t0 as i64) - .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, - &cauldron_p, - Some(t0), - Some(t0), - 10_000, - 200, - ) - .await - .unwrap(); - - // -------- Candle #2 (net zero tokens but non-zero volume) -------- - 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 - }; - - insert_utxo_funding(&mut conn, &vec![ca_a1.clone()], &txid_net0) - .await - .unwrap(); - 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, - &ca_a1, - Some(t1), - Some(t1), - 10_000, - 200, - ) - .await - .unwrap(); - pool::insert_pool_history_entry( - &mut conn, - &pool_b, - &ca_b1, - Some(t1), - Some(t1), - -10_000, - -200, - ) - .await - .unwrap(); - }) - .await; - - // Build Rocket and call endpoint across the two intervals - let rocket = rocket::build() - .manage(mock_db) - .manage(no_ohlcv()) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket) - .await - .expect("valid rocket instance"); - - let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - let start = 1_700_000_000u64; - let end = start + 2 * 600; - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks?start={start}&end={end}&stepsize=600" - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::Ok); - let body = response.into_string().await.expect("No response body"); - let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); - let candles = json["candlesticks"].as_array().unwrap(); - - 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!((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 (net-zero tokens, carry-forward close) --- - let c2 = &candles[1]; - assert_eq!(c2["time"].as_i64().unwrap(), (start + 600) as i64); - // Should carry previous close (50) because signed_tokens==0 but volume>0 - assert!((c2["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); - 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 - 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); - } - - // ── 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 - // TIME_4=1727964200 → hour bucket 1727964000 - // - // Expected OHLCV (bucket 1727960400): open=40, close=80, high=80, 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 - // vol_sats=200_000, vol_tokens=2_000, tx_count=1 - - /// Full range served from ohlcv_1h (materialized_end covers everything). - #[rocket::async_test] - async fn test_ohlcv_fast_path_full_range() { - let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { - setup_mock_db(pool.clone()).await; - ohlcv::create_table(&pool).await; - ohlcv::rebuild_range(&pool, &pool, 0, 1727970000) - .await - .expect("rebuild_range"); - }) - .await; - - let ohlcv_state = Arc::new(OhlcvState { - materialized_end: AtomicI64::new(1727970000), - }); - - let rocket = rocket::build() - .manage(mock_db) - .manage(ohlcv_state) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket).await.expect("valid rocket"); - - let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - - // Hour-aligned range covering both buckets. - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks\ - ?start=1727960400&end=1727967600&stepsize=3600" - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::Ok); - let body = response.into_string().await.unwrap(); - let json: serde_json::Value = serde_json::from_str(&body).unwrap(); - let candles = json["candlesticks"].as_array().unwrap(); - - assert_eq!(candles.len(), 2, "expected two 1h candles"); - - 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["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); - assert_eq!(c1["transaction_count"].as_i64().unwrap(), 3); - - 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_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); - } - - /// First bucket served from ohlcv_1h, second bucket served from raw CTE tail. - /// Both sources should produce identical output to the full-ohlcv test above. - #[rocket::async_test] - async fn test_ohlcv_fast_path_with_raw_tail() { - let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { - setup_mock_db(pool.clone()).await; - ohlcv::create_table(&pool).await; - // Only materialise up to 1727964000 — leaves the second bucket (T4) for raw. - ohlcv::rebuild_range(&pool, &pool, 0, 1727964000) - .await - .expect("rebuild_range"); - }) - .await; - - let ohlcv_state = Arc::new(OhlcvState { - // materialized_end = 1727964000: ohlcv has bucket 1727960400 only. - materialized_end: AtomicI64::new(1727964000), - }); - - let rocket = rocket::build() - .manage(mock_db) - .manage(ohlcv_state) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket).await.expect("valid rocket"); - - let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks\ - ?start=1727960400&end=1727967600&stepsize=3600" - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::Ok); - let body = response.into_string().await.unwrap(); - let json: serde_json::Value = serde_json::from_str(&body).unwrap(); - let candles = json["candlesticks"].as_array().unwrap(); - - assert_eq!(candles.len(), 2, "expected two 1h candles"); - - // Candle 1 came from ohlcv_1h. - 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_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. - 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_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); - } - - /// Non-hour-aligned start must NOT use the ohlcv fast path (alignment guard). - #[rocket::async_test] - async fn test_ohlcv_skipped_for_non_aligned_start() { - let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { - setup_mock_db(pool.clone()).await; - ohlcv::create_table(&pool).await; - ohlcv::rebuild_range(&pool, &pool, 0, 1727970000) - .await - .expect("rebuild_range"); - }) - .await; - - let ohlcv_state = Arc::new(OhlcvState { - materialized_end: AtomicI64::new(1727970000), - }); - - let rocket = rocket::build() - .manage(mock_db) - .manage(ohlcv_state) - .mount("/api", routes![super::price_candlesticks]); - let client = Client::tracked(rocket).await.expect("valid rocket"); - - let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; - - // 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) - let response = client - .get(format!( - "/api/price/{token_id_zero}/candlesticks\ - ?start=1727963300&end=1727970500&stepsize=3600" - )) - .dispatch() - .await; - - assert_eq!(response.status(), Status::Ok); - let body = response.into_string().await.unwrap(); - let json: serde_json::Value = serde_json::from_str(&body).unwrap(); - let candles = json["candlesticks"].as_array().unwrap(); - - // Key assertion: first candle starts at 1727963300, not 1727960400. - // If the ohlcv path were mistakenly used it would start at 1727960400. - 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_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); - assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 0); - } -} diff --git a/src/rpc/candlesticks/mod.rs b/src/rpc/candlesticks/mod.rs new file mode 100644 index 0000000..ed37e84 --- /dev/null +++ b/src/rpc/candlesticks/mod.rs @@ -0,0 +1,136 @@ +// 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::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 rocket::{get, State}; +use serde_json::json; +use serde_json::Value; +use std::sync::atomic::Ordering; +use std::sync::Arc; + +/// Fetch candlesticks in BCH satoshis for a token. +/// +/// If an interval has no trades, it will be omitted from the result. +/// +/// +/// - start: unix timestamp for period start (default 30 days) +/// - end: unix timestamp for period end (default NOW) +/// - stepsize: seconds per interval (default: 3600 seconds) +/// +/// **Response Example:** +/// +/// ```json +/// { +/// "candlesticks": [ +/// {"close":64654136.35714286, +/// "high":87959043.0, +/// "low":58755326.0, +/// "open":87959043.0, +/// "time":1752522150, +/// "transaction_count":4, +/// "volume_sats":3170247594, +/// "volume_tokens":43}, +/// ] +/// } +/// ``` +/// +#[get("/price//candlesticks?&&")] +pub async fn price_candlesticks( + token: &str, + start: Option, + end: Option, + stepsize: Option, + conn: &State, + ohlcv: &State>, +) -> CachedApiResult { + let current_timestamp = time_now(); + + if let Some(end_ts) = end { + if end_ts > current_timestamp { + return Err(bad_request( + ApiErrorCode::FutureTimestamp, + "End timestamp cannot be in the future", + )); + } + } + + let effective_end = end.unwrap_or(current_timestamp); + let effective_start = start.unwrap_or(current_timestamp - 30 * 24 * 3600); + let effective_stepsize = stepsize.unwrap_or(3600); + + if effective_stepsize <= 0 { + return Err(bad_request( + ApiErrorCode::InvalidStepsize, + "stepsize must be > 0", + )); + } + + if effective_start >= effective_end { + return Err(bad_request( + ApiErrorCode::InvalidTimeRange, + "Start timestamp must be before end timestamp", + )); + } + + const MAX_INTERVALS: i64 = 10000; + let total_intervals = (effective_end - effective_start) / effective_stepsize; + if total_intervals > MAX_INTERVALS { + return Err(bad_request( + ApiErrorCode::TooManyIntervals, + &format!("Too many intervals ({total_intervals} > {MAX_INTERVALS})"), + )); + } + + // Windows fully in the past get immutable HTTP cache headers. + let is_historical = end.is_some() && effective_end < current_timestamp - effective_stepsize; + + let ohlcv_end = ohlcv.materialized_end.load(Ordering::Relaxed); + + let candlestick_data = candlesticks( + &conn.cauldron_r, + effective_start, + effective_end, + effective_stepsize, + token, + ohlcv_end, + ) + .await + .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?; + + let candlesticks_json: Vec = candlestick_data + .into_iter() + .map(|c| { + json!({ + "time": c.time, + "high": c.high, + "low": c.low, + "open": c.open, + "close": c.close, + "volume_sats": c.volume_sats, + "volume_tokens": c.volume_tokens, + "transaction_count": c.transaction_count + }) + }) + .collect(); + + let cache_duration = if is_historical { + CACHE_IMMUTABLE + } else { + CACHE_NONE + }; + + Ok(cached_ok( + json!({ "candlesticks": candlesticks_json }), + cache_duration, + )) +} + +#[cfg(test)] +mod tests; diff --git a/src/rpc/candlesticks/tests.rs b/src/rpc/candlesticks/tests.rs new file mode 100644 index 0000000..03756c4 --- /dev/null +++ b/src/rpc/candlesticks/tests.rs @@ -0,0 +1,1016 @@ +// 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}, + tx::{self, insert_block_tx, insert_mempool_tx}, + utxo_funding::{self, insert_utxo_funding}, +}; +use crate::utiltest::mock_db_pool; +use crate::OhlcvState; + +use crate::timeutil::time_now; +use bitcoin_hashes::Hash; +use bitcoincash::{BlockHash, PubkeyHash, TokenID, Txid}; +use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash}; +use rocket::http::Status; +use rocket::local::asynchronous::Client; +use rocket::routes; +use std::sync::atomic::AtomicI64; +use std::sync::Arc; + +fn no_ohlcv() -> Arc { + Arc::new(OhlcvState { + materialized_end: AtomicI64::new(0), + }) +} + +/// Four trades, each 600-second bin is 10 minutes. +/// We'll have two bins: +/// Bin #1: [1727963300..1727963900) +/// T1=1727963300 => ratio=40.0 +/// T2=1727963600 => ratio=60.0 +/// Bin #2: [1727963900..1727964500) +/// T3=1727963900 => ratio=80.0 +/// T4=1727964200 => ratio=100.0 +const TIME_1: u64 = 1727963400; +const TIME_2: u64 = 1727963600; +const TIME_3: u64 = 1727963900; +const TIME_4: u64 = 1727964200; + +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_mock_db(pool: sqlx::SqlitePool) { + utxo_funding::create_table(&pool).await; + tx::create_table(&pool).await; + pool::create_table(&pool).await; + dummy_init_seq(); + + let mut conn = pool.acquire().await.unwrap(); + + let token_zero = TokenID::all_zeros(); + let pkh_zero = PubkeyHash::all_zeros(); + + let txid1 = Txid::from_byte_array([0xf1; 32]); + let txid2 = Txid::from_byte_array([0xf2; 32]); + let txid3 = Txid::from_byte_array([0xf3; 32]); + let txid4 = Txid::from_byte_array([0xf4; 32]); + let utxo1 = OutPointHash::from_byte_array([0xe1; 32]); + let utxo2 = OutPointHash::from_byte_array([0xe2; 32]); + let utxo3 = OutPointHash::from_byte_array([0xe3; 32]); + let utxo4 = OutPointHash::from_byte_array([0xe4; 32]); + + let cauldron1 = dummy_cauldron(&txid1, &utxo1, &token_zero, 80_000, 2_000, &pkh_zero); + let cauldron2 = dummy_cauldron(&txid2, &utxo2, &token_zero, 120_000, 2_000, &pkh_zero); + let cauldron3 = dummy_cauldron(&txid3, &utxo3, &token_zero, 160_000, 2_000, &pkh_zero); + let cauldron4 = dummy_cauldron(&txid4, &utxo4, &token_zero, 200_000, 2_000, &pkh_zero); + + insert_utxo_funding(&mut conn, &vec![cauldron1.clone()], &txid1) + .await + .unwrap(); + insert_utxo_funding(&mut conn, &vec![cauldron2.clone()], &txid2) + .await + .unwrap(); + insert_utxo_funding(&mut conn, &vec![cauldron3.clone()], &txid3) + .await + .unwrap(); + insert_utxo_funding(&mut conn, &vec![cauldron4.clone()], &txid4) + .await + .unwrap(); + + 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(); + + let pool1 = OutPointHash::from_byte_array([0x0a; 32]); + let pool2 = OutPointHash::from_byte_array([0x0b; 32]); + let pool3 = OutPointHash::from_byte_array([0x0c; 32]); + let pool4 = OutPointHash::from_byte_array([0x0d; 32]); + + pool::insert_pool_history_entry( + &mut conn, + &pool1, + &cauldron1, + Some(TIME_1), + Some(TIME_1), + 80_000, + 2_000, + ) + .await + .unwrap(); + pool::insert_pool_history_entry( + &mut conn, + &pool2, + &cauldron2, + Some(TIME_2), + Some(TIME_2), + 120_000, + 2_000, + ) + .await + .unwrap(); + pool::insert_pool_history_entry( + &mut conn, + &pool3, + &cauldron3, + Some(TIME_3), + Some(TIME_3), + 160_000, + 2_000, + ) + .await + .unwrap(); + pool::insert_pool_history_entry( + &mut conn, + &pool4, + &cauldron4, + Some(TIME_4), + Some(TIME_4), + 200_000, + 2_000, + ) + .await + .unwrap(); + + let token1 = TokenID::from_byte_array([0xda; 32]); + 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(); +} + +// ── 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_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(); + + 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; + let rocket = rocket::build() + .manage(mock_db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket) + .await + .expect("valid rocket instance"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + let future_end = time_now() + 1_000_000; + + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks?end={future_end}" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::BadRequest); + let body = response.into_string().await.unwrap_or_default(); + let json: serde_json::Value = serde_json::from_str(&body).expect("Response should be JSON"); + assert_eq!(json["error"]["code"], "FUTURE_TIMESTAMP"); + assert!(json["error"]["message"] + .as_str() + .unwrap() + .contains("future")); +} + +#[rocket::async_test] +async fn test_start_after_end_timestamp() { + let mock_db = mock_db_pool(setup_mock_db).await; + let rocket = rocket::build() + .manage(mock_db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket) + .await + .expect("valid rocket instance"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + let start_after_end = "2000&end=1000"; + + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks?start={start_after_end}" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::BadRequest); + let body = response.into_string().await.unwrap_or_default(); + let json: serde_json::Value = serde_json::from_str(&body).expect("Response should be JSON"); + assert_eq!(json["error"]["code"], "INVALID_TIME_RANGE"); + assert!(json["error"]["message"] + .as_str() + .unwrap() + .contains("before end")); +} + +#[rocket::async_test] +async fn test_multiple_candlesticks_endpoint() { + let mock_db = mock_db_pool(setup_mock_db).await; + + let rocket = rocket::build() + .manage(mock_db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + + let client = Client::tracked(rocket) + .await + .expect("valid rocket instance"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + // We'll query from 1727963300..1727964500 with step=600 => 10 minutes + // This yields 2 candles: one in [3300..3900), another in [3900..4500). + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks?start=1727963300&end=1727964500&stepsize=600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + + let body = response.into_string().await.expect("No response body"); + let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); + + let cndl_array = json["candlesticks"].as_array().unwrap(); + assert_eq!(cndl_array.len(), 2, "Should produce exactly two candles"); + + // ----- Candle #1 ----- + 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["low"].as_f64().unwrap() - 40.0).abs() < f64::EPSILON); + assert!((cndl1["high"].as_f64().unwrap() - 60.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 ----- + 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 + 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); +} + +#[rocket::async_test] +async fn test_single_swap_multiple_pools() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + utxo_funding::create_table(&pool).await; + tx::create_table(&pool).await; + pool::create_table(&pool).await; + dummy_init_seq(); + + let mut conn = pool.acquire().await.unwrap(); + + let token_zero = TokenID::all_zeros(); + let pkh_zero = PubkeyHash::all_zeros(); + let txid_multi = Txid::from_byte_array([0xaa; 32]); + let block_zero = BlockHash::all_zeros(); + + let mut pools = Vec::new(); + let times = [TIME_1, TIME_2]; + for (i, &time) in times.iter().enumerate() { + let utxo = OutPointHash::from_byte_array([0xe1 + i as u8; 32]); + let pool_hash = OutPointHash::from_byte_array([0x0a + i as u8; 32]); + let cauldron = dummy_cauldron( + &txid_multi, + &utxo, + &token_zero, + 100_000 * (i + 1) as u64, + 2_000, + &pkh_zero, + ); + pools.push(pool_hash); + + 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(); + insert_mempool_tx(&mut conn, &txid_multi, time) + .await + .unwrap(); + + pool::insert_pool_history_entry( + &mut conn, + &pool_hash, + &cauldron, + Some(time), + Some(time), + (100_000 * (i + 1)) as i64, + 2_000i64, + ) + .await + .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), + ) + .await + .unwrap(); + } + }) + .await; + + let rocket = rocket::build() + .manage(mock_db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket) + .await + .expect("valid rocket instance"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + let response = client + .get(format!( + "/api/price/{}/candlesticks?start={}&end={}&stepsize=600", + token_id_zero, + TIME_1 - 100, + TIME_4 + 100 + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.expect("No response body"); + let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); + let candles = json["candlesticks"].as_array().unwrap(); + + let first_candle = &candles[0]; + assert_eq!( + first_candle["volume_sats"].as_i64().unwrap(), + 100_000 + 200_000 + ); + assert_eq!(first_candle["volume_tokens"].as_i64().unwrap(), 2000 + 2000); +} + +#[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 --- + utxo_funding::create_table(&pool).await; + tx::create_table(&pool).await; + pool::create_table(&pool).await; + dummy_init_seq(); + + let mut conn = pool.acquire().await.unwrap(); + + let token_zero = TokenID::all_zeros(); + 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]); + + // 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) -------- + 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), + }; + insert_utxo_funding(&mut conn, &vec![cauldron_p.clone()], &txid_price) + .await + .unwrap(); + insert_block_tx(&mut conn, &txid_price, &block_zero, t0 as i64) + .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, + &cauldron_p, + Some(t0), + Some(t0), + 10_000, + 200, + ) + .await + .unwrap(); + + // -------- Candle #2 (net zero tokens but non-zero volume) -------- + 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 + }; + + insert_utxo_funding(&mut conn, &vec![ca_a1.clone()], &txid_net0) + .await + .unwrap(); + 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(); + + pool::insert_pool_history_entry( + &mut conn, + &pool_a, + &ca_a1, + Some(t1), + Some(t1), + 10_000, + 200, + ) + .await + .unwrap(); + // Opposite deltas within the same tx + pool::insert_pool_history_entry( + &mut conn, + &pool_b, + &ca_b1, + Some(t1), + Some(t1), + -10_000, + -200, + ) + .await + .unwrap(); + }) + .await; + + let rocket = rocket::build() + .manage(mock_db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket) + .await + .expect("valid rocket instance"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + let start = 1_700_000_000u64; + let end = start + 2 * 600; + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks?start={start}&end={end}&stepsize=600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.expect("No response body"); + let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON"); + let candles = json["candlesticks"].as_array().unwrap(); + + 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!((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 (net-zero tokens, carry-forward close) --- + let c2 = &candles[1]; + assert_eq!(c2["time"].as_i64().unwrap(), (start + 600) as i64); + // Should carry previous close (50) because signed_tokens==0 but volume>0 + assert!((c2["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); + 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 + 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); +} + +// ── 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 +// TIME_4=1727964200 → hour bucket 1727964000 +// +// Expected OHLCV (bucket 1727960400): open=40, close=80, high=80, 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 +// vol_sats=200_000, vol_tokens=2_000, tx_count=1 + +/// Full range served from ohlcv_1h (materialized_end covers everything). +#[rocket::async_test] +async fn test_ohlcv_fast_path_full_range() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + setup_mock_db(pool.clone()).await; + ohlcv::create_table(&pool).await; + ohlcv::rebuild_range(&pool, &pool, 0, 1727970000) + .await + .expect("rebuild_range"); + }) + .await; + + let ohlcv_state = Arc::new(OhlcvState { + materialized_end: AtomicI64::new(1727970000), + }); + + let rocket = rocket::build() + .manage(mock_db) + .manage(ohlcv_state) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.expect("valid rocket"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + // Hour-aligned range covering both buckets. + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks\ + ?start=1727960400&end=1727967600&stepsize=3600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!(candles.len(), 2, "expected two 1h candles"); + + 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["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); + assert_eq!(c1["transaction_count"].as_i64().unwrap(), 3); + + 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_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); +} + +/// First bucket served from ohlcv_1h, second bucket served from raw CTE tail. +/// Both sources should produce identical output to the full-ohlcv test above. +#[rocket::async_test] +async fn test_ohlcv_fast_path_with_raw_tail() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + setup_mock_db(pool.clone()).await; + ohlcv::create_table(&pool).await; + // Only materialise up to 1727964000 — leaves the second bucket (T4) for raw. + ohlcv::rebuild_range(&pool, &pool, 0, 1727964000) + .await + .expect("rebuild_range"); + }) + .await; + + let ohlcv_state = Arc::new(OhlcvState { + // materialized_end = 1727964000: ohlcv has bucket 1727960400 only. + materialized_end: AtomicI64::new(1727964000), + }); + + let rocket = rocket::build() + .manage(mock_db) + .manage(ohlcv_state) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.expect("valid rocket"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks\ + ?start=1727960400&end=1727967600&stepsize=3600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!(candles.len(), 2, "expected two 1h candles"); + + // Candle 1 came from ohlcv_1h. + 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_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. + 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_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); +} + +/// Non-hour-aligned start must NOT use the ohlcv fast path (alignment guard). +#[rocket::async_test] +async fn test_ohlcv_skipped_for_non_aligned_start() { + let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { + setup_mock_db(pool.clone()).await; + ohlcv::create_table(&pool).await; + ohlcv::rebuild_range(&pool, &pool, 0, 1727970000) + .await + .expect("rebuild_range"); + }) + .await; + + let ohlcv_state = Arc::new(OhlcvState { + materialized_end: AtomicI64::new(1727970000), + }); + + let rocket = rocket::build() + .manage(mock_db) + .manage(ohlcv_state) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.expect("valid rocket"); + + let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000"; + + // 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) + let response = client + .get(format!( + "/api/price/{token_id_zero}/candlesticks\ + ?start=1727963300&end=1727970500&stepsize=3600" + )) + .dispatch() + .await; + + assert_eq!(response.status(), Status::Ok); + let body = response.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + // Key assertion: first candle starts at 1727963300, not 1727960400. + // If the ohlcv path were mistakenly used it would start at 1727960400. + 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_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); + assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 0); +} + +// ── Seeded gap-fill endpoint tests ──────────────────────────────────────── + +/// 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_byte_array([0xBA; 32]); + + let token_copy = token; + 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; + }) + .await; + + let rocket = rocket::build() + .manage(db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.unwrap(); + + let resp = client + .get(format!( + "/api/price/{}/candlesticks?start=2000&end=3000&stepsize=500", + token.to_string() + )) + .dispatch() + .await; + + assert_eq!(resp.status(), Status::Ok); + let body = resp.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!(candles.len(), 2, "both intervals should be gap-filled"); + for c in candles { + assert_eq!(c["transaction_count"].as_i64().unwrap(), 0); + assert!( + (c["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON, + "gap-fill should carry prior close (50), got {}", + c["close"] + ); + } +} + +/// 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() { + let token = TokenID::from_byte_array([0xBB; 32]); + + let token_copy = token; + 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; + }) + .await; + + let rocket = rocket::build() + .manage(db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.unwrap(); + + let resp = client + .get(format!( + "/api/price/{}/candlesticks?start=2000&end=3500&stepsize=500", + token.to_string() + )) + .dispatch() + .await; + + assert_eq!(resp.status(), Status::Ok); + let body = resp.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!(candles.len(), 3); + + assert_eq!(candles[0]["time"].as_i64().unwrap(), 2000); + assert_eq!(candles[0]["transaction_count"].as_i64().unwrap(), 0); + assert!( + (candles[0]["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON, + "seed gap-fill should be 50, got {}", + candles[0]["close"] + ); + + assert_eq!(candles[1]["time"].as_i64().unwrap(), 2500); + assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 1); + assert!( + (candles[1]["close"].as_f64().unwrap() - 75.0).abs() < f64::EPSILON, + "real candle close should be 75, got {}", + candles[1]["close"] + ); + + assert_eq!(candles[2]["time"].as_i64().unwrap(), 3000); + assert_eq!(candles[2]["transaction_count"].as_i64().unwrap(), 0); + assert!( + (candles[2]["close"].as_f64().unwrap() - 75.0).abs() < f64::EPSILON, + "post-trade gap-fill should follow in-window close (75), got {}", + candles[2]["close"] + ); +} + +/// 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_byte_array([0xBC; 32]); + + let token_copy = token; + 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; + }) + .await; + + let rocket = rocket::build() + .manage(db) + .manage(no_ohlcv()) + .mount("/api", routes![super::price_candlesticks]); + let client = Client::tracked(rocket).await.unwrap(); + + let resp = client + .get(format!( + "/api/price/{}/candlesticks?start=2000&end=3500&stepsize=500", + token.to_string() + )) + .dispatch() + .await; + + assert_eq!(resp.status(), Status::Ok); + let body = resp.into_string().await.unwrap(); + let json: serde_json::Value = serde_json::from_str(&body).unwrap(); + let candles = json["candlesticks"].as_array().unwrap(); + + assert_eq!( + candles.len(), + 2, + "only two candles expected (real + post-trade gap-fill)" + ); + assert_eq!( + candles[0]["time"].as_i64().unwrap(), + 2500, + "first candle must be the real trade" + ); + assert_eq!(candles[0]["transaction_count"].as_i64().unwrap(), 1); +}