Price candlesticks by gross volume instead of net deltas
A transaction that trades against several pools of the same token had its price computed as |SUM(sats_delta) / SUM(token_delta)|. Arbitrage routers buy from one pool and sell into the others, so the token deltas very nearly cancel, and the division turned real satoshis into a price no leg ever traded at. Mainnet token NWB printed 30,792,599.5 sats/unit from a 27-pool sweep whose legs all executed between 0.288 and 0.335 — a hundred-million-fold error, and the visible spike on its chart. 820 such transactions exist across 91 tokens. Price is now SUM(ABS(sats_delta)) / SUM(ABS(token_delta)): the volume-weighted average of the prices the transaction's legs actually executed at, which is always bounded by its cheapest and dearest leg. For single-direction transactions — 99.76% of all prints, including the OLA supply-shock crash — this is arithmetically identical to the old formula, so honest history is untouched. Transactions whose legs cancel exactly used to print nothing and let the candle carry the previous close; they now price from their legs like any other trade. ohlcv_1h is materialised with INSERT OR IGNORE and the materialiser only ever moves forward, so contaminated buckets could never be corrected in place. An ohlcv_version config key clears the table once when the pricing rule changes. The synchronous post-IBD backfill is skipped on that pass: it runs before rocket::build() returns, so rebuilding all of history there would refuse connections for the duration instead of falling back to the raw query path. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jakobsn
parent
4a23ce70d4
commit
283c0c3e2c
5 changed files with 346 additions and 98 deletions
|
|
@ -71,7 +71,7 @@ impl PriceInterval {
|
|||
}
|
||||
|
||||
fn aggregate_raw_trades(
|
||||
all_trades: &[(i64, i64, i64, i64, i64)],
|
||||
all_trades: &[(i64, i64, i64)],
|
||||
intervals: Vec<PriceInterval>,
|
||||
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<Vec<(i64, i64, i64, i64, i64)>> {
|
||||
) -> Result<Vec<(i64, i64, i64)>> {
|
||||
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
|
||||
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 < ?
|
||||
),
|
||||
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;
|
||||
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<Option<f64>> {
|
||||
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
|
||||
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 < ?
|
||||
),
|
||||
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
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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<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.
|
||||
#[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;
|
||||
|
|
|
|||
|
|
@ -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<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.
|
||||
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")
|
||||
|
|
@ -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");
|
||||
}
|
||||
}
|
||||
|
|
|
|||
25
src/main.rs
25
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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue