diff --git a/src/main.rs b/src/main.rs index b941957..0a788a9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -334,6 +334,7 @@ fn launch() -> _ { rpc::tokens::list_by_volume, rpc::tokens::search_by_volume, rpc::price::price_history, + rpc::candlesticks::price_candlesticks, rpc::price::price_current, rpc::price::price_at, rpc::pool::list_pools_by_apy, diff --git a/src/rpc/candlesticks.rs b/src/rpc/candlesticks.rs new file mode 100644 index 0000000..ffdc4ab --- /dev/null +++ b/src/rpc/candlesticks.rs @@ -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, + close: Option, + 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 { + 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> { + 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 = 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//candlesticks?&&")] +pub fn price_candlesticks( + token: &str, + start: Option, + end: Option, + stepsize: Option, + conn: &State, +) -> Result, Custom> { + 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 = 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. + } +} diff --git a/src/rpc/mod.rs b/src/rpc/mod.rs index c97fd79..8767c02 100644 --- a/src/rpc/mod.rs +++ b/src/rpc/mod.rs @@ -13,6 +13,7 @@ use rusqlite::{params, Connection}; pub mod apy; pub mod bcmr; +pub mod candlesticks; pub mod contract; pub mod err; pub mod pool;