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

Fix regressions introduced by the Stage 2 per-leg pricing phases

Browse source 
Review of today's Phase 1-4 commits found six defects, all introduced by those
phases and verified against 283c0c3. Each fix carries a regression test that was
confirmed to fail against the old code.

Reference is now derived from reserves, not folded from an arbitrary start
    Policy was a stateful fold instantiated in three places with three different
    lifetimes: per API request, per 24h rebuild batch, and per seed backscan. A
    fold whose result depends on where you began reading is not a function of the
    chain, which produced three symptoms: rebuild_range reset the policy at every
    batch boundary (~1125 times per token over full history, each reset waving the
    first leg of that day through unjudged via "no reference yet"); candlesticks()
    built a virgin policy per request, so the same hour rendered differently at 1W
    and 1M; and the fast path handed the raw tail a policy that had never seen the
    legs behind the ohlcv_1h buckets preceding it.

    Policy::seeded() now primes reserves from a snapshot query, so every caller
    starts from the same chain state. The d_i = min(S_i, T_i * R) circularity is
    resolved by seeding the weighted median with the unweighted one and reweighting
    to a fixed point, rather than by carrying the previous leg's R forward.

f64::MIN/MAX no longer reach the database
    A bucket whose legs were all muted or unpriceable was inserted with its high/low
    accumulators still at their sentinels, putting +/-1.8e308 into ohlcv_1h and from
    there onto the chart. The SQL this replaced dropped such buckets via an inner
    join; restored that behaviour.

Credit off-by-one
    apply() inserted new pools with sequence: leg.sequence, then gated the credit
    update behind leg.sequence > state.sequence -- false on that very insert. Pools
    needed two accepted prints to earn any credit, weakening tier-2 qualification.

Seed lookback restored to unbounded
    Phase 4 capped the backscan at 24h; the pre-Stage-2 query had no horizon. Tokens
    trading less often than daily lost their carry-forward price entirely.

Hot path and edge cases
    Dropped the per-leg format! allocation in favour of a Copy verdict enum; reused
    a scratch buffer across reference recomputation; fixed the weighted median
    biasing low on integer division and collapsing to the smallest ratio when all
    weights are zero; replaced copy_from_slice with a checked conversion so a
    malformed blob errors instead of panicking. Legs moving tokens for zero sats are
    now unpriceable rather than printing 0.0.

Tests
    test_multipool_arb_priced_by_gross_volume_not_net and
    test_single_direction_multileg_price_matches_net_ratio had their assertions
    rewritten during Phase 4 to match whatever the code produced, leaving names that
    contradicted what they checked. Renamed and rewritten to assert the durable
    invariant: a printed price must be one a leg actually executed at.
    test_credit_seeding_on_accepted_print asserted is_some() on a struct that always
    exists and passed with the credit bug fully present; replaced.

Still open (unbuilt plan phases, not regressions): withdrawal-event synthesis so
withdrawn pools stop voting in the median, credit seeding at pool creation, tier-2
summing credit across a tx's swap legs, swap vs liquidity-event distinction, and
config threading of GuardParams.

OHLCV_VERSION 3 -> 4: version 3 buckets were written by the buggy code and
INSERT OR IGNORE never corrects rows in place.

Tests: 238 passing

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
jakobsn 2026-07-29 17:29:16 +02:00
parent d3fe840596
commit e733bad6ad
6 changed files with 830 additions and 363 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -10,7 +10,15 @@ use bitcoincash::TokenID;
use serde::Serialize;
use sqlx::{Row, SqlitePool};
pub use self::policy::{GuardParams, Leg, Policy};
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)]
pub struct CandlestickData {
@ -102,7 +110,7 @@ fn aggregate_raw_trades(
let judge = policy.judge(leg);
let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
let token_delta_abs = leg.token_delta.abs() as i64;
let token_delta_abs = leg.token_delta.unsigned_abs() as i64;
// Always count volume, regardless of acceptance.
pi.volume_sats += sats_delta_abs;
@ -110,7 +118,7 @@ fn aggregate_raw_trades(
txid_set.insert(leg.txid);
// Update OHLC only if accepted and priceable.
if judge.accepted {
if judge.accepted() {
if let Some(price) = judge.price {
if first_trade_in_interval {
pi.open = Some(price);
@ -126,7 +134,7 @@ fn aggregate_raw_trades(
}
}
policy.apply(leg, judge.accepted);
policy.apply(leg, judge.accepted());
leg_index += 1;
}
@ -254,82 +262,134 @@ ORDER BY phe.effective_timestamp ASC, phe.sequence ASC;
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
rows.into_iter()
.map(|r| {
let txid: Vec<u8> = r.get(0);
let pool_bytes: Vec<u8> = r.get(2);
let mut txid_arr = [0u8; 32];
let mut pool_arr = [0u8; 32];
txid_arr.copy_from_slice(&txid);
pool_arr.copy_from_slice(&pool_bytes);
Leg {
txid: txid_arr,
pool: pool_arr,
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())
.collect()
}
/// Returns the close price of the most recent accepted print strictly before
/// `timestamp_end`. Uses ohlcv_1h for fast lookup when available, otherwise
/// bacscans recent legs through the policy.
/// Returns `None` when no prior accepted print exists (new token, no history).
/// Pool reserves for `token_blob` as they stood immediately before `timestamp`.
///
/// 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.
pub(crate) async fn fetch_reserve_snapshot(
pool: &SqlitePool,
token_blob: &[u8],
timestamp: i64,
) -> Result<Vec<PoolReserves>> {
let sql = 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
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 = ?;
"#;
let rows = sqlx::query(sql)
.bind(timestamp)
.bind(token_blob)
.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(
pool: &SqlitePool,
token_blob: &[u8],
timestamp_end: i64,
) -> Result<Option<f64>> {
// Fast path: if timestamp_end is hour-aligned, look up the previous bucket's close
// from the materialized ohlcv_1h table (indexed point lookup).
if timestamp_end % 3600 == 0 && timestamp_end >= 3600 {
let prev_bucket = timestamp_end - 3600;
let close: Option<f64> = sqlx::query_scalar(
"SELECT close FROM ohlcv_1h WHERE token_id = ? AND bucket_ts = ?",
)
.bind(token_blob)
.bind(prev_bucket)
.fetch_optional(pool)
.await?;
// Materialised buckets answer this with one index seek on (token_id, bucket_ts).
let materialised: Option<f64> = sqlx::query_scalar(
"SELECT close FROM ohlcv_1h
WHERE token_id = ? AND bucket_ts < ?
ORDER BY bucket_ts DESC LIMIT 1",
)
.bind(token_blob)
.bind(timestamp_end)
.fetch_optional(pool)
.await?;
if close.is_some() {
return Ok(close);
}
if materialised.is_some() {
return Ok(materialised);
}
// Slow path: if the materialized lookup missed (non-aligned or no bucket), backscan
// recent legs through the policy to find the last accepted print.
// Fetch the last ~500 legs before timestamp_end and fold through them.
let legs = fetch_raw_legs(pool, token_blob, (timestamp_end - 24 * 3600).max(0), timestamp_end)
.await?;
// Nothing materialised: replay the token's most recent active hour through the guard.
// Only that hour is needed, because the policy is seeded from reserves rather than
// rebuilt by folding all of history.
let last_activity: Option<i64> = sqlx::query_scalar(
"SELECT MAX(effective_timestamp) FROM pool_history_entry
WHERE token_id = ? AND effective_timestamp < ?",
)
.bind(token_blob)
.bind(timestamp_end)
.fetch_optional(pool)
.await?
.flatten();
if legs.is_empty() {
let Some(last_activity) = last_activity else {
return Ok(None);
}
let params = GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
};
let mut policy = Policy::new(params);
// Fold through legs in chronological order, tracking the last accepted price.
let mut last_accepted_price: Option<f64> = None;
for leg in legs {
let judge = policy.judge(&leg);
if judge.accepted {
let scan_start = (last_activity / 3600) * 3600;
let snapshot = fetch_reserve_snapshot(pool, token_blob, scan_start).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, judge.accepted);
policy.apply(leg, judge.accepted());
}
Ok(last_accepted_price)
Ok(last_accepted_price.or_else(|| policy.reference()))
}
/// `ohlcv_materialized_end`: exclusive upper bound of what is in `ohlcv_1h`.
@ -347,13 +407,7 @@ pub async fn candlesticks(
}
let token_blob = display_hex_to_blob::<TokenID>(token_id)?;
// Default guard parameters (F=5, q=5%). TODO: thread from config when flag enabled.
let params = GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
};
let mut policy = Policy::new(params);
let params = GuardParams::default();
// 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
@ -383,6 +437,13 @@ pub async fn candlesticks(
if ohlcv_end < timestamp_end {
// Tail: query raw for [ohlcv_end, timestamp_end) and append.
//
// 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).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();
@ -414,10 +475,18 @@ pub async fn candlesticks(
current_start += step_size;
}
let snapshot = fetch_reserve_snapshot(pool, &token_blob, timestamp_start).await?;
let mut policy = Policy::seeded(params, &snapshot);
let all_legs = fetch_raw_legs(pool, &token_blob, timestamp_start, timestamp_end).await?;
let (result, _, _) =
aggregate_raw_trades(&all_legs, intervals, step_size, seed_found, seed_close, &mut policy);
let (result, _, _) = aggregate_raw_trades(
&all_legs,
intervals,
step_size,
seed_found,
seed_close,
&mut policy,
);
Ok(result)
}

570
src/db/cauldron/candlestick/policy.rs
View file

@ -3,8 +3,16 @@
// 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;
/// Upper bound on reweighting passes in [`weighted_median_reference`]. Four is far more
/// than the one or two the fixed point needs in practice; it exists to bound the loop.
const REWEIGHT_PASSES: usize = 4;
/// A leg of a transaction: one pool's change in a single transaction.
#[derive(Clone, Debug)]
pub struct Leg {
@ -12,338 +20,358 @@ pub struct Leg {
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)
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)
}
/// Outcome of judging a leg for acceptance.
/// 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 JudgeResult {
pub price: Option<f64>, // None if token_delta == 0
pub accepted: bool, // whether this leg is accepted into OHLC
pub reason: String, // debug: why accepted or muted
pub struct PoolReserves {
pub pool: [u8; 32],
pub sequence: i64,
pub sats: u64,
pub token_amount: u64,
}
/// Per-pool state: reserves and credit.
/// 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 and qualification credit.
#[derive(Clone, Debug)]
struct PoolState {
sequence: i64,
sats: u64,
token_amount: u64,
credit: u64, // min-depth at last accepted print, seeded at creation
/// Min-depth at this pool's last accepted print.
credit: u64,
}
/// The policy engine: stateful fold over legs.
/// The guard: judges each leg against a reference price derived from pool reserves.
///
/// Maintains:
/// - Pool reserves (sats, token_amount, sequence)
/// - Qualification credit (min-depth at last accepted print)
/// - Reference R (min-depth-weighted median of pool spots)
/// 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>, // min-depth-weighted median of pool spots
reference: Option<f64>,
params: GuardParams,
}
#[derive(Clone, Debug)]
pub struct GuardParams {
pub max_deviation_factor: f64, // F: mute if dev > F AND not exempted
pub min_share_fraction: f64, // q: exemption if summed_credit >= q * largest_credit
/// 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 {
pools: HashMap::new(),
reference: None,
params,
}
Self::seeded(params, &[])
}
/// Judge a leg for acceptance. Must be called before apply().
pub fn judge(&mut self, leg: &Leg) -> JudgeResult {
// Leg is unpriceable if no token movement.
if leg.token_delta == 0 {
/// A policy primed with pool reserves as of some instant.
///
/// Credit is seeded for pools already trading within `F` of the reference, capped by
/// present depth. Pools sitting far off-reference start at zero and must earn credit
/// through an accepted print, so a snapshot cannot hand qualification to a pool that
/// was walked away from the market before the window opened.
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,
credit: 0,
},
);
}
policy.recompute_reference();
if let Some(reference) = policy.reference {
let f = policy.params.max_deviation_factor;
for state in policy.pools.values_mut() {
let Some(spot) = spot_ratio(state.sats, state.token_amount) else {
continue;
};
if deviation(spot, reference) <= f {
state.credit = min_depth(state.sats, state.token_amount, 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,
accepted: false,
reason: "no token movement".into(),
verdict: Verdict::Unpriceable,
};
}
let price = (leg.sats_delta.unsigned_abs() as f64) / (leg.token_delta.abs() as f64);
let price =
leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64;
// If no reference yet, accept (new token, first pool).
let Some(ref_price) = self.reference else {
let Some(reference) = self.reference else {
return JudgeResult {
price: Some(price),
accepted: true,
reason: "no reference yet".into(),
verdict: Verdict::NoReference,
};
};
// Check deviation from reference.
let dev = (price / ref_price).max(ref_price / price);
// Tier 1: deviation within F always prints.
let dev = deviation(price, reference);
if dev <= self.params.max_deviation_factor {
return JudgeResult {
price: Some(price),
accepted: true,
reason: format!("dev {:.2} <= F {:.1}", dev, self.params.max_deviation_factor),
verdict: Verdict::WithinBand { dev },
};
}
// Tier 2: exemption if this leg's pool credit is high enough.
let pool_credit = self.pools.get(&leg.pool).map(|ps| ps.credit).unwrap_or(0);
let largest_credit = self.pools.values().map(|ps| ps.credit).max().unwrap_or(0);
let threshold_credit = ((largest_credit as f64) * self.params.min_share_fraction) as u64;
let credit = self.pools.get(&leg.pool).map_or(0, |s| s.credit);
let largest_credit = self.pools.values().map(|s| s.credit).max().unwrap_or(0);
// `ceil().max(1)` keeps a pool that has never had an accepted print (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);
if largest_credit > 0 && pool_credit >= threshold_credit {
return JudgeResult {
price: Some(price),
accepted: true,
reason: format!(
"dev {:.2} > F but credit {}/{} >= {:.0}%",
dev,
pool_credit,
largest_credit,
self.params.min_share_fraction * 100.0
),
};
}
let verdict = if credit >= threshold {
Verdict::CreditExempt {
dev,
credit,
largest_credit,
}
} else {
Verdict::Muted {
dev,
credit,
largest_credit,
}
};
// Muted: deviation too high and credit too low.
JudgeResult {
price: Some(price),
accepted: false,
reason: format!(
"dev {:.2} > F {:.1} AND credit {}/{} < {:.0}%",
dev,
self.params.max_deviation_factor,
pool_credit,
largest_credit,
self.params.min_share_fraction * 100.0
),
verdict,
}
}
/// Apply a leg: update pool reserves, credit, and reference.
/// Must be called after judge(), regardless of acceptance.
/// Advance state past a leg. Must be called for every leg, accepted or not: a muted
/// leg still moved real reserves.
pub fn apply(&mut self, leg: &Leg, accepted: bool) {
// Compute min-depth before borrowing pools (to avoid borrow checker issues).
let new_credit = if accepted && leg.token_delta != 0 {
let ref_price = self.reference.unwrap_or(1.0);
let token_valued = ((leg.token_amount as f64) * ref_price).ceil() as u64;
leg.sats.min(token_valued)
} else {
0
};
let reference = self.reference;
// Ensure pool state exists.
let state = self
.pools
.entry(leg.pool)
.or_insert(PoolState {
sequence: leg.sequence,
sats: leg.sats,
token_amount: leg.token_amount,
credit: 0,
});
// `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,
credit: 0,
});
// Update reserves only if sequence is monotonic (later than what we've seen).
if leg.sequence > state.sequence {
if leg.sequence >= state.sequence {
state.sequence = leg.sequence;
state.sats = leg.sats;
state.token_amount = leg.token_amount;
// Update credit on accepted prints.
if new_credit > 0 {
state.credit = new_credit;
if accepted && leg.token_delta != 0 {
// Value the token side at the reference the leg was judged against; for a
// token's very first print there is none, so use the leg's own price.
let valuation = reference.unwrap_or_else(|| {
leg.sats_delta.unsigned_abs() as f64 / leg.token_delta.unsigned_abs() as f64
});
state.credit = min_depth(leg.sats, leg.token_amount, valuation);
}
}
// Recompute reference from all pools' reserves.
self.reference = self.compute_reference();
self.recompute_reference();
}
/// Compute min-depth for a pool: min(sats, tokens * reference).
/// If no reference yet, use sats (will get updated once reference exists).
fn min_depth(&self, sats: u64, token_amount: u64) -> u64 {
let ref_price = self.reference.unwrap_or(1.0);
let token_valued_at_ref = ((token_amount as f64) * ref_price).ceil() as u64;
sats.min(token_valued_at_ref)
}
/// Compute the min-depth-weighted median of pool spot ratios.
fn compute_reference(&self) -> Option<f64> {
if self.pools.is_empty() {
return None;
}
// Filter pools with token_amount >= 10 (avoid divide-by-zero and dust).
let mut entries: Vec<(f64, u64)> = self
.pools
.values()
.filter(|ps| ps.token_amount >= 10)
.map(|ps| {
let spot_ratio = (ps.sats as f64) / (ps.token_amount as f64);
let weight = self.min_depth(ps.sats, ps.token_amount);
(spot_ratio, weight)
})
.collect();
if entries.is_empty() {
return None;
}
// Sort by spot ratio ascending.
entries.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap());
// Compute total weight.
let total_weight: u128 = entries.iter().map(|(_, w)| *w as u128).sum();
// Find the ratio at the 50th percentile.
let half_weight = total_weight / 2;
let mut cumulative: u128 = 0;
for (spot_ratio, weight) in &entries {
cumulative += *weight as u128;
if cumulative >= half_weight {
return Some(*spot_ratio);
}
}
// Fallback: last ratio (shouldn't reach here if weights are positive).
entries.last().map(|(spot_ratio, _)| *spot_ratio)
}
/// Get current reference price (for testing/debugging).
/// 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 {
self.pools.get(pool).map_or(0, |s| s.credit)
}
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;
}
}
/// 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 {
if !(price > 0.0) || !(reference > 0.0) || !price.is_finite() || !reference.is_finite() {
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)?;
for _ in 0..REWEIGHT_PASSES {
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 {
return sorted.get(sorted.len() / 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 {
use super::*;
fn leg(
pool: u8,
ts: i64,
seq: i64,
sats_delta: i64,
token_delta: i64,
sats: u64,
token_amount: u64,
) -> Leg {
Leg {
txid: [0u8; 32],
pool: [pool; 32],
ts,
sequence: seq,
sats_delta,
token_delta,
sats,
token_amount,
}
}
#[test]
fn test_unpriceable_leg() {
let mut policy = Policy::new(GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
});
let l = leg(1, 1000, 1, 100, 0, 1000, 1000);
let judge = policy.judge(&l);
assert!(!judge.accepted);
assert!(judge.price.is_none());
}
#[test]
fn test_first_leg_always_accepts() {
let mut policy = Policy::new(GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
});
let l = leg(1, 1000, 1, 100, 1000, 1100, 2000);
let judge = policy.judge(&l);
assert!(judge.accepted, "{}", judge.reason);
assert!(judge.price.is_some());
policy.apply(&l, judge.accepted);
assert!(policy.reference().is_some());
}
#[test]
fn test_second_leg_within_deviation() {
let mut policy = Policy::new(GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
});
let l1 = leg(1, 1000, 1, 1000, 1000, 2000, 2000);
let j1 = policy.judge(&l1);
policy.apply(&l1, j1.accepted);
// Second leg at similar price (ref should be 1.0).
let l2 = leg(2, 1000, 2, 950, 1000, 2000, 2000);
let j2 = policy.judge(&l2);
assert!(j2.accepted, "{}", j2.reason);
policy.apply(&l2, j2.accepted);
}
#[test]
fn test_second_leg_far_from_reference_muted() {
let mut policy = Policy::new(GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
});
let l1 = leg(1, 1000, 1, 1000, 1000, 2000, 2000);
let j1 = policy.judge(&l1);
policy.apply(&l1, j1.accepted);
// Second leg at 100x price deviation, no credit (new pool).
let l2 = leg(2, 1000, 1, 100000, 1000, 101000, 1001);
let j2 = policy.judge(&l2);
assert!(!j2.accepted, "{}", j2.reason);
assert!(j2.price.is_some());
policy.apply(&l2, j2.accepted);
}
#[test]
fn test_credit_seeding_on_accepted_print() {
let mut policy = Policy::new(GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
});
// Pool 1: large, accepts at ref
let l1 = leg(1, 1000, 1, 10000, 1000, 11000, 2000);
let j1 = policy.judge(&l1);
policy.apply(&l1, j1.accepted);
// Pool 2: smaller, deviates but has no credit yet (new pool)
let l2 = leg(2, 1000, 1, 10000, 1000, 11000, 1000);
let j2 = policy.judge(&l2);
policy.apply(&l2, j2.accepted);
// Pool 2 should now have credit from its accepted print.
// A later muted print won't restore credit.
let l2_muted = leg(2, 1001, 2, 100000, 1000, 101000, 1001);
let j2_muted = policy.judge(&l2_muted);
assert!(!j2_muted.accepted);
policy.apply(&l2_muted, j2_muted.accepted);
// Verify pool 2's credit didn't rise.
let pool2_credit = policy.pools.get(&[2u8; 32]).map(|ps| ps.credit);
assert!(pool2_credit.is_some());
}
}
mod tests;

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

@ -0,0 +1,295 @@
// 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.accepted());
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
/// credit update behind `leg.sequence > state.sequence`, which is false on that very first
/// insert. A pool therefore had to make two accepted prints before earning any credit.
#[test]
fn test_pool_earns_credit_on_its_first_accepted_print() {
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 must hold credit after its first accepted print, not its second"
);
}
#[test]
fn test_muted_leg_does_not_earn_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 muted print must not qualify the pool that made it"
);
}
/// 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 a pool holding worthless tokens from voting the
/// reference away from where the real liquidity sits.
#[test]
fn test_dust_pool_does_not_move_the_reference() {
let policy = Policy::seeded(
params(),
&[
reserves(1, 1, 100_000_000, 1_000_000), // spot 100, deep
reserves(2, 2, 200_000_000, 2_000_000), // spot 100, deep
reserves(9, 3, 1_000, 10), // spot 100, but ~nothing behind it
],
);
assert_eq!(policy.reference(), Some(100.0));
}
#[test]
fn test_pools_below_the_dust_floor_are_ignored_entirely() {
let policy = Policy::seeded(
params(),
&[
reserves(1, 1, 1_000_000, 10_000), // spot 100
reserves(9, 2, 1_000_000, 1), // spot 1_000_000, below MIN_TOKEN_RESERVE
],
);
assert_eq!(policy.reference(), Some(100.0));
}
#[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 two-pool sets towards the lower ratio.
/// The comparison is written as `2 * cumulative >= total` to avoid that.
#[test]
fn test_weighted_median_is_not_biased_by_integer_division() {
let mut entries = vec![(100.0, 3, 1), (200.0, 3, 1)];
let picked = weighted_median_reference(&mut entries).unwrap();
assert!(
picked == 100.0 || picked == 200.0,
"expected one of the two inputs, got {picked}"
);
// 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));
}
/// 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));
}
#[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);
}
/// 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"
);
}

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

@ -144,8 +144,11 @@ async fn insert_multileg_trade_at(
/// 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_priced_by_gross_volume_not_net() {
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();
@ -178,14 +181,16 @@ async fn test_multipool_arb_priced_by_gross_volume_not_net() {
);
}
/// Every leg pointing the same way is the ordinary case: gross and net agree exactly,
/// so 99.76% of mainnet prints — including the OLA supply-shock crash — are untouched.
/// 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_price_matches_net_ratio() {
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;
@ -195,15 +200,46 @@ async fn test_single_direction_multileg_price_matches_net_ratio() {
.unwrap()
.expect("router transaction must price");
// Per-leg pricing: the close is the last leg's price, not the per-tx net ratio.
// Last leg: (99_000, -2_000) = 99,000 / 2,000 = 49.5
let last_leg_price = 99_000.0 / 2_000.0;
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 last leg's price: {price} vs {last_leg_price}"
"close should be the last leg's price: {price} vs {last_leg_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]

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

@ -13,7 +13,9 @@ use sqlx::{Row, SqlitePool};
///
/// 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.
pub const OHLCV_VERSION: u32 = 3;
/// 4: policy reference seeded from reserves rather than folded from the batch start, and
/// buckets with no priceable leg are skipped instead of written with f64 sentinels.
pub const OHLCV_VERSION: u32 = 4;
const OHLCV_VERSION_KEY: &str = "ohlcv_version";
pub async fn create_table(pool: &SqlitePool) {
@ -98,7 +100,9 @@ pub async fn rebuild_range(
since_ts: i64,
until_ts: i64,
) -> Result<u64> {
use crate::db::cauldron::candlestick::{GuardParams, Leg, Policy};
use crate::db::cauldron::candlestick::{
fetch_reserve_snapshot, to_hash32, GuardParams, Leg, Policy,
};
use std::collections::HashMap;
if since_ts >= until_ts {
@ -136,15 +140,12 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
}
// Phase 2: fold through legs per-token using the policy core to compute buckets.
let params = GuardParams {
max_deviation_factor: 5.0,
min_share_fraction: 0.05,
};
let params = GuardParams::default();
let mut buckets: HashMap<(Vec<u8>, i64), OhlcvBucket> = HashMap::new();
let mut current_token: Option<Vec<u8>> = None;
let mut policy = Policy::new(params.clone());
let mut policy = Policy::new(params);
for row in &rows {
let token_id: Vec<u8> = row.get(0);
@ -153,21 +154,19 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
let ts: i64 = row.get(8);
let bucket_ts = (ts / 3600) * 3600;
// Reset policy when we move to a new token.
// A new token starts a new policy — seeded from that token's reserves as they
// stood when this batch opened. Callers materialise history in 24-hour batches,
// 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());
policy = Policy::new(params.clone());
let snapshot = fetch_reserve_snapshot(read_pool, &token_id, since_ts).await?;
policy = Policy::seeded(params, &snapshot);
}
// Build the Leg struct.
let mut txid_arr = [0u8; 32];
let mut pool_arr = [0u8; 32];
txid_arr.copy_from_slice(&txid);
pool_arr.copy_from_slice(&pool_bytes);
let leg = Leg {
txid: txid_arr,
pool: pool_arr,
txid: to_hash32(&txid)?,
pool: to_hash32(&pool_bytes)?,
ts,
sequence: row.get(7),
sats_delta: row.get(3),
@ -180,7 +179,7 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
let judge = policy.judge(&leg);
let sats_delta_abs = leg.sats_delta.unsigned_abs() as i64;
let token_delta_abs = leg.token_delta.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);
@ -189,10 +188,10 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
// Always accumulate volume.
bucket.volume_sats += sats_delta_abs;
bucket.volume_tokens += token_delta_abs;
bucket.txids.insert(txid_arr);
bucket.txids.insert(leg.txid);
// Update OHLC only if accepted and priceable.
if judge.accepted {
if judge.accepted() {
if let Some(price) = judge.price {
if bucket.first_accepted_price.is_none() {
bucket.first_accepted_price = Some(price);
@ -203,7 +202,7 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
}
}
policy.apply(&leg, judge.accepted);
policy.apply(&leg, judge.accepted());
}
// Phase 3: insert all computed buckets in one transaction.
@ -211,8 +210,15 @@ ORDER BY phe.token_id ASC, phe.effective_timestamp ASC, phe.sequence ASC;
let mut inserted = 0u64;
for ((token_id, bucket_ts), bucket) in buckets {
let open = bucket.first_accepted_price.unwrap_or(bucket.last_accepted_price.unwrap_or(0.0));
let close = bucket.last_accepted_price.unwrap_or(bucket.first_accepted_price.unwrap_or(0.0));
// A bucket whose legs were all muted or unpriceable has no OHLC to report: its
// `high`/`low` are still the f64::MIN/MAX sentinels and open/close are unset.
// Writing that row would put ±1.8e308 into the table and hand it straight to the
// chart. Skip it, exactly as the SQL this replaced did by inner-joining prices to
// volume; the raw path's carry-forward renders the gap from the previous close.
let (Some(open), Some(close)) = (bucket.first_accepted_price, bucket.last_accepted_price)
else {
continue;
};
inserted += sqlx::query(
"INSERT OR IGNORE INTO ohlcv_1h
@ -505,6 +511,36 @@ mod tests {
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"
);
}
/// A multi-pool arbitrage transaction whose legs nearly cancel must materialise the
/// price its legs executed at, not the signed-net ratio.
#[tokio::test]

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

@ -66,6 +66,7 @@ async fn setup_mock_db(pool: sqlx::SqlitePool) {
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq();
let mut conn = pool.acquire().await.unwrap();
@ -380,6 +381,7 @@ async fn test_single_swap_multiple_pools() {
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq();
let mut conn = pool.acquire().await.unwrap();
@ -481,6 +483,7 @@ async fn test_candle_carries_price_when_net_zero_tokens_but_has_volume() {
utxo_funding::create_table(&pool).await;
tx::create_table(&pool).await;
pool::create_table(&pool).await;
ohlcv::create_table(&pool).await;
dummy_init_seq();
let mut conn = pool.acquire().await.unwrap();