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

Merge branch 'apy' into 'master'

Browse source 
Add aggregated APY endpoint

See merge request riftenlabs/riftenlabs-indexer!17
This commit is contained in:
Dagur Valberg Johannsson 2024-10-24 07:16:57 +00:00
commit 8401ef7362
11 changed files with 706 additions and 246 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.toml
View file

@ -27,7 +27,7 @@ log = "0.4"
stderrlog = "0.6.0"
r2d2 = "0.8.10"
r2d2_sqlite = "0.24.0"
rust_decimal = "1.34"
rust_decimal = { version = "1.34", features = ["maths"] }
rust_decimal_macros = "1.34"
ureq = { version = "2.9", features = ["json"] }
rand = "0.8.5"

45
src/bcmr/mod.rs
View file

@ -4,16 +4,12 @@
// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
use rayon::prelude::*;
use std::{
cmp,
collections::{HashMap, HashSet, VecDeque},
convert::TryInto,
};
use std::{cmp, collections::HashMap, convert::TryInto};
use anyhow::Result;
use bitcoin_hashes::{hex::ToHex, Hash};
use bitcoincash::BlockHash;
use bitcoincash::{Script, TokenID, Transaction, Txid};
use bitcoincash::{Script, TokenID, Transaction};
use log::debug;
use riftenlabs_defi::chainutil::{compute_outpoint_hash, read_push_from_script, OutPointHash};
use rusqlite::Connection;
@ -95,37 +91,6 @@ pub fn parse_bcmr(tx: &Transaction) -> Option<BCMR> {
parse_bcmr_from_opreturn(&bcmr_op_return.script_pubkey)
}
pub fn ttor_sorted(txs: Vec<Transaction>) -> Vec<Transaction> {
let txs = {
let mut queue: VecDeque<Transaction> = txs.into_iter().collect();
let mut queue_txids: HashSet<Txid> = queue.par_iter().map(|tx| tx.txid()).collect();
let mut txs: Vec<Transaction> = Vec::with_capacity(queue.len());
while let Some(tx) = queue.pop_front() {
let mut has_parent = false;
for i in &tx.input {
if queue_txids.contains(&i.previous_output.txid) {
// depends on parent
has_parent = true;
break;
};
}
if has_parent {
queue.push_back(tx);
} else {
queue_txids.remove(&tx.txid());
txs.push(tx)
}
}
txs
};
txs
}
// TODO: Use when we get scriptpubkey filter for electrum.mempoo.get
#[allow(dead_code)]
pub fn mempool_index_genesis(conn: &Connection, txs: &Vec<Transaction>) -> Result<usize> {
@ -157,10 +122,8 @@ pub fn mempool_index_genesis(conn: &Connection, txs: &Vec<Transaction>) -> Resul
pub fn index_bcmr(
conn: &Connection,
blockhash: &BlockHash,
txs: Vec<Transaction>,
sorted: Vec<Transaction>, // ttor sorted!!
) -> Result<usize> {
let sorted = ttor_sorted(txs);
let mut new_tokens: Vec<(Transaction, TokenID)> = Vec::default();
let sorted: Vec<Transaction> = sorted
@ -182,7 +145,7 @@ pub fn index_bcmr(
.flat_map(|(tx_pos, tx)| {
let tx_candidates: Vec<(OutPointHash, (&Transaction, usize))> = tx
.input
.iter()
.par_iter()
.filter_map(|i| {
let prevout = &i.previous_output;
if prevout.vout != 0 {

209
src/db/cauldron/pool.rs
View file

@ -3,13 +3,19 @@
// 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 std::collections::VecDeque;
use std::{
collections::{HashMap, VecDeque},
sync::atomic::AtomicI64,
};
use anyhow::{Context, Result};
use anyhow::Result;
use bitcoin_hashes::hex::{FromHex, ToHex};
use log::{debug, info};
use log::{debug, info, warn};
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
use rusqlite::{params, Connection};
use rusqlite::{params, Connection, Row};
use rust_decimal::prelude::Zero;
use crate::rpc::apy::PoolSnapshot;
pub fn create_table(conn: &Connection) {
conn.execute(
@ -31,7 +37,10 @@ pub fn create_table(conn: &Connection) {
txid TEXT REFERENCES tx(txid) ON DELETE CASCADE,
tx_pos TEXT NOT NULL,
mtp_timestamp BIGINT,
first_seen_timestamp BIGINT
first_seen_timestamp BIGINT,
sequence BIGINT NOT NULL,
sats BIGINT,
token_amount BIGINT
)",
[],
)
@ -52,7 +61,7 @@ fn get_pool_by_utxo(conn: &Connection, utxo_hash: &OutPointHash) -> Result<Optio
}
}
fn flag_as_withdrawn(
pub fn flag_as_withdrawn(
conn: &Connection,
pool_utxo: &OutPointHash,
cauldron: &ParsedContract,
@ -61,12 +70,12 @@ fn flag_as_withdrawn(
"UPDATE pool SET withdrawn_in_utxo = ? WHERE creation_utxo = ?",
params![cauldron.spent_utxo_hash.to_hex(), pool_utxo.to_hex()],
)
.context("flagging pool as withdrawn")?;
.map_err(|e| anyhow::anyhow!("failed flag pool as withdrawn. Original error: {:?}", e))?;
Ok(())
}
fn insert_new_pool(conn: &Connection, cauldron: &ParsedContract) -> Result<()> {
pub fn insert_new_pool(conn: &Connection, cauldron: &ParsedContract) -> Result<()> {
// or replace, as it could have been added in mempool, then block
conn.execute(
"INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
@ -74,29 +83,57 @@ fn insert_new_pool(conn: &Connection, cauldron: &ParsedContract) -> Result<()> {
cauldron.new_utxo_hash.expect("outpoint hash for new pool missing").to_hex(),
cauldron.pkh.to_hex(),
cauldron.token_id.expect("token id for new pool missing").to_hex(),
None::<String>
None::<String>,
]
).context("inserting new pool")?;
).map_err(|e| {
anyhow::anyhow!(
"failed to insert new pool. Original error: {:?}",
e
)
})?;
Ok(())
}
fn insert_pool_history_entry(
/// Next sequence number in the `pool_history_entry` table
static NEXT_SEQUENCE: AtomicI64 = AtomicI64::new(-10);
pub fn initialize_seq(conn: &Connection) {
let mut s = conn
.prepare("SELECT IFNULL(MAX(sequence), 0) + 1 FROM pool_history_entry")
.unwrap();
let mut q = s.query(params![]).unwrap();
let seq: i64 = q.next().unwrap().unwrap().get(0).unwrap();
NEXT_SEQUENCE.store(seq, std::sync::atomic::Ordering::SeqCst);
}
#[allow(dead_code)] // for unit tests
pub fn dummy_init_seq() {
if NEXT_SEQUENCE.load(std::sync::atomic::Ordering::SeqCst) < 0 {
NEXT_SEQUENCE.store(42, std::sync::atomic::Ordering::SeqCst);
}
}
pub fn insert_pool_history_entry(
conn: &Connection,
pool: &OutPointHash,
cauldron: &ParsedContract,
mtp_timestamp: Option<u64>,
first_seen_timestamp: Option<u64>,
) -> Result<()> {
let next_seq = NEXT_SEQUENCE.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
assert!(next_seq >= 0);
conn.execute(
"INSERT INTO pool_history_entry (utxo, pool, txid, tx_pos, mtp_timestamp, first_seen_timestamp)
VALUES (?, ?, ?, ?, ?, ?)
"INSERT INTO pool_history_entry (utxo, pool, txid, tx_pos, mtp_timestamp, first_seen_timestamp, sequence, sats, token_amount)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(utxo) DO UPDATE SET
pool = excluded.pool,
txid = excluded.txid,
tx_pos = excluded.tx_pos,
mtp_timestamp = COALESCE(excluded.mtp_timestamp, pool_history_entry.mtp_timestamp),
first_seen_timestamp = COALESCE(excluded.first_seen_timestamp, pool_history_entry.first_seen_timestamp)",
first_seen_timestamp = COALESCE(excluded.first_seen_timestamp, pool_history_entry.first_seen_timestamp),
sequence = excluded.sequence",
params![
cauldron
.new_utxo_hash
@ -112,9 +149,17 @@ fn insert_pool_history_entry(
.expect("utxo index of new pool history entry"),
mtp_timestamp,
first_seen_timestamp,
next_seq,
cauldron.sats,
cauldron.token_amount
],
)
.context("inserting pool_history_entry")?;
.map_err(|e| {
anyhow::anyhow!(
"failed to insert pool history entry. Original error: {:?}",
e
)
})?;
Ok(())
}
@ -196,3 +241,137 @@ pub fn update_pool_history(
Ok(())
}
/// Strategy for which pool state to select when two pool states are available for a timestamp
#[derive(PartialEq, Eq)]
enum SnapshotSelection {
UseBefore,
UseAfter,
}
/// Get pools and their state nearest to given timestamp.
fn get_nearest_entries(
conn: &Connection,
timestamp: i64,
token_id: Option<&str>,
owner_pkh: Option<&str>,
resolution: SnapshotSelection,
) -> Result<HashMap<String, PoolSnapshot>> {
let extra_filters = match (token_id.is_some(), owner_pkh.is_some()) {
(true, true) => "token_id = ?2 AND owner_pkh = ?3",
(true, false) => "token_id = ?2",
(false, true) => "owner_pkh = ?2",
(false, false) => "1=1",
};
let params = match (token_id.is_some(), owner_pkh.is_some()) {
(true, true) => params![timestamp, token_id, owner_pkh],
(true, false) => params![timestamp, token_id],
(false, true) => params![timestamp, owner_pkh],
(false, false) => params![timestamp],
};
let ts = "COALESCE(first_seen_timestamp, mtp_timestamp)";
let preferred = match resolution {
SnapshotSelection::UseBefore => "ts ASC", // Prefer the lower timestamp first
SnapshotSelection::UseAfter => "ts DESC", // Prefer the higher timestamp first
};
let query = format!(
"WITH NearestEntries AS (
-- Get the closest lower or equal to the timestamp (MAX for <= timestamp)
SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
FROM pool_history_entry phe
WHERE {ts} <= ?1
AND phe.pool IN (
SELECT creation_utxo
FROM pool
WHERE {extra_filters}
)
GROUP BY phe.pool
HAVING MAX({ts})
UNION ALL
-- Get the closest greater than the timestamp (MIN for > timestamp)
SELECT phe.pool, phe.sats, phe.token_amount, {ts} AS ts, phe.sequence
FROM pool_history_entry phe
WHERE {ts} > ?1
AND phe.pool IN (
SELECT creation_utxo
FROM pool
WHERE {extra_filters}
)
GROUP BY phe.pool
HAVING MIN({ts})
)
SELECT * FROM NearestEntries
ORDER BY {preferred}",
);
let mut stmt = conn.prepare(&query)?;
let mut rows = stmt.query(params)?;
let from_row = |row: &Row<'_>| -> Result<PoolSnapshot> {
Ok(PoolSnapshot {
pool_id: row.get(0)?,
sats: row.get(1)?,
token_amount: row.get(2)?,
timestamp: row.get(3)?,
})
};
let mut pools: HashMap<String, PoolSnapshot> = HashMap::default();
while let Some(row) = rows.next()? {
let pool_snapshot = from_row(row)?;
if !pools.contains_key(&pool_snapshot.pool_id) {
let existed = pools.insert(pool_snapshot.pool_id.clone(), pool_snapshot.clone());
assert!(existed.is_none());
}
}
Ok(pools)
}
/// Returns
pub fn get_pool_period_snapshot(
conn: &Connection,
token_id: Option<&str>,
owner_pkh: Option<&str>,
start: i64,
end: i64,
) -> Result<Vec<(PoolSnapshot, PoolSnapshot)>> {
let pools_start = get_nearest_entries(
conn,
start,
token_id,
owner_pkh,
SnapshotSelection::UseBefore,
)?;
let mut pools_end =
get_nearest_entries(conn, end, token_id, owner_pkh, SnapshotSelection::UseAfter)?;
let mut pools: Vec<(PoolSnapshot, PoolSnapshot)> = Vec::default();
for (start_pool_id, start_pool) in pools_start {
let end_pool = match pools_end.remove(&start_pool_id) {
Some(end) => end,
None => {
warn!("Found no 'end pool' for {}", start_pool_id);
continue;
}
};
let duration = end_pool.timestamp.saturating_sub(start_pool.timestamp);
if !duration.is_zero() {
pools.push((start_pool, end_pool))
}
}
if !pools_end.is_empty() {
warn!("Found {} end pools not in start pools", pools_end.len());
}
Ok(pools)
}

17
src/index.rs
View file

@ -33,6 +33,7 @@ use crate::{
DBPool, DB,
},
electrum::{electrum_fetch_mempool, electrum_get_tip, electrum_get_tx},
utiltx::ttor_sorted,
CASHTOKEN_ACTIVATION_HEIGHT, KEY_LAST_INDEXED,
};
use anyhow::{Context, Result};
@ -65,6 +66,8 @@ pub fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
)
.collect();
let txs_to_add = ttor_sorted(txs_to_add);
let mut db_conn = db.get().unwrap();
let db_tx = db_conn.transaction()?;
@ -98,8 +101,7 @@ pub fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
all_cauldrons.extend(cauldrons);
}
db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons, None, Some(current_timestamp))
.context("update pool history")?;
db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons, None, Some(current_timestamp))?;
Ok(db_tx.commit()?)
}
@ -222,7 +224,9 @@ pub fn index_blocks(
let mut all_cauldrons = vec![];
for tx in &block.txdata {
let sorted_txs = ttor_sorted(block.txdata);
for tx in &sorted_txs {
let cauldrons = parse_cauldrons(tx);
if cauldrons.is_empty() {
continue;
@ -242,22 +246,21 @@ pub fn index_blocks(
}
// Figuring out initial utxo needs to be done on all cauldrons in a block.
db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons, Some(mtp), None)
.context("update pool history")?;
db::cauldron::pool::update_pool_history(&db_tx, all_cauldrons, Some(mtp), None)?;
config_set(&db_tx, KEY_LAST_INDEXED, &blockhash.to_hex());
{
// crc20
let mut conn = db.crc20_w.get().context("failed to get crc20 db")?;
let tx = conn.transaction()?;
index_crc20(&tx, &block.txdata)?;
index_crc20(&tx, &sorted_txs)?;
tx.commit()?;
}
let autheader_updates = if bcmr_enabled {
let mut conn = db.bcmr_w.get().context("failed to get bcmr db")?;
let bcmr_db_tx = conn.transaction()?;
let updates = index_bcmr(&bcmr_db_tx, &blockhash, block.txdata)?;
let updates = index_bcmr(&bcmr_db_tx, &blockhash, sorted_txs)?;
bcmr_db_tx.commit()?;
updates as i64
} else {

5
src/main.rs
View file

@ -53,6 +53,7 @@ mod index;
mod rpc;
mod timeutil;
mod utiltoken;
mod utiltx;
fn set_panic_hook() {
panic::set_hook(Box::new(|panic_info| {
@ -142,6 +143,9 @@ fn start_program() -> Result<(DB, BCMRDownloader, WellKnownDownloader, CRC20Fetc
let genesis: Block = deserialize(&hex::decode(genesis.as_str().unwrap()).unwrap()).unwrap();
let chain = Arc::new(Mutex::new(chain::Chain::new(genesis.header)));
// initialize insert sequence for pool history
db::cauldron::pool::initialize_seq(&cauldron_db_read.get().unwrap());
info!("Loading block headers...");
let all_headers = load_all_headers(&cauldron_db_read.get().unwrap()).unwrap();
info!("Initializing {} headers...", all_headers.len());
@ -269,6 +273,7 @@ fn launch() -> _ {
rpc::price::price_at,
rpc::pool::list_pools_by_apy,
rpc::pool::list_active_pools,
rpc::apy::aggregate_apy,
rpc::contract::contract_count_token,
rpc::contract::contract_count_all,
rpc::contract::contract_volume,

37
src/rpc/apy/apyaggregator.rs Normal file
View file

@ -0,0 +1,37 @@
// 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 anyhow::Result;
use rust_decimal::Decimal;
use super::poolperiod::PoolPeriod;
pub struct APYAggregator;
impl APYAggregator {
pub fn aggregate_apy<I>(pools: I, period_start: Option<u64>) -> Result<Decimal>
where
I: Iterator<Item = PoolPeriod>,
{
let mut weighted_apy_sum = Decimal::ZERO;
let mut total_active_time = Decimal::ZERO;
for pool in pools {
let days_active = pool.duration_days(&period_start)?;
if !days_active.is_zero() {
let (_pool_yield, apy) = pool.yield_and_apy(&period_start)?;
weighted_apy_sum += apy * days_active;
total_active_time += days_active;
}
}
if total_active_time.is_zero() {
Ok(Decimal::ZERO)
} else {
Ok(weighted_apy_sum / total_active_time)
}
}
}

85
src/rpc/apy/mod.rs Normal file
View file

@ -0,0 +1,85 @@
// 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 poolperiod::PoolPeriod;
use rocket::{get, http::Status, response::status::Custom, serde::json::Json, State};
use serde_json::Value;
use crate::{
db::{cauldron::pool::get_pool_period_snapshot, DB},
timeutil::time_now,
};
use serde_json::json;
pub mod apyaggregator;
pub mod poolperiod;
#[derive(Clone, Debug)]
pub struct PoolSnapshot {
pub pool_id: String,
pub timestamp: u64,
pub sats: u64,
pub token_amount: u64,
}
impl PoolSnapshot {
#[allow(dead_code)] // used in unit tests
pub fn dummy(timestamp: u64, sats: u64, token_amount: u64) -> Self {
Self {
pool_id: "dummy".to_string(),
timestamp,
sats,
token_amount,
}
}
}
#[get("/pool/aggregated_apy?<token>&<pkh>&<start>&<end>")]
pub fn aggregate_apy(
token: Option<&str>,
pkh: Option<&str>,
start: Option<i64>, // default 30 days before end
end: Option<i64>, // default now
db: &State<DB>,
) -> Result<Json<Value>, Custom<String>> {
let end = end.unwrap_or(time_now());
let start = start.unwrap_or(end - (3600 * 24 * 30)); // 30 days
if end < start {
return Err(Custom(
Status::BadRequest,
"end time cannot be less than start time".to_string(),
));
}
if start < 0 {
return Err(Custom(
Status::BadRequest,
"start time cannot be negative".to_string(),
));
}
let conn = db
.cauldron_r
.get()
.map_err(|e| Custom(Status::InternalServerError, format!("DB error: {}", e)))?;
let pools: anyhow::Result<Vec<PoolPeriod>> =
get_pool_period_snapshot(&conn, token, pkh, start, end)
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?
.into_iter()
.map(|(start, end)| PoolPeriod::new(start, end))
.collect();
let pools = pools.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;
let pools_count = pools.len();
let apy = apyaggregator::APYAggregator::aggregate_apy(pools.into_iter(), Some(start as u64))
.map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;
Ok(Json(json!({
"apy": apy.to_string(),
"pools": pools_count,
})))
}

139
src/rpc/apy/poolperiod.rs Normal file
View file

@ -0,0 +1,139 @@
// 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 anyhow::{bail, Context, Result};
use rust_decimal::MathematicalOps;
use rust_decimal::{prelude::FromPrimitive, Decimal};
use rust_decimal_macros::dec;
use super::PoolSnapshot;
const SECONDS_IN_DAY: Decimal = dec!(86400.0);
const DAYS_IN_YEAR: Decimal = dec!(365.25);
#[derive(Debug, Clone)]
pub struct PoolPeriod {
pub start: PoolSnapshot,
pub end: PoolSnapshot,
start_k: Decimal,
end_k: Decimal,
}
impl PoolPeriod {
pub fn new(start: PoolSnapshot, end: PoolSnapshot) -> Result<Self> {
if start.timestamp > end.timestamp {
bail!(
"start timestamp ({}) > end timestamp ({})",
start.timestamp,
end.timestamp
)
}
let start_k = Decimal::from_u64(start.sats * start.token_amount)
.context("failed to convert inital_k to decimal")?;
let end_k = Decimal::from_u64(end.sats * end.token_amount)
.context("failed to convert final_k to decimal")?;
Ok(Self {
start,
end,
start_k,
end_k,
})
}
/// Duration in seconds of this pool period
/// starting: If provided; we assume that when the real starting position was BEFORE this timestamp,
/// then this was also the state of the pool at this time.
pub fn duration(&self, starting: &Option<u64>) -> u64 {
let start = match *starting {
Some(starting) => {
if starting > self.start.timestamp {
starting
} else {
self.start.timestamp
}
}
None => self.start.timestamp,
};
self.end.timestamp.saturating_sub(start)
}
pub fn durationd(&self, starting: &Option<u64>) -> Result<Decimal> {
Decimal::from_u64(self.duration(starting)).context("duration to decimal")
}
pub fn duration_days(&self, starting: &Option<u64>) -> Result<Decimal> {
Ok(self.durationd(starting)? / SECONDS_IN_DAY)
}
pub fn days_over_year(&self, starting: &Option<u64>) -> Result<Decimal> {
Ok(DAYS_IN_YEAR
.checked_div(self.duration_days(starting)?)
.unwrap_or_default())
}
pub fn pool_yield(&self) -> Result<Decimal> {
let start_k_sr = self.start_k.sqrt().context("failed to sqrt start")?;
let end_k_sr = self.end_k.sqrt().context("failed to sqrt end")?;
Ok(((end_k_sr - start_k_sr) / start_k_sr) * Decimal::ONE_HUNDRED)
}
pub fn yield_and_apy(&self, starting: &Option<u64>) -> Result<(Decimal, Decimal)> {
let pool_yield = self.pool_yield()?;
let years_elapsed = self.days_over_year(starting)?;
// to avoid powd overflow; don't calculate for pools < 6 hour old
if self.duration(starting) < 3600 * 6 {
return Ok((pool_yield, Decimal::ZERO));
}
let apy = if years_elapsed.is_zero() {
Decimal::ZERO
} else {
(((pool_yield / Decimal::ONE_HUNDRED) + Decimal::ONE)
.checked_powd(years_elapsed)
.context("powd overflow")?
- Decimal::ONE)
* Decimal::ONE_HUNDRED
};
Ok((pool_yield, apy))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_apy_and_yield() {
// http://localhost:3000/position/f67ff489fb3b8efaf5db1a2cf9e3faa07fdfd7903079262a534c921e7e7d0d2c17
// APY: 0.2815061492818 numberOfDaysInPeriod: 28.668907708332494 first timestamp 1727201259 last trade 1729678252.626 first sats 1648241n first tokens 116674414n last sats 1759576n last tokens 109340225n yield 0.02206714530974295
// compare numbers to our existing implementation
let start = PoolSnapshot::dummy(1727201259, 1648241, 116674414);
let end = PoolSnapshot::dummy(1729678252, 1759576, 109340225);
let expected_yield = dec!(0.02206714530974295);
let expected_apy = dec!(0.2815061492818);
let period = PoolPeriod::new(start, end).unwrap();
let (pool_yield, pool_apy) = period.yield_and_apy(&None).unwrap();
assert!((expected_yield - pool_yield).abs() < dec!(1e-12));
assert!(
(expected_apy - pool_apy).abs() < dec!(1e-7),
"expected {} != actual {}",
expected_apy,
pool_apy
);
}
}

1
src/rpc/mod.rs
View file

@ -16,6 +16,7 @@ use crate::{
db::bcmr::{get_token_bcmr, get_well_known_bcmr},
};
pub mod apy;
pub mod bcmr;
pub mod contract;
pub mod pool;

371
src/rpc/price.rs
View file

@ -353,141 +353,179 @@ pub fn price_history(
#[cfg(test)]
mod tests {
use crate::db::cauldron::{pool, tx, utxo_funding};
use crate::db::cauldron::{
pool::{
self, dummy_init_seq, flag_as_withdrawn, insert_new_pool, insert_pool_history_entry,
},
tx::{self, insert_block_tx, insert_mempool_tx},
utxo_funding::{self, insert_utxo_funding},
};
use super::*;
use bitcoin_hashes::Hash;
use bitcoincash::{BlockHash, PubkeyHash, Txid};
use r2d2::Pool;
use r2d2_sqlite::SqliteConnectionManager;
use riftenlabs_defi::{cauldron::ParsedContract, chainutil::OutPointHash};
use rocket::http::Status;
use rocket::local::blocking::Client;
use rocket::routes;
use rusqlite::{params, Connection};
use rusqlite::Connection;
use std::sync::Arc;
fn setup_mock_db(connection: &Connection) {
utxo_funding::create_table(connection);
tx::create_table(connection);
pool::create_table(connection);
const TIME_1: u64 = 1727963300;
const TIME_2: u64 = 1727963350;
const TIME_3: u64 = 1727963400;
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) {
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();
let txid1 = Txid::from_inner([0xf0; 32]);
let txid2 = Txid::from_inner([0xf1; 32]);
let txid3 = Txid::from_inner([0xf2; 32]);
let utxo1 = OutPointHash::from_inner([0xe0; 32]);
let utxo2 = OutPointHash::from_inner([0xe1; 32]);
let utxo3 = OutPointHash::from_inner([0xe2; 32]);
// Insert mock data into utxo_funding
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params!["mock_utxo_hash1", "test_txid1", 50000, 1000, "0000000000000000000000000000000000000000000000000000000000000000"],
)
.expect("Failed to insert test data 1 into utxo_funding");
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params!["mock_utxo_hash2", "test_txid2", 60000, 2000, "0000000000000000000000000000000000000000000000000000000000000000"],
)
.expect("Failed to insert test data 2 into utxo_funding");
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params!["mock_utxo_hash3", "test_txid3", 90000, 3000, "0000000000000000000000000000000000000000000000000000000000000000"],
)
.expect("Failed to insert test data 3 into utxo_funding");
let cauldron1 = dummy_cauldron(&txid1, &utxo1, &token_zero, 50000, 1000, &pkh_zero);
let cauldron2 = dummy_cauldron(&txid2, &utxo2, &token_zero, 60000, 2000, &pkh_zero);
let cauldron3 = dummy_cauldron(&txid2, &utxo3, &token_zero, 90000, 3000, &pkh_zero);
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 mock data into tx table
connection
.execute(
"INSERT INTO tx (txid, first_seen_timestamp, mtp_timestamp) VALUES (?, ?, ?)",
params!["test_txid1", 1727963300, 1727963300],
)
.expect("Failed to insert test data 1 into tx");
connection
.execute(
"INSERT INTO tx (txid, first_seen_timestamp, mtp_timestamp) VALUES (?, ?, ?)",
params!["test_txid2", 1727963350, 1727963350],
)
.expect("Failed to insert test data 2 into tx");
connection
.execute(
"INSERT INTO tx (txid, first_seen_timestamp, mtp_timestamp) VALUES (?, ?, ?)",
params!["test_txid3", 1727963400, 1727963400],
)
.expect("Failed to insert test data 3 into tx");
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 mock data into `pool_history_entry`
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params!["pool1", "mock_utxo_hash1", "test_txid1", "tx_pos1", 1727963300, 1727963300],
)
.expect("Failed to insert test data 1 into pool_history_entry");
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params!["pool2", "mock_utxo_hash2", "test_txid2", "tx_pos2", 1727963350, 1727963350],
)
.expect("Failed to insert test data 2 into pool_history_entry");
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params!["pool3", "mock_utxo_hash3", "test_txid3", "tx_pos3", 1727963400, 1727963400],
)
.expect("Failed to insert test data 3 into 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 txid1_newer = Txid::from_inner([0xf3; 32]);
let utxo1_newer = OutPointHash::from_inner([0xe3; 32]);
insert_pool_history_entry(&conn, &pool1, &cauldron1, Some(TIME_1), Some(TIME_1)).unwrap();
insert_pool_history_entry(&conn, &pool2, &cauldron2, Some(TIME_2), Some(TIME_2)).unwrap();
insert_pool_history_entry(&conn, &pool3, &cauldron3, Some(TIME_3), Some(TIME_3)).unwrap();
// Insert newer entry for pool1
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params!["pool1", "mock_utxo_hash1_newer", "test_txid1_newer", "tx_pos1_newer", 1727963500, 1727963500],
let cauldron1_newer = dummy_cauldron(
&txid1_newer,
&utxo1_newer,
&token_zero,
70000,
1500,
&pkh_zero,
);
insert_pool_history_entry(
&conn,
&pool1,
&cauldron1_newer,
Some(1727963500),
Some(1727963500),
)
.expect("Failed to insert newer test data into pool_history_entry");
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params!["mock_utxo_hash1_newer", "test_txid1_newer", 70000, 1500, "0000000000000000000000000000000000000000000000000000000000000000"],
)
.expect("Failed to insert newer test data into utxo_funding");
.unwrap();
insert_utxo_funding(&conn, &vec![cauldron1_newer], &txid1_newer, true).unwrap();
// Insert active pools with required columns (owner_pkh and token_id)
connection
.execute(
"INSERT INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
params!["pool1", "dummy_owner_pkh1", "dummy_token_id1", Option::<String>::None], // Active pool
let token1 = TokenID::from_inner([0xda; 32]);
let token2 = TokenID::from_inner([0xdb; 32]);
let token3 = TokenID::from_inner([0xdc; 32]);
let pkh1 = PubkeyHash::from_inner([0xca; 20]);
let pkh2 = PubkeyHash::from_inner([0xcb; 20]);
let pkh3 = PubkeyHash::from_inner([0xcc; 20]);
insert_new_pool(
&conn,
&dummy_cauldron(&Txid::all_zeros(), &pool1, &token1, 0, 0, &pkh1),
)
.expect("Failed to insert test data 1 into pool");
connection
.execute(
"INSERT INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
params!["pool2", "dummy_owner_pkh2", "dummy_token_id2", Option::<String>::None], // Active pool
.unwrap();
insert_new_pool(
&conn,
&dummy_cauldron(&Txid::all_zeros(), &pool2, &token2, 0, 0, &pkh2),
)
.expect("Failed to insert test data 2 into pool");
connection
.execute(
"INSERT INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
params!["pool3", "dummy_owner_pkh3", "dummy_token_id3", Option::<String>::None], // Active pool
.unwrap();
insert_new_pool(
&conn,
&dummy_cauldron(&Txid::all_zeros(), &pool3, &token3, 0, 0, &pkh3),
)
.expect("Failed to insert test data 3 into pool");
.unwrap();
// Insert an inactive pool
connection
.execute(
"INSERT INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)",
params!["inactive_pool", "dummy_owner_pkh_inactive", "dummy_token_id_inactive", "inactive_utxo"], // Inactive pool
)
.expect("Failed to insert inactive pool");
let inactive_pool = OutPointHash::from_inner([0xaa; 32]);
//let inactive_withdrawn_utxo = OutPointHash::from_inner([0xab; 32]);
let inactive_token = TokenID::from_inner([0xac; 32]);
let inactive_pkh = PubkeyHash::from_inner([0xac; 20]);
let inactive_txid = Txid::from_inner([0xad; 32]);
let inactive_cauldron = dummy_cauldron(
&inactive_txid,
&inactive_pool,
&inactive_token,
80000,
2500,
&inactive_pkh,
);
let inactive_cauldron_withdraw = ParsedContract {
pkh: inactive_pkh,
is_withdrawn: true,
spent_utxo_hash: inactive_pool,
new_utxo_hash: None,
new_utxo_txid: None,
new_utxo_n: None,
token_id: None,
sats: None,
token_amount: None,
};
// Insert corresponding utxo_funding for the inactive pool
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params!["inactive_utxo", "inactive_txid", 80000, 2500, "0000000000000000000000000000000000000000000000000000000000000000"],
insert_new_pool(&conn, &inactive_cauldron).unwrap();
insert_utxo_funding(
&conn,
&vec![inactive_cauldron.clone()],
&inactive_txid,
true,
)
.expect("Failed to insert inactive utxo_funding");
// Insert pool_history_entry for the inactive pool
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params!["inactive_pool", "inactive_utxo", "inactive_txid", "tx_pos_inactive", 1727963350, 1727963350],
.unwrap();
insert_pool_history_entry(
&conn,
&inactive_pool,
&inactive_cauldron,
Some(1727963350),
Some(1727963350),
)
.expect("Failed to insert inactive pool_history_entry");
.unwrap();
flag_as_withdrawn(&conn, &inactive_pool, &inactive_cauldron_withdraw).unwrap();
}
// Mock function to create a DB pool
@ -520,7 +558,7 @@ mod tests {
let token_id = TokenID::all_zeros().to_hex();
// Test 1: Querying at a timestamp that exactly matches test_txid1
let timestamp = 1727963300;
let timestamp = TIME_1;
let response = client
.get(format!("/cauldron/price/{}/at/{}", token_id, timestamp))
.dispatch();
@ -537,7 +575,7 @@ mod tests {
assert!((actual_price - expected_price).abs() < 0.01);
// Test 2: Querying at a timestamp that includes Pool 1 and Pool 2
let timestamp = 1727963350;
let timestamp = TIME_2;
let response = client
.get(format!("/cauldron/price/{}/at/{}", token_id, timestamp))
.dispatch();
@ -550,10 +588,15 @@ mod tests {
// The combined price should be 36.67 (rounded)
let expected_price = 36.67;
assert!((actual_price - expected_price).abs() < 0.01);
assert!(
(actual_price - expected_price).abs() < 0.01,
"expected {} != actual {}",
expected_price,
actual_price
);
// Test 3: Querying a timestamp that includes Pool 1, Pool 2, and Pool 3
let timestamp = 1727963400;
let timestamp = TIME_3;
let response = client
.get(format!("/cauldron/price/{}/at/{}", token_id, timestamp))
.dispatch();
@ -596,7 +639,12 @@ mod tests {
// Expected price based on the total from pool1, pool2, and pool3
let expected_price = 33.85; // Rounded to 2 decimal places
assert!((actual_price - expected_price).abs() < 0.01);
assert!(
(actual_price - expected_price).abs() < 0.01,
"expected {} != actual {}",
expected_price,
actual_price
);
}
#[test]
@ -612,25 +660,25 @@ mod tests {
let token_id = TokenID::all_zeros().to_hex();
// Insert additional entries with the same timestamp for an existing pool (e.g., pool1)
let connection = mock_db.cauldron_r.get().expect("Failed to get connection.");
let conn = mock_db.cauldron_w.get().expect("Failed to get connection.");
// These entries should have the same timestamp as previous mockdata and be counted in the price calculation
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params!["pool1", "mock_utxo_hash1_extra", "test_txid1_extra", "0", 1727963300, 1727963300],
)
.expect("Failed to insert extra test data into pool_history_entry");
let pool1 = OutPointHash::from_inner([0x0a; 32]);
let txid = Txid::hash("txid_extra".as_bytes());
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params!["mock_utxo_hash1_extra", "test_txid1_extra", 40000, 500, "0000000000000000000000000000000000000000000000000000000000000000"],
)
.expect("Failed to insert extra test data into utxo_funding");
let cauldron = dummy_cauldron(
&txid,
&OutPointHash::hash("utxo_extra".as_bytes()),
&TokenID::all_zeros(),
40000,
500,
&PubkeyHash::all_zeros(),
);
insert_pool_history_entry(&conn, &pool1, &cauldron, Some(TIME_1), Some(TIME_1)).unwrap();
insert_utxo_funding(&conn, &vec![cauldron], &txid, true).unwrap();
// Test: Query at the timestamp matching test_txid1 (1727963300) and check the price
let timestamp = 1727963300;
let timestamp = TIME_1;
let response = client
.get(format!("/cauldron/price/{}/at/{}", token_id, timestamp))
.dispatch();
@ -727,69 +775,28 @@ mod tests {
let client = Client::tracked(rocket).expect("valid rocket instance");
let connection = mock_db.cauldron_r.get().expect("Failed to get connection.");
// Setup mock data for this specific test
connection
.execute(
"CREATE TABLE IF NOT EXISTS utxo_funding (
new_utxo_hash TEXT PRIMARY KEY,
txid TEXT,
sats BIGINT,
token_amount BIGINT,
token_id TEXT
)",
[],
)
.expect("Failed to create utxo_funding table");
connection
.execute(
"CREATE TABLE IF NOT EXISTS pool_history_entry (
pool TEXT,
utxo TEXT PRIMARY KEY,
txid TEXT,
tx_pos TEXT,
mtp_timestamp BIGINT,
first_seen_timestamp BIGINT
)",
[],
)
.expect("Failed to create pool_history_entry table");
let conn = mock_db.cauldron_w.get().expect("Failed to get connection.");
// Insert high tokens and low sats
connection
.execute(
"INSERT INTO utxo_funding (new_utxo_hash, txid, sats, token_amount, token_id) VALUES (?, ?, ?, ?, ?)",
params![
"mock_utxo_hash_high_tokens",
"test_txid_high_tokens",
1_i64, // Low sats
9999999999999999_i64, // Very high token amount
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
],
)
.expect("Failed to insert high-token data into utxo_funding");
let pool = OutPointHash::hash("many tokens pool".as_bytes());
let txid = Txid::hash("many tokens txid".as_bytes());
let utxo = OutPointHash::hash("many tokens utxo".as_bytes());
let token = TokenID::hash("many tokens tokenid".as_bytes());
connection
.execute(
"INSERT INTO pool_history_entry (pool, utxo, txid, tx_pos, mtp_timestamp, first_seen_timestamp) VALUES (?, ?, ?, ?, ?, ?)",
params![
"pool_high_tokens",
"mock_utxo_hash_high_tokens",
"test_txid_high_tokens",
"tx_pos_high_tokens",
1727963300,
1727963300
],
)
.expect("Failed to insert high-token data into pool_history_entry");
let cauldron = dummy_cauldron(
&txid,
&utxo,
&token,
1, /* low sats */
9999999999999999, /* many tokens */
&PubkeyHash::all_zeros(),
);
insert_utxo_funding(&conn, &vec![cauldron.clone()], &txid, true).unwrap();
insert_pool_history_entry(&conn, &pool, &cauldron, Some(TIME_1), Some(TIME_1)).unwrap();
// Test: Querying the price with high tokens and low sats
let token_id = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let timestamp = 1727963300;
let response = client
.get(format!("/cauldron/price/{}/at/{}", token_id, timestamp))
.get(format!("/cauldron/price/{}/at/{}", token.to_hex(), TIME_1))
.dispatch();
// Expect that we could not handle the calculation/conversion.

41
src/utiltx.rs Normal file
View file

@ -0,0 +1,41 @@
// 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 std::collections::{HashSet, VecDeque};
use bitcoincash::{Transaction, Txid};
use rayon::prelude::*;
// TTOR sort a list of transactions
pub fn ttor_sorted(txs: Vec<Transaction>) -> Vec<Transaction> {
let txs = {
let mut queue: VecDeque<Transaction> = txs.into_iter().collect();
let mut queue_txids: HashSet<Txid> = queue.par_iter().map(|tx| tx.txid()).collect();
let mut txs: Vec<Transaction> = Vec::with_capacity(queue.len());
while let Some(tx) = queue.pop_front() {
let mut has_parent = false;
for i in &tx.input {
if queue_txids.contains(&i.previous_output.txid) {
// depends on parent
has_parent = true;
break;
};
}
if has_parent {
queue.push_back(tx);
} else {
queue_txids.remove(&tx.txid());
txs.push(tx)
}
}
txs
};
txs
}