Organise candlesticks

src/rpc/candlesticks/mod.rs

@@ -0,0 +1,549 @@
+// Copyright (C) 2024-2026 Whiterun LLC
+//
+// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
+// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
+
+use crate::db::blob::display_hex_to_blob;
+use crate::db::cauldron::ohlcv;
+use crate::db::DB;
+use crate::rpc::err::{bad_request, ApiErrorCode, CachedApiResult};
+use crate::rpc::response::{cached_ok, CACHE_IMMUTABLE, CACHE_NONE};
+use crate::timeutil::time_now;
+use crate::OhlcvState;
+use anyhow::{bail, Result};
+use bitcoincash::TokenID;
+use rocket::{get, State};
+use serde::Serialize;
+use serde_json::json;
+use serde_json::Value;
+use sqlx::{Row, SqlitePool};
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+#[derive(Debug, Serialize)]
+pub struct CandlestickData {
+    pub time: i64, // start of the interval
+    pub open: f64,
+    pub close: f64,
+    pub high: f64,
+    pub low: f64,
+    pub volume_sats: i64,
+    pub volume_tokens: i64,
+    pub transaction_count: i64,
+}
+
+struct PriceInterval {
+    start: i64,
+    step: i64,
+    low: f64,
+    high: f64,
+    open: Option<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.
+///
+///
+/// - start: unix timestamp for period start (default 30 days)
+/// - end: unix timestamp for period end (default NOW)
+/// - stepsize: seconds per interval (default: 3600 seconds)
+///
+/// **Response Example:**
+///
+/// ```json
+/// {
+///     "candlesticks": [
+///         {"close":64654136.35714286,
+///          "high":87959043.0,
+///          "low":58755326.0,
+///          "open":87959043.0,
+///          "time":1752522150,
+///          "transaction_count":4,
+///          "volume_sats":3170247594,
+///          "volume_tokens":43},
+///     ]
+/// }
+/// ```
+///
+#[get("/price/<token>/candlesticks?<start>&<end>&<stepsize>")]
+pub async fn price_candlesticks(
+    token: &str,
+    start: Option<i64>,
+    end: Option<i64>,
+    stepsize: Option<i64>,
+    conn: &State<DB>,
+    ohlcv: &State<Arc<OhlcvState>>,
+) -> CachedApiResult<Value> {
+    let current_timestamp = time_now();
+
+    if let Some(end_ts) = end {
+        if end_ts > current_timestamp {
+            return Err(bad_request(
+                ApiErrorCode::FutureTimestamp,
+                "End timestamp cannot be in the future",
+            ));
+        }
+    }
+
+    let effective_end = end.unwrap_or(current_timestamp);
+    let effective_start = start.unwrap_or(current_timestamp - 30 * 24 * 3600);
+    let effective_stepsize = stepsize.unwrap_or(3600);
+
+    if effective_stepsize <= 0 {
+        return Err(bad_request(
+            ApiErrorCode::InvalidStepsize,
+            "stepsize must be > 0",
+        ));
+    }
+
+    if effective_start >= effective_end {
+        return Err(bad_request(
+            ApiErrorCode::InvalidTimeRange,
+            "Start timestamp must be before end timestamp",
+        ));
+    }
+
+    const MAX_INTERVALS: i64 = 10000;
+    let total_intervals = (effective_end - effective_start) / effective_stepsize;
+    if total_intervals > MAX_INTERVALS {
+        return Err(bad_request(
+            ApiErrorCode::TooManyIntervals,
+            &format!("Too many intervals ({total_intervals} > {MAX_INTERVALS})"),
+        ));
+    }
+
+    // Windows fully in the past get immutable HTTP cache headers.
+    let is_historical = end.is_some() && effective_end < current_timestamp - effective_stepsize;
+
+    let ohlcv_end = ohlcv.materialized_end.load(Ordering::Relaxed);
+
+    let candlestick_data = candlesticks(
+        &conn.cauldron_r,
+        effective_start,
+        effective_end,
+        effective_stepsize,
+        token,
+        ohlcv_end,
+    )
+    .await
+    .map_err(|e| bad_request(ApiErrorCode::InvalidParameters, &e.to_string()))?;
+
+    let candlesticks_json: Vec<Value> = candlestick_data
+        .into_iter()
+        .map(|c| {
+            json!({
+                "time": c.time,
+                "high": c.high,
+                "low": c.low,
+                "open": c.open,
+                "close": c.close,
+                "volume_sats": c.volume_sats,
+                "volume_tokens": c.volume_tokens,
+                "transaction_count": c.transaction_count
+            })
+        })
+        .collect();
+
+    let cache_duration = if is_historical {
+        CACHE_IMMUTABLE
+    } else {
+        CACHE_NONE
+    };
+
+    Ok(cached_ok(
+        json!({ "candlesticks": candlesticks_json }),
+        cache_duration,
+    ))
+}
+
+#[cfg(test)]
+mod tests;