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

Merge branch 'candlesticks2' into 'master'

Browse source 
Candlesticks2

See merge request riftenlabs/riftenlabs-indexer!20
This commit is contained in:
Dagur Valberg Johannsson 2025-06-17 13:22:46 +00:00
commit 3d83bdee47
3 changed files with 660 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/main.rs
View file

@ -349,6 +349,7 @@ fn launch() -> _ {
rpc::tokens::list_by_volume, rpc::tokens::list_by_volume,
rpc::tokens::search_by_volume, rpc::tokens::search_by_volume,
rpc::price::price_history, rpc::price::price_history,
rpc::candlesticks::price_candlesticks,
rpc::price::price_current, rpc::price::price_current,
rpc::price::price_at, rpc::price::price_at,
rpc::pool::list_pools_by_apy, rpc::pool::list_pools_by_apy,

658
src/rpc/candlesticks.rs Normal file
View file

@ -0,0 +1,658 @@
// Copyright (C) 2024 Riften Labs AS
//
// 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::DB;
use crate::timeutil::time_now;
use anyhow::{bail, Result};
use rocket::{get, http::Status, response::status::Custom, serde::json::Json, State};
use rusqlite::{params, Connection};
use serde::Serialize;
use serde_json::json;
use serde_json::Value;
#[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,
sats: i64,
tokens: 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,
sats: 0,
tokens: 0,
low: f64::MAX,
high: f64::MIN,
open: None,
close: None,
volume_sats: 0,
volume_tokens: 0,
transaction_count: 0,
}
}
pub fn add_transaction(&mut self, sats: i64, tokens: i64, is_first: bool, is_last: bool) {
if tokens == 0 {
return;
}
let price = sats as f64 / tokens as f64;
// Only set open if it's currently None
if is_first && self.open.is_none() {
self.open = Some(price);
self.high = price;
self.low = price;
}
if is_last {
self.close = Some(price);
}
if price.is_finite() {
self.high = self.high.max(price);
self.low = self.low.min(price);
}
self.sats += sats;
self.tokens += tokens;
self.volume_sats += sats;
self.volume_tokens += tokens;
self.transaction_count += 1;
}
pub fn to_candlestick_data(&self) -> Option<CandlestickData> {
if self.tokens == 0 {
None
} else {
Some(CandlestickData {
time: self.start,
open: self.open.unwrap(),
close: self.close.unwrap(),
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
}
}
pub fn candlesticks(
connection: &Connection,
timestamp_start: i64,
timestamp_end: i64,
step_size: i64,
token_id: &str,
) -> Result<Vec<CandlestickData>> {
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 tx_trades AS (
SELECT
tx.txid,
COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) AS effective_timestamp,
SUM(phe.sats) AS sats,
SUM(phe.token_amount) AS token_amount
FROM pool_history_entry phe
JOIN utxo_funding uf ON phe.utxo = uf.new_utxo_hash
JOIN tx ON tx.txid = phe.txid
WHERE
uf.token_id = ?
AND COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) >= ?
AND COALESCE(tx.first_seen_timestamp, tx.mtp_timestamp) < ?
GROUP BY tx.txid
)
SELECT
effective_timestamp,
sats,
token_amount
FROM tx_trades
ORDER BY effective_timestamp ASC;
"#;
let mut statement = connection.prepare(sql)?;
let mut rows = statement.query(params![token_id, timestamp_start, timestamp_end])?;
let mut all_trades = Vec::new();
while let Some(row) = rows.next()? {
let timestamp: i64 = row.get(0)?;
let sats: i64 = row.get(1)?;
let tokens: i64 = row.get(2)?;
all_trades.push((timestamp, sats, tokens));
}
let mut found_first_trade = false;
let mut last_close_price: Option<f64> = None;
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, sats, tokens) = all_trades[trade_index];
if ts < interval_start {
trade_index += 1;
continue;
} else if ts >= interval_end {
break;
} else {
if tokens != 0 {
let price = sats as f64 / tokens as f64;
if first_trade_in_interval {
pi.open = Some(price);
first_trade_in_interval = false;
}
pi.close = Some(price);
}
pi.add_transaction(sats, tokens, false, false);
trade_index += 1;
}
}
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,
});
}
}
}
Ok(result)
}
#[get("/price/<token>/candlesticks?<start>&<end>&<stepsize>")]
pub fn price_candlesticks(
token: &str,
start: Option<i64>,
end: Option<i64>,
stepsize: Option<i64>,
conn: &State<DB>,
) -> Result<Json<Value>, Custom<String>> {
let current_timestamp = time_now();
// Validate that the provided end timestamp is not in the future.
if let Some(end_ts) = end {
if end_ts > current_timestamp {
return Err(Custom(
Status::BadRequest,
"End timestamp cannot be in the future".to_string(),
));
}
}
// Determine effective start and end values.
let effective_end = end.unwrap_or(current_timestamp);
let effective_start = start.unwrap_or(current_timestamp - 30 * 24 * 3600); // default 30 days ago
let effective_stepsize = stepsize.unwrap_or(3600);
// Validate that start is before end.
if effective_start >= effective_end {
return Err(Custom(
Status::BadRequest,
"Start timestamp must be before end timestamp".to_string(),
));
}
// Check for too many intervals
const MAX_INTERVALS: i64 = 10000;
let total_intervals = (effective_end - effective_start) / effective_stepsize;
if total_intervals > MAX_INTERVALS {
return Err(Custom(
Status::BadRequest,
format!(
"Too many intervals ({} > {})",
total_intervals, MAX_INTERVALS
),
));
}
let db = conn
.cauldron_r
.get()
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;
let candlesticks = candlesticks(
&db,
effective_start,
effective_end,
stepsize.unwrap_or(3600), // default 1 hour
token,
)
.map_err(|e| Custom(Status::BadRequest, format!("Error: {}", e)))?;
let candlesticks_json: Vec<Value> = candlesticks
.iter()
.map(|candlestick| {
json!({
"time": candlestick.time,
"high": candlestick.high,
"low": candlestick.low,
"open": candlestick.open,
"close": candlestick.close,
"volume_sats": candlestick.volume_sats,
"volume_tokens": candlestick.volume_tokens,
"transaction_count": candlestick.transaction_count
})
})
.collect();
Ok(Json(json!({ "candlesticks": candlesticks_json })))
}
#[cfg(test)]
mod tests {
use crate::db::cauldron::{
pool::{self, dummy_init_seq, insert_new_pool},
tx::{self, insert_block_tx, insert_mempool_tx},
utxo_funding::{self, insert_utxo_funding},
};
use crate::utiltest::mock_db_pool;
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::blocking::Client;
use rocket::routes;
use rusqlite::Connection;
/// 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.clone(),
is_withdrawn: false,
spent_utxo_hash: OutPointHash::all_zeros(),
new_utxo_hash: Some(utxo.clone()),
new_utxo_txid: Some(txid.clone()),
new_utxo_n: Some(0),
token_id: Some(token.clone()),
sats: Some(sats),
token_amount: Some(tokens),
}
}
fn setup_mock_db(conn: &Connection) {
// Create tables
utxo_funding::create_table(conn);
tx::create_table(conn);
pool::create_table(conn);
dummy_init_seq();
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]);
let txid4 = Txid::from_inner([0xf4; 32]);
let utxo1 = OutPointHash::from_inner([0xe1; 32]);
let utxo2 = OutPointHash::from_inner([0xe2; 32]);
let utxo3 = OutPointHash::from_inner([0xe3; 32]);
let utxo4 = OutPointHash::from_inner([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);
// Notice we wrap each single-item array with a Vec!
insert_utxo_funding(conn, &vec![cauldron1.clone()], &txid1, true).unwrap();
insert_utxo_funding(conn, &vec![cauldron2.clone()], &txid2, true).unwrap();
insert_utxo_funding(conn, &vec![cauldron3.clone()], &txid3, true).unwrap();
insert_utxo_funding(conn, &vec![cauldron4.clone()], &txid4, true).unwrap();
// Insert tx rows
let block_zero = BlockHash::all_zeros();
insert_block_tx(conn, &txid1, &block_zero, TIME_1 as i64).unwrap();
insert_mempool_tx(conn, &txid1, TIME_1).unwrap();
insert_block_tx(conn, &txid2, &block_zero, TIME_2 as i64).unwrap();
insert_mempool_tx(conn, &txid2, TIME_2).unwrap();
insert_block_tx(conn, &txid3, &block_zero, TIME_3 as i64).unwrap();
insert_mempool_tx(conn, &txid3, TIME_3).unwrap();
insert_block_tx(conn, &txid4, &block_zero, TIME_4 as i64).unwrap();
insert_mempool_tx(conn, &txid4, TIME_4).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]);
let pool4 = OutPointHash::from_inner([0x0d; 32]);
pool::insert_pool_history_entry(conn, &pool1, &cauldron1, Some(TIME_1), Some(TIME_1))
.unwrap();
pool::insert_pool_history_entry(conn, &pool2, &cauldron2, Some(TIME_2), Some(TIME_2))
.unwrap();
pool::insert_pool_history_entry(conn, &pool3, &cauldron3, Some(TIME_3), Some(TIME_3))
.unwrap();
pool::insert_pool_history_entry(conn, &pool4, &cauldron4, Some(TIME_4), Some(TIME_4))
.unwrap();
// Insert pools
let token1 = TokenID::from_inner([0xda; 32]);
let pkh1 = PubkeyHash::from_inner([0xca; 20]);
insert_new_pool(
conn,
&dummy_cauldron(&Txid::all_zeros(), &pool1, &token1, 0, 0, &pkh1),
)
.unwrap();
insert_new_pool(
conn,
&dummy_cauldron(&Txid::all_zeros(), &pool2, &token1, 0, 0, &pkh1),
)
.unwrap();
insert_new_pool(
conn,
&dummy_cauldron(&Txid::all_zeros(), &pool3, &token1, 0, 0, &pkh1),
)
.unwrap();
insert_new_pool(
conn,
&dummy_cauldron(&Txid::all_zeros(), &pool4, &token1, 0, 0, &pkh1),
)
.unwrap();
}
#[test]
fn test_future_end_timestamp() {
// Set up the database and Rocket instance
let mock_db = mock_db_pool(setup_mock_db);
let rocket = rocket::build()
.manage(mock_db)
.mount("/api", routes![super::price_candlesticks]);
let client = Client::tracked(rocket).expect("valid rocket instance");
// Use a far-future timestamp for 'end'
let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000";
let future_end = time_now() + 1_000_000; // current time + offset
let response = client
.get(format!(
"/api/price/{}/candlesticks?end={}",
token_id_zero, future_end
))
.dispatch();
assert_eq!(response.status(), Status::BadRequest);
let body = response.into_string().unwrap_or_default();
assert!(body.contains("End timestamp cannot be in the future"));
}
#[test]
fn test_start_after_end_timestamp() {
// Set up the database and Rocket instance
let mock_db = mock_db_pool(setup_mock_db);
let rocket = rocket::build()
.manage(mock_db)
.mount("/api", routes![super::price_candlesticks]);
let client = Client::tracked(rocket).expect("valid rocket instance");
let token_id_zero = "0000000000000000000000000000000000000000000000000000000000000000";
// Provide a start timestamp that is after the end timestamp
let start_after_end = "2000&end=1000"; // start=2000, end=1000
let response = client
.get(format!(
"/api/price/{}/candlesticks?start={}",
token_id_zero, start_after_end
))
.dispatch();
assert_eq!(response.status(), Status::BadRequest);
let body = response.into_string().unwrap_or_default();
assert!(body.contains("Start timestamp must be before end timestamp"));
}
#[test]
fn test_multiple_candlesticks_endpoint() {
// Set up DB with 4 trades across 2 bins
let mock_db = mock_db_pool(setup_mock_db);
// Build Rocket
let rocket = rocket::build()
.manage(mock_db)
.mount("/api", routes![super::price_candlesticks]);
let client = Client::tracked(rocket).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/{}/candlesticks?start=1727963300&end=1727964500&stepsize=600",
token_id_zero
))
.dispatch();
assert_eq!(response.status(), Status::Ok);
let body = response.into_string().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);
}
#[test]
fn test_single_swap_multiple_pools() {
// Set up a fresh mock DB
let mock_db = mock_db_pool(|conn| {
// Create required tables
utxo_funding::create_table(conn);
tx::create_table(conn);
pool::create_table(conn);
dummy_init_seq();
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 cauldrons = Vec::new();
let mut pools = Vec::new();
let times = [TIME_1, TIME_2]; // Use two different times within same transaction, for simplicity
for (i, &time) in times.iter().enumerate() {
let utxo = OutPointHash::from_inner([0xe1 + i as u8; 32]);
let pool_hash = OutPointHash::from_inner([0x0a + i as u8; 32]);
let cauldron = dummy_cauldron(
&txid_multi,
&utxo,
&token_zero,
100_000 * (i + 1) as u64,
2_000,
&pkh_zero,
);
cauldrons.push(cauldron.clone());
pools.push(pool_hash);
insert_utxo_funding(conn, &vec![cauldron.clone()], &txid_multi, true).unwrap();
// Insert a tx row for the transaction
insert_block_tx(conn, &txid_multi, &block_zero, time as i64).unwrap();
insert_mempool_tx(conn, &txid_multi, time as u64).unwrap();
// Insert pool_history_entry for each pool related to the transaction
pool::insert_pool_history_entry(
conn,
&pool_hash,
&cauldron,
Some(time as u64),
Some(time as u64),
)
.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 {
insert_new_pool(
conn,
&dummy_cauldron(&Txid::all_zeros(), pool_hash, &token1, 0, 0, &pkh1),
)
.unwrap();
}
});
// Build Rocket instance with our endpoint
let rocket = rocket::build()
.manage(mock_db)
.mount("/api", routes![super::price_candlesticks]);
let client = Client::tracked(rocket).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();
assert_eq!(response.status(), Status::Ok);
let body = response.into_string().expect("No response body");
let json: serde_json::Value = serde_json::from_str(&body).expect("Invalid JSON");
let candles = json["candlesticks"].as_array().unwrap();
// Since all pool trades occurred under one transaction,
// the candlestick aggregation should consider them as a single swap per interval.
// Validate that the volume_sats and volume_tokens aggregate values from all trades under the same transaction.
// Depending on how intervals align with our test times, check the results.
// For simplicity, assume our interval covers all trades in one candle.
let first_candle = &candles[0];
let expected_volume_sats = 100_000 + 200_000; // Sum of trades from both pools in the single tx
let expected_volume_tokens = 2000 + 2000; // Sum of token amounts from both pools
assert_eq!(
first_candle["volume_sats"].as_i64().unwrap(),
expected_volume_sats
);
assert_eq!(
first_candle["volume_tokens"].as_i64().unwrap(),
expected_volume_tokens
);
// Additional assertions can be made regarding open, close, low, high values if required.
}
}

1
src/rpc/mod.rs
View file

@ -13,6 +13,7 @@ use rusqlite::{params, Connection};
pub mod apy; pub mod apy;
pub mod bcmr; pub mod bcmr;
pub mod candlesticks;
pub mod contract; pub mod contract;
pub mod err; pub mod err;
pub mod oracle; pub mod oracle;