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

Merge branch 'candlestick-robust-pricing' into 'master'

Browse source 
Price candlesticks by gross volume instead of net deltas

See merge request riftenlabs/riftenlabs-indexer!95
This commit is contained in:
jakobsn 2026-07-30 11:35:02 +00:00
commit 049a19dc72
7 changed files with 1911 additions and 242 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

396
src/db/cauldron/candlestick/mod.rs
View file

@ -10,6 +10,16 @@ use bitcoincash::TokenID;
use serde::Serialize; use serde::Serialize;
use sqlx::{Row, SqlitePool}; use sqlx::{Row, SqlitePool};
pub use self::policy::{GuardParams, Leg, Policy, PoolReserves};
/// Interpret a database blob as a 32-byte hash, erroring rather than panicking on a
/// short or oversized value.
pub(crate) fn to_hash32(bytes: &[u8]) -> Result<[u8; 32]> {
bytes
.try_into()
.map_err(|_| anyhow::anyhow!("expected a 32-byte hash, got {} bytes", bytes.len()))
}
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
pub struct CandlestickData { pub struct CandlestickData {
pub time: i64, // start of the interval pub time: i64, // start of the interval
@ -71,14 +81,16 @@ impl PriceInterval {
} }
fn aggregate_raw_trades( fn aggregate_raw_trades(
all_trades: &[(i64, i64, i64, i64, i64)], all_legs: &[Leg],
intervals: Vec<PriceInterval>, intervals: Vec<PriceInterval>,
step_size: i64, step_size: i64,
mut found_first_trade: bool, mut found_first_trade: bool,
mut last_close_price: Option<f64>, mut last_close_price: Option<f64>,
policy: &mut Policy,
) -> (Vec<CandlestickData>, bool, Option<f64>) { ) -> (Vec<CandlestickData>, bool, Option<f64>) {
let mut result = Vec::with_capacity(intervals.len()); let mut result = Vec::with_capacity(intervals.len());
let mut trade_index = 0; let mut leg_index = 0;
let mut txid_set = std::collections::HashSet::new();
for interval in intervals { for interval in intervals {
let interval_start = interval.start; let interval_start = interval.start;
@ -86,38 +98,51 @@ fn aggregate_raw_trades(
let mut pi = PriceInterval::new(interval_start, step_size); let mut pi = PriceInterval::new(interval_start, step_size);
let mut first_trade_in_interval = true; let mut first_trade_in_interval = true;
while trade_index < all_trades.len() { while leg_index < all_legs.len() {
let (ts, signed_sats, signed_tokens, vol_sats, vol_tokens) = all_trades[trade_index]; let leg = &all_legs[leg_index];
if ts < interval_start { if leg.ts < interval_start {
trade_index += 1; leg_index += 1;
continue; continue;
} }
if ts >= interval_end { if leg.ts >= interval_end {
break; break;
} }
if signed_tokens != 0 { let judge = policy.judge(leg);
let price = (signed_sats as f64 / signed_tokens as f64).abs(); let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
if first_trade_in_interval { let token_delta_abs = leg.token_delta.unsigned_abs() as i64;
pi.open = Some(price);
pi.high = price; // Always count volume, regardless of acceptance.
pi.low = price; pi.volume_sats += sats_delta_abs;
first_trade_in_interval = false; pi.volume_tokens += token_delta_abs;
} txid_set.insert(leg.txid);
pi.close = Some(price);
if price.is_finite() { // Update OHLC only if accepted and priceable.
pi.high = pi.high.max(price); if judge.accepted() {
pi.low = pi.low.min(price); if let Some(price) = judge.price {
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; policy.apply(leg);
pi.volume_tokens += vol_tokens; leg_index += 1;
pi.transaction_count += 1;
trade_index += 1;
} }
// Carry forward last close when volume exists but net tokens are zero. // Set tx_count to unique txids in this interval.
pi.transaction_count = txid_set.len() as i64;
txid_set.clear();
// Carry forward last close when no accepted prints in this interval.
if pi.transaction_count > 0 && (pi.open.is_none() || pi.close.is_none()) { if pi.transaction_count > 0 && (pi.open.is_none() || pi.close.is_none()) {
if let Some(prev) = last_close_price { if let Some(prev) = last_close_price {
if pi.open.is_none() { if pi.open.is_none() {
@ -206,59 +231,29 @@ fn fill_ohlcv_candles(
(result, found_first_trade, last_close) (result, found_first_trade, last_close)
} }
async fn fetch_raw_trades( /// Returns per-leg data for policy evaluation.
/// Each row is: (txid, effective_timestamp, pool, sats_delta, token_delta, sats, token_amount, sequence)
async fn fetch_raw_legs(
pool: &SqlitePool, pool: &SqlitePool,
token_blob: &[u8], token_blob: &[u8],
timestamp_start: i64, timestamp_start: i64,
timestamp_end: i64, timestamp_end: i64,
) -> Result<Vec<(i64, i64, i64, i64, i64)>> { ) -> Result<Vec<Leg>> {
let sql = r#" 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 SELECT
effective_timestamp, phe.txid,
signed_sats, phe.effective_timestamp,
signed_tokens, phe.pool,
volume_sats, phe.sats_delta,
volume_tokens phe.token_delta,
FROM tx_trades phe.sats,
ORDER BY effective_timestamp ASC, min_sequence ASC; phe.token_amount,
phe.sequence
FROM pool_history_entry AS phe
WHERE phe.token_id = ?
AND phe.effective_timestamp >= ?
AND phe.effective_timestamp < ?
ORDER BY phe.effective_timestamp ASC, phe.sequence ASC;
"#; "#;
let rows = sqlx::query(sql) let rows = sqlx::query(sql)
.bind(token_blob) .bind(token_blob)
@ -267,67 +262,191 @@ ORDER BY effective_timestamp ASC, min_sequence ASC;
.fetch_all(pool) .fetch_all(pool)
.await?; .await?;
Ok(rows rows.into_iter()
.into_iter() .map(|r| {
.map(|r| (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4))) let txid: Vec<u8> = r.get(0);
.collect()) let pool_bytes: Vec<u8> = r.get(2);
Ok(Leg {
txid: to_hash32(&txid)?,
pool: to_hash32(&pool_bytes)?,
ts: r.get(1),
sequence: r.get(7),
sats_delta: r.get(3),
token_delta: r.get(4),
sats: r.get(5),
token_amount: r.get(6),
})
})
.collect()
} }
/// Returns the close price of the most recent priceable trade strictly before /// Pool reserves for `token_blob` as they stood immediately before `timestamp`.
/// `timestamp_end`, using the same per-tx aggregation as `fetch_raw_trades`. ///
/// Returns `None` when no prior trade exists (new token, no history). /// Seeding a [`Policy`] with this is what makes its verdicts independent of the caller's
/// window: the raw path, a rebuild batch, and a 1-week versus 1-month request all start
/// from the same chain state rather than from whichever legs happened to be in range.
///
/// Enumerates the token's pools and seeks each one's latest history row, rather than
/// scanning every row the token ever produced — a busy token has hundreds of thousands,
/// and this runs once per query.
///
/// Pools already withdrawn at `timestamp` are excluded. A withdrawal leaves no history
/// row behind, so a drained pool's last entry still shows its full pre-withdrawal
/// reserves; counting those would let a pool that no longer holds anything go on voting
/// the reference. Pools withdrawn *after* `timestamp` are kept, since they were live then
/// — the same rule `poolvisitor` applies.
///
/// `confirmed_only` must match whatever the caller's leg query does. `rebuild_range` folds
/// only confirmed legs, so seeding it from reserves that include mempool rows would make
/// materialised output a function of mempool contents at rebuild time — and `INSERT OR
/// IGNORE` would then freeze whichever run happened first. The live query paths do not
/// filter their legs, so they pass `false`.
pub(crate) async fn fetch_reserve_snapshot(
pool: &SqlitePool,
token_blob: &[u8],
timestamp: i64,
confirmed_only: bool,
) -> Result<Vec<PoolReserves>> {
// Interpolated, not bound: a constant fragment keeps the query planner able to use
// idx_pool_history_entry_pool_timestamp instead of evaluating a flag per row.
let confirmed_join = if confirmed_only {
"JOIN tx AS prior_tx ON prior_tx.txid = prior.txid AND prior_tx.blockhash IS NOT NULL"
} else {
""
};
let sql = format!(
r#"
SELECT phe.pool, phe.sequence, phe.sats, phe.token_amount
FROM pool AS p
JOIN pool_history_entry AS phe ON phe.utxo = (
SELECT prior.utxo
FROM pool_history_entry AS prior
{confirmed_join}
WHERE prior.pool = p.creation_utxo
AND prior.effective_timestamp < ?
ORDER BY prior.effective_timestamp DESC, prior.sequence DESC
LIMIT 1
)
WHERE p.token_id = ?
AND (p.withdrawn_in_utxo IS NULL OR (
SELECT t.effective_timestamp
FROM utxo_spending AS us
JOIN tx AS t ON us.txid = t.txid
WHERE us.spent_utxo_hash = p.withdrawn_in_utxo
) >= ?);
"#
);
let rows = sqlx::query(&sql)
.bind(timestamp)
.bind(token_blob)
.bind(timestamp)
.fetch_all(pool)
.await?;
rows.into_iter()
.map(|r| {
let pool_bytes: Vec<u8> = r.get(0);
Ok(PoolReserves {
pool: to_hash32(&pool_bytes)?,
sequence: r.get(1),
sats: r.get(2),
token_amount: r.get(3),
})
})
.collect()
}
/// A price to carry into a window that opens on silence, so that requesting 1W and 1M
/// yields the same candles over the period they share.
///
/// Prefers the most recent accepted print strictly before `timestamp_end`. Falls back to
/// the reference implied by pool reserves when the token's last activity was entirely
/// muted or unpriceable — reserves still say what the market was, and a carried price
/// beats a hole in the chart.
///
/// Returns `None` only when the token has no history at all before `timestamp_end`.
///
/// The lookback is deliberately unbounded. An earlier revision capped the scan at 24
/// hours, which silently dropped the seed for any token trading less often than daily.
async fn fetch_last_close_before( async fn fetch_last_close_before(
pool: &SqlitePool, pool: &SqlitePool,
token_blob: &[u8], token_blob: &[u8],
timestamp_end: i64, timestamp_end: i64,
) -> Result<Option<f64>> { ) -> Result<Option<f64>> {
let sql = r#" // Locate the token's most recent activity first. Trusting `ohlcv_1h` before knowing
WITH per_pool_tx_raw AS ( // this would let a stale bucket shadow newer legs: the table is structurally at least
SELECT // three hours behind the tip, and is empty for the whole of a post-version-bump
phe.txid, // rebuild, so "newest materialised bucket" is routinely far older than the real last
phe.effective_timestamp, // print.
phe.utxo, let last_activity: Option<i64> = sqlx::query_scalar(
phe.sats_delta, "SELECT MAX(effective_timestamp) FROM pool_history_entry
phe.token_delta, WHERE token_id = ? AND effective_timestamp < ?",
phe.sequence )
FROM pool_history_entry AS phe .bind(token_blob)
WHERE phe.token_id = ? .bind(timestamp_end)
AND phe.effective_timestamp < ? .fetch_optional(pool)
), .await?
per_pool_tx AS ( .flatten();
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))) let Some(last_activity) = last_activity else {
return Ok(None);
};
let scan_start = (last_activity / 3600) * 3600;
// The last accepted print lives in the hour containing that activity. If that exact
// hour is materialised and closes at or before the cutoff, its close is the answer.
//
// The `scan_start + 3600 <= timestamp_end` test matters: a bucket straddling the cutoff
// closes on a print that happened *after* the instant being asked about, which an
// earlier revision leaked for any non-hour-aligned `timestamp_end`.
if scan_start + 3600 <= timestamp_end {
let exact: Option<f64> =
sqlx::query_scalar("SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?")
.bind(token_blob)
.bind(scan_start)
.fetch_optional(pool)
.await?;
if exact.is_some() {
return Ok(exact);
}
}
// Replay that hour through the guard. Only one hour is needed, because the policy is
// seeded from reserves rather than rebuilt by folding all of history.
let snapshot = fetch_reserve_snapshot(pool, token_blob, scan_start, false).await?;
let mut policy = Policy::seeded(GuardParams::default(), &snapshot);
let legs = fetch_raw_legs(pool, token_blob, scan_start, timestamp_end).await?;
let mut last_accepted_price = None;
for leg in &legs {
let judge = policy.judge(leg);
if judge.accepted() {
last_accepted_price = judge.price;
}
policy.apply(leg);
}
if last_accepted_price.is_some() {
return Ok(last_accepted_price);
}
// That hour was entirely muted or unpriceable. Reach further back through the
// materialised table — unbounded, but only now that it cannot shadow newer legs.
let earlier: Option<f64> = sqlx::query_scalar(
"SELECT close FROM ohlcv_1h
WHERE token_id = ? AND bucket_ts + 3600 <= ?
ORDER BY bucket_ts DESC LIMIT 1",
)
.bind(token_blob)
.bind(timestamp_end)
.fetch_optional(pool)
.await?;
// Last resort: the price implied by current reserves. Not a executed print, but a
// carried price beats a hole in the chart.
Ok(earlier.or_else(|| policy.reference()))
} }
/// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`. /// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`.
@ -345,6 +464,7 @@ pub async fn candlesticks(
} }
let token_blob = display_hex_to_blob::<TokenID>(token_id)?; let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
let params = GuardParams::default();
// Seed gap-fill with the last known close price before this window so that // 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 // switching between timeframes (e.g. 1W vs 1M) produces consistent prices
@ -374,7 +494,14 @@ pub async fn candlesticks(
if ohlcv_end < timestamp_end { if ohlcv_end < timestamp_end {
// Tail: query raw for [ohlcv_end, timestamp_end) and append. // Tail: query raw for [ohlcv_end, timestamp_end) and append.
let raw_trades = fetch_raw_trades(pool, &token_blob, ohlcv_end, timestamp_end).await?; //
// The policy is seeded at `ohlcv_end`, not at `timestamp_start`: the buckets
// before it came from `ohlcv_1h` and never passed through this policy, so a
// policy built from the window's start would enter the tail believing the
// token had no history and wave its first leg through unjudged.
let snapshot = fetch_reserve_snapshot(pool, &token_blob, ohlcv_end, false).await?;
let mut policy = Policy::seeded(params, &snapshot);
let raw_legs = fetch_raw_legs(pool, &token_blob, ohlcv_end, timestamp_end).await?;
let mut tail_intervals = Vec::new(); let mut tail_intervals = Vec::new();
let mut t = ohlcv_end; let mut t = ohlcv_end;
@ -384,11 +511,12 @@ pub async fn candlesticks(
} }
let (tail, _, _) = aggregate_raw_trades( let (tail, _, _) = aggregate_raw_trades(
&raw_trades, &raw_legs,
tail_intervals, tail_intervals,
step_size, step_size,
found_first, found_first,
last_close, last_close,
&mut policy,
); );
result.extend(tail); result.extend(tail);
} }
@ -404,12 +532,22 @@ pub async fn candlesticks(
current_start += step_size; current_start += step_size;
} }
let all_trades = fetch_raw_trades(pool, &token_blob, timestamp_start, timestamp_end).await?; let snapshot = fetch_reserve_snapshot(pool, &token_blob, timestamp_start, false).await?;
let mut policy = Policy::seeded(params, &snapshot);
let all_legs = fetch_raw_legs(pool, &token_blob, timestamp_start, timestamp_end).await?;
let (result, _, _) = let (result, _, _) = aggregate_raw_trades(
aggregate_raw_trades(&all_trades, intervals, step_size, seed_found, seed_close); &all_legs,
intervals,
step_size,
seed_found,
seed_close,
&mut policy,
);
Ok(result) Ok(result)
} }
pub mod policy;
#[cfg(test)] #[cfg(test)]
mod tests; mod tests;

412
src/db/cauldron/candlestick/policy.rs Normal file
View file

@ -0,0 +1,412 @@
// 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 std::cmp::Ordering;
use std::collections::HashMap;
/// Pools holding fewer tokens than this are dust and do not vote on the reference.
const MIN_TOKEN_RESERVE: u64 = 10;
/// A leg of a transaction: one pool's change in a single transaction.
#[derive(Clone, Debug)]
pub struct Leg {
pub txid: [u8; 32],
pub pool: [u8; 32],
pub ts: i64,
pub sequence: i64,
pub sats_delta: i64, // sats change at this pool
pub token_delta: i64, // token change at this pool
pub sats: u64, // post-change sats (reserves)
pub token_amount: u64, // post-change token_amount (reserves)
}
/// One pool's reserves at an instant, used to seed a [`Policy`].
///
/// Seeding is what makes the policy's verdicts a function of the chain rather than of
/// where the caller happened to start reading: a request for one hour and a request for
/// one month both begin from the same reserve state and so agree on the overlap.
#[derive(Clone, Debug)]
pub struct PoolReserves {
pub pool: [u8; 32],
pub sequence: i64,
pub sats: u64,
pub token_amount: u64,
}
/// Why a leg was accepted or muted.
///
/// Deliberately allocation-free: `judge` runs once per leg, and a full `ohlcv_1h` rebuild
/// walks millions of them.
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum Verdict {
/// No token moved, or no sats moved, so the leg has no price.
Unpriceable,
/// No pool held usable reserves, so there is nothing to deviate from.
NoReference,
/// Within `max_deviation_factor` of the reference.
WithinBand { dev: f64 },
/// Outside the band, but the pool holds enough qualification credit to print anyway.
CreditExempt {
dev: f64,
credit: u64,
largest_credit: u64,
},
/// Outside the band with too little credit: counted in volume, excluded from OHLC.
Muted {
dev: f64,
credit: u64,
largest_credit: u64,
},
}
impl Verdict {
pub fn accepted(self) -> bool {
matches!(
self,
Verdict::NoReference | Verdict::WithinBand { .. } | Verdict::CreditExempt { .. }
)
}
}
/// Outcome of judging a leg.
#[derive(Clone, Copy, Debug)]
pub struct JudgeResult {
/// The leg's executed price, or `None` when it has none.
pub price: Option<f64>,
pub verdict: Verdict,
}
impl JudgeResult {
pub fn accepted(&self) -> bool {
self.verdict.accepted()
}
}
#[derive(Clone, Copy, Debug)]
pub struct GuardParams {
/// `F`: a leg further than this factor from the reference needs credit to print.
pub max_deviation_factor: f64,
/// `q`: credit needed to qualify, as a fraction of the largest pool's credit.
pub min_share_fraction: f64,
}
impl Default for GuardParams {
/// Calibrated against a production replay: `F = 5` sits well above the worst real
/// non-qualifying print and well below the weakest observed attack print.
///
/// Single source of truth for every call site until these are threaded through
/// `configure_me`; changing them requires an `OHLCV_VERSION` bump.
fn default() -> Self {
Self {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
}
}
}
/// Per-pool state: reserves only.
///
/// Qualification credit is *derived* from these on demand (see [`credit_from`]) rather than
/// stored, which is what keeps a policy's verdicts a function of the reserves it holds
/// rather than of the order it learned them in.
#[derive(Clone, Debug)]
struct PoolState {
sequence: i64,
sats: u64,
token_amount: u64,
}
/// The guard: judges each leg against a reference price derived from pool reserves.
///
/// Reserves are advanced by [`Policy::apply`] as legs stream past, but the starting
/// state comes from a snapshot (see [`Policy::seeded`]) rather than from whatever legs
/// happen to fall inside the caller's window. That is what keeps a bucket's verdict
/// identical whether it was reached through the raw path, through a rebuild batch, or
/// through a request for a different timeframe.
pub struct Policy {
pools: HashMap<[u8; 32], PoolState>,
reference: Option<f64>,
params: GuardParams,
/// Reused across `recompute_reference` calls so the hot path does not allocate.
scratch: Vec<(f64, u64, u64)>,
}
impl Policy {
/// A policy with no prior knowledge. Only correct when the caller genuinely starts
/// at the token's first trade; otherwise use [`Policy::seeded`].
pub fn new(params: GuardParams) -> Self {
Self::seeded(params, &[])
}
/// A policy primed with pool reserves as of some instant.
///
/// Nothing beyond reserves needs priming: credit is derived from them (see
/// [`credit_from`]), so a seeded policy and one that folded its way to the same
/// reserves are indistinguishable by construction.
pub fn seeded(params: GuardParams, snapshot: &[PoolReserves]) -> Self {
let mut policy = Self {
pools: HashMap::with_capacity(snapshot.len()),
reference: None,
params,
scratch: Vec::with_capacity(snapshot.len()),
};
for r in snapshot {
policy.pools.insert(
r.pool,
PoolState {
sequence: r.sequence,
sats: r.sats,
token_amount: r.token_amount,
},
);
}
policy.recompute_reference();
policy
}
/// Judge a leg. Pure: call [`Policy::apply`] afterwards to advance state.
pub fn judge(&self, leg: &Leg) -> JudgeResult {
// A leg that moved no tokens, or bought them for nothing, has no price to print.
if leg.token_delta == 0 || leg.sats_delta == 0 {
return JudgeResult {
price: None,
verdict: Verdict::Unpriceable,
};
}
let price = leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64;
let Some(reference) = self.reference else {
return JudgeResult {
price: Some(price),
verdict: Verdict::NoReference,
};
};
let dev = deviation(price, reference);
if dev <= self.params.max_deviation_factor {
return JudgeResult {
price: Some(price),
verdict: Verdict::WithinBand { dev },
};
}
// Both sides derived from present reserves, so this scan is the whole of tier 2's
// state. It is O(pools), but only reached by off-band legs — `WithinBand` returns
// above — so the common path stays a single comparison.
let f = self.params.max_deviation_factor;
let credit = self
.pools
.get(&leg.pool)
.map_or(0, |s| credit_from(s.sats, s.token_amount, reference, f));
let largest_credit = self
.pools
.values()
.map(|s| credit_from(s.sats, s.token_amount, reference, f))
.max()
.unwrap_or(0);
// `ceil().max(1)` keeps a pool with no usable depth (credit 0) from qualifying just
// because the largest credit is small.
let threshold =
((largest_credit as f64 * self.params.min_share_fraction).ceil() as u64).max(1);
let verdict = if credit >= threshold {
Verdict::CreditExempt {
dev,
credit,
largest_credit,
}
} else {
Verdict::Muted {
dev,
credit,
largest_credit,
}
};
JudgeResult {
price: Some(price),
verdict,
}
}
/// Advance state past a leg. Must be called for every leg, accepted or not: a muted
/// leg still moved real reserves.
///
/// Takes no verdict, because none is needed — credit follows the reserves this records.
pub fn apply(&mut self, leg: &Leg) {
// `i64::MIN` so a pool's very first leg clears the monotonicity guard below.
let state = self.pools.entry(leg.pool).or_insert(PoolState {
sequence: i64::MIN,
sats: 0,
token_amount: 0,
});
// Strict `>`: the `i64::MIN` sentinel above is what lets a pool's first leg through,
// so there is no reason to also relax the monotonicity rule for later ones.
if leg.sequence > state.sequence {
state.sequence = leg.sequence;
state.sats = leg.sats;
state.token_amount = leg.token_amount;
}
self.recompute_reference();
}
/// Current reference price.
pub fn reference(&self) -> Option<f64> {
self.reference
}
/// Qualification credit held by a pool. Exposed for tests and diagnostics.
#[allow(dead_code)]
pub fn credit_of(&self, pool: &[u8; 32]) -> u64 {
let Some(reference) = self.reference else {
return 0;
};
self.pools.get(pool).map_or(0, |s| {
credit_from(
s.sats,
s.token_amount,
reference,
self.params.max_deviation_factor,
)
})
}
fn recompute_reference(&mut self) {
// Detach the scratch buffer so `self.pools` can be borrowed while filling it.
let mut scratch = std::mem::take(&mut self.scratch);
scratch.clear();
scratch.extend(self.pools.values().filter_map(|s| {
spot_ratio(s.sats, s.token_amount).map(|spot| (spot, s.sats, s.token_amount))
}));
self.reference = weighted_median_reference(&mut scratch);
self.scratch = scratch;
}
}
/// Qualification credit: a pool's min-depth, but only while its own spot sits within `F` of
/// the reference.
///
/// Stateless by construction, and that is the point. An earlier revision stored "min-depth
/// at the pool's last accepted print", which made credit path-dependent — a reserve
/// snapshot cannot recover it, so [`Policy::seeded`] had to approximate, and a seeded policy
/// returned `CreditExempt` where a folded one returned `Muted` on the very same leg. That
/// broke the invariant the whole seeding design exists to hold, and because `ohlcv_1h` is
/// written with `INSERT OR IGNORE`, whichever path ran first would freeze its answer in
/// permanently.
///
/// Little security was given up. The path-dependent rule was a speed bump rather than a
/// wall: an attacker needed one cheap in-band trade to convert a fresh pool into a credited
/// one. What remains is a pure depth gate — printing far off-reference requires a pool
/// holding a real share of the token's liquidity, which is the part that actually costs an
/// attacker money. A pool that has been walked off-market is still excluded outright, so a
/// snapshot cannot launder an attacker's pool into qualification.
fn credit_from(sats: u64, token_amount: u64, reference: f64, f: f64) -> u64 {
let Some(spot) = spot_ratio(sats, token_amount) else {
return 0;
};
if deviation(spot, reference) > f {
return 0;
}
min_depth(sats, token_amount, reference)
}
/// A pool's spot price, or `None` if it is too small or one-sided to quote one.
fn spot_ratio(sats: u64, token_amount: u64) -> Option<f64> {
if token_amount < MIN_TOKEN_RESERVE || sats == 0 {
return None;
}
Some(sats as f64 / token_amount as f64)
}
/// How far apart two prices are, as a factor >= 1 in whichever direction.
fn deviation(price: f64, reference: f64) -> f64 {
// Finiteness is tested first so the `<=` comparisons below are total: NaN fails
// `is_finite` and returns here, which is why plain `<=` is safe rather than needing
// `!(price > 0.0)` to catch it.
if !price.is_finite() || !reference.is_finite() || price <= 0.0 || reference <= 0.0 {
return f64::INFINITY;
}
(price / reference).max(reference / price)
}
/// `d = min(S, T * R)`: the side of the pool an attacker would have to actually fund.
///
/// Valuing the token side at the reference is what stops a pool stuffed with worthless
/// tokens from out-voting a pool holding real sats.
fn min_depth(sats: u64, token_amount: u64, reference: f64) -> u64 {
let valued = (token_amount as f64 * reference).ceil();
if !valued.is_finite() {
return sats;
}
// Rust saturates on out-of-range float-to-int casts, so a huge reference clamps to
// u64::MAX and `min` still picks the sats side.
sats.min(valued as u64)
}
/// Min-depth-weighted median of pool spot ratios.
///
/// `d_i = min(S_i, T_i * R)` needs an `R` to value the token side, and `R` is what we are
/// solving for. Carrying the previous leg's `R` forward would resolve that circularity,
/// but it would also make the answer depend on where the caller started reading. Instead
/// seed with the unweighted median and reweight to a fixed point, which depends on
/// nothing but the reserves themselves.
fn weighted_median_reference(entries: &mut [(f64, u64, u64)]) -> Option<f64> {
if entries.is_empty() {
return None;
}
entries.sort_unstable_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(Ordering::Equal));
let mut reference = median_by(entries, |_| 1)?;
// Raising `reference` can only raise each `min(S, T * reference)` weight, and only for
// pools not yet capped by their sats side — which are exactly the high-spot tail. So
// the chosen index moves monotonically upwards and never revisits a position, giving a
// fixed point within `entries.len()` passes. A fixed pass count would instead return a
// truncated iterate: on a set of ten pools spread over many decades, stopping at four
// passes yields 411_155 where the fixed point is 501_964, and wider spreads diverge by
// more than an order of magnitude.
for _ in 0..entries.len() {
let Some(next) = median_by(entries, |e| min_depth(e.1, e.2, reference)) else {
break;
};
if next == reference {
break;
}
reference = next;
}
Some(reference)
}
/// Lower weighted median of pre-sorted `entries`: the first value whose cumulative weight
/// reaches half the total.
///
/// Compares `2 * cumulative >= total` rather than `cumulative >= total / 2` so integer
/// division cannot bias the pick downwards. When every weight is zero it falls back to
/// the positional median, so an all-dust set does not collapse to its smallest ratio.
fn median_by(sorted: &[(f64, u64, u64)], weight: impl Fn(&(f64, u64, u64)) -> u64) -> Option<f64> {
let total: u128 = sorted.iter().map(|e| weight(e) as u128).sum();
if total == 0 {
// Defensive only: `spot_ratio` rejects `sats == 0`, so every live weight is >= 1.
// Index chosen to match what the loop below returns for equal weights, rather than
// `len / 2`, which is the upper median for an even count.
return sorted.get(sorted.len().saturating_sub(1) / 2).map(|e| e.0);
}
let mut cumulative: u128 = 0;
for entry in sorted {
cumulative += weight(entry) as u128;
if cumulative * 2 >= total {
return Some(entry.0);
}
}
sorted.last().map(|e| e.0)
}
#[cfg(test)]
mod tests;

444
src/db/cauldron/candlestick/policy/tests.rs Normal file
View file

@ -0,0 +1,444 @@
// 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 super::*;
fn params() -> GuardParams {
GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
}
}
/// A leg at `pool`, leaving that pool holding `sats`/`tokens` afterwards.
fn leg(
pool: u8,
sequence: i64,
sats_delta: i64,
token_delta: i64,
sats: u64,
token_amount: u64,
) -> Leg {
Leg {
txid: [pool; 32],
pool: [pool; 32],
ts: 1000 + sequence,
sequence,
sats_delta,
token_delta,
sats,
token_amount,
}
}
fn reserves(pool: u8, sequence: i64, sats: u64, token_amount: u64) -> PoolReserves {
PoolReserves {
pool: [pool; 32],
sequence,
sats,
token_amount,
}
}
/// Feed a leg through judge-then-apply, returning the verdict.
fn step(policy: &mut Policy, l: &Leg) -> JudgeResult {
let judged = policy.judge(l);
policy.apply(l);
judged
}
#[test]
fn test_leg_with_no_token_movement_is_unpriceable() {
let policy = Policy::new(params());
let judged = policy.judge(&leg(1, 1, 100, 0, 1_000_000, 10_000));
assert_eq!(judged.verdict, Verdict::Unpriceable);
assert!(judged.price.is_none());
assert!(!judged.accepted());
}
/// A leg that moved tokens for zero sats has no meaningful price; printing it as 0.0
/// would drag a candle's low to the floor.
#[test]
fn test_leg_with_no_sats_movement_is_unpriceable() {
let policy = Policy::new(params());
let judged = policy.judge(&leg(1, 1, 0, -1_000, 1_000_000, 10_000));
assert_eq!(judged.verdict, Verdict::Unpriceable);
assert!(judged.price.is_none());
}
#[test]
fn test_first_leg_of_a_new_token_prints_and_sets_the_reference() {
let mut policy = Policy::new(params());
let judged = step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
assert_eq!(judged.verdict, Verdict::NoReference);
assert!(judged.accepted());
assert_eq!(judged.price, Some(100.0));
assert_eq!(policy.reference(), Some(100.0));
}
#[test]
fn test_leg_near_the_reference_prints() {
let mut policy = Policy::new(params());
step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
let judged = step(&mut policy, &leg(2, 2, 190_000, -2_000, 2_000_000, 20_000));
assert!(matches!(judged.verdict, Verdict::WithinBand { .. }));
assert!(judged.accepted());
}
#[test]
fn test_leg_far_from_the_reference_without_credit_is_muted() {
let mut policy = Policy::new(params());
step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
// A fresh pool printing 100x off market, with no accepted print behind it.
let judged = step(
&mut policy,
&leg(9, 2, 10_000_000, -1_000, 10_001_000, 1_010),
);
assert!(
matches!(judged.verdict, Verdict::Muted { .. }),
"{:?}",
judged.verdict
);
assert!(!judged.accepted());
// Muted, but the price was still computed — the caller still counts its volume.
assert!(judged.price.is_some());
}
/// Regression: `apply` inserted new pools with `sequence: leg.sequence` and then gated the
/// reserve update behind `leg.sequence > state.sequence`, which is false on that very first
/// insert — so a pool's opening leg left it recorded with zero reserves, and therefore zero
/// credit. The `i64::MIN` sentinel is what fixes it.
#[test]
fn test_pool_holds_credit_after_its_very_first_leg() {
let mut policy = Policy::new(params());
step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
let second = leg(2, 2, 200_000, -2_000, 2_000_000, 20_000);
let judged = step(&mut policy, &second);
assert!(judged.accepted(), "{:?}", judged.verdict);
assert!(
policy.credit_of(&[2; 32]) > 0,
"a pool's opening leg must record its reserves, not leave it at zero"
);
}
/// A wild print leaves the pool that made it sitting off-market, and an off-market pool
/// holds no credit — so the pool cannot use one absurd leg to qualify the next.
#[test]
fn test_pool_left_off_market_by_its_own_print_holds_no_credit() {
let mut policy = Policy::new(params());
step(&mut policy, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
step(
&mut policy,
&leg(9, 2, 10_000_000, -1_000, 10_001_000, 1_010),
);
assert_eq!(
policy.credit_of(&[9; 32]),
0,
"a pool 99x off the reference must not qualify for tier 2"
);
}
/// Regression for the seeded/folded credit divergence — the reason credit is derived from
/// reserves rather than remembered.
///
/// Pool 2's only leg is a one-token trade at 100x the market. It is muted, but it is far too
/// small to move the pool's own spot, so the pool is left deep and in-band. Under the old
/// "min-depth at the last accepted print" rule the folded policy held pool 2 at zero credit
/// while a policy seeded from the resulting reserves granted it full credit — and the two
/// then returned `Muted` and `CreditExempt` for the identical next leg. Since `ohlcv_1h` is
/// written with `INSERT OR IGNORE`, whichever path ran first would have frozen its answer in.
#[test]
fn test_seeded_and_folded_agree_on_credit_after_a_muted_print() {
let mut folded = Policy::new(params());
step(&mut folded, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
// Pool 2's opening leg: absurd price, negligible size, so it ends in-band.
step(&mut folded, &leg(2, 2, 10_000, -1, 2_010_000, 19_999));
let seeded = Policy::seeded(
params(),
&[
reserves(1, 1, 1_000_000, 10_000),
reserves(2, 2, 2_010_000, 19_999),
],
);
assert_eq!(folded.reference(), seeded.reference());
assert_eq!(
folded.credit_of(&[2; 32]),
seeded.credit_of(&[2; 32]),
"credit must not depend on whether the policy folded history or was seeded from it"
);
assert!(
folded.credit_of(&[2; 32]) > 0,
"the test is vacuous unless the pool actually ends up in-band and credited"
);
// The consequence that made this worth fixing: identical verdicts on the next leg.
let next = leg(2, 3, 500_000, -1, 2_510_000, 19_998);
assert_eq!(folded.judge(&next).verdict, seeded.judge(&next).verdict);
}
/// The property the whole seeded design exists for: the reference depends only on pool
/// reserves, never on how many legs the caller happened to read to arrive at them.
#[test]
fn test_reference_is_a_function_of_reserves_not_of_history_read() {
let mut folded = Policy::new(params());
step(&mut folded, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
step(&mut folded, &leg(2, 2, 190_000, -2_000, 2_000_000, 20_000));
step(&mut folded, &leg(1, 3, 105_000, -1_000, 1_105_000, 9_000));
// Same end state, reached by seeding instead of folding.
let seeded = Policy::seeded(
params(),
&[
reserves(1, 3, 1_105_000, 9_000),
reserves(2, 2, 2_000_000, 20_000),
],
);
assert_eq!(
folded.reference(),
seeded.reference(),
"a seeded policy must agree with a folded one on identical reserves"
);
}
/// Corollary: two callers whose windows start at different points must reach the same
/// verdict on a leg they both see.
#[test]
fn test_seeded_and_folded_policies_agree_on_a_shared_leg() {
let mut folded = Policy::new(params());
step(&mut folded, &leg(1, 1, 100_000, -1_000, 1_000_000, 10_000));
step(&mut folded, &leg(2, 2, 190_000, -2_000, 2_000_000, 20_000));
let seeded = Policy::seeded(
params(),
&[
reserves(1, 1, 1_000_000, 10_000),
reserves(2, 2, 2_000_000, 20_000),
],
);
let shared = leg(1, 3, 300_000, -1_000, 1_300_000, 9_000);
assert_eq!(
folded.judge(&shared).accepted(),
seeded.judge(&shared).accepted(),
"verdict must not depend on where the caller started reading"
);
}
/// Seeding hands credit to pools already trading at market, but withholds it from one
/// that has been walked far away — otherwise a snapshot would launder an attacker's pool
/// straight into qualification.
#[test]
fn test_seeding_withholds_credit_from_off_market_pools() {
let policy = Policy::seeded(
params(),
&[
reserves(1, 1, 1_000_000, 10_000), // spot 100, at market
reserves(2, 2, 2_000_000, 20_000), // spot 100, at market
reserves(9, 3, 10_000_000, 1_000), // spot 10_000, 100x off
],
);
assert_eq!(policy.reference(), Some(100.0));
assert!(policy.credit_of(&[1; 32]) > 0);
assert!(policy.credit_of(&[2; 32]) > 0);
assert_eq!(
policy.credit_of(&[9; 32]),
0,
"a pool sitting 100x off market must not be seeded with credit"
);
}
/// Min-depth weighting is what stops shallow pools from voting the reference away from
/// where the real liquidity sits.
///
/// The dust pools deliberately OUTNUMBER the real one, so an unweighted median lands on
/// them (10_000) and only the weighting pulls the answer back to 100. A test where all
/// pools quote the same price would pass no matter what the weighting did.
#[test]
fn test_dust_pools_do_not_outvote_one_deep_pool() {
let policy = Policy::seeded(
params(),
&[
reserves(1, 1, 100_000_000, 1_000_000), // spot 100, deep
reserves(8, 2, 100_000, 10), // spot 10_000, shallow
reserves(9, 3, 100_000, 10), // spot 10_000, shallow
],
);
assert_eq!(
policy.reference(),
Some(100.0),
"two shallow pools must not outvote one holding a thousand times the depth"
);
}
/// The dust floor is a guard against degenerate arithmetic, not a weighting mechanism —
/// min-depth already suppresses thin pools. What it must do is keep a pool that cannot
/// quote a ratio from producing an infinity.
#[test]
fn test_pools_that_cannot_quote_a_ratio_are_excluded() {
let policy = Policy::seeded(
params(),
&[
reserves(1, 1, 1_000_000, 10_000), // spot 100
reserves(8, 2, 1_000_000, 0), // no tokens: ratio would divide by zero
reserves(9, 3, 0, 10_000), // no sats: ratio would be 0.0
],
);
let reference = policy
.reference()
.expect("the one quotable pool sets the reference");
assert!(
reference.is_finite() && reference == 100.0,
"got {reference}"
);
}
#[test]
fn test_policy_with_no_usable_pools_has_no_reference() {
let policy = Policy::seeded(params(), &[reserves(1, 1, 0, 0)]);
assert_eq!(policy.reference(), None);
}
/// `cumulative >= total / 2` truncates, which tips the pick towards the lower ratio.
///
/// Weights 2 and 3 are the minimal case that separates the two forms: the old test is
/// `2 >= 5/2 == 2`, true, so it stops on the first entry; the correct `2*2 >= 5` is false,
/// so it advances to the second. Driving `median_by` directly is what makes this
/// falsifiable — going through `weighted_median_reference` cannot pin the weights.
#[test]
fn test_weighted_median_is_not_biased_by_integer_division() {
let sorted = [(100.0, 2u64, 0u64), (200.0, 3u64, 0u64)];
assert_eq!(
median_by(&sorted, |e| e.1),
Some(200.0),
"weight 3 holds the majority, so the median is 200, not 100"
);
// An odd, clearly-weighted set must land on the true weighted median.
let mut skewed = vec![(10.0, 1, 1), (100.0, 1_000_000, 1_000_000), (1000.0, 1, 1)];
assert_eq!(weighted_median_reference(&mut skewed), Some(100.0));
}
/// Regression: the reweighting loop ran a fixed four passes and returned whatever iterate
/// it had reached. Because raising the reference only ever raises weights, the pick climbs
/// one position per pass, so a set spread over many decades stops short — this one settled
/// on 411_155 against a true fixed point of 501_964, and wider spreads diverged by more
/// than an order of magnitude.
#[test]
fn test_reference_reaches_its_fixed_point_not_a_truncated_iterate() {
let pools: [(u64, u64); 10] = [
(962, 249_939_293_128_076),
(3_198, 71_909_139_611_364),
(455_109_248_120, 243_581_757_429_499),
(279, 9_934),
(432_371, 2_608),
(208_541_212_143, 728_829_157),
(925_517_097_739, 737_748_546),
(34_778_861_631_604, 885_976_135),
(89_743_615_643_767, 218_272_037),
(499_655_065_278_927, 995_401_018),
];
let mut entries: Vec<(f64, u64, u64)> = pools
.iter()
.map(|(sats, tokens)| (*sats as f64 / *tokens as f64, *sats, *tokens))
.collect();
let reference = weighted_median_reference(&mut entries).unwrap();
// `entries` is left sorted, so one more pass is a direct fixed-point check.
let next = median_by(&entries, |e| min_depth(e.1, e.2, reference)).unwrap();
assert_eq!(
next, reference,
"reweighting must converge; {reference} moves to {next} on the next pass"
);
}
/// All-zero weights used to make the cumulative test pass on the first entry, returning
/// the smallest ratio rather than the middle one.
#[test]
fn test_all_zero_weights_fall_back_to_the_positional_median() {
let mut entries = vec![(10.0, 0, 0), (100.0, 0, 0), (1000.0, 0, 0)];
assert_eq!(weighted_median_reference(&mut entries), Some(100.0));
}
/// Documents a KNOWN GAP in tier-2 rather than endorsing it: the threshold is a share of
/// `largest_credit`, and the pool holding that largest credit trivially clears a share of
/// itself. So the deepest pool — and every pool of a single-pool token — can print any
/// price it likes. Left as-is pending a decision on the tier-2 rule; this test exists so
/// the behaviour is visible and any change to it is deliberate.
#[test]
fn test_known_gap_deepest_pool_self_qualifies_for_tier_two() {
let mut policy = Policy::seeded(params(), &[reserves(1, 1, 10_000_000, 100_000)]);
assert_eq!(policy.reference(), Some(100.0));
assert!(policy.credit_of(&[1; 32]) > 0);
// The same pool now prints ten million times off its own spot.
let absurd = leg(1, 2, 1_000_000_000, -1, 11_000_000_000, 99_999);
let judged = policy.judge(&absurd);
assert!(
matches!(judged.verdict, Verdict::CreditExempt { .. }),
"expected the known self-qualification gap, got {:?}",
judged.verdict
);
assert!(judged.accepted());
let _ = &mut policy;
}
#[test]
fn test_min_depth_values_the_token_side_at_the_reference() {
// Token side is the thinner one: 10 tokens * 100 = 1_000 < 50_000 sats.
assert_eq!(min_depth(50_000, 10, 100.0), 1_000);
// Sats side is the thinner one.
assert_eq!(min_depth(500, 10, 100.0), 500);
}
#[test]
fn test_min_depth_saturates_instead_of_overflowing() {
assert_eq!(min_depth(42, u64::MAX, f64::MAX), 42);
}
#[test]
fn test_deviation_is_symmetric_and_rejects_degenerate_input() {
assert_eq!(deviation(200.0, 100.0), 2.0);
assert_eq!(deviation(50.0, 100.0), 2.0);
assert_eq!(deviation(0.0, 100.0), f64::INFINITY);
assert_eq!(deviation(100.0, 0.0), f64::INFINITY);
// NaN and the infinities must land on INFINITY (i.e. "out of band"), never sneak
// through as a small deviation. `<=` alone is false for NaN, so the finiteness test
// has to run first — this is what pins that ordering.
assert_eq!(deviation(f64::NAN, 100.0), f64::INFINITY);
assert_eq!(deviation(100.0, f64::NAN), f64::INFINITY);
assert_eq!(deviation(f64::INFINITY, 100.0), f64::INFINITY);
assert_eq!(deviation(100.0, f64::INFINITY), f64::INFINITY);
assert_eq!(deviation(-100.0, 100.0), f64::INFINITY);
assert_eq!(deviation(100.0, -100.0), f64::INFINITY);
}
/// A leg replayed at or below a pool's known sequence must not rewind its reserves.
#[test]
fn test_out_of_order_leg_does_not_rewind_reserves() {
let mut policy = Policy::new(params());
step(&mut policy, &leg(1, 10, 100_000, -1_000, 1_000_000, 10_000));
let after_current = policy.reference();
step(&mut policy, &leg(1, 5, 100_000, -1_000, 7, 7));
assert_eq!(
policy.reference(),
after_current,
"a stale leg must not overwrite newer reserves"
);
}

351
src/db/cauldron/candlestick/tests.rs
View file

@ -9,6 +9,7 @@ use crate::db::cauldron::{
pool::{self, dummy_init_seq}, pool::{self, dummy_init_seq},
tx::{self, insert_block_tx, insert_mempool_tx}, tx::{self, insert_block_tx, insert_mempool_tx},
utxo_funding::{self, insert_utxo_funding}, utxo_funding::{self, insert_utxo_funding},
utxo_spending,
}; };
use crate::utiltest::mock_db_pool; use crate::utiltest::mock_db_pool;
use bitcoin_hashes::Hash; use bitcoin_hashes::Hash;
@ -38,12 +39,48 @@ fn dummy_cauldron(
async fn setup_db(pool: sqlx::SqlitePool) { async fn setup_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await; utxo_funding::create_table(&pool).await;
utxo_spending::create_table(&pool).await;
tx::create_table(&pool).await; tx::create_table(&pool).await;
pool::create_table(&pool).await; pool::create_table(&pool).await;
ohlcv::create_table(&pool).await; ohlcv::create_table(&pool).await;
dummy_init_seq(); dummy_init_seq();
} }
/// Register a pool row so the reserve snapshot can find it, optionally marking it
/// withdrawn by the transaction `withdrawn_by` at `withdrawn_at`.
async fn register_pool(
conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
pool_hash: &OutPointHash,
token: &TokenID,
withdrawal: Option<(Txid, i64)>,
) {
let withdrawn_utxo = withdrawal.map(|(txid, ts)| {
let spent = OutPointHash::from_byte_array(*txid.as_byte_array());
(spent, txid, ts)
});
sqlx::query("INSERT OR REPLACE INTO pool (creation_utxo, owner_pkh, token_id, withdrawn_in_utxo) VALUES (?, ?, ?, ?)")
.bind(pool_hash.to_blob())
.bind(PubkeyHash::all_zeros().to_blob())
.bind(token.to_blob())
.bind(withdrawn_utxo.as_ref().map(|(spent, _, _)| spent.to_blob()))
.execute(&mut **conn)
.await
.unwrap();
if let Some((spent, txid, ts)) = withdrawn_utxo {
insert_mempool_tx(&mut **conn, &txid, ts as u64)
.await
.unwrap();
sqlx::query("INSERT OR REPLACE INTO utxo_spending (spent_utxo_hash, txid) VALUES (?, ?)")
.bind(spent.to_blob())
.bind(txid.to_blob())
.execute(&mut **conn)
.await
.unwrap();
}
}
async fn insert_trade_at( async fn insert_trade_at(
conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>, conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
token: &TokenID, token: &TokenID,
@ -86,6 +123,320 @@ async fn insert_trade_at(
.unwrap(); .unwrap();
} }
/// Insert one transaction that touches several pools, each leg with its own
/// `(sats_delta, token_delta)`. Models a router/arbitrage transaction.
async fn insert_multileg_trade_at(
conn: &mut sqlx::pool::PoolConnection<sqlx::Sqlite>,
token: &TokenID,
txid_byte: u8,
ts: u64,
legs: &[(i64, i64)],
) {
let txid = Txid::from_byte_array([txid_byte; 32]);
let block = BlockHash::all_zeros();
insert_mempool_tx(&mut **conn, &txid, ts).await.unwrap();
insert_block_tx(&mut **conn, &txid, &block, ts as i64)
.await
.unwrap();
for (i, (sats_delta, token_delta)) in legs.iter().enumerate() {
let mut utxo_bytes = [txid_byte; 32];
utxo_bytes[0] = i as u8;
let utxo = OutPointHash::from_byte_array(utxo_bytes);
let mut pool_bytes = [txid_byte.wrapping_add(0x80); 32];
pool_bytes[0] = i as u8;
let pool_hash = OutPointHash::from_byte_array(pool_bytes);
// Post-trade reserves large enough to look like a real pool.
let cauldron = dummy_cauldron(
&txid,
&utxo,
token,
sats_delta.unsigned_abs() * 10,
token_delta.abs() * 10,
&PubkeyHash::all_zeros(),
);
insert_utxo_funding(&mut **conn, &vec![cauldron.clone()], &txid)
.await
.unwrap();
pool::insert_pool_history_entry(
&mut **conn,
&pool_hash,
&cauldron,
Some(ts),
Some(ts),
*sats_delta,
*token_delta,
)
.await
.unwrap();
}
}
/// Regression for the netting artifact: a multi-pool arbitrage transaction that buys
/// from one pool and sells into another nets its token movement to almost nothing.
/// Dividing the signed sums produced a price no leg traded at — on mainnet token NWB
/// (tx 1E84F4E9…1916, 27 legs) that printed 30,792,599.5 sats/unit against legs that
/// actually executed between 0.288 and 0.335.
///
/// The durable invariant is that a printed price must be one a leg actually executed at.
/// Under per-leg pricing the close is specifically the last accepted leg's price.
#[tokio::test]
async fn test_multipool_arb_prices_from_its_legs_not_their_net() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC1; 32]);
let token_blob = token.to_blob();
// Real leg totals from the NWB transaction, collapsed to two legs.
let sell = (-446_491_239i64, 1_334_527_069i64); // executes at 0.334569
let buy = (384_906_040i64, -1_334_527_067i64); // executes at 0.288421
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_multileg_trade_at(&mut conn, &token, 0x21, 1000, &[sell, buy]).await;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000)
.await
.unwrap()
.expect("arb transaction must price");
// Per-leg pricing: each leg executes at its own price, not a net ratio.
// The last leg is the buy leg: 384_906_040 / 1_334_527_067 = 0.288421
let low_leg = 384_906_040.0 / 1_334_527_067.0;
let high_leg = 446_491_239.0 / 1_334_527_069.0;
assert!(
price >= low_leg && price <= high_leg,
"price {price} must lie within the executed leg range [{low_leg}, {high_leg}]"
);
// The close should be the last leg's price (buy leg).
let expected = 384_906_040.0 / 1_334_527_067.0;
assert!(
(price - expected).abs() < 1e-9,
"expected {expected} (last leg price), got {price}"
);
}
/// Every leg pointing the same way is the ordinary case — the one covering 99.76% of
/// mainnet prints, including the OLA supply-shock crash. Nothing here should be muted or
/// distorted: each leg executed at a genuine price and the candle closes on the last.
#[tokio::test]
async fn test_single_direction_multileg_closes_on_its_last_leg() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC2; 32]);
let token_blob = token.to_blob();
// Leg prices: 50.0, 50.0, 49.5 — a tight, ordinary spread.
let legs = [(150_000i64, -3_000i64), (50_000, -1_000), (99_000, -2_000)];
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_multileg_trade_at(&mut conn, &token, 0x22, 1000, &legs).await;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000)
.await
.unwrap()
.expect("router transaction must price");
let leg_prices: Vec<f64> = legs
.iter()
.map(|(s, t)| *s as f64 / t.unsigned_abs() as f64)
.collect();
let lowest = leg_prices.iter().cloned().fold(f64::MAX, f64::min);
let highest = leg_prices.iter().cloned().fold(f64::MIN, f64::max);
assert!(
price >= lowest && price <= highest,
"price {price} must be one the legs actually executed at, within [{lowest}, {highest}]"
);
let last_leg_price = *leg_prices.last().unwrap();
assert!(
(price - last_leg_price).abs() < f64::EPSILON,
"close should be the last leg's price: {price} vs {last_leg_price}"
);
}
/// A pool that has been drained leaves its last history row showing full pre-withdrawal
/// reserves, because a withdrawal writes no new row. Seeding the policy from those would
/// let a pool holding nothing keep voting the reference — the OLA case, where a 10.9B-sat
/// pool was withdrawn shortly before a crash and a lingering ghost would have muted it.
#[tokio::test]
async fn test_reserve_snapshot_drops_pools_withdrawn_before_the_instant() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC6; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_trade_at(&mut conn, &token, 0x41, 1000, 100_000, 2_000).await;
let pool_hash = OutPointHash::from_byte_array([0x41u8.wrapping_add(0x80); 32]);
register_pool(&mut conn, &pool_hash, &token, None).await;
let live = super::fetch_reserve_snapshot(&db.cauldron_r, &token_blob, 5000, false)
.await
.unwrap();
assert_eq!(live.len(), 1, "a live pool must be in the snapshot");
// Withdraw it at ts=2000 and ask again afterwards.
let withdrawal_tx = Txid::from_byte_array([0x42; 32]);
register_pool(&mut conn, &pool_hash, &token, Some((withdrawal_tx, 2000))).await;
let after = super::fetch_reserve_snapshot(&db.cauldron_r, &token_blob, 5000, false)
.await
.unwrap();
assert!(
after.is_empty(),
"a pool withdrawn at 2000 must not vote on the reference at 5000: {after:?}"
);
// But it was live at ts=1500, so a historical query must still see it.
let before = super::fetch_reserve_snapshot(&db.cauldron_r, &token_blob, 1500, false)
.await
.unwrap();
assert_eq!(
before.len(),
1,
"a pool withdrawn later was still live earlier and must remain visible"
);
}
/// Regression: the seed consulted `ohlcv_1h` first and returned on any hit. Since that
/// table is structurally at least three hours behind the tip — and completely empty for
/// the whole of a post-version-bump rebuild — a stale bucket shadowed every newer leg.
#[tokio::test]
async fn test_seed_prefers_recent_legs_over_a_stale_materialised_bucket() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC7; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
sqlx::query(
"INSERT INTO ohlcv_1h
(token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count)
VALUES (?, 0, 10.0, 10.0, 10.0, 10.0, 1, 1, 1)",
)
.bind(&token_blob)
.execute(&mut *conn)
.await
.unwrap();
// A real print at 50.0 two hours later that the table has not caught up with.
insert_trade_at(&mut conn, &token, 0x51, 7200, 100_000, 2_000).await;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 10800)
.await
.unwrap()
.expect("a seed must exist");
assert!(
(price - 50.0).abs() < f64::EPSILON,
"stale bucket must not shadow the newer print: got {price}"
);
}
/// Regression: the materialised lookup took any bucket with `bucket_ts < timestamp_end`,
/// so for a non-hour-aligned cutoff it could return a bucket straddling that cutoff —
/// closing on a print from *after* the instant being asked about.
#[tokio::test]
async fn test_seed_does_not_leak_a_price_from_after_the_cutoff() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC8; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
// Bucket [7200, 10800) closes at 80.0 — a price set late in that hour.
sqlx::query(
"INSERT INTO ohlcv_1h
(token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count)
VALUES (?, 7200, 50.0, 80.0, 50.0, 80.0, 1, 1, 1)",
)
.bind(&token_blob)
.execute(&mut *conn)
.await
.unwrap();
// What actually happened by ts=8000 was a single print at 50.0.
insert_trade_at(&mut conn, &token, 0x52, 7300, 100_000, 2_000).await;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 8000)
.await
.unwrap()
.expect("a seed must exist");
assert!(
(price - 50.0).abs() < f64::EPSILON,
"must not return the 80.0 close of a bucket that ends after the cutoff: got {price}"
);
}
/// Regression: the seed lookback was briefly capped at 24 hours, which silently dropped
/// the carry-forward price for any token trading less often than daily and left a hole
/// where the leading candles should be.
#[tokio::test]
async fn test_seed_lookback_reaches_past_a_long_silence() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC5; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_trade_at(&mut conn, &token, 0x31, 1000, 100_000, 2_000).await;
// Ask 60 days later. The only trade is far outside any 24-hour window.
let sixty_days = 1000 + 60 * 24 * 3600;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, sixty_days)
.await
.unwrap()
.expect("a dormant token must still carry its last known price");
assert!((price - 50.0).abs() < f64::EPSILON, "got {price}");
}
/// A transaction whose legs cancel exactly used to print nothing at all (the chart
/// carried the previous close). Its legs are real executions and now price normally.
#[tokio::test]
async fn test_exactly_cancelling_legs_still_price() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC3; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_multileg_trade_at(
&mut conn,
&token,
0x23,
1000,
&[(-9_000, 1_000), (11_000, -1_000)],
)
.await;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000)
.await
.unwrap()
.expect("zero-net transaction must still price from its legs");
// Per-leg pricing: the close is the last leg's price.
// Last leg: (11_000, -1_000) = 11,000 / 1,000 = 11.0
assert!(
(price - 11.0).abs() < f64::EPSILON,
"close should be last leg's price: got {price}"
);
}
/// Legs that move no tokens cannot produce a price (division by zero volume).
#[tokio::test]
async fn test_token_less_legs_do_not_price() {
let db = mock_db_pool(setup_db).await;
let token = TokenID::from_byte_array([0xC4; 32]);
let token_blob = token.to_blob();
let mut conn = db.cauldron_w.acquire().await.unwrap();
insert_multileg_trade_at(&mut conn, &token, 0x24, 1000, &[(5_000, 0), (7_000, 0)]).await;
let price = super::fetch_last_close_before(&db.cauldron_r, &token_blob, 2000)
.await
.unwrap();
assert!(price.is_none(), "no token movement means no price");
}
#[tokio::test] #[tokio::test]
async fn test_fetch_last_close_before_no_trades() { async fn test_fetch_last_close_before_no_trades() {
let db = mock_db_pool(setup_db).await; let db = mock_db_pool(setup_db).await;

494
src/db/cauldron/ohlcv.rs
View file

@ -3,9 +3,31 @@
// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. // 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 // 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::cauldron::config::{config_get, config_set};
use anyhow::Result; use anyhow::Result;
use sqlx::{Row, SqlitePool}; use sqlx::{Row, SqlitePool};
/// Bumped whenever the maths that fills `ohlcv_1h` changes, so buckets materialised by
/// an older rule are discarded instead of being served forever — `rebuild_range` uses
/// `INSERT OR IGNORE`, so existing rows are never corrected in place.
///
/// 2: price switched from the signed net ratio to the gross volume ratio.
/// 3: pricing switched from per-transaction to per-leg with policy-based acceptance.
/// 4: policy reference seeded from reserves rather than folded from the batch start;
/// f64 sentinels no longer written; credit earned on a pool's first accepted print;
/// zero-sats legs unpriceable; weighted-median rounding and zero-weight fixes;
/// pools withdrawn before the snapshot instant excluded from the reference.
/// 5: reweighting iterates to its fixed point instead of stopping at four passes, so the
/// reference moves on multi-decade pool spreads; buckets with no accepted price are
/// written flat at the carried close instead of dropped, restoring their volume; the
/// rebuild seeds from confirmed-only reserves, so output no longer depends on mempool
/// contents at rebuild time.
/// 6: qualification credit derived from present reserves instead of remembered from a
/// pool's last accepted print, so a seeded policy and a folded one can no longer
/// disagree on the same leg. Tier 2 is now a pure depth gate.
pub const OHLCV_VERSION: u32 = 6;
const OHLCV_VERSION_KEY: &str = "ohlcv_version";
pub async fn create_table(pool: &SqlitePool) { pub async fn create_table(pool: &SqlitePool) {
sqlx::query( sqlx::query(
"CREATE TABLE IF NOT EXISTS ohlcv_1h ( "CREATE TABLE IF NOT EXISTS ohlcv_1h (
@ -26,6 +48,30 @@ pub async fn create_table(pool: &SqlitePool) {
.expect("failed to create ohlcv_1h table"); .expect("failed to create ohlcv_1h table");
} }
/// Discards `ohlcv_1h` when it was materialised under an older pricing rule.
///
/// Returns `true` when the table was cleared. The caller must then leave repopulation
/// to the incremental background task: re-materialising the whole history inline would
/// hold up Rocket's startup for as long as it takes, and the raw query path already
/// serves correct candles from `pool_history_entry` while the table refills.
pub async fn migrate_if_stale(read_pool: &SqlitePool, write_pool: &SqlitePool) -> Result<bool> {
let stored = config_get(read_pool, OHLCV_VERSION_KEY)
.await?
.and_then(|v| v.parse::<u32>().ok());
if stored == Some(OHLCV_VERSION) {
return Ok(false);
}
let mut tx = write_pool.begin().await?;
sqlx::query("DELETE FROM ohlcv_1h")
.execute(&mut *tx)
.await?;
config_set(&mut *tx, OHLCV_VERSION_KEY, &OHLCV_VERSION.to_string()).await;
tx.commit().await?;
Ok(true)
}
/// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty. /// Returns the highest `bucket_ts` in `ohlcv_1h`, or `None` if the table is empty.
pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result<Option<i64>> { pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result<Option<i64>> {
let row: Option<(Option<i64>,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h") let row: Option<(Option<i64>,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h")
@ -51,8 +97,11 @@ pub async fn get_min_trade_bucket_ts(pool: &SqlitePool) -> Result<Option<i64>> {
/// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls /// Materialise all 1-hour OHLCV buckets for confirmed trades whose effective timestamp falls
/// in `[since_ts, until_ts)`. /// in `[since_ts, until_ts)`.
/// ///
/// Two-phase approach: the slow aggregation SELECT runs against `read_pool` (no write lock), /// Three-phase approach:
/// then the pre-computed rows are bulk-inserted via `write_pool` (write lock held briefly). /// 1. Fetch per-leg data for all tokens in the range (no write lock)
/// 2. Fold through each token's legs using the Policy to compute buckets (no write lock)
/// 3. Insert pre-computed buckets in one transaction (brief write lock)
///
/// Uses INSERT OR IGNORE so existing rows are never overwritten. /// Uses INSERT OR IGNORE so existing rows are never overwritten.
/// Returns the number of rows inserted. /// Returns the number of rows inserted.
pub async fn rebuild_range( pub async fn rebuild_range(
@ -61,102 +110,36 @@ pub async fn rebuild_range(
since_ts: i64, since_ts: i64,
until_ts: i64, until_ts: i64,
) -> Result<u64> { ) -> Result<u64> {
use crate::db::cauldron::candlestick::{
fetch_reserve_snapshot, to_hash32, GuardParams, Leg, Policy,
};
use std::collections::HashMap;
if since_ts >= until_ts { if since_ts >= until_ts {
return Ok(0); return Ok(0);
} }
// Phase 1: aggregate using the read pool — no write lock held during the slow CTE. // Phase 1: fetch all confirmed legs for the time range.
let select_sql = r#" let sql = r#"
WITH per_pool_tx_raw AS (
SELECT
phe.token_id,
phe.txid,
phe.effective_timestamp AS ts,
phe.utxo,
phe.sats_delta,
phe.token_delta,
phe.sequence
FROM pool_history_entry AS phe
JOIN tx ON tx.txid = phe.txid
WHERE tx.blockhash IS NOT NULL
AND phe.effective_timestamp >= ?
AND phe.effective_timestamp < ?
),
per_pool_tx AS (
SELECT
token_id,
txid,
ts,
(ts / 3600) * 3600 AS bucket_ts,
utxo,
MIN(sequence) AS min_sequence,
SUM(sats_delta) AS signed_sats,
SUM(token_delta) AS signed_tokens,
SUM(ABS(sats_delta)) AS vol_sats,
SUM(ABS(token_delta)) AS vol_tokens
FROM per_pool_tx_raw
GROUP BY token_id, txid, ts, utxo
),
tx_trades AS (
SELECT
token_id,
txid,
ts,
bucket_ts,
MIN(min_sequence) AS min_sequence,
SUM(signed_sats) AS signed_sats,
SUM(signed_tokens) AS signed_tokens,
SUM(vol_sats) AS vol_sats,
SUM(vol_tokens) AS vol_tokens
FROM per_pool_tx
GROUP BY token_id, txid, ts
),
priceable AS (
SELECT
token_id,
bucket_ts,
ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price,
ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts ASC, min_sequence ASC) AS rn_asc,
ROW_NUMBER() OVER (PARTITION BY token_id, bucket_ts ORDER BY ts DESC, min_sequence DESC) AS rn_desc
FROM tx_trades
WHERE signed_tokens != 0
),
ohlc AS (
SELECT
token_id,
bucket_ts,
MAX(CASE WHEN rn_asc = 1 THEN price END) AS open,
MAX(CASE WHEN rn_desc = 1 THEN price END) AS close,
MAX(price) AS high,
MIN(price) AS low
FROM priceable
GROUP BY token_id, bucket_ts
),
vol AS (
SELECT
token_id,
bucket_ts,
SUM(vol_sats) AS volume_sats,
SUM(vol_tokens) AS volume_tokens,
COUNT(*) AS tx_count
FROM tx_trades
GROUP BY token_id, bucket_ts
)
SELECT SELECT
ohlc.token_id, phe.token_id,
ohlc.bucket_ts, phe.txid,
ohlc.open, phe.pool,
ohlc.high, phe.sats_delta,
ohlc.low, phe.token_delta,
ohlc.close, phe.sats,
vol.volume_sats, phe.token_amount,
vol.volume_tokens, phe.sequence,
vol.tx_count phe.effective_timestamp
FROM ohlc FROM pool_history_entry AS phe
JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts JOIN tx ON tx.txid = phe.txid
WHERE tx.blockhash IS NOT NULL
AND phe.effective_timestamp >= ?
AND phe.effective_timestamp < ?
ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
"#; "#;
let rows = sqlx::query(select_sql) let rows = sqlx::query(sql)
.bind(since_ts) .bind(since_ts)
.bind(until_ts) .bind(until_ts)
.fetch_all(read_pool) .fetch_all(read_pool)
@ -166,35 +149,132 @@ JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts
return Ok(0); return Ok(0);
} }
// Phase 2: insert pre-computed rows inside a single transaction. // Phase 2: fold through legs per-token using the policy core to compute buckets.
// The write lock is held only for these fast INSERTs, not during aggregation. let params = GuardParams::default();
let mut tx = write_pool.begin().await?;
let mut inserted = 0u64; let mut buckets: HashMap<(Vec<u8>, i64), OhlcvBucket> = HashMap::new();
let mut current_token: Option<Vec<u8>> = None;
let mut policy = Policy::new(params);
// Reserve-implied price for each token as the batch opened, used to price a leading
// bucket whose legs were all muted so its volume is still recorded.
let mut entry_price: HashMap<Vec<u8>, Option<f64>> = HashMap::new();
for row in &rows { for row in &rows {
let token_id: Vec<u8> = row.get(0); let token_id: Vec<u8> = row.get(0);
let bucket_ts: i64 = row.get(1); let txid: Vec<u8> = row.get(1);
let open: f64 = row.get(2); let pool_bytes: Vec<u8> = row.get(2);
let high: f64 = row.get(3); let ts: i64 = row.get(8);
let low: f64 = row.get(4); let bucket_ts = (ts / 3600) * 3600;
let close: f64 = row.get(5);
let volume_sats: i64 = row.get(6); // A new token starts a new policy — seeded from that token's reserves as they
let volume_tokens: i64 = row.get(7); // stood when this batch opened. Callers materialise history in 24-hour batches,
let tx_count: i64 = row.get(8); // so an unseeded policy would forget everything at every batch boundary and wave
// through the first leg of every day of history unjudged.
if current_token.as_ref() != Some(&token_id) {
current_token = Some(token_id.clone());
let snapshot = fetch_reserve_snapshot(read_pool, &token_id, since_ts, true).await?;
policy = Policy::seeded(params, &snapshot);
entry_price.insert(token_id.clone(), policy.reference());
}
let leg = Leg {
txid: to_hash32(&txid)?,
pool: to_hash32(&pool_bytes)?,
ts,
sequence: row.get(7),
sats_delta: row.get(3),
token_delta: row.get(4),
sats: row.get(5),
token_amount: row.get(6),
};
// Evaluate the leg through the policy.
let judge = policy.judge(&leg);
let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
let token_delta_abs = leg.token_delta.unsigned_abs() as i64;
// Get or create the bucket for this (token, bucket_ts).
let bucket_key = (token_id.clone(), bucket_ts);
let bucket = buckets.entry(bucket_key).or_insert_with(OhlcvBucket::new);
// Always accumulate volume.
bucket.volume_sats += sats_delta_abs;
bucket.volume_tokens += token_delta_abs;
bucket.txids.insert(leg.txid);
// Update OHLC only if accepted and priceable.
if judge.accepted() {
if let Some(price) = judge.price {
if bucket.first_accepted_price.is_none() {
bucket.first_accepted_price = Some(price);
}
bucket.last_accepted_price = Some(price);
bucket.high = bucket.high.max(price);
bucket.low = bucket.low.min(price);
}
}
policy.apply(&leg);
}
// Phase 3: insert all computed buckets in one transaction.
//
// Ordered by (token, bucket_ts) so a token's buckets can carry their close forward.
// A bucket whose legs were all muted or unpriceable has no price of its own, but it
// does have real volume, and the guard makes such buckets ordinary rather than
// exotic — a manipulation burst filling an hour produces exactly one. Writing it with
// the carried close keeps that volume, where writing the raw accumulators would put
// the f64::MIN/MAX sentinels into the table and skipping the row entirely would make
// the materialised path report zero volume for an hour the raw path reports in full.
let mut ordered: Vec<((Vec<u8>, i64), OhlcvBucket)> = buckets.into_iter().collect();
ordered.sort_unstable_by(|a, b| a.0.cmp(&b.0));
let mut tx = write_pool.begin().await?;
let mut inserted = 0u64;
let mut carry_token: Option<Vec<u8>> = None;
let mut carry: Option<f64> = None;
for ((token_id, bucket_ts), bucket) in ordered {
if carry_token.as_ref() != Some(&token_id) {
carry_token = Some(token_id.clone());
carry = entry_price.get(&token_id).copied().flatten();
}
let (open, close, high, low) = match bucket.last_accepted_price {
Some(last) => {
carry = Some(last);
(
bucket.first_accepted_price.unwrap_or(last),
last,
bucket.high,
bucket.low,
)
}
// No price anywhere in this bucket. Flat at the carried close if one exists;
// if the token has no known price yet at all, there is nothing to plot and the
// row is skipped, as the pre-policy SQL did by inner-joining prices to volume.
None => match carry {
Some(prev) => (prev, prev, prev, prev),
None => continue,
},
};
inserted += sqlx::query( inserted += sqlx::query(
"INSERT OR IGNORE INTO ohlcv_1h "INSERT OR IGNORE INTO ohlcv_1h
(token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count) (token_id, bucket_ts, open, high, low, close, volume_sats, volume_tokens, tx_count)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
) )
.bind(token_id) .bind(&token_id)
.bind(bucket_ts) .bind(bucket_ts)
.bind(open) .bind(open)
.bind(high) .bind(high)
.bind(low) .bind(low)
.bind(close) .bind(close)
.bind(volume_sats) .bind(bucket.volume_sats)
.bind(volume_tokens) .bind(bucket.volume_tokens)
.bind(tx_count) .bind(bucket.txids.len() as i64)
.execute(&mut *tx) .execute(&mut *tx)
.await? .await?
.rows_affected(); .rows_affected();
@ -204,6 +284,31 @@ JOIN vol ON ohlc.token_id = vol.token_id AND ohlc.bucket_ts = vol.bucket_ts
Ok(inserted) Ok(inserted)
} }
/// Temporary bucket structure for accumulating OHLCV data per (token, bucket_ts).
struct OhlcvBucket {
first_accepted_price: Option<f64>,
last_accepted_price: Option<f64>,
high: f64,
low: f64,
volume_sats: i64,
volume_tokens: i64,
txids: std::collections::HashSet<[u8; 32]>,
}
impl OhlcvBucket {
fn new() -> Self {
Self {
first_accepted_price: None,
last_accepted_price: None,
high: f64::MIN,
low: f64::MAX,
volume_sats: 0,
volume_tokens: 0,
txids: std::collections::HashSet::new(),
}
}
}
pub struct OhlcvRow { pub struct OhlcvRow {
pub bucket_ts: i64, pub bucket_ts: i64,
pub open: f64, pub open: f64,
@ -253,7 +358,7 @@ pub async fn get_active_candles(
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding}; use crate::db::cauldron::{pool as cauldron_pool, tx, utxo_funding, utxo_spending};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
@ -271,6 +376,7 @@ mod tests {
async fn setup_db(pool: &SqlitePool) { async fn setup_db(pool: &SqlitePool) {
tx::create_table(pool).await; tx::create_table(pool).await;
utxo_funding::create_table(pool).await; utxo_funding::create_table(pool).await;
utxo_spending::create_table(pool).await;
cauldron_pool::create_table(pool).await; cauldron_pool::create_table(pool).await;
create_table(pool).await; // ohlcv_1h + idx_phe_txid create_table(pool).await; // ohlcv_1h + idx_phe_txid
} }
@ -286,7 +392,7 @@ mod tests {
token_delta: i64, token_delta: i64,
) { ) {
let blockhash = [0xAA_u8; 32]; let blockhash = [0xAA_u8; 32];
sqlx::query("INSERT INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)") sqlx::query("INSERT OR IGNORE INTO tx (txid, blockhash, mtp_timestamp) VALUES (?, ?, ?)")
.bind(txid.as_slice()) .bind(txid.as_slice())
.bind(blockhash.as_slice()) .bind(blockhash.as_slice())
.bind(mtp_ts) .bind(mtp_ts)
@ -446,4 +552,174 @@ mod tests {
.unwrap(); .unwrap();
assert_eq!(n, 0, "mempool trades must not be materialised"); assert_eq!(n, 0, "mempool trades must not be materialised");
} }
/// Regression: a bucket whose legs are all unpriceable used to be written out with its
/// `high`/`low` accumulators still at the `f64::MIN`/`f64::MAX` sentinels, putting
/// ±1.8e308 into the table and from there straight onto the chart. Such a bucket has
/// no OHLC to report and must simply not be materialised.
#[tokio::test]
async fn test_rebuild_range_skips_buckets_with_no_priceable_leg() {
let pool = test_pool().await;
setup_db(&pool).await;
// A confirmed leg that moves sats but no tokens: real volume, no price.
insert_confirmed_trade(
&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 0,
)
.await;
let n = rebuild_range(&pool, &pool, 1727960400, 1727967600)
.await
.unwrap();
assert_eq!(n, 0, "a bucket with no priceable leg must not be written");
let rows: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ohlcv_1h")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
rows, 0,
"no row should exist, let alone one carrying f64 sentinels"
);
}
/// Once a token has a known price, an hour whose legs are all unpriceable or muted
/// still carries real volume. Dropping the row made the materialised path report zero
/// volume for an hour the raw path reports in full — and the guard makes such hours
/// ordinary, since one manipulation burst can fill an entire bucket.
#[tokio::test]
async fn test_rebuild_range_keeps_volume_for_a_bucket_with_no_accepted_price() {
let pool = test_pool().await;
setup_db(&pool).await;
let token = [0x03_u8; 32];
// Hour 1727960400: an ordinary priced trade at 1000/25 = 40.
insert_confirmed_trade(&pool, [0x01; 32], [0x02; 32], token, 1727960500, -1000, 25).await;
// Hour 1727964000: tokens move but no sats, so the leg has no price.
insert_confirmed_trade(&pool, [0x11; 32], [0x12; 32], token, 1727964100, 0, 25).await;
let n = rebuild_range(&pool, &pool, 1727960400, 1727967600)
.await
.unwrap();
assert_eq!(n, 2, "both hours must be materialised");
let (volume, open, close, high, low): (i64, f64, f64, f64, f64) = sqlx::query_as(
"SELECT volume_tokens, open, close, high, low FROM ohlcv_1h WHERE bucket_ts = 1727964000",
)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(
volume, 25,
"volume must survive a bucket with no accepted price"
);
assert_eq!(
(open, close, high, low),
(40.0, 40.0, 40.0, 40.0),
"the bucket must sit flat at the carried close, not at a sentinel"
);
}
/// A multi-pool arbitrage transaction whose legs nearly cancel must materialise the
/// price its legs executed at, not the signed-net ratio.
#[tokio::test]
async fn test_rebuild_range_prices_arb_by_gross_volume() {
let pool = test_pool().await;
setup_db(&pool).await;
let token = [0x03_u8; 32];
let txid = [0x01_u8; 32];
// Same transaction, two pools, opposite directions netting to +2 token units.
insert_confirmed_trade(
&pool,
txid,
[0x02; 32],
token,
1727963400,
-446_491_239,
1_334_527_069,
)
.await;
insert_confirmed_trade(
&pool,
txid,
[0x04; 32],
token,
1727963400,
384_906_040,
-1_334_527_067,
)
.await;
rebuild_range(&pool, &pool, 1727960400, 1727964000)
.await
.unwrap();
let (close, open, high, low): (f64, f64, f64, f64) =
sqlx::query_as("SELECT close, open, high, low FROM ohlcv_1h WHERE token_id = ?")
.bind(token.as_slice())
.fetch_one(&pool)
.await
.unwrap();
// Per-leg pricing: each leg executes at its own price.
// Leg 1 (buy): 446,491,239 / 1,334,527,069 = 0.334569
// Leg 2 (sell): 384,906,040 / 1,334,527,067 = 0.288421
// Close is the last leg, open is the first, high/low are the extremes.
let leg1_price = 446_491_239.0 / 1_334_527_069.0;
let leg2_price = 384_906_040.0 / 1_334_527_067.0;
assert!(
(close - leg2_price).abs() < 1e-9,
"materialised close {close} should be leg2 price {leg2_price}"
);
assert!(
(open - leg1_price).abs() < 1e-9,
"materialised open {open} should be leg1 price {leg1_price}"
);
assert!(
(high - leg1_price).abs() < 1e-9 && (low - leg2_price).abs() < 1e-9,
"high {high} and low {low} should span the leg prices"
);
}
/// The version key gates the wipe: stale tables are cleared exactly once.
#[tokio::test]
async fn test_migrate_if_stale_clears_once() {
let pool = test_pool().await;
setup_db(&pool).await;
crate::db::cauldron::config::create_table(&pool).await;
insert_confirmed_trade(
&pool, [0x01; 32], [0x02; 32], [0x03; 32], 1727963400, -1000, 25,
)
.await;
rebuild_range(&pool, &pool, 1727960400, 1727964000)
.await
.unwrap();
assert!(
migrate_if_stale(&pool, &pool).await.unwrap(),
"unversioned table must be wiped"
);
let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ohlcv_1h")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(remaining, 0, "stale buckets must be gone");
rebuild_range(&pool, &pool, 1727960400, 1727964000)
.await
.unwrap();
assert!(
!migrate_if_stale(&pool, &pool).await.unwrap(),
"a table at the current version must survive"
);
let kept: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ohlcv_1h")
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(kept, 1, "rebuilt buckets must not be wiped again");
}
} }

42
src/main.rs
View file

@ -419,6 +419,25 @@ async fn launch() -> _ {
// Ensure the OHLCV pre-aggregation table exists (safe on both new and existing DBs). // Ensure the OHLCV pre-aggregation table exists (safe on both new and existing DBs).
db::cauldron::ohlcv::create_table(&dbpool.cauldron_w).await; db::cauldron::ohlcv::create_table(&dbpool.cauldron_w).await;
// Discard buckets materialised under a superseded pricing rule.
let ohlcv_wiped =
match db::cauldron::ohlcv::migrate_if_stale(&dbpool.cauldron_r, &dbpool.cauldron_w).await {
Ok(wiped) => {
if wiped {
info!(
"ohlcv: cleared for rebuild at version {}; \
serving the raw path until the background task catches up",
db::cauldron::ohlcv::OHLCV_VERSION
);
}
wiped
}
Err(e) => {
warn!("ohlcv: version check failed, leaving table as-is: {e}");
false
}
};
// Bootstrap OhlcvState from whatever is already in the table (survives restarts). // Bootstrap OhlcvState from whatever is already in the table (survives restarts).
let max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r) let max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r)
.await .await
@ -431,7 +450,11 @@ async fn launch() -> _ {
// Synchronous post-IBD backfill: run the full ohlcv_1h materialisation before // Synchronous post-IBD backfill: run the full ohlcv_1h materialisation before
// allowing metrics_cache and other background writers to start. We reuse the // allowing metrics_cache and other background writers to start. We reuse the
// indexing_in_progress flag so metrics_cache backs off during this window. // indexing_in_progress flag so metrics_cache backs off during this window.
{ //
// Skipped after a version wipe: the backfill runs before `rocket::build()` returns,
// so re-materialising all of history here would refuse connections for the whole
// rebuild rather than degrading to the (correct, slower) raw path.
if !ohlcv_wiped {
const BACKFILL_BATCH_SECS: i64 = 24 * 3600; const BACKFILL_BATCH_SECS: i64 = 24 * 3600;
const BACKFILL_SAFETY_SECS: i64 = 3 * 3600; const BACKFILL_SAFETY_SECS: i64 = 3 * 3600;
@ -492,18 +515,33 @@ async fn launch() -> _ {
} }
// Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive. // Background task: incrementally materialise new 1-hour OHLCV buckets as blocks arrive.
// The full historical backfill above already ran; this task only handles the tail. // After a version wipe this is also what repopulates history, since the synchronous
// backfill above is skipped in that case.
// Only processes buckets older than 3 hours (well beyond BCH reorg depth). // Only processes buckets older than 3 hours (well beyond BCH reorg depth).
{ {
let ohlcv_write = dbpool.cauldron_w.clone(); let ohlcv_write = dbpool.cauldron_w.clone();
let ohlcv_read = dbpool.cauldron_r.clone(); let ohlcv_read = dbpool.cauldron_r.clone();
let ohlcv_state_bg = ohlcv_state.clone(); let ohlcv_state_bg = ohlcv_state.clone();
let ohlcv_ibd = ibd_state.clone();
tokio::spawn(async move { tokio::spawn(async move {
// Batch size: 1 day per SQL call to keep each write short. // Batch size: 1 day per SQL call to keep each write short.
const BATCH_SECS: i64 = 24 * 3600; const BATCH_SECS: i64 = 24 * 3600;
// Safety margin: only materialise buckets older than this many seconds. // Safety margin: only materialise buckets older than this many seconds.
const SAFETY_SECS: i64 = 3 * 3600; const SAFETY_SECS: i64 = 3 * 3600;
// Wait for IBD before materialising anything.
//
// The synchronous backfill above waits too, but it is skipped whenever
// `ohlcv_wiped` is set — which includes every fresh database, since a missing
// version key reads as stale. Without this the task would sweep from the
// first trade all the way to `now - 3h` while indexing is still years behind,
// writing nothing, contending with block writes for the cauldron write lock,
// and advancing `materialized_end` to roughly now against an empty table — at
// which point `candlesticks()` would take the fast path over nothing.
while !ohlcv_ibd.initial_sync_complete.load(Ordering::Relaxed) {
tokio::time::sleep(Duration::from_secs(1)).await;
}
loop { loop {
let now = crate::timeutil::time_now(); let now = crate::timeutil::time_now();
// Floor to 1-hour boundary, 3 hours ago. // Floor to 1-hour boundary, 3 hours ago.

14
src/rpc/candlesticks/tests.rs
View file

@ -8,6 +8,7 @@ use crate::db::cauldron::{
pool::{self, dummy_init_seq, insert_new_pool}, pool::{self, dummy_init_seq, insert_new_pool},
tx::{self, insert_block_tx, insert_mempool_tx}, tx::{self, insert_block_tx, insert_mempool_tx},
utxo_funding::{self, insert_utxo_funding}, utxo_funding::{self, insert_utxo_funding},
utxo_spending,
}; };
use crate::utiltest::mock_db_pool; use crate::utiltest::mock_db_pool;
use crate::OhlcvState; use crate::OhlcvState;
@ -64,8 +65,10 @@ fn dummy_cauldron(
async fn setup_mock_db(pool: sqlx::SqlitePool) { async fn setup_mock_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await; utxo_funding::create_table(&pool).await;
utxo_spending::create_table(&pool).await;
tx::create_table(&pool).await; tx::create_table(&pool).await;
pool::create_table(&pool).await; pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq(); dummy_init_seq();
let mut conn = pool.acquire().await.unwrap(); let mut conn = pool.acquire().await.unwrap();
@ -200,6 +203,7 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
async fn setup_seed_db(pool: sqlx::SqlitePool) { async fn setup_seed_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await; utxo_funding::create_table(&pool).await;
utxo_spending::create_table(&pool).await;
tx::create_table(&pool).await; tx::create_table(&pool).await;
pool::create_table(&pool).await; pool::create_table(&pool).await;
ohlcv::create_table(&pool).await; ohlcv::create_table(&pool).await;
@ -378,8 +382,10 @@ async fn test_multiple_candlesticks_endpoint() {
async fn test_single_swap_multiple_pools() { async fn test_single_swap_multiple_pools() {
let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move {
utxo_funding::create_table(&pool).await; utxo_funding::create_table(&pool).await;
utxo_spending::create_table(&pool).await;
tx::create_table(&pool).await; tx::create_table(&pool).await;
pool::create_table(&pool).await; pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq(); dummy_init_seq();
let mut conn = pool.acquire().await.unwrap(); let mut conn = pool.acquire().await.unwrap();
@ -479,8 +485,10 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move { let mock_db = mock_db_pool(|pool: sqlx::SqlitePool| async move {
// --- boilerplate setup --- // --- boilerplate setup ---
utxo_funding::create_table(&pool).await; utxo_funding::create_table(&pool).await;
utxo_spending::create_table(&pool).await;
tx::create_table(&pool).await; tx::create_table(&pool).await;
pool::create_table(&pool).await; pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq(); dummy_init_seq();
let mut conn = pool.acquire().await.unwrap(); let mut conn = pool.acquire().await.unwrap();
@ -664,10 +672,12 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 200); assert_eq!(c1["volume_tokens"].as_i64().unwrap(), 200);
assert_eq!(c1["transaction_count"].as_i64().unwrap(), 1); assert_eq!(c1["transaction_count"].as_i64().unwrap(), 1);
// --- Candle #2 (net-zero tokens, carry-forward close) --- // --- Candle #2 (legs cancel to zero net tokens) ---
let c2 = &candles[1]; let c2 = &candles[1];
assert_eq!(c2["time"].as_i64().unwrap(), (start + 600) as i64); assert_eq!(c2["time"].as_i64().unwrap(), (start + 600) as i64);
// Should carry previous close (50) because signed_tokens==0 but volume>0 // Both legs executed at 50, so the gross ratio 20_000/400 prices the tx at 50
// directly. (Before gross pricing this candle carried the previous close because
// the net token delta was zero; the value coincides, the derivation does not.)
assert!((c2["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert!((c2["open"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON);
assert!((c2["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert!((c2["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON);
assert!((c2["low"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert!((c2["low"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON);