diff --git a/src/db/cauldron/mod.rs b/src/db/cauldron/mod.rs index 6d21b0b..977c1d0 100644 --- a/src/db/cauldron/mod.rs +++ b/src/db/cauldron/mod.rs @@ -16,6 +16,7 @@ use crate::db::cauldron::tokenlist::db_utils::{ pub mod config; pub mod header; pub mod mempool; +pub mod ohlcv; pub mod pool; pub mod poolvisitor; pub mod tokenlist; diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs new file mode 100644 index 0000000..fa32178 --- /dev/null +++ b/src/db/cauldron/ohlcv.rs @@ -0,0 +1,449 @@ +// 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 anyhow::Result; +use sqlx::{Row, SqlitePool}; + +pub async fn create_table(pool: &SqlitePool) { + sqlx::query( + "CREATE TABLE IF NOT EXISTS ohlcv_1h ( + token_id BLOB NOT NULL, + bucket_ts INTEGER NOT NULL, + open REAL NOT NULL, + high REAL NOT NULL, + low REAL NOT NULL, + close REAL NOT NULL, + volume_sats INTEGER NOT NULL, + volume_tokens INTEGER NOT NULL, + tx_count INTEGER NOT NULL, + PRIMARY KEY (token_id, bucket_ts) + )", + ) + .execute(pool) + .await + .expect("failed to create ohlcv_1h table"); +} + +/// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty. +pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result> { + let row: Option<(Option,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h") + .fetch_optional(pool) + .await?; + Ok(row.and_then(|r| r.0)) +} + +/// Returns the earliest confirmed trade timestamp floored to the nearest 1-hour bucket, +/// or `None` if there are no confirmed trades. Used to seed the initial backfill start +/// so the background task doesn't scan from Unix epoch 0. +pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result> { + let row: Option<(Option,)> = sqlx::query_as( + "SELECT (MIN(phe.effective_timestamp) / 3600) * 3600 + FROM pool_history_entry AS phe + WHERE phe.mtp_timestamp IS NOT NULL", + ) + .fetch_optional(pool) + .await?; + Ok(row.and_then(|r| r.0)) +} + +/// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls +/// in `[since_ts, until_ts)`. +/// +/// Two-phase approach: the slow aggregation SELECT runs against `read_pool` (no write lock), +/// then the pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly). +/// Uses INSERT OR IGNORE so existing rows are never overwritten. +/// Returns the number of rows inserted. +pub async fn rebuild_range( + read_pool: &SqlitePool, + write_pool: &SqlitePool, + since_ts: i64, + until_ts: i64, +) -> Result { + if since_ts >= until_ts { + return Ok(0); + } + + // Phase 1: aggregate using the read pool — no write lock held during the slow CTE. + let select_sql = r#" +WITH per_pool_tx_raw AS ( + SELECT + phe.token_id, + phe.txid, + phe.effective_timestamp AS ts, + phe.utxo, + phe.sats_delta, + phe.token_delta, + phe.sequence + FROM pool_history_entry AS phe + JOIN tx ON tx.txid = phe.txid + WHERE tx.blockhash IS NOT NULL + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? +), +per_pool_tx AS ( + SELECT + token_id, + txid, + ts, + (ts / 3600) * 3600 AS bucket_ts, + utxo, + MIN(sequence) AS min_sequence, + SUM(sats_delta) AS signed_sats, + SUM(token_delta) AS signed_tokens, + SUM(ABS(sats_delta)) AS vol_sats, + SUM(ABS(token_delta)) AS vol_tokens + FROM per_pool_tx_raw + GROUP BY token_id, txid, ts, utxo +), +tx_trades AS ( + SELECT + token_id, + txid, + ts, + bucket_ts, + MIN(min_sequence) AS min_sequence, + SUM(signed_sats) AS signed_sats, + SUM(signed_tokens) AS signed_tokens, + SUM(vol_sats) AS vol_sats, + SUM(vol_tokens) AS vol_tokens + FROM per_pool_tx + GROUP BY token_id, txid, ts +), +priceable AS ( + SELECT + token_id, + bucket_ts, + ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, min_sequence ASC) AS rn_asc, + ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, min_sequence DESC) AS rn_desc + FROM tx_trades + WHERE signed_tokens != 0 +), +ohlc AS ( + SELECT + token_id, + bucket_ts, + MAX(CASE WHEN rn_asc = 1 THEN price END) AS open, + MAX(CASE WHEN rn_desc = 1 THEN price END) AS close, + MAX(price) AS high, + MIN(price) AS low + FROM priceable + GROUP BY token_id, bucket_ts +), +vol AS ( + SELECT + token_id, + bucket_ts, + SUM(vol_sats) AS volume_sats, + SUM(vol_tokens) AS volume_tokens, + COUNT(*) AS tx_count + FROM tx_trades + GROUP BY token_id, bucket_ts +) +SELECT + ohlc.token_id, + ohlc.bucket_ts, + ohlc.open, + ohlc.high, + ohlc.low, + ohlc.close, + vol.volume_sats, + vol.volume_tokens, + vol.tx_count +FROM ohlc +JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts +"#; + + let rows = sqlx::query(select_sql) + .bind(since_ts) + .bind(until_ts) + .fetch_all(read_pool) + .await?; + + if rows.is_empty() { + return Ok(0); + } + + // Phase 2: insert pre-computed rows inside a single transaction. + // The write lock is held only for these fast INSERTs, not during aggregation. + let mut tx = write_pool.begin().await?; + let mut inserted = 0u64; + for row in &rows { + let token_id: Vec = row.get(0); + let bucket_ts: i64 = row.get(1); + let open: f64 = row.get(2); + let high: f64 = row.get(3); + let low: f64 = row.get(4); + let close: f64 = row.get(5); + let volume_sats: i64 = row.get(6); + let volume_tokens: i64 = row.get(7); + let tx_count: i64 = row.get(8); + + inserted += sqlx::query( + "INSERT OR IGNORE INTO ohlcv_1h + (token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(token_id) + .bind(bucket_ts) + .bind(open) + .bind(high) + .bind(low) + .bind(close) + .bind(volume_sats) + .bind(volume_tokens) + .bind(tx_count) + .execute(&mut *tx) + .await? + .rows_affected(); + } + tx.commit().await?; + + Ok(inserted) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding}; + use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; + use std::sync::atomic::{AtomicU64, Ordering}; + + static OHLCV_TEST_COUNTER: AtomicU64 = AtomicU64::new(0); + + async fn test_pool() -> SqlitePool { + let id = OHLCV_TEST_COUNTER.fetch_add(1, Ordering::SeqCst); + let uri = format!("file:ohlcv_test_{}?mode=memory&cache=shared", id); + let opts = SqliteConnectOptions::new() + .filename(&uri) + .foreign_keys(false); + SqlitePoolOptions::new().connect_with(opts).await.unwrap() + } + + async fn setup_db(pool: &SqlitePool) { + tx::create_table(pool).await; + utxo_funding::create_table(pool).await; + cauldron_pool::create_table(pool).await; + create_table(pool).await; // ohlcv_1h + idx_phe_txid + } + + /// Insert a single confirmed trade via raw SQL (FK disabled in tests). + async fn insert_confirmed_trade( + pool: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + mtp_ts: i64, + sats_delta: i64, + token_delta: i64, + ) { + let blockhash = [0xAA_u8; 32]; + sqlx::query("INSERT INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)") + .bind(txid.as_slice()) + .bind(blockhash.as_slice()) + .bind(mtp_ts) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) + VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + ) + .bind(utxo.as_slice()) + .bind(txid.as_slice()) + .bind([0u8; 32].as_slice()) + .bind(txid.as_slice()) + .bind(token_id.as_slice()) + .execute(pool) + .await + .unwrap(); + let seq: i64 = + sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pool_history_entry + (utxo, pool, token_id, txid, tx_pos, mtp_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, ?, ?)", + ) + .bind(utxo.as_slice()) + .bind([0xBB_u8; 32].as_slice()) // dummy pool hash (FK disabled) + .bind(token_id.as_slice()) + .bind(txid.as_slice()) + .bind(mtp_ts) + .bind(seq) + .bind(sats_delta) + .bind(token_delta) + .execute(pool) + .await + .unwrap(); + } + + /// Insert a mempool-only trade (no blockhash on the tx row). + async fn insert_mempool_trade( + pool: &SqlitePool, + txid: [u8; 32], + utxo: [u8; 32], + token_id: [u8; 32], + first_seen_ts: i64, + ) { + sqlx::query("INSERT INTO tx (txid, first_seen_timestamp) VALUES (?, ?)") + .bind(txid.as_slice()) + .bind(first_seen_ts) + .execute(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO utxo_funding (new_utxo_hash, txid, spent_utxo_hash, new_utxo_txid, new_utxo_n, sats, token_amount, token_id) + VALUES (?, ?, ?, ?, 0, 1000, 1000, ?)", + ) + .bind(utxo.as_slice()) + .bind(txid.as_slice()) + .bind([0u8; 32].as_slice()) + .bind(txid.as_slice()) + .bind(token_id.as_slice()) + .execute(pool) + .await + .unwrap(); + let seq: i64 = + sqlx::query_scalar("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry") + .fetch_one(pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO pool_history_entry + (utxo, pool, token_id, txid, tx_pos, first_seen_timestamp, sequence, sats, token_amount, sats_delta, token_delta) + VALUES (?, ?, ?, ?, 0, ?, ?, 1000, 1000, -1000, 25)", + ) + .bind(utxo.as_slice()) + .bind([0xBB_u8; 32].as_slice()) + .bind(token_id.as_slice()) + .bind(txid.as_slice()) + .bind(first_seen_ts) + .bind(seq) + .execute(pool) + .await + .unwrap(); + } + + /// `get_min_trade_bucket_ts` should floor a mid-hour timestamp to the hour boundary. + #[tokio::test] + async fn test_get_min_trade_bucket_ts_floors_to_hour() { + let pool = test_pool().await; + setup_db(&pool).await; + + // Trade at 1727963400 — not hour-aligned; floor to 1727960400 + insert_confirmed_trade( + &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25, + ) + .await; + + let result = get_min_trade_bucket_ts(&pool).await.unwrap(); + assert_eq!( + result, + Some(1727960400), + "1727963400 should floor to 1727960400" + ); + } + + /// `get_min_trade_bucket_ts` returns `None` when only unconfirmed (mempool) trades exist. + #[tokio::test] + async fn test_get_min_trade_bucket_ts_no_confirmed_trades() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await; + + let result = get_min_trade_bucket_ts(&pool).await.unwrap(); + assert_eq!(result, None, "mempool-only trades must not be returned"); + } + + /// Running `rebuild_range` twice on the same range inserts nothing on the second call + /// because INSERT OR IGNORE skips rows that already exist. + #[tokio::test] + async fn test_rebuild_range_idempotent() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_confirmed_trade( + &pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25, + ) + .await; + + let n1 = rebuild_range(&pool, &pool, 1727960400, 1727964000) + .await + .unwrap(); + assert!(n1 > 0, "first rebuild should insert at least one bucket"); + + let n2 = rebuild_range(&pool, &pool, 1727960400, 1727964000) + .await + .unwrap(); + assert_eq!( + n2, 0, + "second rebuild must insert nothing (INSERT OR IGNORE)" + ); + } + + /// Unconfirmed trades (tx.blockhash IS NULL) must not appear in `ohlcv_1h`. + #[tokio::test] + async fn test_rebuild_range_excludes_mempool() { + let pool = test_pool().await; + setup_db(&pool).await; + + insert_mempool_trade(&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400).await; + + let n = rebuild_range(&pool, &pool, 1727960400, 1727967600) + .await + .unwrap(); + assert_eq!(n, 0, "mempool trades must not be materialised"); + } +} + +pub struct OhlcvRow { + pub bucket_ts: i64, + pub open: f64, + pub high: f64, + pub low: f64, + pub close: f64, + pub volume_sats: i64, + pub volume_tokens: i64, + pub tx_count: i64, +} + +/// Return the materialised 1-hour candles for a single token in `[start, end)`. +/// Only buckets that had at least one trade are returned (gaps must be filled by the caller). +pub async fn get_active_candles( + pool: &SqlitePool, + token_blob: &[u8], + start: i64, + end: i64, +) -> Result> { + let rows = sqlx::query( + "SELECT bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count + FROM ohlcv_1h + WHERE token_id = ? AND bucket_ts >= ? AND bucket_ts < ? + ORDER BY bucket_ts ASC", + ) + .bind(token_blob) + .bind(start) + .bind(end) + .fetch_all(pool) + .await?; + + Ok(rows + .into_iter() + .map(|r| OhlcvRow { + bucket_ts: r.get(0), + open: r.get(1), + high: r.get(2), + low: r.get(3), + close: r.get(4), + volume_sats: r.get(5), + volume_tokens: r.get(6), + tx_count: r.get(7), + }) + .collect()) +} diff --git a/src/db/cauldron/pool.rs b/src/db/cauldron/pool.rs index 5cfd741..9c7d807 100644 --- a/src/db/cauldron/pool.rs +++ b/src/db/cauldron/pool.rs @@ -83,6 +83,16 @@ pub async fn create_table(pool: &SqlitePool) { .execute(pool) .await .unwrap(); + sqlx::query("CREATE INDEX IF NOT EXISTS idx_phe_txid ON pool_history_entry(txid)") + .execute(pool) + .await + .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_phe_token_id_ts ON pool_history_entry(token_id, effective_timestamp)", + ) + .execute(pool) + .await + .unwrap(); } async fn get_pool_by_utxo( diff --git a/src/db/cauldron/utxo_funding.rs b/src/db/cauldron/utxo_funding.rs index cbc8452..3e3a43c 100644 --- a/src/db/cauldron/utxo_funding.rs +++ b/src/db/cauldron/utxo_funding.rs @@ -27,6 +27,12 @@ pub async fn create_table(pool: &SqlitePool) { .execute(pool) .await .unwrap(); + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_utxo_funding_token_txid ON utxo_funding(token_id, txid)", + ) + .execute(pool) + .await + .unwrap(); } pub async fn insert_utxo_funding( diff --git a/src/main.rs b/src/main.rs index 08f01ca..a6612fe 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,7 @@ use electrum_client_netagnostic::{Client, ElectrumApi, Param}; use log::{error, info, warn}; use rocket::{launch, routes}; use rocket_cors::{AllowedHeaders, AllowedOrigins}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering}; use std::{ backtrace::Backtrace, panic, process, @@ -23,6 +23,14 @@ use std::{ }; use stderrlog::LogLevelNum; +/// Tracks how far the OHLCV pre-aggregation table has been populated. +/// `materialized_end` is the exclusive upper bound: `ohlcv_1h` contains +/// data for all complete 1-hour buckets whose `bucket_ts + 3600 ≤ materialized_end`. +/// Value 0 means nothing has been materialised yet. +pub struct OhlcvState { + pub materialized_end: AtomicI64, +} + /// State tracking for Initial Block Download (IBD). /// Used to return 503 errors while the indexer is catching up. pub struct IbdState { @@ -111,6 +119,7 @@ async fn start_program( WellKnownDownloader, CRC20Fetcher, Arc, + Arc, // indexing_in_progress )> { let network = match config.network.to_lowercase().as_str() { "mainnet" => Network::Bitcoin, @@ -264,13 +273,10 @@ async fn start_program( }; }; indexing_in_progress_clone.store(false, Ordering::Relaxed); - - ibd_state_clone - .initial_sync_complete - .store(true, Ordering::Relaxed); info!("Initial block download complete"); - // Update query planner statistics so joins pick optimal order + // Run ANALYZE before signalling initial_sync_complete so the ohlcv post-IBD + // backfill (which waits for that flag) doesn't race with this write. info!("Running ANALYZE on cauldron database..."); if let Err(e) = sqlx::query("ANALYZE;").execute(&db.cauldron_w).await { warn!("ANALYZE failed: {e}"); @@ -278,6 +284,10 @@ async fn start_program( info!("ANALYZE complete"); } + ibd_state_clone + .initial_sync_complete + .store(true, Ordering::Relaxed); + // Follow chain loop { if signal::shutdown_requested() { @@ -343,6 +353,7 @@ async fn start_program( wellknowndownloader, crc20fetcher, ibd_state, + indexing_in_progress, )) } @@ -362,16 +373,22 @@ async fn launch() -> _ { config }; - let (dbpool, bcmrdownloader, wellknowndownloader, crc20fetcher, ibd_state) = - match start_program(config).await { - Ok(db) => db, - Err(e) => { - let backtrace = Backtrace::capture(); - error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}"); - error!("Error: {e}"); - panic!("Failed at program startup") - } - }; + let ( + dbpool, + bcmrdownloader, + wellknowndownloader, + crc20fetcher, + ibd_state, + indexing_in_progress, + ) = match start_program(config).await { + Ok(db) => db, + Err(e) => { + let backtrace = Backtrace::capture(); + error!("Backtrace (if RUST_BACKTRACE=1):\n{backtrace}"); + error!("Error: {e}"); + panic!("Failed at program startup") + } + }; let allowed_origins = AllowedOrigins::all(); let cors = rocket_cors::CorsOptions { @@ -391,11 +408,163 @@ async fn launch() -> _ { .await .expect("ensure cached_token_metrics exists"); + // Ensure the OHLCV pre-aggregation table exists (safe on both new and existing DBs). + db::cauldron::ohlcv::create_table(&dbpool.cauldron_w).await; + + // Bootstrap OhlcvState from whatever is already in the table (survives restarts). + let max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r) + .await + .unwrap_or(None); + let initial_ohlcv_end = max_bucket_ts.map(|ts| ts + 3600).unwrap_or(0); + let ohlcv_state = Arc::new(OhlcvState { + materialized_end: AtomicI64::new(initial_ohlcv_end), + }); + + // Synchronous post-IBD backfill: run the full ohlcv_1h materialisation before + // allowing metrics_cache and other background writers to start. We reuse the + // indexing_in_progress flag so metrics_cache backs off during this window. + { + const BACKFILL_BATCH_SECS: i64 = 24 * 3600; + const BACKFILL_SAFETY_SECS: i64 = 3 * 3600; + + // Wait for IBD to finish — ohlcv_1h data is only useful for confirmed blocks. + while !ibd_state.initial_sync_complete.load(Ordering::Relaxed) { + tokio::time::sleep(Duration::from_secs(1)).await; + } + + // Gate metrics_cache so it doesn't compete for cauldron_w during backfill. + indexing_in_progress.store(true, Ordering::Relaxed); + info!("ohlcv: starting post-IBD full backfill"); + + let now = crate::timeutil::time_now(); + let cutoff = (now - BACKFILL_SAFETY_SECS) / 3600 * 3600; + let since_opt = match max_bucket_ts { + Some(ts) => Some(ts + 3600), + None => match db::cauldron::ohlcv::get_min_trade_bucket_ts(&dbpool.cauldron_r).await { + Ok(v) => v, + Err(e) => { + warn!("ohlcv backfill: could not read min trade ts: {e}"); + None + } + }, + }; + if since_opt.is_none() { + info!("ohlcv backfill: no confirmed trades found, skipping"); + } + if let Some(mut batch_start) = since_opt { + while batch_start < cutoff { + let batch_end = (batch_start + BACKFILL_BATCH_SECS).min(cutoff); + match db::cauldron::ohlcv::rebuild_range( + &dbpool.cauldron_r, + &dbpool.cauldron_w, + batch_start, + batch_end, + ) + .await + { + Ok(n) => { + info!("ohlcv backfill: {n} buckets [{batch_start}, {batch_end})"); + ohlcv_state + .materialized_end + .store(batch_end, Ordering::Relaxed); + } + Err(e) => { + warn!("ohlcv backfill failed at [{batch_start}, {batch_end}): {e}"); + break; + } + } + batch_start = batch_end; + // Brief yield so new block writes are not starved. + tokio::time::sleep(Duration::from_millis(200)).await; + } + } + + info!("ohlcv: post-IBD backfill complete"); + indexing_in_progress.store(false, Ordering::Relaxed); + } + + // Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive. + // The full historical backfill above already ran; this task only handles the tail. + // Only processes buckets older than 3 hours (well beyond BCH reorg depth). + { + let ohlcv_write = dbpool.cauldron_w.clone(); + let ohlcv_read = dbpool.cauldron_r.clone(); + let ohlcv_state_bg = ohlcv_state.clone(); + tokio::spawn(async move { + // Batch size: 1 day per SQL call to keep each write short. + const BATCH_SECS: i64 = 24 * 3600; + // Safety margin: only materialise buckets older than this many seconds. + const SAFETY_SECS: i64 = 3 * 3600; + + loop { + let now = crate::timeutil::time_now(); + // Floor to 1-hour boundary, 3 hours ago. + let cutoff = (now - SAFETY_SECS) / 3600 * 3600; + + let since = match db::cauldron::ohlcv::get_max_bucket_ts(&ohlcv_read).await { + Ok(Some(max_ts)) => max_ts + 3600, + Ok(None) => { + // Table is empty: start from the first confirmed trade rather than + // scanning from Unix epoch 0 through thousands of empty batches. + match db::cauldron::ohlcv::get_min_trade_bucket_ts(&ohlcv_read).await { + Ok(Some(min_ts)) => min_ts, + Ok(None) => { + // No confirmed trades yet; wait before retrying. + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + Err(e) => { + error!("ohlcv rebuild (min trade ts): {e}"); + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + } + } + Err(e) => { + error!("ohlcv rebuild: {e}"); + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + }; + + let mut batch_start = since; + while batch_start < cutoff { + let batch_end = (batch_start + BATCH_SECS).min(cutoff); + match db::cauldron::ohlcv::rebuild_range( + &ohlcv_read, + &ohlcv_write, + batch_start, + batch_end, + ) + .await + { + Ok(n) => { + info!("ohlcv: materialised {n} buckets [{batch_start}, {batch_end})"); + ohlcv_state_bg + .materialized_end + .store(batch_end, Ordering::Relaxed); + } + Err(e) => { + error!("ohlcv rebuild failed: {e}"); + break; + } + } + batch_start = batch_end; + // Yield between batches so block indexing writes can proceed. + tokio::time::sleep(Duration::from_millis(200)).await; + } + + tokio::time::sleep(Duration::from_secs(600)).await; + } + }); + } + rocket::build() .attach(signal::ShutdownFairing) .attach(signal::IbdCheckFairing) .manage(dbpool) .manage(ibd_state) + .manage(ohlcv_state) // give rocket ownership of downloader to ensure thread isn't dropped .manage(bcmrdownloader) .manage(wellknowndownloader) diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs index 28987a3..11a25ea 100644 --- a/src/rpc/candlesticks.rs +++ b/src/rpc/candlesticks.rs @@ -4,10 +4,12 @@ // 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}; @@ -15,6 +17,8 @@ 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 { @@ -77,91 +81,15 @@ impl PriceInterval { } } -pub async fn candlesticks( - pool: &SqlitePool, - timestamp_start: i64, - timestamp_end: i64, +/// 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, - token_id: &str, -) -> Result> { - if timestamp_start > timestamp_end { - bail!("Start cannot be higher than end"); - } - - 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 sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - tx.txid, - COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS effective_timestamp, - phe.utxo, - phe.sats_delta, - phe.token_delta - FROM pool_history_entry AS phe - JOIN utxo_funding AS uf ON phe.utxo = uf.new_utxo_hash - JOIN tx ON tx.txid = phe.txid - WHERE uf.token_id = ? -), -per_pool_tx AS ( - SELECT - txid, - effective_timestamp, - utxo, - 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 - WHERE effective_timestamp >= ? AND effective_timestamp < ? - GROUP BY txid, effective_timestamp, utxo -), -tx_trades AS ( - SELECT - txid, - effective_timestamp, - 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; -"#; - // Use proper blob conversion for Bitcoin hash types (handles byte reversal) - let token_blob = display_hex_to_blob::(token_id)?; - let rows = sqlx::query(sql) - .bind(token_blob) - .bind(timestamp_start) - .bind(timestamp_end) - .fetch_all(pool) - .await?; - - let mut all_trades = Vec::new(); - for row in rows { - let timestamp: i64 = row.get(0); - let signed_sats: i64 = row.get(1); - let signed_tokens: i64 = row.get(2); - let vol_sats: i64 = row.get(3); - let vol_tokens: i64 = row.get(4); - all_trades.push((timestamp, signed_sats, signed_tokens, vol_sats, vol_tokens)); - } - - let mut found_first_trade = false; - let mut last_close_price: Option = None; + 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; @@ -183,9 +111,7 @@ ORDER BY effective_timestamp ASC; } if signed_tokens != 0 { - // Price from signed ratio; make it positive for OHLC let price = (signed_sats as f64 / signed_tokens as f64).abs(); - if first_trade_in_interval { pi.open = Some(price); pi.high = price; @@ -199,17 +125,13 @@ ORDER BY effective_timestamp ASC; } } - // Accumulate volume using ABS deltas (already computed in SQL) pi.volume_sats += vol_sats; pi.volume_tokens += vol_tokens; - pi.transaction_count += 1; trade_index += 1; } - // If there were trades in this interval (volume), but net signed_tokens was 0 - // so we never set a price, carry forward the previous close so the candle - // keeps its (correct) volume and doesn't get dropped. + // 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() { @@ -249,6 +171,191 @@ ORDER BY effective_timestamp ASC; } } + (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) } @@ -285,6 +392,7 @@ pub async fn price_candlesticks( end: Option, stepsize: Option, conn: &State, + ohlcv: &State>, ) -> CachedApiResult { let current_timestamp = time_now(); @@ -324,18 +432,24 @@ pub async fn price_candlesticks( )); } + // 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 - .iter() + .into_iter() .map(|c| { json!({ "time": c.time, @@ -350,8 +464,7 @@ pub async fn price_candlesticks( }) .collect(); - let cache_duration = if end.is_some() && effective_end < current_timestamp - effective_stepsize - { + let cache_duration = if is_historical { CACHE_IMMUTABLE } else { CACHE_NONE @@ -366,11 +479,13 @@ pub async fn price_candlesticks( #[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; @@ -379,6 +494,14 @@ mod tests { 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: @@ -455,27 +578,27 @@ mod tests { .await .unwrap(); - // Insert tx rows + // 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, &txid1, TIME_1).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, &txid2, TIME_2).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, &txid3, TIME_3).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_mempool_tx(&mut *conn, &txid4, TIME_4).await.unwrap(); // Insert pool_history_entry let pool1 = OutPointHash::from_inner([0x0a; 32]); @@ -562,6 +685,7 @@ mod tests { 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 @@ -593,6 +717,7 @@ mod tests { 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 @@ -624,6 +749,7 @@ mod tests { let rocket = rocket::build() .manage(mock_db) + .manage(no_ohlcv()) .mount("/api", routes![super::price_candlesticks]); let client = Client::tracked(rocket) @@ -760,6 +886,7 @@ mod tests { // 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 @@ -957,6 +1084,7 @@ mod tests { // 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 @@ -1004,4 +1132,189 @@ mod tests { 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); + } }