BitcoinCash/riftenlabs-indexer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Organise candlesticks

Browse source
This commit is contained in:
Jakob Notland 2026-06-05 14:12:56 +02:00
parent ac4564fb09
commit 5f844e5158
5 changed files with 622 additions and 575 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

415
src/db/cauldron/candlestick/mod.rs Normal file
View file

@ -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<f64>,
close: Option<f64>,
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<CandlestickData> {
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<PriceInterval>,
step_size: i64,
mut found_first_trade: bool,
mut last_close_price: Option<f64>,
) -> (Vec<CandlestickData>, bool, Option<f64>) {
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<ohlcv::OhlcvRow>,
start: i64,
end: i64,
mut found_first_trade: bool,
mut last_close: Option<f64>,
) -> (Vec<CandlestickData>, bool, Option<f64>) {
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<Vec<(i64, i64, i64, i64, i64)>> {
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<Option<f64>> {
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::<f64, _>(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<Vec<CandlestickData>> {
if timestamp_start > timestamp_end {
bail!("Start cannot be higher than end");
}
let token_blob = display_hex_to_blob::<TokenID>(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;

135
src/db/cauldron/candlestick/tests.rs Normal file
View file

@ -0,0 +1,135 @@
// Copyright (C) 2024-2026 Whiterun LLC
//
// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
use crate::db::blob::ToBlob;
use crate::db::cauldron::{
ohlcv,
pool::{self, dummy_init_seq},
tx::{self, insert_block_tx, insert_mempool_tx},
utxo_funding::{self, insert_utxo_funding},
};
use crate::utiltest::mock_db_pool;
use bitcoin_hashes::Hash;
use bitcoincash::{BlockHash, PubkeyHash, TokenID, Txid};
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
fn dummy_cauldron(
txid: &Txid,
utxo: &OutPointHash,
token: &TokenID,
sats: u64,
tokens: i64,
pkh: &PubkeyHash,
) -> ParsedContract {
ParsedContract {
pkh: *pkh,
is_withdrawn: false,
spent_utxo_hash: OutPointHash::all_zeros(),
new_utxo_hash: Some(*utxo),
new_utxo_txid: Some(*txid),
new_utxo_n: Some(0),
token_id: Some(*token),
sats: Some(sats),
token_amount: Some(tokens),
}
}
async fn setup_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq();
}
async fn insert_trade_at(
conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
token: &TokenID,
txid_byte: u8,
ts: u64,
sats_delta: i64,
token_delta: i64,
) {
let txid = Txid::from_inner([txid_byte; 32]);
let utxo = OutPointHash::from_inner([txid_byte; 32]);
let pool_hash = OutPointHash::from_inner([txid_byte.wrapping_add(0x80); 32]);
let block = BlockHash::all_zeros();
let cauldron = dummy_cauldron(
&txid,
&utxo,
token,
sats_delta.unsigned_abs(),
token_delta,
&PubkeyHash::all_zeros(),
);
insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid)
.await
.unwrap();
insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap();
insert_block_tx(&mut **conn, &txid, &block, ts as i64)
.await
.unwrap();
pool::insert_pool_history_entry(
&mut **conn,
&pool_hash,
&cauldron,
Some(ts),
Some(ts),
sats_delta,
token_delta,
)
.await
.unwrap();
}
#[tokio::test]
async fn test_fetch_last_close_before_no_trades() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_inner([0xAA; 32]);
let token_blob = token.to_blob();
let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2_000_000_000)
.await
.unwrap();
assert!(result.is_none());
}
#[tokio::test]
async fn test_fetch_last_close_before_only_future_trades() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_inner([0xAB; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_trade_at(&mut conn, &token, 0x01, 2000, 100_000, 2_000).await;
let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 1000)
.await
.unwrap();
assert!(result.is_none());
}
#[tokio::test]
async fn test_fetch_last_close_before_returns_most_recent() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_inner([0xAC; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
// ts=1000: price = 40_000/2_000 = 20
insert_trade_at(&mut conn, &token, 0x01, 1000, 40_000, 2_000).await;
// ts=2000: price = 100_000/2_000 = 50 ← most recent before 3000
insert_trade_at(&mut conn, &token, 0x02, 2000, 100_000, 2_000).await;
// ts=4000: after cutoff, must be excluded
insert_trade_at(&mut conn, &token, 0x03, 4000, 200_000, 2_000).await;
let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 3000)
.await
.unwrap();
assert!(result.is_some());
assert!((result.unwrap() - 50.0).abs() < f64::EPSILON);
}

1
src/db/cauldron/mod.rs
View file

@ -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;

415
src/rpc/candlesticks/mod.rs
View file

@ -3,431 +3,18 @@
// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
use crate::db::blob::display_hex_to_blob;
use crate::db::cauldron::ohlcv;
use crate::db::cauldron::candlestick::candlesticks;
use crate::db::DB;
use crate::rpc::err::{bad_request, ApiErrorCode, CachedApiResult};
use crate::rpc::response::{cached_ok, CACHE_IMMUTABLE, CACHE_NONE};
use crate::timeutil::time_now;
use crate::OhlcvState;
use anyhow::{bail, Result};
use bitcoincash::TokenID;
use rocket::{get, State};
use serde::Serialize;
use serde_json::json;
use serde_json::Value;
use sqlx::{Row, SqlitePool};
use std::sync::atomic::Ordering;
use std::sync::Arc;
#[derive(Debug, Serialize)]
pub struct CandlestickData {
pub time: i64, // start of the interval
pub open: f64,
pub close: f64,
pub high: f64,
pub low: f64,
pub volume_sats: i64,
pub volume_tokens: i64,
pub transaction_count: i64,
}
struct PriceInterval {
start: i64,
step: i64,
low: f64,
high: f64,
open: Option<f64>,
close: Option<f64>,
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<CandlestickData> {
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<PriceInterval>,
step_size: i64,
mut found_first_trade: bool,
mut last_close_price: Option<f64>,
) -> (Vec<CandlestickData>, bool, Option<f64>) {
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<ohlcv::OhlcvRow>,
start: i64,
end: i64,
mut found_first_trade: bool,
mut last_close: Option<f64>,
) -> (Vec<CandlestickData>, bool, Option<f64>) {
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<Vec<(i64, i64, i64, i64, i64)>> {
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<Option<f64>> {
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::<f64, _>(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<Vec<CandlestickData>> {
if timestamp_start > timestamp_end {
bail!("Start cannot be higher than end");
}
let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
// Seed gap-fill with the last known close price before this window so that
// switching between timeframes (e.g. 1W vs 1M) produces consistent prices
// for any overlapping period.
let seed_close = fetch_last_close_before(pool, &token_blob, timestamp_start).await?;
let seed_found = seed_close.is_some();
// Fast path: use pre-materialised ohlcv_1h when step_size is exactly 1 hour,
// ohlcv covers at least part of the range, AND the start is hour-aligned.
// ohlcv_1h buckets are always aligned to multiples of 3600, so a non-aligned
// start would produce candles whose boundaries disagree with the raw path.
if step_size == 3600 && ohlcv_materialized_end > timestamp_start && timestamp_start % 3600 == 0
{
let ohlcv_end = ohlcv_materialized_end.min(timestamp_end);
// timestamp_start is guaranteed hour-aligned by the entry condition above.
let ohlcv_rows =
ohlcv::get_active_candles(pool, &token_blob, timestamp_start, ohlcv_end).await?;
let (mut result, found_first, last_close) = fill_ohlcv_candles(
ohlcv_rows,
timestamp_start,
ohlcv_end,
seed_found,
seed_close,
);
if ohlcv_end < timestamp_end {
// Tail: query raw for [ohlcv_end, timestamp_end) and append.
let raw_trades = fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?;
let mut tail_intervals = Vec::new();
let mut t = ohlcv_end;
while t < timestamp_end {
tail_intervals.push(PriceInterval::new(t, step_size));
t += step_size;
}
let (tail, _, _) = aggregate_raw_trades(
&raw_trades,
tail_intervals,
step_size,
found_first,
last_close,
);
result.extend(tail);
}
return Ok(result);
}
// Raw path: full CTE scan (all non-3600 step sizes, or when ohlcv is not ready).
let mut intervals = Vec::new();
let mut current_start = timestamp_start;
while current_start < timestamp_end {
intervals.push(PriceInterval::new(current_start, step_size));
current_start += step_size;
}
let all_trades = fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?;
let (result, _, _) =
aggregate_raw_trades(&all_trades, intervals, step_size, seed_found, seed_close);
Ok(result)
}
/// Fetch candlesticks in BCH satoshis for a token.
///
/// If an interval has no trades, it will be omitted from the result.

231
src/rpc/candlesticks/tests.rs
View file

@ -1,4 +1,8 @@
use crate::db::blob::ToBlob;
// Copyright (C) 2024-2026 Whiterun LLC
//
// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
use crate::db::cauldron::{
ohlcv,
pool::{self, dummy_init_seq, insert_new_pool},
@ -38,7 +42,6 @@ const TIME_2: u64 = 1727963600;
const TIME_3: u64 = 1727963900;
const TIME_4: u64 = 1727964200;
/// Helper to build a ParsedContract
fn dummy_cauldron(
txid: &Txid,
utxo: &OutPointHash,
@ -61,7 +64,6 @@ fn dummy_cauldron(
}
async fn setup_mock_db(pool: sqlx::SqlitePool) {
// Create tables
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
@ -72,7 +74,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
let token_zero = TokenID::all_zeros();
let pkh_zero = PubkeyHash::all_zeros();
// We'll make 4 trades with distinct times
let txid1 = Txid::from_inner([0xf1; 32]);
let txid2 = Txid::from_inner([0xf2; 32]);
let txid3 = Txid::from_inner([0xf3; 32]);
@ -100,29 +101,24 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
.await
.unwrap();
// Insert tx rows in realistic order: mempool first, then confirmed.
let block_zero = BlockHash::all_zeros();
insert_mempool_tx(&mut *conn, &txid1, TIME_1).await.unwrap();
insert_block_tx(&mut *conn, &txid1, &block_zero, TIME_1 as i64)
.await
.unwrap();
insert_mempool_tx(&mut *conn, &txid2, TIME_2).await.unwrap();
insert_block_tx(&mut *conn, &txid2, &block_zero, TIME_2 as i64)
.await
.unwrap();
insert_mempool_tx(&mut *conn, &txid3, TIME_3).await.unwrap();
insert_block_tx(&mut *conn, &txid3, &block_zero, TIME_3 as i64)
.await
.unwrap();
insert_mempool_tx(&mut *conn, &txid4, TIME_4).await.unwrap();
insert_block_tx(&mut *conn, &txid4, &block_zero, TIME_4 as i64)
.await
.unwrap();
// Insert pool_history_entry
let pool1 = OutPointHash::from_inner([0x0a; 32]);
let pool2 = OutPointHash::from_inner([0x0b; 32]);
let pool3 = OutPointHash::from_inner([0x0c; 32]);
@ -173,7 +169,6 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
.await
.unwrap();
// Insert pools
let token1 = TokenID::from_inner([0xda; 32]);
let pkh1 = PubkeyHash::from_inner([0xca; 20]);
insert_new_pool(
@ -202,6 +197,60 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
.unwrap();
}
// ── Seeded gap-fill helpers ───────────────────────────────────────────────
async fn setup_seed_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq();
}
async fn insert_trade_at(
conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
token: &TokenID,
txid_byte: u8,
ts: u64,
sats_delta: i64,
token_delta: i64,
) {
let txid = Txid::from_inner([txid_byte; 32]);
let utxo = OutPointHash::from_inner([txid_byte; 32]);
let pool_hash = OutPointHash::from_inner([txid_byte.wrapping_add(0x80); 32]);
let block = BlockHash::all_zeros();
let cauldron = dummy_cauldron(
&txid,
&utxo,
token,
sats_delta.unsigned_abs(),
token_delta,
&PubkeyHash::all_zeros(),
);
insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid)
.await
.unwrap();
insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap();
insert_block_tx(&mut **conn, &txid, &block, ts as i64)
.await
.unwrap();
pool::insert_pool_history_entry(
&mut **conn,
&pool_hash,
&cauldron,
Some(ts),
Some(ts),
sats_delta,
token_delta,
)
.await
.unwrap();
}
// ── Endpoint tests ────────────────────────────────────────────────────────
#[rocket::async_test]
async fn test_future_end_timestamp() {
let mock_db = mock_db_pool(setup_mock_db).await;
@ -213,7 +262,6 @@ async fn test_future_end_timestamp() {
.await
.expect("valid rocket instance");
// Use a far-future timestamp for 'end'
let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000";
let future_end = time_now() + 1_000_000;
@ -299,7 +347,6 @@ async fn test_multiple_candlesticks_endpoint() {
// ----- Candle #1 -----
let cndl1 = &cndl_array[0];
println!("First candlestick: {cndl1:?}");
// Candle #1 => time=1727963300
// trades at 1727963300 => ratio=40, 1727963600 => ratio=60
// open=40, close=60, low=40, high=60, volume_sats=200k, volume_tokens=4k, transaction_count=2
@ -314,7 +361,6 @@ async fn test_multiple_candlesticks_endpoint() {
// ----- Candle #2 -----
let cndl2 = &cndl_array[1];
println!("Second candlestick: {cndl2:?}");
// Candle #2 => time=1727963900
// trades at 1727963900 => ratio=80, 1727964200 => ratio=100
// open=80, close=100, low=80, high=100, volume_sats=360k, volume_tokens=4k, transaction_count=2
@ -323,7 +369,6 @@ async fn test_multiple_candlesticks_endpoint() {
assert!((cndl2["close"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON);
assert!((cndl2["low"].as_f64().unwrap() - 80.0).abs() < f64::EPSILON);
assert!((cndl2["high"].as_f64().unwrap() - 100.0).abs() < f64::EPSILON);
// volume_sats=160k+200k=360k, volume_tokens=2k+2k=4k, transaction_count=2
assert_eq!(cndl2["volume_sats"].as_i64().unwrap(), 160_000 + 200_000);
assert_eq!(cndl2["volume_tokens"].as_i64().unwrap(), 4_000);
@ -332,9 +377,7 @@ async fn test_multiple_candlesticks_endpoint() {
#[rocket::async_test]
async fn test_single_swap_multiple_pools() {
// Set up a fresh mock DB
let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move {
// Create required tables
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
@ -344,12 +387,9 @@ async fn test_single_swap_multiple_pools() {
let token_zero = TokenID::all_zeros();
let pkh_zero = PubkeyHash::all_zeros();
// Create a single transaction that will be used for multiple pool trades
let txid_multi = Txid::from_inner([0xaa; 32]);
let block_zero = BlockHash::all_zeros();
// Create multiple pool and pool_history_entry records for the same txid
let mut pools = Vec::new();
let times = [TIME_1, TIME_2];
for (i, &time) in times.iter().enumerate() {
@ -368,7 +408,6 @@ async fn test_single_swap_multiple_pools() {
insert_utxo_funding(&mut *conn, &vec![cauldron.clone()], &txid_multi)
.await
.unwrap();
insert_block_tx(&mut *conn, &txid_multi, &block_zero, time as i64)
.await
.unwrap();
@ -376,23 +415,19 @@ async fn test_single_swap_multiple_pools() {
.await
.unwrap();
let sats_delta = (100_000 * (i + 1)) as i64;
let token_delta = 2_000i64;
pool::insert_pool_history_entry(
&mut *conn,
&pool_hash,
&cauldron,
Some(time),
Some(time),
sats_delta,
token_delta,
(100_000 * (i + 1)) as i64,
2_000i64,
)
.await
.unwrap();
}
// Insert pools into the pool table
let token1 = TokenID::from_inner([0xda; 32]);
let pkh1 = PubkeyHash::from_inner([0xca; 20]);
for pool_hash in &pools {
@ -406,7 +441,6 @@ async fn test_single_swap_multiple_pools() {
})
.await;
// Build Rocket instance with our endpoint
let rocket = rocket::build()
.manage(mock_db)
.manage(no_ohlcv())
@ -417,7 +451,6 @@ async fn test_single_swap_multiple_pools() {
let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000";
// Query a time range that includes our test transactions
let response = client
.get(format!(
"/api/price/{}/candlesticks?start={}&end={}&stepsize=600",
@ -434,17 +467,11 @@ async fn test_single_swap_multiple_pools() {
let candles = json["candlesticks"].as_array().unwrap();
let first_candle = &candles[0];
let expected_volume_sats = 100_000 + 200_000;
let expected_volume_tokens = 2000 + 2000;
assert_eq!(
first_candle["volume_sats"].as_i64().unwrap(),
expected_volume_sats
);
assert_eq!(
first_candle["volume_tokens"].as_i64().unwrap(),
expected_volume_tokens
100_000 + 200_000
);
assert_eq!(first_candle["volume_tokens"].as_i64().unwrap(), 2000 + 2000);
}
#[rocket::async_test]
@ -572,13 +599,11 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
insert_utxo_funding(&mut *conn, &vec![ca_b1.clone()], &txid_net0)
.await
.unwrap();
insert_block_tx(&mut *conn, &txid_net0, &block_zero, t1 as i64)
.await
.unwrap();
insert_mempool_tx(&mut *conn, &txid_net0, t1).await.unwrap();
// Opposite deltas within the same tx
pool::insert_pool_history_entry(
&mut *conn,
&pool_a,
@ -590,6 +615,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
)
.await
.unwrap();
// Opposite deltas within the same tx
pool::insert_pool_history_entry(
&mut *conn,
&pool_b,
@ -604,7 +630,6 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
})
.await;
// Build Rocket and call endpoint across the two intervals
let rocket = rocket::build()
.manage(mock_db)
.manage(no_ohlcv())
@ -656,7 +681,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
assert_eq!(c2["transaction_count"].as_i64().unwrap(), 1);
}
// ── ohlcv_1h fast-path tests ─────────────────────────────────────────────
// ── ohlcv_1h fast-path tests ──────────────────────────────────────────────
//
// Test data from setup_mock_db — four trades at:
// TIME_1=1727963400, TIME_2=1727963600, TIME_3=1727963900 → hour bucket 1727960400
@ -841,115 +866,10 @@ async fn test_ohlcv_skipped_for_non_aligned_start() {
assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 0);
}
// ── fetch_last_close_before + gap-fill seeding tests ───────────────────
// ── Seeded gap-fill endpoint tests ────────────────────────────────────────
/// Insert a single confirmed trade for a token.
/// `txid_byte` is used to derive unique txid/utxo hashes.
async fn insert_trade_at(
conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
token: &TokenID,
txid_byte: u8,
ts: u64,
sats_delta: i64,
token_delta: i64,
) {
let txid = Txid::from_inner([txid_byte; 32]);
let utxo = OutPointHash::from_inner([txid_byte; 32]);
let pool_hash = OutPointHash::from_inner([txid_byte.wrapping_add(0x80); 32]);
let block = BlockHash::all_zeros();
let cauldron = dummy_cauldron(
&txid,
&utxo,
token,
sats_delta.unsigned_abs(),
token_delta,
&PubkeyHash::all_zeros(),
);
insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid)
.await
.unwrap();
insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap();
insert_block_tx(&mut **conn, &txid, &block, ts as i64)
.await
.unwrap();
pool::insert_pool_history_entry(
&mut **conn,
&pool_hash,
&cauldron,
Some(ts),
Some(ts),
sats_delta,
token_delta,
)
.await
.unwrap();
}
async fn setup_seed_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq();
}
/// fetch_last_close_before returns None when there are no trades at all.
#[tokio::test]
async fn test_fetch_last_close_before_no_trades() {
let db = mock_db_pool(setup_seed_db).await;
let token = TokenID::from_inner([0xAA; 32]);
let token_blob = token.to_blob();
let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2_000_000_000)
.await
.unwrap();
assert!(result.is_none());
}
/// fetch_last_close_before returns None when all trades are after timestamp_end.
#[tokio::test]
async fn test_fetch_last_close_before_only_future_trades() {
let db = mock_db_pool(setup_seed_db).await;
let token = TokenID::from_inner([0xAB; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
// Trade at ts=2000, query for ts < 1000 → None
insert_trade_at(&mut conn, &token, 0x01, 2000, 100_000, 2_000).await;
let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 1000)
.await
.unwrap();
assert!(result.is_none());
}
/// fetch_last_close_before returns the most recent trade before the cutoff,
/// not earlier ones, and computes the correct price.
#[tokio::test]
async fn test_fetch_last_close_before_returns_most_recent() {
let db = mock_db_pool(setup_seed_db).await;
let token = TokenID::from_inner([0xAC; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
// ts=1000: price = 40_000/2_000 = 20
insert_trade_at(&mut conn, &token, 0x01, 1000, 40_000, 2_000).await;
// ts=2000: price = 100_000/2_000 = 50 ← most recent before 3000
insert_trade_at(&mut conn, &token, 0x02, 2000, 100_000, 2_000).await;
// ts=4000: after cutoff, must be excluded
insert_trade_at(&mut conn, &token, 0x03, 4000, 200_000, 2_000).await;
let result = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 3000)
.await
.unwrap();
assert!(result.is_some());
assert!((result.unwrap() - 50.0).abs() < f64::EPSILON);
}
/// Raw path: when the window has no trades but there is a prior trade,
/// all candles in the window are flat gap-fill at the prior close price.
/// When the window has no trades but there is a prior trade,
/// all candles are flat gap-fill at the prior close price.
#[rocket::async_test]
async fn test_raw_path_seeded_gap_fill_no_in_window_trades() {
let token = TokenID::from_inner([0xBA; 32]);
@ -958,7 +878,6 @@ async fn test_raw_path_seeded_gap_fill_no_in_window_trades() {
let db = mock_db_pool(move |pool: sqlx::SqlitePool| async move {
setup_seed_db(pool.clone()).await;
let mut conn = pool.acquire().await.unwrap();
// price = 100_000 / 2_000 = 50 — BEFORE the query window
insert_trade_at(&mut conn, &token_copy, 0x10, 1_000, 100_000, 2_000).await;
})
.await;
@ -969,7 +888,6 @@ async fn test_raw_path_seeded_gap_fill_no_in_window_trades() {
.mount("/api", routes![super::price_candlesticks]);
let client = Client::tracked(rocket).await.unwrap();
// Window: [2000, 3000), step=500 — no in-window trades
let resp = client
.get(format!(
"/api/price/{}/candlesticks?start=2000&end=3000&stepsize=500",
@ -994,7 +912,7 @@ async fn test_raw_path_seeded_gap_fill_no_in_window_trades() {
}
}
/// Raw path: pre-window seed fills the gap before the first in-window trade,
/// Pre-window seed fills the gap before the first in-window trade,
/// then real candle and post-trade gap-fill use the in-window close.
#[rocket::async_test]
async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
@ -1004,9 +922,7 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
let db = mock_db_pool(move |pool: sqlx::SqlitePool| async move {
setup_seed_db(pool.clone()).await;
let mut conn = pool.acquire().await.unwrap();
// Pre-window trade: price = 50
insert_trade_at(&mut conn, &token_copy, 0x20, 1_000, 100_000, 2_000).await;
// In-window trade at ts=2500: price = 150_000 / 2_000 = 75
insert_trade_at(&mut conn, &token_copy, 0x21, 2_500, 150_000, 2_000).await;
})
.await;
@ -1017,7 +933,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
.mount("/api", routes![super::price_candlesticks]);
let client = Client::tracked(rocket).await.unwrap();
// Window [2000, 3500), step=500 → intervals: [2000,2500), [2500,3000), [3000,3500)
let resp = client
.get(format!(
"/api/price/{}/candlesticks?start=2000&end=3500&stepsize=500",
@ -1033,7 +948,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
assert_eq!(candles.len(), 3);
// First interval: gap-fill seeded from pre-window close (50)
assert_eq!(candles[0]["time"].as_i64().unwrap(), 2000);
assert_eq!(candles[0]["transaction_count"].as_i64().unwrap(), 0);
assert!(
@ -1042,7 +956,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
candles[0]["close"]
);
// Second interval: real in-window trade (75)
assert_eq!(candles[1]["time"].as_i64().unwrap(), 2500);
assert_eq!(candles[1]["transaction_count"].as_i64().unwrap(), 1);
assert!(
@ -1051,7 +964,6 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
candles[1]["close"]
);
// Third interval: gap-fill from in-window close (75), not from seed (50)
assert_eq!(candles[2]["time"].as_i64().unwrap(), 3000);
assert_eq!(candles[2]["transaction_count"].as_i64().unwrap(), 0);
assert!(
@ -1061,8 +973,7 @@ async fn test_raw_path_seeded_gap_fill_then_in_window_trade() {
);
}
/// Without a pre-window trade, no candles appear before the first in-window trade.
/// This confirms the seed is only active when a prior trade actually exists.
/// Without a pre-window trade no candles appear before the first in-window trade.
#[rocket::async_test]
async fn test_raw_path_no_seed_no_prefill() {
let token = TokenID::from_inner([0xBC; 32]);
@ -1071,7 +982,6 @@ async fn test_raw_path_no_seed_no_prefill() {
let db = mock_db_pool(move |pool: sqlx::SqlitePool| async move {
setup_seed_db(pool.clone()).await;
let mut conn = pool.acquire().await.unwrap();
// Only an in-window trade at ts=2500: price=75 — no pre-window trade
insert_trade_at(&mut conn, &token_copy, 0x30, 2_500, 150_000, 2_000).await;
})
.await;
@ -1095,7 +1005,6 @@ async fn test_raw_path_no_seed_no_prefill() {
let json: serde_json::Value = serde_json::from_str(&body).unwrap();
let candles = json["candlesticks"].as_array().unwrap();
// [2000,2500) must be absent — no seed, no prior close to gap-fill from
assert_eq!(
candles.len(),
2,