using oracles cash feed

src/db/cauldron/tokenlist/token_utils.rs

@@ -11,6 +11,7 @@ use sqlx::SqlitePool;
 use crate::db::blob::display_hex_to_blob;
 use crate::db::cauldron::pool::{get_injections_between, get_pool_period_snapshot};
 use crate::db::oracle::get_closest;
+use crate::db::oracle::oracle_cash::get_oracle_cash_closest;
 use crate::rpc::apy::apyaggregator::APYAggregator;
 use crate::rpc::apy::poolperiod::split_at_injections;
 use malachite::base::num::arithmetic::traits::FloorSqrt;
@@ -54,9 +55,16 @@ pub fn dec_to_f64_bounded(d: Decimal) -> f64 {
 }
 
 pub const SATS_PER_BCH: i64 = 100_000_000;
-pub const ORACLE_SCALE: i64 = 1_000_000;
+/// On-chain Delphi oracle scale: prices are stored in cents
+/// (e.g. $384.24 → 38424). Same unit as the oracles.cash feed.
+pub const ORACLE_SCALE: i64 = 100;
+/// Timestamps before this value are looked up in the on-chain Delphi indexed table.
+/// Timestamps on or after are looked up in the oracle_cash table (oracles.cash feed).
+/// Value: 2026-04-23 00:00:00 UTC (last reliable Delphi oracle update).
+const ORACLE_CUTOFF_TS: i64 = 1776902400;
 
 pub async fn usd_per_bch_at_or_before(oracle_pool: &SqlitePool, ts: i64) -> Decimal {
+    if ts < ORACLE_CUTOFF_TS {
         match get_closest(oracle_pool, &None, ts).await {
             Ok(Some(e)) => {
                 Decimal::from_i64(e.oracle_price).unwrap_or(Decimal::ZERO)
@@ -64,6 +72,15 @@ pub async fn usd_per_bch_at_or_before(oracle_pool: &SqlitePool, ts: i64) -> Deci
             }
             _ => Decimal::ZERO,
         }
+    } else {
+        // oracles.cash prices are in cents; divide by 100 to get USD/BCH
+        match get_oracle_cash_closest(oracle_pool, ts).await {
+            Ok(Some(e)) => {
+                Decimal::from_i64(e.oracle_price).unwrap_or(Decimal::ZERO) / Decimal::from(100)
+            }
+            _ => Decimal::ZERO,
+        }
+    }
 }
 
 #[inline]

src/db/init.rs

@@ -16,6 +16,7 @@ use crate::db::cauldron::config::check_db_version;
 use crate::db::cauldron::prepare_tables as cauldron_prepare_tables;
 use crate::db::crc20::prepare_tables as crc20_prepare_tables;
 use crate::db::moria::prepare_tables as moria_prepare_tables;
+use crate::db::oracle::oracle_cash::prepare_oracle_cash_tables;
 use crate::db::oracle::prepare_tables as oracle_prepare_tables;
 
 /// Create read and write database pools for a given database path
@@ -124,6 +125,8 @@ pub async fn initialize_databases(network: &str, read_slots: ReadSlots) -> Resul
     if !db_exists {
         oracle_prepare_tables(&oracle_db_write).await;
     }
+    // Always-run migration: safe on both new and existing oracle.db
+    prepare_oracle_cash_tables(&oracle_db_write).await;
 
     // Initialize moria lending database
     let (db_exists, moria_db_write, moria_db_read) =

src/db/oracle/mod.rs

@@ -3,6 +3,8 @@
 // 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
 
+pub mod oracle_cash;
+
 use anyhow::{bail, Result};
 use bitcoin_hashes::{hex::ToHex, Hash};
 use bitcoincash::{BlockHash, TokenID, Transaction, Txid};

src/db/oracle/oracle_cash.rs

@@ -0,0 +1,160 @@
+// Copyright (C) 2024-2026 Whiterun LLC
+//
+// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
+// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
+
+use anyhow::{bail, Result};
+use sqlx::{Row, SqlitePool};
+
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct OracleCashPrice {
+    pub oracle_timestamp: i64,
+    /// Price in cents (same convention as the on-chain Delphi oracle)
+    pub oracle_price: i64,
+    pub message_sequence: Option<i64>,
+}
+
+/// Creates the oracle_cash_price table if it doesn't already exist.
+/// Safe to call on both new and existing oracle.db instances.
+pub async fn prepare_oracle_cash_tables(pool: &SqlitePool) {
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS oracle_cash_price (
+            oracle_timestamp INTEGER NOT NULL PRIMARY KEY,
+            oracle_price     INTEGER NOT NULL,
+            message_sequence INTEGER
+        )",
+    )
+    .execute(pool)
+    .await
+    .expect("failed to create oracle_cash_price table");
+
+    sqlx::query(
+        "CREATE INDEX IF NOT EXISTS idx_oracle_cash_seq
+         ON oracle_cash_price(message_sequence)
+         WHERE message_sequence IS NOT NULL",
+    )
+    .execute(pool)
+    .await
+    .expect("failed to create oracle_cash_seq index");
+}
+
+pub async fn upsert_oracle_cash_price(
+    pool: &SqlitePool,
+    oracle_timestamp: i64,
+    oracle_price: i64,
+    message_sequence: Option<i64>,
+) -> Result<()> {
+    sqlx::query(
+        "INSERT OR REPLACE INTO oracle_cash_price
+             (oracle_timestamp, oracle_price, message_sequence)
+         VALUES (?, ?, ?)",
+    )
+    .bind(oracle_timestamp)
+    .bind(oracle_price)
+    .bind(message_sequence)
+    .execute(pool)
+    .await
+    .map_err(|e| anyhow::anyhow!("failed to upsert oracle_cash_price: {}", e))?;
+    Ok(())
+}
+
+/// Returns the earliest oracle_timestamp stored, or None if the table is empty.
+pub async fn get_oracle_cash_min_timestamp(pool: &SqlitePool) -> Result<Option<i64>> {
+    let row: (Option<i64>,) = sqlx::query_as("SELECT MIN(oracle_timestamp) FROM oracle_cash_price")
+        .fetch_one(pool)
+        .await?;
+    Ok(row.0)
+}
+
+pub async fn get_oracle_cash_closest(
+    pool: &SqlitePool,
+    timestamp: i64,
+) -> Result<Option<OracleCashPrice>> {
+    let row = sqlx::query(
+        "SELECT oracle_timestamp, oracle_price, message_sequence
+         FROM oracle_cash_price
+         WHERE oracle_timestamp <= ?
+         ORDER BY oracle_timestamp DESC
+         LIMIT 1",
+    )
+    .bind(timestamp)
+    .fetch_optional(pool)
+    .await?;
+
+    Ok(row.map(|r| OracleCashPrice {
+        oracle_timestamp: r.get(0),
+        oracle_price: r.get(1),
+        message_sequence: r.get(2),
+    }))
+}
+
+pub async fn get_oracle_cash_range(
+    pool: &SqlitePool,
+    start: i64,
+    end: i64,
+) -> Result<Vec<OracleCashPrice>> {
+    let rows = sqlx::query(
+        "SELECT oracle_timestamp, oracle_price, message_sequence
+         FROM oracle_cash_price
+         WHERE oracle_timestamp BETWEEN ? AND ?
+         ORDER BY oracle_timestamp ASC",
+    )
+    .bind(start)
+    .bind(end)
+    .fetch_all(pool)
+    .await?;
+
+    Ok(rows
+        .iter()
+        .map(|r| OracleCashPrice {
+            oracle_timestamp: r.get(0),
+            oracle_price: r.get(1),
+            message_sequence: r.get(2),
+        })
+        .collect())
+}
+
+pub async fn get_oracle_cash_range_with_step(
+    pool: &SqlitePool,
+    start: i64,
+    end: i64,
+    step: i64,
+) -> Result<Vec<OracleCashPrice>> {
+    let total_intervals = (end - start) / step;
+    const MAX_INTERVALS: i64 = 10_000;
+    if total_intervals > MAX_INTERVALS {
+        bail!(
+            "Too many intervals ({} > {})",
+            total_intervals,
+            MAX_INTERVALS
+        );
+    }
+
+    let all = get_oracle_cash_range(pool, start, end).await?;
+
+    let mut buckets: Vec<OracleCashPrice> = Vec::new();
+    let mut next_threshold = start + step;
+    let mut last_entry: Option<OracleCashPrice> = None;
+
+    for entry in all {
+        while entry.oracle_timestamp >= next_threshold {
+            if let Some(e) = last_entry.take() {
+                buckets.push(e);
+            }
+            next_threshold += step;
+        }
+        last_entry = Some(entry);
+    }
+
+    if let Some(e) = last_entry {
+        if buckets
+            .last()
+            .map(|b| b.oracle_timestamp != e.oracle_timestamp)
+            .unwrap_or(true)
+        {
+            buckets.push(e);
+        }
+    }
+
+    Ok(buckets)
+}

src/db/search/mod.rs

@@ -163,7 +163,7 @@ pub async fn search_tokens_by_volume(
     ]
     .concat();
 
-    combined_tokens.sort_by(|a, b| a.0.to_lowercase().cmp(&b.0.to_lowercase()));
+    combined_tokens.sort_by_key(|a| a.0.to_lowercase());
 
     combined_tokens.dedup_by(|a, b| {
         if a.0.eq_ignore_ascii_case(&b.0) {
@@ -179,7 +179,7 @@ pub async fn search_tokens_by_volume(
 
     let mut result = token_volume(cauldron_pool, combined_tokens).await?;
 
-    result.sort_by(|a, b| b.3.cmp(&a.3));
+    result.sort_by_key(|b| std::cmp::Reverse(b.3));
     Ok(result)
 }
 

src/main.rs

@@ -79,6 +79,7 @@ mod db;
 mod def;
 mod electrum;
 mod index;
+mod oracle_cash;
 mod rpc;
 mod signal;
 mod timeutil;
@@ -118,6 +119,7 @@ async fn start_program(
     BCMRDownloader,
     WellKnownDownloader,
     CRC20Fetcher,
+    oracle_cash::OracleCashFetcher,
     Arc<IbdState>,
     Arc<AtomicBool>, // indexing_in_progress
 )> {
@@ -352,11 +354,15 @@ async fn start_program(
 
     spawn_token_metrics_updater(db.clone(), indexing_in_progress.clone());
 
+    let mut oracle_cash_fetcher = oracle_cash::OracleCashFetcher::new();
+    oracle_cash_fetcher.start(db.oracle_w.clone(), db.oracle_r.clone());
+
     Ok((
         db,
         bcmrdownloader,
         wellknowndownloader,
         crc20fetcher,
+        oracle_cash_fetcher,
         ibd_state,
         indexing_in_progress,
     ))
@@ -383,6 +389,7 @@ async fn launch() -> _ {
         bcmrdownloader,
         wellknowndownloader,
         crc20fetcher,
+        oracle_cash_fetcher,
         ibd_state,
         indexing_in_progress,
     ) = match start_program(config).await {
@@ -570,10 +577,11 @@ async fn launch() -> _ {
         .manage(dbpool)
         .manage(ibd_state)
         .manage(ohlcv_state)
-        // give rocket ownership of downloader to ensure thread isn't dropped
+        // give rocket ownership of downloaders/fetchers to ensure threads aren't dropped
         .manage(bcmrdownloader)
         .manage(wellknowndownloader)
         .manage(crc20fetcher)
+        .manage(oracle_cash_fetcher)
         .mount(
             "/cauldron/",
             routes![
@@ -611,7 +619,9 @@ async fn launch() -> _ {
             routes![
                 rpc::oracle::oracle_get_closest,
                 rpc::oracle::oracle_get_range,
-                rpc::oracle::oracle_get_history
+                rpc::oracle::oracle_get_history,
+                rpc::oracle::oracle_cash_closest,
+                rpc::oracle::oracle_cash_history,
             ],
         )
         .mount(

src/oracle_cash.rs

@@ -0,0 +1,230 @@
+// 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
+
+//! Background task that polls oracles.cash and keeps oracle_cash_price up-to-date.
+//! This is a parallel feed alongside the on-chain Delphi oracle; both write to
+//! oracle.db but to separate tables.
+
+use std::{
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    },
+    time::Duration,
+};
+
+use log::{info, warn};
+use serde::Deserialize;
+use sqlx::SqlitePool;
+use tokio::task::JoinHandle;
+
+use crate::db::oracle::oracle_cash::{get_oracle_cash_min_timestamp, upsert_oracle_cash_price};
+use crate::timeutil::time_now;
+
+const ORACLES_CASH_URL: &str = "https://oracles.generalprotocols.com";
+/// General Protocols BCH/USD oracle public key
+const BCH_USD_ORACLE_PUBKEY: &str =
+    "02d09db08af1ff4e8453919cc866a4be427d7bfe18f2c05e5444c196fcf6fd2818";
+
+const POLL_INTERVAL: Duration = Duration::from_secs(300); // 5 minutes
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(20);
+const BACKFILL_AGGREGATION: i64 = 3600; // 1-hour buckets
+/// Backfill oracle_cash data starting from this timestamp.
+/// Everything before this point is served from the on-chain Delphi indexed table,
+/// so oracle_cash only needs to cover from here onward.
+/// Value: 2026-04-23 00:00:00 UTC (last reliable Delphi oracle update).
+const ORACLE_CUTOFF_TS: i64 = 1776902400;
+
+// ── API response types ────────────────────────────────────────────────────────
+
+#[derive(Deserialize, Debug)]
+struct MessageMetrics {
+    #[serde(rename = "maxMessageSequence")]
+    max_message_sequence: i64,
+    /// Price in cents (same unit as on-chain Delphi: 38424 = $384.24)
+    #[serde(rename = "currentPrice")]
+    current_price: i64,
+    #[serde(rename = "maxMessageTimestamp")]
+    max_message_timestamp: i64,
+}
+
+#[derive(Deserialize, Debug)]
+struct OracleEntry {
+    #[serde(rename = "publicKey")]
+    public_key: String,
+    #[serde(rename = "messageMetrics")]
+    message_metrics: Option<MessageMetrics>,
+}
+
+#[derive(Deserialize, Debug)]
+struct OraclesResponse {
+    oracles: Vec<OracleEntry>,
+}
+
+#[derive(Deserialize, Debug)]
+struct PriceGraphPoint {
+    #[serde(rename = "averageTimestamp")]
+    average_timestamp: f64,
+    /// Average price in cents over the aggregation window
+    #[serde(rename = "averagePrice")]
+    average_price: f64,
+}
+
+#[derive(Deserialize, Debug)]
+struct PriceGraphResponse {
+    #[serde(rename = "priceGraphPoints")]
+    price_graph_points: Vec<PriceGraphPoint>,
+}
+
+// ── Fetch helpers ─────────────────────────────────────────────────────────────
+
+/// Returns (timestamp_secs, price_cents, message_sequence)
+async fn fetch_current_price(client: &reqwest::Client) -> anyhow::Result<(i64, i64, i64)> {
+    let url = format!("{}/api/v1/oracles", ORACLES_CASH_URL);
+    let resp: OraclesResponse = client
+        .get(&url)
+        .timeout(REQUEST_TIMEOUT)
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+
+    let oracle = resp
+        .oracles
+        .into_iter()
+        .find(|o| o.public_key == BCH_USD_ORACLE_PUBKEY)
+        .ok_or_else(|| anyhow::anyhow!("BCH/USD oracle not found in oracles.cash response"))?;
+
+    let m = oracle
+        .message_metrics
+        .ok_or_else(|| anyhow::anyhow!("No messageMetrics for BCH/USD oracle"))?;
+
+    Ok((
+        m.max_message_timestamp,
+        m.current_price,
+        m.max_message_sequence,
+    ))
+}
+
+/// Fetches hourly price history from ORACLE_CUTOFF_TS to now and inserts into the DB.
+async fn backfill_history(client: &reqwest::Client, pool: &SqlitePool) -> anyhow::Result<()> {
+    let now = time_now();
+    let min_ts = ORACLE_CUTOFF_TS;
+
+    let url = format!(
+        "{}/api/v2/priceGraphPoints?publicKey={}&minMessageTimestamp={}&maxMessageTimestamp={}&aggregationPeriod={}",
+        ORACLES_CASH_URL, BCH_USD_ORACLE_PUBKEY, min_ts, now, BACKFILL_AGGREGATION
+    );
+
+    let resp: PriceGraphResponse = client
+        .get(&url)
+        .timeout(Duration::from_secs(30))
+        .send()
+        .await?
+        .error_for_status()?
+        .json()
+        .await?;
+
+    let count = resp.price_graph_points.len();
+    for point in resp.price_graph_points {
+        let ts = point.average_timestamp as i64;
+        let price = point.average_price as i64;
+        // No real sequence available for aggregated points
+        upsert_oracle_cash_price(pool, ts, price, None).await?;
+    }
+
+    info!("oracle_cash: backfilled {count} hourly price points");
+    Ok(())
+}
+
+// ── Fetcher ───────────────────────────────────────────────────────────────────
+
+pub struct OracleCashFetcher {
+    keep_running: Arc<AtomicBool>,
+    task: Option<JoinHandle<()>>,
+}
+
+impl OracleCashFetcher {
+    pub fn new() -> Self {
+        Self {
+            keep_running: Arc::new(AtomicBool::new(true)),
+            task: None,
+        }
+    }
+
+    pub fn start(&mut self, oracle_w: SqlitePool, oracle_r: SqlitePool) {
+        let keep_running = self.keep_running.clone();
+
+        self.task = Some(tokio::spawn(async move {
+            let client = reqwest::Client::new();
+
+            // Backfill if the table is empty or if the earliest entry is after
+            // ORACLE_CUTOFF_TS (meaning we're missing data from when the Delphi oracle
+            // stopped being updated).
+            let needs_backfill = match get_oracle_cash_min_timestamp(&oracle_r).await {
+                Ok(None) => {
+                    info!("oracle_cash: table empty, will backfill from ORACLE_CUTOFF_TS");
+                    true
+                }
+                Ok(Some(min_ts)) => {
+                    if min_ts > ORACLE_CUTOFF_TS {
+                        info!(
+                            "oracle_cash: earliest entry {min_ts} is after cutoff {ORACLE_CUTOFF_TS} — will backfill"
+                        );
+                        true
+                    } else {
+                        info!(
+                            "oracle_cash: earliest entry {min_ts} covers cutoff, skipping backfill"
+                        );
+                        false
+                    }
+                }
+                Err(e) => {
+                    warn!("oracle_cash: could not check earliest timestamp: {e}");
+                    false
+                }
+            };
+
+            if needs_backfill {
+                if let Err(e) = backfill_history(&client, &oracle_w).await {
+                    warn!("oracle_cash: backfill failed: {e}");
+                }
+            }
+
+            loop {
+                if !keep_running.load(Ordering::Relaxed) {
+                    info!("oracle_cash: exiting fetch task");
+                    return;
+                }
+
+                match fetch_current_price(&client).await {
+                    Ok((ts, price_cents, seq)) => {
+                        match upsert_oracle_cash_price(&oracle_w, ts, price_cents, Some(seq)).await
+                        {
+                            Ok(()) => info!(
+                                "oracle_cash: stored price={price_cents} cents ts={ts} seq={seq}"
+                            ),
+                            Err(e) => warn!("oracle_cash: failed to store price: {e}"),
+                        }
+                    }
+                    Err(e) => warn!("oracle_cash: failed to fetch current price: {e}"),
+                }
+
+                tokio::time::sleep(POLL_INTERVAL).await;
+            }
+        }));
+    }
+}
+
+impl Drop for OracleCashFetcher {
+    fn drop(&mut self) {
+        self.keep_running.store(false, Ordering::SeqCst);
+        if let Some(task) = self.task.take() {
+            task.abort();
+            info!("oracle_cash fetch task aborted");
+        }
+    }
+}

src/rpc/oracle.rs

@@ -3,6 +3,9 @@
 // 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::oracle::oracle_cash::{
+    get_oracle_cash_closest, get_oracle_cash_range, get_oracle_cash_range_with_step,
+};
 use crate::db::oracle::{get_closest, get_range, get_range_with_step};
 use crate::db::DB;
 use crate::rpc::err::{bad_request, db_error, ApiErrorCode, CachedApiResult};
@@ -177,3 +180,83 @@ pub async fn oracle_get_history(
         CACHE_AGGREGATE,
     ))
 }
+
+/// Get the closest BCH/USD price from the oracles.cash feed for a given timestamp.
+///
+/// Status: Stable
+///
+/// - timestamp: Unix timestamp in seconds (optional, defaults to now)
+///
+/// Returns `null` if no data is available.
+///
+/// `oracle_price` is in **cents**. Divide by 100 to get USD.
+#[get("/cash/closest?<timestamp>")]
+pub async fn oracle_cash_closest(
+    timestamp: Option<i64>,
+    db: &State<DB>,
+) -> CachedApiResult<serde_json::Value> {
+    let ts = timestamp.unwrap_or_else(time_now);
+    let entry = get_oracle_cash_closest(&db.oracle_r, ts)
+        .await
+        .map_err(db_error)?;
+    Ok(cached_ok(
+        entry.map_or(serde_json::Value::Null, |e| {
+            serde_json::to_value(e).unwrap()
+        }),
+        CACHE_AGGREGATE,
+    ))
+}
+
+/// Get historical BCH/USD prices from the oracles.cash feed.
+///
+/// Status: Stable
+///
+/// - start: Unix timestamp for start of period
+/// - end: Unix timestamp for end of period (optional, defaults to now)
+/// - stepsize: Seconds per interval (optional)
+///
+/// `oracle_price` values are in **cents**.
+///
+/// **Response Example:**
+/// ```json
+/// [
+///   { "oracle_timestamp": 1709468902, "oracle_price": 38424, "message_sequence": 12345 },
+///   { "oracle_timestamp": 1709472502, "oracle_price": 38501, "message_sequence": null }
+/// ]
+/// ```
+#[get("/cash/history?<start>&<end>&<stepsize>")]
+pub async fn oracle_cash_history(
+    start: Option<i64>,
+    end: Option<i64>,
+    stepsize: Option<i64>,
+    db: &State<DB>,
+) -> CachedApiResult<Value> {
+    let current_timestamp = time_now();
+    let start_ts = start.unwrap_or(current_timestamp - 30 * 24 * 3600);
+    let end_ts = end.unwrap_or(current_timestamp);
+
+    let entries = if let Some(step) = stepsize {
+        get_oracle_cash_range_with_step(&db.oracle_r, start_ts, end_ts, step)
+            .await
+            .map_err(|e| {
+                bad_request(
+                    ApiErrorCode::InvalidParameters,
+                    &format!("Query error: {e}"),
+                )
+            })?
+    } else {
+        get_oracle_cash_range(&db.oracle_r, start_ts, end_ts)
+            .await
+            .map_err(|e| {
+                bad_request(
+                    ApiErrorCode::InvalidParameters,
+                    &format!("Query error: {e}"),
+                )
+            })?
+    };
+
+    Ok(cached_ok(
+        serde_json::to_value(entries).unwrap(),
+        CACHE_AGGREGATE,
+    ))
+}