diff --git a/src/db/cauldron/candlestick/mod.rs b/src/db/cauldron/candlestick/mod.rs index 5e8745a..bd40d46 100644 --- a/src/db/cauldron/candlestick/mod.rs +++ b/src/db/cauldron/candlestick/mod.rs @@ -71,7 +71,7 @@ impl PriceInterval { } fn aggregate_raw_trades( - all_trades: &[(i64, i64, i64, i64, i64)], + all_trades: &[(i64, i64, i64)], intervals: Vec, step_size: i64, mut found_first_trade: bool, @@ -87,7 +87,7 @@ fn aggregate_raw_trades( let mut first_trade_in_interval = true; while trade_index < all_trades.len() { - let (ts, signed_sats, signed_tokens, vol_sats, vol_tokens) = all_trades[trade_index]; + let (ts, vol_sats, vol_tokens) = all_trades[trade_index]; if ts < interval_start { trade_index += 1; continue; @@ -96,8 +96,8 @@ fn aggregate_raw_trades( break; } - if signed_tokens != 0 { - let price = (signed_sats as f64 / signed_tokens as f64).abs(); + if vol_tokens != 0 { + let price = vol_sats as f64 / vol_tokens as f64; if first_trade_in_interval { pi.open = Some(price); pi.high = price; @@ -206,59 +206,33 @@ fn fill_ohlcv_candles( (result, found_first_trade, last_close) } +/// Returns one row per transaction: `(effective_timestamp, volume_sats, volume_tokens)`. +/// +/// Volumes are gross sums of the absolute per-leg deltas, so the derived price +/// `volume_sats / volume_tokens` is the volume-weighted average of the prices actually +/// executed by that transaction's legs, and is therefore always bounded by the cheapest +/// and dearest leg. Summing the *signed* deltas instead lets a multi-pool arbitrage +/// transaction — which buys from one pool and sells into others — cancel almost all of +/// its token movement and divide real satoshis by a near-zero remainder, fabricating a +/// price no leg ever traded at. async fn fetch_raw_trades( pool: &SqlitePool, token_blob: &[u8], timestamp_start: i64, timestamp_end: i64, -) -> Result> { +) -> Result> { let sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - phe.txid, - phe.effective_timestamp, - phe.utxo, - phe.sats_delta, - phe.token_delta, - phe.sequence - FROM pool_history_entry AS phe - WHERE phe.token_id = ? - AND phe.effective_timestamp >= ? - AND phe.effective_timestamp < ? -), -per_pool_tx AS ( - SELECT - txid, - effective_timestamp, - utxo, - MIN(sequence) AS min_sequence, - SUM(sats_delta) AS signed_sats_pool, - SUM(token_delta) AS signed_tokens_pool, - SUM(ABS(sats_delta)) AS volume_sats_pool, - SUM(ABS(token_delta)) AS volume_tokens_pool - FROM per_pool_tx_raw - GROUP BY txid, effective_timestamp, utxo -), -tx_trades AS ( - SELECT - txid, - effective_timestamp, - MIN(min_sequence) AS min_sequence, - SUM(signed_sats_pool) AS signed_sats, - SUM(signed_tokens_pool) AS signed_tokens, - SUM(volume_sats_pool) AS volume_sats, - SUM(volume_tokens_pool) AS volume_tokens - FROM per_pool_tx - GROUP BY txid, effective_timestamp -) SELECT - effective_timestamp, - signed_sats, - signed_tokens, - volume_sats, - volume_tokens -FROM tx_trades -ORDER BY effective_timestamp ASC, min_sequence ASC; + phe.effective_timestamp, + SUM(ABS(phe.sats_delta)) AS volume_sats, + SUM(ABS(phe.token_delta)) AS volume_tokens, + MIN(phe.sequence) AS min_sequence +FROM pool_history_entry AS phe +WHERE phe.token_id = ? + AND phe.effective_timestamp >= ? + AND phe.effective_timestamp < ? +GROUP BY phe.txid, phe.effective_timestamp +ORDER BY phe.effective_timestamp ASC, min_sequence ASC; "#; let rows = sqlx::query(sql) .bind(token_blob) @@ -269,7 +243,7 @@ ORDER BY effective_timestamp ASC, min_sequence ASC; Ok(rows .into_iter() - .map(|r| (r.get(0), r.get(1), r.get(2), r.get(3), r.get(4))) + .map(|r| (r.get(0), r.get(1), r.get(2))) .collect()) } @@ -282,43 +256,14 @@ async fn fetch_last_close_before( timestamp_end: i64, ) -> Result> { let sql = r#" -WITH per_pool_tx_raw AS ( - SELECT - phe.txid, - phe.effective_timestamp, - phe.utxo, - phe.sats_delta, - phe.token_delta, - phe.sequence - FROM pool_history_entry AS phe - WHERE phe.token_id = ? - AND phe.effective_timestamp < ? -), -per_pool_tx AS ( - SELECT - txid, - effective_timestamp, - utxo, - MIN(sequence) AS min_sequence, - SUM(sats_delta) AS signed_sats_pool, - SUM(token_delta) AS signed_tokens_pool - FROM per_pool_tx_raw - GROUP BY txid, effective_timestamp, utxo -), -tx_trades AS ( - SELECT - txid, - effective_timestamp, - MIN(min_sequence) AS min_sequence, - SUM(signed_sats_pool) AS signed_sats, - SUM(signed_tokens_pool) AS signed_tokens - FROM per_pool_tx - GROUP BY txid, effective_timestamp -) -SELECT ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS close_price -FROM tx_trades -WHERE signed_tokens != 0 -ORDER BY effective_timestamp DESC, min_sequence DESC +SELECT + CAST(SUM(ABS(phe.sats_delta)) AS REAL) / CAST(SUM(ABS(phe.token_delta)) AS REAL) AS close_price +FROM pool_history_entry AS phe +WHERE phe.token_id = ? + AND phe.effective_timestamp < ? +GROUP BY phe.txid, phe.effective_timestamp +HAVING SUM(ABS(phe.token_delta)) != 0 +ORDER BY phe.effective_timestamp DESC, MIN(phe.sequence) DESC LIMIT 1 "#; let row = sqlx::query(sql) diff --git a/src/db/cauldron/candlestick/tests.rs b/src/db/cauldron/candlestick/tests.rs index f6e1409..7ddeb4f 100644 --- a/src/db/cauldron/candlestick/tests.rs +++ b/src/db/cauldron/candlestick/tests.rs @@ -86,6 +86,168 @@ async fn insert_trade_at( .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, + 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. +#[tokio::test] +async fn test_multipool_arb_priced_by_gross_volume_not_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"); + + 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}]" + ); + + // Net-ratio pricing would divide 61,585,199 sats by 2 token units. + let net_ratio = 61_585_199.0 / 2.0; + assert!( + price < net_ratio / 1000.0, + "price {price} must not resemble the netting artifact {net_ratio}" + ); + + let expected = 831_397_279.0 / 2_669_054_136.0; + assert!((price - expected).abs() < 1e-9, "expected {expected}"); +} + +/// 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. +#[tokio::test] +async fn test_single_direction_multileg_price_matches_net_ratio() { + let db = mock_db_pool(setup_db).await; + let token = TokenID::from_byte_array([0xC2; 32]); + let token_blob = token.to_blob(); + + 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 signed_sats: i64 = legs.iter().map(|l| l.0).sum(); + let signed_tokens: i64 = legs.iter().map(|l| l.1).sum(); + let net_ratio = (signed_sats as f64 / signed_tokens as f64).abs(); + assert!( + (price - net_ratio).abs() < f64::EPSILON, + "single-direction transactions must be unaffected: {price} vs {net_ratio}" + ); +} + +/// 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"); + + // 20_000 gross sats over 2_000 gross tokens, between the 9 and 11 leg prices. + assert!((price - 10.0).abs() < f64::EPSILON, "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] async fn test_fetch_last_close_before_no_trades() { let db = mock_db_pool(setup_db).await; diff --git a/src/db/cauldron/ohlcv.rs b/src/db/cauldron/ohlcv.rs index 68b3c6a..fb168c0 100644 --- a/src/db/cauldron/ohlcv.rs +++ b/src/db/cauldron/ohlcv.rs @@ -3,9 +3,18 @@ // This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later. // A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html +use crate::db::cauldron::config::{config_get, config_set}; use anyhow::Result; 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. +pub const OHLCV_VERSION: u32 = 2; +const OHLCV_VERSION_KEY: &str = "ohlcv_version"; + pub async fn create_table(pool: &SqlitePool) { sqlx::query( "CREATE TABLE IF NOT EXISTS ohlcv_1h ( @@ -26,6 +35,30 @@ pub async fn create_table(pool: &SqlitePool) { .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 { + let stored = config_get(read_pool, OHLCV_VERSION_KEY) + .await? + .and_then(|v| v.parse::().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. pub async fn get_max_bucket_ts(pool: &SqlitePool) -> Result> { let row: Option<(Option,)> = sqlx::query_as("SELECT MAX(bucket_ts) FROM ohlcv_1h") @@ -90,8 +123,6 @@ per_pool_tx AS ( (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 @@ -104,8 +135,6 @@ tx_trades AS ( 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 @@ -115,11 +144,11 @@ priceable AS ( SELECT token_id, bucket_ts, - ABS(CAST(signed_sats AS REAL) / CAST(signed_tokens AS REAL)) AS price, + CAST(vol_sats AS REAL) / CAST(vol_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 + WHERE vol_tokens != 0 ), ohlc AS ( SELECT @@ -286,7 +315,7 @@ mod tests { token_delta: i64, ) { 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(blockhash.as_slice()) .bind(mtp_ts) @@ -446,4 +475,91 @@ mod tests { .unwrap(); assert_eq!(n, 0, "mempool trades must not be materialised"); } + + /// 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: f64 = sqlx::query_scalar("SELECT close FROM ohlcv_1h WHERE token_id = ?") + .bind(token.as_slice()) + .fetch_one(&pool) + .await + .unwrap(); + + let expected = 831_397_279.0 / 2_669_054_136.0; + assert!( + (close - expected).abs() < 1e-9, + "materialised close {close} should be the gross ratio {expected}" + ); + } + + /// 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"); + } } diff --git a/src/main.rs b/src/main.rs index 29f76c5..a464023 100644 --- a/src/main.rs +++ b/src/main.rs @@ -419,6 +419,25 @@ async fn launch() -> _ { // Ensure the OHLCV pre-aggregation table exists (safe on both new and existing DBs). 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). let max_bucket_ts = db::cauldron::ohlcv::get_max_bucket_ts(&dbpool.cauldron_r) .await @@ -431,7 +450,11 @@ async fn launch() -> _ { // Synchronous post-IBD backfill: run the full ohlcv_1h materialisation before // allowing metrics_cache and other background writers to start. We reuse the // 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_SAFETY_SECS: i64 = 3 * 3600; diff --git a/src/rpc/candlesticks/tests.rs b/src/rpc/candlesticks/tests.rs index 03756c4..d7fbc1f 100644 --- a/src/rpc/candlesticks/tests.rs +++ b/src/rpc/candlesticks/tests.rs @@ -664,10 +664,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["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]; 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["close"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON); assert!((c2["low"].as_f64().unwrap() - 50.0).abs() < f64::EPSILON);