From 51a303032cb3fffe5b353c53f19a6ae881e11db8 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Fri, 18 Sep 2026 13:34:05 +0200 Subject: [PATCH 1/7] feat(lore-0290): index SushiSwap V3, reusing the pair swap decode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SushiSwap V3 pools emit the same CLMM shape Soroswap's already do — topics [Symbol("swap")], signed amount0/amount1 — so the extractor is made venue-neutral (TokenPair / PairPoolRegistry / PairSwapExtractor, with the Soroswap* names kept as aliases) rather than copied. Only the venue stamped on the TradeRow differs. learn_factory gains a pool_created arm. It matches on shape like every other arm, so the three earlier factory generations whose pools still trade register through the same path as the live factory — no factory address is hardcoded anywhere. The two pair-backed venues keep separate registries, so a contract_id can never resolve to the other venue's tokens. unregistered_pool_venue must key on the DATA, not the topic: SushiSwap's routers emit the identical [Symbol("swap")] topic and differ only in carrying amount_in/amount_out against the pool's amount0/amount1. Matching the topic alone would count every routed swap as a missing pool — the double-counting shape 0285 flagged. --- packages/extractors-core/src/lib.rs | 7 + packages/ledger-processor/src/dispatch.rs | 30 ++- .../prices-ingest-core/src/registry_io.rs | 31 ++- packages/prices-ingest-core/src/soroban.rs | 225 +++++++++++++++++- packages/soroswap-extractor/src/lib.rs | 76 ++++-- 5 files changed, 337 insertions(+), 32 deletions(-) diff --git a/packages/extractors-core/src/lib.rs b/packages/extractors-core/src/lib.rs index 44f09b47..c808c64d 100644 --- a/packages/extractors-core/src/lib.rs +++ b/packages/extractors-core/src/lib.rs @@ -5,6 +5,11 @@ pub enum Venue { Soroswap, Aquarius, Phoenix, + /// SushiSwap V3 — a Uniswap-v3-style concentrated-liquidity venue (task + /// 0290). Its pool `swap` carries signed `amount0`/`amount1`, the same + /// shape [`Venue::Soroswap`]'s CLMM pools use, so both decode through + /// [`TokenPair`]-backed extraction. + Sushiswap, } impl Venue { @@ -15,6 +20,7 @@ impl Venue { Venue::Soroswap => "soroswap", Venue::Aquarius => "aquarius", Venue::Phoenix => "phoenix", + Venue::Sushiswap => "sushiswap", } } @@ -25,6 +31,7 @@ impl Venue { "soroswap" => Some(Venue::Soroswap), "aquarius" => Some(Venue::Aquarius), "phoenix" => Some(Venue::Phoenix), + "sushiswap" => Some(Venue::Sushiswap), _ => None, } } diff --git a/packages/ledger-processor/src/dispatch.rs b/packages/ledger-processor/src/dispatch.rs index 50e6a614..0247a485 100644 --- a/packages/ledger-processor/src/dispatch.rs +++ b/packages/ledger-processor/src/dispatch.rs @@ -6,7 +6,7 @@ use phoenix_extractor::{ PHOENIX_STABLE_EVENT_COUNT, PHOENIX_XYK_MIN_EVENT_COUNT, POOL_TYPE_XYK, PhoenixPoolRegistry, PhoenixXykExtractor, }; -use soroswap_extractor::{SoroswapPairExtractor, SoroswapPoolRegistry}; +use soroswap_extractor::{PairSwapExtractor, SoroswapPairExtractor, SoroswapPoolRegistry}; #[derive(Debug, thiserror::Error)] pub enum DispatchError { @@ -74,14 +74,18 @@ pub fn dispatch_phoenix( /// Top-level dispatcher: routes events by venue, then by pool shape for Phoenix. /// -/// Soroswap requires the pool→tokens registry to resolve token identities; an -/// unresolved pool (created before the indexed window) yields no trades rather -/// than an error. Aquarius and Phoenix carry tokens inline. +/// Soroswap and SushiSwap require the pool→tokens registry to resolve token +/// identities; an unresolved pool (created before the indexed window) yields no +/// trades rather than an error. Aquarius and Phoenix carry tokens inline. +/// +/// The two pair-backed venues keep SEPARATE registries so a contract_id can +/// never resolve to the wrong venue's tokens (task 0290). pub fn dispatch( rows: &[SorobanEventRow], venue_registry: &VenueRegistry, phoenix_registry: &PhoenixPoolRegistry, soroswap_registry: &SoroswapPoolRegistry, + sushiswap_registry: &SoroswapPoolRegistry, ) -> Result, DispatchError> { if rows.is_empty() { return Ok(vec![]); @@ -96,6 +100,12 @@ pub fn dispatch( Some(pair) => Ok(SoroswapPairExtractor::new(pair).extract(rows)?.trades), None => Ok(vec![]), }, + Some(Venue::Sushiswap) => match sushiswap_registry.lookup(contract_id) { + Some(pair) => Ok(PairSwapExtractor::with_venue(Venue::Sushiswap, pair) + .extract(rows)? + .trades), + None => Ok(vec![]), + }, Some(Venue::Aquarius) => Ok(AquariusPoolExtractor.extract(rows)?.trades), None => Ok(vec![]), } @@ -128,6 +138,7 @@ mod tests { &venue_reg, &phoenix_reg, &SoroswapPoolRegistry::new(), + &SoroswapPoolRegistry::new(), ) .unwrap(); assert_eq!(trades.len(), 1); @@ -146,6 +157,7 @@ mod tests { &venue_reg, &phoenix_reg, &SoroswapPoolRegistry::new(), + &SoroswapPoolRegistry::new(), ) .unwrap(); assert_eq!(trades.len(), 1); @@ -197,6 +209,7 @@ mod tests { &venue_reg, &phoenix_reg, &SoroswapPoolRegistry::new(), + &SoroswapPoolRegistry::new(), ) .unwrap(); assert!(trades.is_empty()); @@ -207,7 +220,14 @@ mod tests { let phoenix_reg = phoenix_registry_both_wasm_variants(); let venue_reg = venue_registry_phoenix(&[XLM_USDC_POOL]); - let trades = dispatch(&[], &venue_reg, &phoenix_reg, &SoroswapPoolRegistry::new()).unwrap(); + let trades = dispatch( + &[], + &venue_reg, + &phoenix_reg, + &SoroswapPoolRegistry::new(), + &SoroswapPoolRegistry::new(), + ) + .unwrap(); assert!(trades.is_empty()); } } diff --git a/packages/prices-ingest-core/src/registry_io.rs b/packages/prices-ingest-core/src/registry_io.rs index 0dbb8e87..59a29075 100644 --- a/packages/prices-ingest-core/src/registry_io.rs +++ b/packages/prices-ingest-core/src/registry_io.rs @@ -61,6 +61,12 @@ impl Registries { } } } + Venue::Sushiswap => { + if let Some(p) = self.sushiswap.lookup(contract_id) { + row.token0 = p.token0.clone(); + row.token1 = p.token1.clone(); + } + } Venue::Aquarius => {} } row @@ -111,6 +117,13 @@ impl Registries { row.token1.clone(), ); } + Venue::Sushiswap => { + self.sushiswap.register( + row.contract_id.clone(), + row.token0.clone(), + row.token1.clone(), + ); + } Venue::Phoenix => match hex_decode32(&row.wasm_hash) { Some(hash) => self.phoenix.register_with_wasm( row.contract_id.clone(), @@ -168,9 +181,13 @@ mod tests { reg.phoenix .register_with_wasm("CPHOENIX".into(), 0, [0xab; 32]); reg.venue.insert("CAQUA".into(), Venue::Aquarius); + // Task 0290: pair-backed like Soroswap, but its OWN registry. + reg.venue.insert("CSUSHI".into(), Venue::Sushiswap); + reg.sushiswap + .register("CSUSHI".into(), "CSUSHI0".into(), "CSUSHI1".into()); let rows = reg.to_pool_rows(); - assert_eq!(rows.len(), 3); + assert_eq!(rows.len(), 4); // Sorted, stable order. assert_eq!(rows[0].contract_id, "CAQUA"); @@ -187,6 +204,18 @@ mod tests { ); let ph = loaded.phoenix.lookup("CPHOENIX").expect("phoenix pool"); assert_eq!(ph.wasm_hash, Some([0xab; 32])); + + // The sushiswap pool round-trips into its own registry, and the two + // pair-backed venues stay disjoint (task 0290). + assert_eq!(loaded.venue.get("CSUSHI"), Some(&Venue::Sushiswap)); + let su = loaded.sushiswap.lookup("CSUSHI").expect("sushiswap pair"); + assert_eq!( + (su.token0.as_str(), su.token1.as_str()), + ("CSUSHI0", "CSUSHI1") + ); + assert!(!loaded.soroswap.contains("CSUSHI")); + assert!(!loaded.sushiswap.contains("CSOROSWAP")); + assert_eq!(loaded.pool_count(), reg.pool_count()); } diff --git a/packages/prices-ingest-core/src/soroban.rs b/packages/prices-ingest-core/src/soroban.rs index aed170b4..039958d2 100644 --- a/packages/prices-ingest-core/src/soroban.rs +++ b/packages/prices-ingest-core/src/soroban.rs @@ -47,6 +47,10 @@ pub struct Registries { pub venue: VenueRegistry, pub phoenix: PhoenixPoolRegistry, pub soroswap: SoroswapPoolRegistry, + /// SushiSwap V3 pools (task 0290). Same type as `soroswap` — both are + /// pair-backed — but a SEPARATE instance, so a contract_id can never + /// resolve to the other venue's tokens. + pub sushiswap: SoroswapPoolRegistry, } impl Default for Registries { @@ -55,6 +59,7 @@ impl Default for Registries { venue: VenueRegistry::new(), phoenix: PhoenixPoolRegistry::new(), soroswap: SoroswapPoolRegistry::new(), + sushiswap: SoroswapPoolRegistry::new(), } } } @@ -65,7 +70,7 @@ impl Registries { } pub fn pool_count(&self) -> usize { - self.soroswap.pool_count() + self.phoenix.pool_count() + self.soroswap.pool_count() + self.phoenix.pool_count() + self.sushiswap.pool_count() } } @@ -532,7 +537,13 @@ fn classify_amm_groups( }; let source = venue.as_source(); - match dispatch(&rows, ®.venue, ®.phoenix, ®.soroswap) { + match dispatch( + &rows, + ®.venue, + ®.phoenix, + ®.soroswap, + ®.sushiswap, + ) { Ok(trades) => { for t in trades { if let Some(tick) = amm_trade_to_tick(&t, transaction_index, closed_at, assets) @@ -569,7 +580,14 @@ fn classify_amm_groups( // deliberately not recorded here. Inferring the gap from "no tick" would // flood `unresolved_pools` with false positives on healthy pools and // grow the run's in-memory `unresolved` unboundedly. - if matches!(venue, Venue::Soroswap) && !reg.soroswap.contains(&contract_id) { + let pair_unresolved = match venue { + Venue::Soroswap => !reg.soroswap.contains(&contract_id), + // SushiSwap is pair-backed too (task 0290), so the same miss is + // possible and must be just as loud. + Venue::Sushiswap => !reg.sushiswap.contains(&contract_id), + Venue::Aquarius | Venue::Phoenix => false, + }; + if pair_unresolved { if let Some(rec) = unresolved_from_swaps(contract_id, &swaps, ledger_seq) { out.unresolved.push(rec); } @@ -606,9 +624,18 @@ fn unresolved_from_swaps( /// - Phoenix XYK swap: a group of `[String("swap"), String()]` rows, of /// which exactly one carries `sell_token`, so that row stands for the swap. /// -/// The router `swap` summaries (Aquarius, Soroswap) and the unindexed -/// Uniswap-v3-style venue (task 0290) match none of these, so on a complete -/// registry this counts nothing. +/// - SushiSwap V3 pool `swap`: `[Symbol("swap")]` **with `amount0`/`amount1` in +/// the data** (task 0290). +/// +/// The router `swap` summaries match none of these, so on a complete registry +/// this counts nothing. +/// +/// ⚠️ The SushiSwap arm CANNOT key on the topic alone: its two routers +/// (`CDMIM23W…`, `CAUF4DFY…`) emit the identical `[Symbol("swap")]` topic and +/// differ only in the data — `{ amount_in, amount_out }` against the pool's +/// `{ amount0, amount1, liquidity, sqrt_price_x96, tick }`. Matching the topic +/// alone would count every router swap as a missing pool, which is the +/// double-counting shape task 0285 warned about. fn unregistered_pool_venue(row: &SorobanEventRow) -> Option { let t0 = row.topics.first().and_then(|t| t.as_str())?; let is_address = |i: usize| row.topics.get(i).and_then(|t| t.as_address()).is_some(); @@ -616,10 +643,24 @@ fn unregistered_pool_venue(row: &SorobanEventRow) -> Option { "trade" if is_address(1) && is_address(2) => Some(Venue::Aquarius), "SoroswapPair" if topic_str(row, 1) == Some("swap") => Some(Venue::Soroswap), "swap" if topic_str(row, 1) == Some("sell_token") => Some(Venue::Phoenix), + // Ordered after Phoenix's `swap`: both start at the same topic, and + // Phoenix's is decided by topic[1] before the data is consulted. + "swap" if has_data_key(row, "amount0") && has_data_key(row, "amount1") => { + Some(Venue::Sushiswap) + } _ => None, } } +/// Whether the event's data map carries `key`. Used to tell a CLMM pool `swap` +/// from a router `swap` that shares its topic shape. +fn has_data_key(row: &SorobanEventRow, key: &str) -> bool { + match &row.data { + TaggedValue::Map(m) => m.iter().any(|(k, _)| k.as_str() == Some(key)), + _ => false, + } +} + /// The Symbol/String value of topic `i`. `TaggedValue::as_str` also answers for /// an Address, which must not pass for an action or field name. fn topic_str(row: &SorobanEventRow, i: usize) -> Option<&str> { @@ -707,6 +748,31 @@ fn learn_factory(topics: &Value, data: &Value, reg: &mut Registries) { reg.venue.insert(pair, Venue::Soroswap); } } + return; + } + + // SushiSwap V3 factory: [Symbol("pool_created")], data + // { fee, pool_address, sender, tick_spacing, token0, token1 } (task 0290). + // + // Matched by SHAPE, like every arm here — no factory address is hardcoded — + // which is what makes all four deployed factory generations (three on wasm + // FC9B0DF0, the live one on 9F94C577) register through this one arm. All + // three addresses are required, so a differently-shaped `pool_created` from + // some other protocol cannot register a pool with empty tokens. + if sig0 == Some("pool_created") { + if let TaggedValue::Map(m) = json_to_tagged(data) { + let get = |k: &str| { + m.iter() + .find(|(key, _)| key.as_str() == Some(k)) + .and_then(|(_, v)| v.as_address().map(String::from)) + }; + if let (Some(pool), Some(t0), Some(t1)) = + (get("pool_address"), get("token0"), get("token1")) + { + reg.sushiswap.register(pool.clone(), t0, t1); + reg.venue.insert(pool, Venue::Sushiswap); + } + } } } @@ -922,6 +988,67 @@ mod tests { assert_eq!(topic_symbol(&topics, 1), Some("new_pair")); } + /// Task 0290. The SushiSwap V3 factory's `pool_created` carries the token + /// pair in the EVENT, so a pool is learned exactly like a Soroswap + /// `new_pair` — no contract-storage read. + /// + /// Payload is a real production event from the live factory + /// `CD3KRKGD…GLYF` at ledger 64,116,662. Because `learn_factory` matches on + /// SHAPE and never on a factory address, this one arm also covers the three + /// earlier factory generations (wasm `FC9B0DF0`) whose pools still trade. + #[test] + fn sushiswap_factory_pool_created_learns_the_pair() { + let topics = json!([{"type":"sym","value":"pool_created"}]); + let data = json!({"type":"map","value":[ + {"key":{"type":"sym","value":"fee"},"value":{"type":"u32","value":500}}, + {"key":{"type":"sym","value":"pool_address"}, + "value":{"type":"address","value":"CBVHBZSZOS6KRDJ4D44FU2YLIENOVSSLM3UGKW6XQMVIFUAMWIWCVH2U"}}, + {"key":{"type":"sym","value":"sender"}, + "value":{"type":"address","value":"CD3KRKGDRVWPXVB3VXLUMQKMX6XZ6Q2H334IVZD4XXNAMKSRVQL5GLYF"}}, + {"key":{"type":"sym","value":"tick_spacing"},"value":{"type":"i32","value":10}}, + {"key":{"type":"sym","value":"token0"}, + "value":{"type":"address","value":"CBSJZEIO5C7KC2SF3MKSNXXJSW5G3VTNBX4ATMKUI3B2MR4JKM4R26YF"}}, + {"key":{"type":"sym","value":"token1"}, + "value":{"type":"address","value":"CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75"}} + ]}); + + let mut reg = Registries::new(); + learn_factory(&topics, &data, &mut reg); + + let pool = "CBVHBZSZOS6KRDJ4D44FU2YLIENOVSSLM3UGKW6XQMVIFUAMWIWCVH2U"; + assert_eq!(reg.venue.get(pool), Some(&Venue::Sushiswap)); + let pair = reg.sushiswap.lookup(pool).expect("pair registered"); + assert_eq!( + pair.token0, + "CBSJZEIO5C7KC2SF3MKSNXXJSW5G3VTNBX4ATMKUI3B2MR4JKM4R26YF" + ); + assert_eq!( + pair.token1, + "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75" + ); + // It must not leak into the Soroswap registry — separate instances. + assert!(!reg.soroswap.contains(pool)); + } + + /// A `pool_created` missing an address registers nothing, rather than a + /// pool with empty tokens that would later price against asset "". + #[test] + fn sushiswap_pool_created_without_tokens_registers_nothing() { + let topics = json!([{"type":"sym","value":"pool_created"}]); + let data = json!({"type":"map","value":[ + {"key":{"type":"sym","value":"pool_address"}, + "value":{"type":"address","value":"CBVHBZSZOS6KRDJ4D44FU2YLIENOVSSLM3UGKW6XQMVIFUAMWIWCVH2U"}}, + {"key":{"type":"sym","value":"token0"}, + "value":{"type":"address","value":"CBSJZEIO5C7KC2SF3MKSNXXJSW5G3VTNBX4ATMKUI3B2MR4JKM4R26YF"}} + ]}); + + let mut reg = Registries::new(); + learn_factory(&topics, &data, &mut reg); + + assert_eq!(reg.pool_count(), 0); + assert!(reg.venue.is_empty()); + } + #[test] fn reflector_resolves_quote_symbols_to_canonical_identities() { assert_eq!( @@ -1181,7 +1308,29 @@ mod tests { ], 0, ); - // The Uniswap-v3-style venue (task 0290): a bare `swap` with map data. + // ⚠️ A SushiSwap V3 ROUTER (task 0290): the SAME bare `swap` topic its + // pools use, told apart only by the data — routers carry + // `amount_in`/`amount_out`, pools carry `amount0`/`amount1`. Counting + // this would double-count every routed swap alongside the pool swap it + // wraps, which is the hazard task 0285 flagged. + let sushiswap_router = SorobanEventRow { + data: TaggedValue::Map(vec![ + ( + TaggedValue::Symbol("amount_in".into()), + TaggedValue::I128(2_539_492_571), + ), + ( + TaggedValue::Symbol("amount_out".into()), + TaggedValue::I128(13_410_007_617), + ), + ]), + ..event( + "CAUF4DFYSX52L2KJ4J7OFW3WDQMEUDVXNB7PG5VIC4VVOA3BCLWXDO2E", + vec![TaggedValue::Symbol("swap".into())], + 0, + ) + }; + // A bare `swap` carrying no amounts at all resolves to nothing either. let clmm = event("CCLMM", vec![TaggedValue::Symbol("swap".into())], 0); // A `trade` whose topics are not token addresses. let other_trade = event( @@ -1202,6 +1351,7 @@ mod tests { for row in [ aquarius_router, soroswap_router, + sushiswap_router, clmm, other_trade, odd_swap, @@ -1210,6 +1360,67 @@ mod tests { } } + /// The counterpart of the row above: the SushiSwap V3 POOL shape, which + /// shares the router's `[Symbol("swap")]` topic and is distinguished only + /// by `amount0`/`amount1`, must be counted (task 0290). + /// + /// Payload is a real production event — pool + /// `CCR2CH4GQVCZHG7CHFVMNANCK45CU5DVKXZIIITDZQAU3CEJZ7RQH2MQ` (XLM/USDC) at + /// ledger 64,488,316. + #[test] + fn a_sushiswap_pool_swap_is_counted_but_its_router_twin_is_not() { + let pool_swap = SorobanEventRow { + data: TaggedValue::Map(vec![ + ( + TaggedValue::Symbol("amount0".into()), + TaggedValue::I128(-10_002_738_052), + ), + ( + TaggedValue::Symbol("amount1".into()), + TaggedValue::I128(1_886_660_000), + ), + ( + TaggedValue::Symbol("liquidity".into()), + TaggedValue::I128(22_083_689_118_901), + ), + ]), + ..event( + "CCR2CH4GQVCZHG7CHFVMNANCK45CU5DVKXZIIITDZQAU3CEJZ7RQH2MQ", + vec![TaggedValue::Symbol("swap".into())], + 0, + ) + }; + assert_eq!( + unregistered_pool_venue(&pool_swap), + Some(Venue::Sushiswap), + "a pool swap carrying amount0/amount1 must be counted" + ); + + // Same topic, router data — must stay uncounted. + let router_swap = SorobanEventRow { + data: TaggedValue::Map(vec![ + ( + TaggedValue::Symbol("amount_in".into()), + TaggedValue::I128(2_539_492_571), + ), + ( + TaggedValue::Symbol("amount_out".into()), + TaggedValue::I128(13_410_007_617), + ), + ]), + ..event( + "CDMIM23WOUL5CZBKX3GOA3V5R5AMVIMTCP52KCDQORWELAPLJ27WZCHL", + vec![TaggedValue::Symbol("swap".into())], + 0, + ) + }; + assert_eq!( + unregistered_pool_venue(&router_swap), + None, + "the router shares the topic and must NOT be counted" + ); + } + #[test] fn a_registered_pool_is_not_counted_as_unregistered() { use phoenix_extractor::test_fixtures::{ diff --git a/packages/soroswap-extractor/src/lib.rs b/packages/soroswap-extractor/src/lib.rs index fdf26e71..1e8bfc42 100644 --- a/packages/soroswap-extractor/src/lib.rs +++ b/packages/soroswap-extractor/src/lib.rs @@ -13,6 +13,16 @@ //! Router / aggregator wrapper `swap` events (simple {amount_in, amount_out}) //! are dropped upstream (VenueRegistry maps only pool contract_ids) to avoid //! double-counting; they carry no token/direction info on their own. +//! +//! # Shared with SushiSwap V3 (task 0290) +//! +//! SushiSwap V3 pools emit the **same** CLMM shape — `topics = [Symbol("swap")]`, +//! `data = Map{ amount0, amount1 (signed), liquidity, sqrt_price_x96, tick, … }` +//! — so they decode through the very same path rather than a second copy of it. +//! The types below are therefore venue-neutral ([`TokenPair`], +//! [`PairPoolRegistry`], [`PairSwapExtractor`]); the `Soroswap*` names remain as +//! aliases so existing call sites keep compiling. Only the `venue` stamped on +//! the emitted [`TradeRow`] differs, which is why the extractor carries it. use std::collections::HashMap; @@ -20,32 +30,43 @@ use extractors_core::{ ExtractError, ExtractResult, SorobanEventRow, SwapExtractor, TaggedValue, TradeRow, Venue, }; -/// A Soroswap pair's two tokens, in canonical (token0, token1) order as -/// reported by the factory `new_pair` event. +/// A pool's two tokens, in the canonical (token0, token1) order its factory +/// reports — Soroswap's `new_pair`, SushiSwap V3's `pool_created`. #[derive(Debug, Clone)] -pub struct SoroswapPair { +pub struct TokenPair { pub token0: String, pub token1: String, } -/// pool_address → (token0, token1). Populated from Soroswap factory `new_pair` -/// events (`[String("SoroswapFactory"), Symbol("new_pair")]`, -/// data = NewPairEvent{ token_0, token_1, pair, … }). +/// Back-compat alias — [`TokenPair`] is venue-neutral (task 0290). +pub type SoroswapPair = TokenPair; + +/// pool_address → (token0, token1). Populated from factory events: Soroswap's +/// `[String("SoroswapFactory"), Symbol("new_pair")]` with +/// data = NewPairEvent{ token_0, token_1, pair, … }, or SushiSwap V3's +/// `[Symbol("pool_created")]` with data = { token0, token1, pool_address, … }. +/// +/// One registry instance holds ONE venue's pools — `Registries` keeps a +/// separate instance per venue so two venues can never collide on a +/// contract_id. #[derive(Debug, Default)] -pub struct SoroswapPoolRegistry { - pools: HashMap, +pub struct PairPoolRegistry { + pools: HashMap, } -impl SoroswapPoolRegistry { +/// Back-compat alias — [`PairPoolRegistry`] is venue-neutral (task 0290). +pub type SoroswapPoolRegistry = PairPoolRegistry; + +impl PairPoolRegistry { pub fn new() -> Self { Self::default() } pub fn register(&mut self, pair: String, token0: String, token1: String) { - self.pools.insert(pair, SoroswapPair { token0, token1 }); + self.pools.insert(pair, TokenPair { token0, token1 }); } - pub fn lookup(&self, pair: &str) -> Option<&SoroswapPair> { + pub fn lookup(&self, pair: &str) -> Option<&TokenPair> { self.pools.get(pair) } @@ -89,14 +110,31 @@ fn swap_action(row: &SorobanEventRow) -> Option<&str> { } } -/// Extracts Soroswap swaps for a single pool, using its registered token pair. -pub struct SoroswapPairExtractor<'a> { - pub pair: &'a SoroswapPair, +/// Extracts swaps for a single pool, using its registered token pair. +/// +/// `venue` is stamped on every emitted [`TradeRow`] and is the ONLY thing that +/// differs between Soroswap and SushiSwap V3 here — the data shapes are +/// identical (task 0290). +pub struct PairSwapExtractor<'a> { + pub pair: &'a TokenPair, + pub venue: Venue, } -impl<'a> SoroswapPairExtractor<'a> { - pub fn new(pair: &'a SoroswapPair) -> Self { - Self { pair } +/// Back-compat alias — [`PairSwapExtractor`] is venue-neutral (task 0290). +pub type SoroswapPairExtractor<'a> = PairSwapExtractor<'a>; + +impl<'a> PairSwapExtractor<'a> { + /// Soroswap, for the call sites that predate task 0290. + pub fn new(pair: &'a TokenPair) -> Self { + Self { + pair, + venue: Venue::Soroswap, + } + } + + /// The same decode, stamped with an explicit venue. + pub fn with_venue(venue: Venue, pair: &'a TokenPair) -> Self { + Self { pair, venue } } fn decode_swap(&self, row: &SorobanEventRow) -> Result { @@ -171,7 +209,7 @@ impl<'a> SoroswapPairExtractor<'a> { .map(String::from); Ok(TradeRow { - venue: Venue::Soroswap, + venue: self.venue.clone(), contract_id: row.contract_id.clone(), transaction_id: row.transaction_id.clone(), ledger_sequence: row.ledger_sequence, @@ -186,7 +224,7 @@ impl<'a> SoroswapPairExtractor<'a> { } } -impl SwapExtractor for SoroswapPairExtractor<'_> { +impl SwapExtractor for PairSwapExtractor<'_> { fn extract(&self, rows: &[SorobanEventRow]) -> Result { if rows.is_empty() { return Err(ExtractError::InsufficientRows { From 291c21cb61f39885dbcd4801935ffe1a432b35ae Mon Sep 17 00:00:00 2001 From: karczuRF Date: Fri, 18 Sep 2026 17:01:06 +0200 Subject: [PATCH 2/7] feat(lore-0290): discover SushiSwap V3 pools from pool_created --discover-pools read only the Aquarius, Phoenix and Soroswap factory events, so it could never seed a SushiSwap pool. pool_created has a Symbol topic, so signature carries it; adding it to the filter is enough. With no emitter filter one read covers all four factory generations. Checked on production over ledgers 60M-65M: 133 pools announced, every contract on the two known pool wasms among them. Another protocol's token-less pool_created is read too and pinned as learning nothing. --- packages/events-backfill/src/discover.rs | 57 +++++++++++++++++++++--- 1 file changed, 52 insertions(+), 5 deletions(-) diff --git a/packages/events-backfill/src/discover.rs b/packages/events-backfill/src/discover.rs index c4c9a185..a6dde56f 100644 --- a/packages/events-backfill/src/discover.rs +++ b/packages/events-backfill/src/discover.rs @@ -4,7 +4,7 @@ //! The reprice reads only the events of contracts already in the registry, so it //! can never find a pool the registry is missing. This mode reads the factory //! events themselves — Aquarius `add_pool`, Phoenix `create`, Soroswap -//! `new_pair` — and runs them through the same `learn_factory` the live +//! `new_pair`, SushiSwap V3 `pool_created` (task 0290) — and runs them through the same `learn_factory` the live //! processor uses, so a row written here is the row live would have learned. //! Only rows not already in the table are written; a re-run writes nothing. //! @@ -38,8 +38,11 @@ pub struct FactoryEventRow { /// The factory-event read for `[start, end]`, as text. /// /// Filters are a superset of what `learn_factory` accepts. BE fills `signature` -/// only when topic[0] is a Symbol, and two of the three factories use Strings: +/// only when topic[0] is a Symbol, and two of the four factories use Strings: /// - Aquarius `add_pool`: Symbol topics, so `signature = 'add_pool'`; +/// - SushiSwap V3 `pool_created`: Symbol topics, so `signature = 'pool_created'` +/// (checked on production 2026-09-18). Every factory generation emits it, and +/// with no emitter filter one read learns the pools of all four; /// - Phoenix `create`/`liquidity_pool`: **String** topics, so `signature` is NULL /// — a `signature`-only filter finds none of the 20 Phoenix pools; /// - Soroswap `SoroswapFactory`/`new_pair`: String topics, the action in topic[1]. @@ -59,7 +62,7 @@ pub(crate) fn factory_events_sql(start: u32, end: u32) -> String { data_xdr \ FROM default.soroban_events \ WHERE ledger_sequence BETWEEN {start} AND {end} \ - AND (signature IN ('add_pool', 'create') \ + AND (signature IN ('add_pool', 'create', 'pool_created') \ OR (signature IS NULL \ AND (JSONExtractString(topics_xdr, 1, 'value') IN ('add_pool', 'create') \ OR JSONExtractString(topics_xdr, 2, 'value') = 'new_pair'))) \ @@ -179,12 +182,17 @@ mod tests { const CREATE_DATA: &str = r#"{"type":"address","value":"CBHCRSVX3ZZ7EGTSYMKPEFGZNWRVCSESQR3UABET4MIW52N4EVU6BIZX"}"#; + // SushiSwap V3's live factory `CD3KRKGD…GLYF`, ledger 64,116,662. + const POOL_CREATED_TOPICS: &str = r#"[{"type":"sym","value":"pool_created"}]"#; + const POOL_CREATED_DATA: &str = r#"{"type":"map","value":[{"key":{"type":"sym","value":"fee"},"value":{"type":"u32","value":500}},{"key":{"type":"sym","value":"pool_address"},"value":{"type":"address","value":"CBVHBZSZOS6KRDJ4D44FU2YLIENOVSSLM3UGKW6XQMVIFUAMWIWCVH2U"}},{"key":{"type":"sym","value":"sender"},"value":{"type":"address","value":"CD3KRKGDRVWPXVB3VXLUMQKMX6XZ6Q2H334IVZD4XXNAMKSRVQL5GLYF"}},{"key":{"type":"sym","value":"tick_spacing"},"value":{"type":"i32","value":10}},{"key":{"type":"sym","value":"token0"},"value":{"type":"address","value":"CBSJZEIO5C7KC2SF3MKSNXXJSW5G3VTNBX4ATMKUI3B2MR4JKM4R26YF"}},{"key":{"type":"sym","value":"token1"},"value":{"type":"address","value":"CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75"}}]}"#; + #[test] - fn learns_all_three_factory_shapes_from_real_payloads() { + fn learns_all_four_factory_shapes_from_real_payloads() { let mut reg = Registries::new(); learn_from_row(&row(NEW_PAIR_TOPICS, NEW_PAIR_DATA), &mut reg); learn_from_row(&row(ADD_POOL_TOPICS, ADD_POOL_DATA), &mut reg); learn_from_row(&row(CREATE_TOPICS, CREATE_DATA), &mut reg); + learn_from_row(&row(POOL_CREATED_TOPICS, POOL_CREATED_DATA), &mut reg); let pair = "CAZ4Z273BBAAFL5NYNQJKEMZDQBRCPKAS4GOXDUFXPSE56M4ONBJUOVD"; assert_eq!(reg.venue.get(pair), Some(&Venue::Soroswap)); @@ -207,6 +215,45 @@ mod tests { .get("CBHCRSVX3ZZ7EGTSYMKPEFGZNWRVCSESQR3UABET4MIW52N4EVU6BIZX"), Some(&Venue::Phoenix) ); + let pool = "CBVHBZSZOS6KRDJ4D44FU2YLIENOVSSLM3UGKW6XQMVIFUAMWIWCVH2U"; + assert_eq!(reg.venue.get(pool), Some(&Venue::Sushiswap)); + let p = reg.sushiswap.lookup(pool).expect("pool tokens learned"); + assert_eq!( + p.token0, + "CBSJZEIO5C7KC2SF3MKSNXXJSW5G3VTNBX4ATMKUI3B2MR4JKM4R26YF" + ); + assert_eq!( + p.token1, + "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75" + ); + } + + /// Another protocol also emits `pool_created` (emitter `CBMBKXI7…Q227`, + /// wasm `ED0D122C`, ledger 63,173,214): fixed-denomination pools with no + /// token pair. The read selects it, so the learner must refuse it — a + /// pool registered with empty tokens would price against asset "". + #[test] + fn a_pool_created_without_a_token_pair_learns_nothing() { + let topics = r#"[{"type":"sym","value":"pool_created"},{"type":"address","value":"CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA"}]"#; + let data = r#"{"type":"map","value":[{"key":{"type":"sym","value":"denomination"},"value":{"type":"i128","value":"1000000"}},{"key":{"type":"sym","value":"generation"},"value":{"type":"u32","value":0}},{"key":{"type":"sym","value":"pool"},"value":{"type":"address","value":"CC2SYHPYVRQ24IS6BQVA5WUYWDR3GK46W5ADY2UASC2D4IH6J5AXSEQ5"}}]}"#; + let mut reg = Registries::new(); + learn_from_row(&row(topics, data), &mut reg); + assert!(reg.venue.is_empty()); + } + + #[test] + fn a_new_sushiswap_pool_is_written_as_sushiswap() { + let mut reg = Registries::new(); + let persisted = snapshot(®); + learn_from_row(&row(POOL_CREATED_TOPICS, POOL_CREATED_DATA), &mut reg); + + let rows = reg.pool_rows_unpersisted(&persisted); + assert_eq!(rows.len(), 1); + assert_eq!( + rows[0].contract_id, + "CBVHBZSZOS6KRDJ4D44FU2YLIENOVSSLM3UGKW6XQMVIFUAMWIWCVH2U" + ); + assert_eq!(rows[0].venue, "sushiswap"); } #[test] @@ -248,7 +295,7 @@ mod tests { let sql = factory_events_sql(63_000_000, 63_319_999); assert!(sql.contains("ledger_sequence BETWEEN 63000000 AND 63319999")); let sql = sql.split_whitespace().collect::>().join(" "); - assert!(sql.contains("signature IN ('add_pool', 'create')")); + assert!(sql.contains("signature IN ('add_pool', 'create', 'pool_created')")); // String-topic factories leave `signature` NULL: Phoenix's action is in // topic[0], Soroswap's in topic[1] (1-based indexes 1 and 2). assert!(sql.contains( From 7021ae2b43b31d1e4cfe97eae7685f2e390d4c4e Mon Sep 17 00:00:00 2001 From: karczuRF Date: Fri, 18 Sep 2026 17:15:52 +0200 Subject: [PATCH 3/7] test(lore-0290): pin that a routed SushiSwap trade prices once A routed trade emits two swap events in one transaction, the pool's and the router's summary of the same trade. The test drives a real one (ledger 64,481,111) through process_soroban_event_rows and requires a single tick, no registered router, and no missing-pool count. A router wrongly registered as a pool still adds no tick. The seed runbook now lists pool_created and SushiSwap's earlier start. --- docs/runbooks/seed-pool-registry.md | 6 +- packages/prices-ingest-core/src/soroban.rs | 104 +++++++++++++++++++++ 2 files changed, 108 insertions(+), 2 deletions(-) diff --git a/docs/runbooks/seed-pool-registry.md b/docs/runbooks/seed-pool-registry.md index 6010ec10..0024358f 100644 --- a/docs/runbooks/seed-pool-registry.md +++ b/docs/runbooks/seed-pool-registry.md @@ -150,7 +150,7 @@ API returned a contract of an unexpected type — investigate before seeding it. `events-backfill --discover-pools` reads the AMM factory events in a ledger range from BE's `default.soroban_events` (Aquarius `add_pool`, Phoenix `create`, -Soroswap `new_pair`), runs them through the same `learn_factory` the live +Soroswap `new_pair`, SushiSwap V3 `pool_created` — task 0290), runs them through the same `learn_factory` the live processor uses, and writes **only the pools `prices.pool_registry` does not already hold**. No API key, no rewrite of existing rows, no candles. A re-run writes nothing. @@ -183,7 +183,9 @@ scp target/x86_64-unknown-linux-musl/release/events-backfill :~/event factories (Phoenix and Soroswap leave `signature` NULL), 2-4 s per 320k-ledger chunk on the shared box. As of 2026-09-17 every missing pool was created after ledger 63,000,000 (checked per venue over the whole Soroban era), so the -catch-up only needs `63000000` to the tip. +catch-up only needs `63000000` to the tip. **Task 0290 is the exception:** +SushiSwap V3's pools go back to ledger 60,147,305, so its run starts at +`60000000` — the exact command is in the 0290 task file. ```bash # On the prod host, under tmux: diff --git a/packages/prices-ingest-core/src/soroban.rs b/packages/prices-ingest-core/src/soroban.rs index 039958d2..3992f1b2 100644 --- a/packages/prices-ingest-core/src/soroban.rs +++ b/packages/prices-ingest-core/src/soroban.rs @@ -1752,6 +1752,110 @@ mod tests { assert!(out.unresolved.is_empty(), "a priced pool is not unresolved"); } + /// Task 0290, acceptance criterion "its routers stay unindexed". A routed + /// SushiSwap trade emits TWO `[Symbol("swap")]` events in one transaction: + /// the pool's, and the router's summary of the same trade. Only the pool's + /// may become a tick, or every routed trade is counted twice. + /// + /// Real production transaction, ledger 64,481,111: pool `CCR2CH4G…H2MQ` + /// (XLM/USDC) at event 4, router `CDMIM23W…ZCHL` at event 5 carrying the + /// same amounts as `amount_in`/`amount_out`. + #[test] + fn a_routed_sushiswap_trade_prices_once_from_the_pool_not_the_router() { + const SEQ: u32 = 64_481_111; + const CLOSED_AT: i64 = 1_700_000_000; + const POOL: &str = "CCR2CH4GQVCZHG7CHFVMNANCK45CU5DVKXZIIITDZQAU3CEJZ7RQH2MQ"; + const ROUTER: &str = "CDMIM23WOUL5CZBKX3GOA3V5R5AMVIMTCP52KCDQORWELAPLJ27WZCHL"; + const XLM: &str = "CAS3J7GYLGXMF6TDJBBYYSE3HQ6BBSMLNUQ34T6TZMYMW2EVH34XOWMA"; + const USDC: &str = "CCW67TSZV3SSS2HXMBQ5JFGCKJNXKZM7UQUWUZPUTHXSTZLEO7SJMI75"; + + let event = |contract_id: &str, event_index: u32, data: Value| RawSorobanEvent { + contract_id: contract_id.to_string(), + transaction_id: "4737599797939127393".to_string(), + transaction_index: 0, + ledger_sequence: SEQ, + event_index, + topics: json!([{"type":"sym","value":"swap"}]), + data, + }; + let pool_swap = event( + POOL, + 4, + json!({"type":"map","value":[ + {"key":{"type":"sym","value":"amount0"},"value":{"type":"i128","value":"10951822930"}}, + {"key":{"type":"sym","value":"amount1"},"value":{"type":"i128","value":"-2000561352"}}, + {"key":{"type":"sym","value":"liquidity"},"value":{"type":"u128","value":"22083689118901"}}, + {"key":{"type":"sym","value":"recipient"},"value":{"type":"address","value":"GCBYPF2OVPSZ7NJSXIOINCBMOSMY7I6KOWHPZV36OH4OH5R2A63DKUKY"}}, + {"key":{"type":"sym","value":"sender"},"value":{"type":"address","value":"GCBYPF2OVPSZ7NJSXIOINCBMOSMY7I6KOWHPZV36OH4OH5R2A63DKUKY"}}, + {"key":{"type":"sym","value":"sqrt_price_x96"},"value":{"type":"u256","value":"00000000000000000000000000000000000000006d911cd1319d4706470e4080"}}, + {"key":{"type":"sym","value":"tick"},"value":{"type":"i32","value":-16974}} + ]}), + ); + let router_swap = event( + ROUTER, + 5, + json!({"type":"map","value":[ + {"key":{"type":"sym","value":"amount_in"},"value":{"type":"i128","value":"10951822930"}}, + {"key":{"type":"sym","value":"amount_out"},"value":{"type":"i128","value":"2000561352"}}, + {"key":{"type":"sym","value":"recipient"},"value":{"type":"address","value":"GCBYPF2OVPSZ7NJSXIOINCBMOSMY7I6KOWHPZV36OH4OH5R2A63DKUKY"}} + ]}), + ); + + let registry = || { + let mut reg = Registries::new(); + reg.venue.insert(POOL.to_string(), Venue::Sushiswap); + reg.sushiswap + .register(POOL.to_string(), XLM.to_string(), USDC.to_string()); + reg + }; + + // As in production: the pool is registered, the router is not. + let mut reg = registry(); + let mut assets = AssetRegistry::from_existing(vec![]); + let mut out = LedgerSoroban::default(); + process_soroban_event_rows( + SEQ, + CLOSED_AT, + &[pool_swap.clone(), router_swap.clone()], + &mut reg, + &mut assets, + &mut out, + ); + assert_eq!(out.amm_ticks.len(), 1, "one routed trade, one tick"); + assert_eq!(out.amm_ticks[0].0, "sushiswap"); + assert!( + !reg.venue.contains_key(ROUTER), + "nothing in a routed trade may register the router" + ); + assert_eq!( + out.unregistered_pool_events, + Vec::<(&str, u32)>::new(), + "the router's swap is not a missing pool" + ); + + // Second line of defence: even a router wrongly registered as a + // SushiSwap pool prices nothing, because its summary has no + // amount0/amount1 to decode. + let mut reg = registry(); + reg.venue.insert(ROUTER.to_string(), Venue::Sushiswap); + reg.sushiswap + .register(ROUTER.to_string(), XLM.to_string(), USDC.to_string()); + let mut out = LedgerSoroban::default(); + process_soroban_event_rows( + SEQ, + CLOSED_AT, + &[pool_swap, router_swap], + &mut reg, + &mut assets, + &mut out, + ); + assert_eq!( + out.amm_ticks.len(), + 1, + "a registered router must still add no tick" + ); + } + #[test] fn seam_learns_factory_and_skips_oracle_across_transactions() { // The seam's own responsibilities beyond classify: (1) group by From 6c3ed47798e27cc3fe7eba8696a327cde1ff4702 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 21 Sep 2026 12:14:33 +0200 Subject: [PATCH 4/7] fix(lore-0290): roll sushiswap candles into the coarse tables Every statement in the AMM reprice pre-roll scoped itself to `source IN ('aquarius', 'phoenix', 'soroswap')`, so once the backfill writes sushiswap history into price_ohlcv_1m the re-roll would have dropped every one of those rows from _15m/_1h/_4h/_1d/_1w/_1M. AMM history is read from _1d/_1h rather than _1m, so the loss would have been invisible on the table that looked correct. Pins the filter against the canonical Venue list: adding a variant now breaks a test instead of a backfill. --- Cargo.lock | 1 + packages/prices-clickhouse/Cargo.toml | 6 + .../schema/preroll-amm-reprice.sql | 203 +++++++++--------- packages/prices-clickhouse/src/lib.rs | 51 +++++ 4 files changed, 163 insertions(+), 98 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 735fe4d4..cb9faaba 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2872,6 +2872,7 @@ name = "prices-clickhouse" version = "0.1.0" dependencies = [ "clickhouse", + "extractors-core", "hyper-rustls 0.27.9", "hyper-util", "reqwest", diff --git a/packages/prices-clickhouse/Cargo.toml b/packages/prices-clickhouse/Cargo.toml index 46ae00ec..2c956a25 100644 --- a/packages/prices-clickhouse/Cargo.toml +++ b/packages/prices-clickhouse/Cargo.toml @@ -55,3 +55,9 @@ rustls-pemfile = { version = "2", optional = true } rustls-pki-types = { version = "1", optional = true } webpki-roots = { version = "1", optional = true } reqwest = { version = "0.12", default-features = false, features = ["json"], optional = true } + +[dev-dependencies] +# Test-only: the AMM pre-roll's `source IN (...)` filter is pinned against the +# canonical `Venue` list, so a new venue cannot be added without the filter +# moving with it (task 0290 review). +extractors-core = { path = "../extractors-core" } diff --git a/packages/prices-clickhouse/schema/preroll-amm-reprice.sql b/packages/prices-clickhouse/schema/preroll-amm-reprice.sql index ab8bc0d3..332eb5b2 100644 --- a/packages/prices-clickhouse/schema/preroll-amm-reprice.sql +++ b/packages/prices-clickhouse/schema/preroll-amm-reprice.sql @@ -42,12 +42,19 @@ -- so ~824k historical swaps produced nothing. The coarse tables still reflect -- that gap and must be re-rolled from the corrected `1m`. -- +-- `sushiswap` joined the filter for task 0290. Its history is written by a +-- separate backfill, not by 0097's reprice, but it lands in the same +-- `price_ohlcv_1m` and needs the same coarse re-roll — and since AMM history +-- is read from `_1d`/`_1h` rather than `_1m`, a sushiswap row missing from +-- the filter would be invisible to consumers while `_1m` looked correct. +-- -- SAFETY — why this cannot disturb SDEX coarse: -- Every OHLCV table is `ORDER BY (asset_id, quote_asset_id, source, timestamp)` -- with `source` IN the key, so an `sdex` row and an `aquarius` row for the same -- minute+pair are DISTINCT rows, never RMT-merge candidates. Scoping every --- statement to `source IN ('aquarius','phoenix','soroswap')` therefore makes --- SDEX coarse — including the expensive pre-Soroban tail — untouchable here. +-- statement to `source IN ('aquarius','phoenix','soroswap','sushiswap')` +-- therefore makes SDEX coarse — including the expensive pre-Soroban tail — +-- untouchable here. -- -- MEMORY (0090 + 0097 findings): ch-prod-01 enforces a ~5.59 GiB per-query quota -- (0097's coverage probe hit it as MEMORY_LIMIT_EXCEEDED on a full-range scan). @@ -226,7 +233,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 15 MINUTE) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= {start_ts:DateTime} AND t.timestamp < '2025-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source; @@ -242,7 +249,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 15 MINUTE) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-01-01' AND t.timestamp < '2026-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source; @@ -258,7 +265,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 15 MINUTE) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-01-01' AND t.timestamp < {end_ts:DateTime} GROUP BY timestamp, asset_id, quote_asset_id, source; @@ -314,7 +321,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= {start_ts:DateTime} AND t.timestamp < '2024-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -331,7 +338,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-03-01' AND t.timestamp < '2024-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -348,7 +355,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-04-01' AND t.timestamp < '2024-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -365,7 +372,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-05-01' AND t.timestamp < '2024-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -382,7 +389,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-06-01' AND t.timestamp < '2024-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -399,7 +406,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-07-01' AND t.timestamp < '2024-08-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -416,7 +423,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-08-01' AND t.timestamp < '2024-09-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -433,7 +440,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-09-01' AND t.timestamp < '2024-10-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -450,7 +457,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-10-01' AND t.timestamp < '2024-11-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -467,7 +474,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-11-01' AND t.timestamp < '2024-12-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -484,7 +491,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-12-01' AND t.timestamp < '2025-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -501,7 +508,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-01-01' AND t.timestamp < '2025-02-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -518,7 +525,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-02-01' AND t.timestamp < '2025-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -535,7 +542,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-03-01' AND t.timestamp < '2025-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -552,7 +559,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-04-01' AND t.timestamp < '2025-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -569,7 +576,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-05-01' AND t.timestamp < '2025-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -586,7 +593,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-06-01' AND t.timestamp < '2025-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -603,7 +610,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-07-01' AND t.timestamp < '2025-08-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -620,7 +627,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-08-01' AND t.timestamp < '2025-09-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -637,7 +644,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-09-01' AND t.timestamp < '2025-10-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -654,7 +661,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-10-01' AND t.timestamp < '2025-11-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -671,7 +678,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-11-01' AND t.timestamp < '2025-12-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -688,7 +695,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-12-01' AND t.timestamp < '2026-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -705,7 +712,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-01-01' AND t.timestamp < '2026-02-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -722,7 +729,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-02-01' AND t.timestamp < '2026-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -739,7 +746,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-03-01' AND t.timestamp < '2026-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -756,7 +763,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-04-01' AND t.timestamp < '2026-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -773,7 +780,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-05-01' AND t.timestamp < '2026-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -790,7 +797,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-06-01' AND t.timestamp < '2026-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -807,7 +814,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_15m AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-07-01' AND t.timestamp < {end_ts:DateTime} GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -828,7 +835,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= {start_ts:DateTime} AND t.timestamp < '2024-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -845,7 +852,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-03-01' AND t.timestamp < '2024-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -862,7 +869,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-04-01' AND t.timestamp < '2024-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -879,7 +886,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-05-01' AND t.timestamp < '2024-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -896,7 +903,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-06-01' AND t.timestamp < '2024-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -913,7 +920,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-07-01' AND t.timestamp < '2024-08-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -930,7 +937,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-08-01' AND t.timestamp < '2024-09-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -947,7 +954,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-09-01' AND t.timestamp < '2024-10-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -964,7 +971,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-10-01' AND t.timestamp < '2024-11-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -981,7 +988,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-11-01' AND t.timestamp < '2024-12-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -998,7 +1005,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-12-01' AND t.timestamp < '2025-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1015,7 +1022,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-01-01' AND t.timestamp < '2025-02-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1032,7 +1039,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-02-01' AND t.timestamp < '2025-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1049,7 +1056,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-03-01' AND t.timestamp < '2025-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1066,7 +1073,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-04-01' AND t.timestamp < '2025-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1083,7 +1090,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-05-01' AND t.timestamp < '2025-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1100,7 +1107,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-06-01' AND t.timestamp < '2025-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1117,7 +1124,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-07-01' AND t.timestamp < '2025-08-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1134,7 +1141,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-08-01' AND t.timestamp < '2025-09-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1151,7 +1158,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-09-01' AND t.timestamp < '2025-10-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1168,7 +1175,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-10-01' AND t.timestamp < '2025-11-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1185,7 +1192,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-11-01' AND t.timestamp < '2025-12-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1202,7 +1209,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-12-01' AND t.timestamp < '2026-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1219,7 +1226,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-01-01' AND t.timestamp < '2026-02-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1236,7 +1243,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-02-01' AND t.timestamp < '2026-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1253,7 +1260,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-03-01' AND t.timestamp < '2026-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1270,7 +1277,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-04-01' AND t.timestamp < '2026-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1287,7 +1294,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-05-01' AND t.timestamp < '2026-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1304,7 +1311,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-06-01' AND t.timestamp < '2026-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1321,7 +1328,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 4 HOUR) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-07-01' AND t.timestamp < {end_ts:DateTime} GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1342,7 +1349,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= {start_ts:DateTime} AND t.timestamp < '2024-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1359,7 +1366,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-03-01' AND t.timestamp < '2024-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1376,7 +1383,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-04-01' AND t.timestamp < '2024-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1393,7 +1400,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-05-01' AND t.timestamp < '2024-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1410,7 +1417,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-06-01' AND t.timestamp < '2024-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1427,7 +1434,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-07-01' AND t.timestamp < '2024-08-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1444,7 +1451,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-08-01' AND t.timestamp < '2024-09-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1461,7 +1468,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-09-01' AND t.timestamp < '2024-10-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1478,7 +1485,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-10-01' AND t.timestamp < '2024-11-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1495,7 +1502,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-11-01' AND t.timestamp < '2024-12-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1512,7 +1519,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2024-12-01' AND t.timestamp < '2025-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1529,7 +1536,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-01-01' AND t.timestamp < '2025-02-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1546,7 +1553,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-02-01' AND t.timestamp < '2025-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1563,7 +1570,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-03-01' AND t.timestamp < '2025-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1580,7 +1587,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-04-01' AND t.timestamp < '2025-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1597,7 +1604,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-05-01' AND t.timestamp < '2025-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1614,7 +1621,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-06-01' AND t.timestamp < '2025-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1631,7 +1638,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-07-01' AND t.timestamp < '2025-08-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1648,7 +1655,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-08-01' AND t.timestamp < '2025-09-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1665,7 +1672,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-09-01' AND t.timestamp < '2025-10-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1682,7 +1689,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-10-01' AND t.timestamp < '2025-11-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1699,7 +1706,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-11-01' AND t.timestamp < '2025-12-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1716,7 +1723,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2025-12-01' AND t.timestamp < '2026-01-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1733,7 +1740,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-01-01' AND t.timestamp < '2026-02-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1750,7 +1757,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-02-01' AND t.timestamp < '2026-03-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1767,7 +1774,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-03-01' AND t.timestamp < '2026-04-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1784,7 +1791,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-04-01' AND t.timestamp < '2026-05-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1801,7 +1808,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-05-01' AND t.timestamp < '2026-06-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1818,7 +1825,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-06-01' AND t.timestamp < '2026-07-01' GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1835,7 +1842,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 DAY) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_4h AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= '2026-07-01' AND t.timestamp < {end_ts:DateTime} GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1855,7 +1862,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 WEEK) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1d AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= {start_ts:DateTime} AND t.timestamp < {end_ts:DateTime} GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1875,7 +1882,7 @@ SELECT toStartOfInterval(t.timestamp, INTERVAL 1 MONTH) AS timestamp, volume_quote / nullIf(volume_base, 0) AS vwap, sum(trade_count) AS trade_count, max(version) AS version FROM prices.price_ohlcv_1w AS t FINAL -WHERE t.source IN ('aquarius', 'phoenix', 'soroswap') +WHERE t.source IN ('aquarius', 'phoenix', 'soroswap', 'sushiswap') AND t.timestamp >= {start_ts:DateTime} AND t.timestamp < {end_ts:DateTime} GROUP BY timestamp, asset_id, quote_asset_id, source SETTINGS max_threads = 4; @@ -1900,7 +1907,7 @@ SETTINGS max_threads = 4; -- total BELOW 1m means buckets were lost or an RMT tie kept a stale row -- (OPEN QUESTION 1); ABOVE means double-counting. -- SELECT source, sum(volume_base) FROM prices.price_ohlcv_1m FINAL --- WHERE source IN ('aquarius','phoenix','soroswap') +-- WHERE source IN ('aquarius','phoenix','soroswap','sushiswap') -- AND timestamp >= {start_ts:DateTime} AND timestamp < {end_ts:DateTime} -- GROUP BY source; -- -- then the same against price_ohlcv_1d; the two must match per source. diff --git a/packages/prices-clickhouse/src/lib.rs b/packages/prices-clickhouse/src/lib.rs index d1cc72c8..75c5bb33 100644 --- a/packages/prices-clickhouse/src/lib.rs +++ b/packages/prices-clickhouse/src/lib.rs @@ -375,6 +375,7 @@ pub(crate) fn split_statements(sql: &str) -> Vec { #[cfg(test)] mod tests { use super::*; + use extractors_core::Venue; /// Task 0215: `max_execution_time = 0` is ClickHouse's spelling of /// UNLIMITED, not "no delay". A knob zeroed by a typo, or by an operator @@ -927,6 +928,56 @@ mod tests { ); } + /// Every `source IN (...)` in the AMM reprice pre-roll must name EVERY AMM + /// venue. + /// + /// This is the task 0290 defect: `sushiswap` became a candle `source`, the + /// pre-roll's three-venue allowlist did not move with it, and the backfill + /// would have written history into `price_ohlcv_1m` that never reached the + /// coarse tables — invisible to consumers, which read AMM history from + /// `_1d`/`_1h`, while `_1m` looked correct. Pinning the filter against + /// `Venue` means the next venue breaks this test instead of a backfill. + #[test] + fn the_amm_preroll_source_filter_names_every_amm_venue() { + // Adding a `Venue` variant makes this match non-exhaustive — a compile + // error here is the reminder to widen the pre-roll filter as well. + let venues = [ + Venue::Aquarius, + Venue::Phoenix, + Venue::Soroswap, + Venue::Sushiswap, + ]; + for v in &venues { + match v { + Venue::Aquarius | Venue::Phoenix | Venue::Soroswap | Venue::Sushiswap => {} + } + } + + let filters: Vec<&str> = PREROLL_AMM_REPRICE_SQL + .match_indices("source IN (") + .map(|(i, _)| { + let rest = &PREROLL_AMM_REPRICE_SQL[i..]; + let end = rest.find(')').expect("unterminated source IN (...)"); + &rest[..=end] + }) + .collect(); + + assert!( + !filters.is_empty(), + "no `source IN (...)` found — the scoping this test guards is gone" + ); + + for filter in &filters { + for venue in &venues { + let name = venue.as_source(); + assert!( + filter.contains(&format!("'{name}'")), + "AMM venue '{name}' missing from a pre-roll filter: {filter}" + ); + } + } + } + // ------------------------------------------------------------------ // Task 0286 / ADR 0287 §2–§5 — the three generated SQL files. // From 74912b848c5308f78801864daaf4072c2e4620f0 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 21 Sep 2026 12:14:39 +0200 Subject: [PATCH 5/7] refactor(lore-0290): name the pair registries at dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `soroswap_registry` and `sushiswap_registry` were adjacent parameters of the same type, so transposing them compiled silently and priced every Soroswap pool against SushiSwap's token table and vice versa — no error, just candles for the wrong asset pair. They now travel as one PairRegistries with named fields, built only by Registries::pair_registries. --- packages/ledger-processor/src/dispatch.rs | 52 ++++++++++++++++------ packages/prices-ingest-core/src/soroban.rs | 22 +++++---- 2 files changed, 52 insertions(+), 22 deletions(-) diff --git a/packages/ledger-processor/src/dispatch.rs b/packages/ledger-processor/src/dispatch.rs index 0247a485..dd7a0d45 100644 --- a/packages/ledger-processor/src/dispatch.rs +++ b/packages/ledger-processor/src/dispatch.rs @@ -6,7 +6,7 @@ use phoenix_extractor::{ PHOENIX_STABLE_EVENT_COUNT, PHOENIX_XYK_MIN_EVENT_COUNT, POOL_TYPE_XYK, PhoenixPoolRegistry, PhoenixXykExtractor, }; -use soroswap_extractor::{PairSwapExtractor, SoroswapPairExtractor, SoroswapPoolRegistry}; +use soroswap_extractor::{PairPoolRegistry, PairSwapExtractor, SoroswapPairExtractor}; #[derive(Debug, thiserror::Error)] pub enum DispatchError { @@ -72,6 +72,22 @@ pub fn dispatch_phoenix( } } +/// The two pair-backed venues' pool registries, carried as ONE argument with +/// NAMED fields. +/// +/// Both are a [`PairPoolRegistry`], so as two adjacent positional parameters +/// they were freely interchangeable: transposing them compiled silently and +/// produced no runtime error — every Soroswap pool would have resolved against +/// SushiSwap's token table and vice versa, emitting candles for the wrong asset +/// pair. Naming the fields makes that mistake something you have to write on +/// purpose, and `Registries::pair_registries` in `prices-ingest-core` is the +/// only place that wires them (task 0290 review). +#[derive(Clone, Copy)] +pub struct PairRegistries<'a> { + pub soroswap: &'a PairPoolRegistry, + pub sushiswap: &'a PairPoolRegistry, +} + /// Top-level dispatcher: routes events by venue, then by pool shape for Phoenix. /// /// Soroswap and SushiSwap require the pool→tokens registry to resolve token @@ -79,13 +95,13 @@ pub fn dispatch_phoenix( /// trades rather than an error. Aquarius and Phoenix carry tokens inline. /// /// The two pair-backed venues keep SEPARATE registries so a contract_id can -/// never resolve to the wrong venue's tokens (task 0290). +/// never resolve to the wrong venue's tokens (task 0290); they arrive together +/// in [`PairRegistries`], which names them rather than ordering them. pub fn dispatch( rows: &[SorobanEventRow], venue_registry: &VenueRegistry, phoenix_registry: &PhoenixPoolRegistry, - soroswap_registry: &SoroswapPoolRegistry, - sushiswap_registry: &SoroswapPoolRegistry, + pairs: PairRegistries<'_>, ) -> Result, DispatchError> { if rows.is_empty() { return Ok(vec![]); @@ -96,11 +112,11 @@ pub fn dispatch( match venue { Some(Venue::Phoenix) => dispatch_phoenix(rows, phoenix_registry), - Some(Venue::Soroswap) => match soroswap_registry.lookup(contract_id) { + Some(Venue::Soroswap) => match pairs.soroswap.lookup(contract_id) { Some(pair) => Ok(SoroswapPairExtractor::new(pair).extract(rows)?.trades), None => Ok(vec![]), }, - Some(Venue::Sushiswap) => match sushiswap_registry.lookup(contract_id) { + Some(Venue::Sushiswap) => match pairs.sushiswap.lookup(contract_id) { Some(pair) => Ok(PairSwapExtractor::with_venue(Venue::Sushiswap, pair) .extract(rows)? .trades), @@ -137,8 +153,10 @@ mod tests { &rows, &venue_reg, &phoenix_reg, - &SoroswapPoolRegistry::new(), - &SoroswapPoolRegistry::new(), + PairRegistries { + soroswap: &PairPoolRegistry::new(), + sushiswap: &PairPoolRegistry::new(), + }, ) .unwrap(); assert_eq!(trades.len(), 1); @@ -156,8 +174,10 @@ mod tests { &rows, &venue_reg, &phoenix_reg, - &SoroswapPoolRegistry::new(), - &SoroswapPoolRegistry::new(), + PairRegistries { + soroswap: &PairPoolRegistry::new(), + sushiswap: &PairPoolRegistry::new(), + }, ) .unwrap(); assert_eq!(trades.len(), 1); @@ -208,8 +228,10 @@ mod tests { &rows, &venue_reg, &phoenix_reg, - &SoroswapPoolRegistry::new(), - &SoroswapPoolRegistry::new(), + PairRegistries { + soroswap: &PairPoolRegistry::new(), + sushiswap: &PairPoolRegistry::new(), + }, ) .unwrap(); assert!(trades.is_empty()); @@ -224,8 +246,10 @@ mod tests { &[], &venue_reg, &phoenix_reg, - &SoroswapPoolRegistry::new(), - &SoroswapPoolRegistry::new(), + PairRegistries { + soroswap: &PairPoolRegistry::new(), + sushiswap: &PairPoolRegistry::new(), + }, ) .unwrap(); assert!(trades.is_empty()); diff --git a/packages/prices-ingest-core/src/soroban.rs b/packages/prices-ingest-core/src/soroban.rs index 3992f1b2..bf51d361 100644 --- a/packages/prices-ingest-core/src/soroban.rs +++ b/packages/prices-ingest-core/src/soroban.rs @@ -17,7 +17,7 @@ use stellar_xdr::{LedgerCloseMeta, TransactionMeta}; use tracing::{debug, warn}; use extractors_core::{SorobanEventRow, TaggedValue, Venue, VenueRegistry}; -use ledger_processor::dispatch::dispatch; +use ledger_processor::dispatch::{PairRegistries, dispatch}; use phoenix_extractor::PhoenixPoolRegistry; use soroswap_extractor::SoroswapPoolRegistry; use xdr_parser::extract_events; @@ -72,6 +72,18 @@ impl Registries { pub fn pool_count(&self) -> usize { self.soroswap.pool_count() + self.phoenix.pool_count() + self.sushiswap.pool_count() } + + /// The two pair-backed registries, wired to their venues by name. + /// + /// `soroswap` and `sushiswap` are the same type, so this is the ONE place + /// the two can be crossed; every dispatch goes through it instead of + /// passing them as adjacent positional arguments (task 0290 review). + pub fn pair_registries(&self) -> PairRegistries<'_> { + PairRegistries { + soroswap: &self.soroswap, + sushiswap: &self.sushiswap, + } + } } /// Resolve a Reflector `update`-event asset key to a canonical `AssetIdentity`. @@ -537,13 +549,7 @@ fn classify_amm_groups( }; let source = venue.as_source(); - match dispatch( - &rows, - ®.venue, - ®.phoenix, - ®.soroswap, - ®.sushiswap, - ) { + match dispatch(&rows, ®.venue, ®.phoenix, reg.pair_registries()) { Ok(trades) => { for t in trades { if let Some(tick) = amm_trade_to_tick(&t, transaction_index, closed_at, assets) From 6a990e75cafa7f384035f08fe225b13331ca5e19 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 21 Sep 2026 12:14:49 +0200 Subject: [PATCH 6/7] fix(lore-0290): keep a blank persisted pair unresolved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A pool_registry row naming a pair-backed venue with empty token0/token1 was registered anyway, so contains() answered true, classify_amm_groups cleared pair_unresolved, and the pool was priced against asset "" instead of being recorded in unresolved_pools. It now loads its venue only — the same outcome the learn side already gives a pool_created with no pair. --- .../prices-ingest-core/src/registry_io.rs | 69 ++++++++++++++++++- 1 file changed, 67 insertions(+), 2 deletions(-) diff --git a/packages/prices-ingest-core/src/registry_io.rs b/packages/prices-ingest-core/src/registry_io.rs index 59a29075..7967de4a 100644 --- a/packages/prices-ingest-core/src/registry_io.rs +++ b/packages/prices-ingest-core/src/registry_io.rs @@ -103,27 +103,39 @@ impl Registries { /// Rehydrate registries from persisted rows (merged into `self`, so a load /// can seed a run that then keeps discovering). Rows with an unknown venue /// string are skipped. + /// + /// A pair-backed row (Soroswap, SushiSwap) whose `token0`/`token1` are + /// blank registers its VENUE but not its pair. Registering a blank pair + /// would be worse than not loading the row at all: `contains()` would + /// answer true, `classify_amm_groups` would clear `pair_unresolved`, and + /// the pool would be priced against asset `""` instead of being recorded in + /// `unresolved_pools`. Leaving the pair out sends it down the + /// venue-known-but-unpriced branch, which is exactly what a missing pair + /// is. This mirrors the learn side, where a `pool_created` without a token + /// pair learns nothing (task 0290 review). pub fn load_pool_rows(&mut self, rows: &[PoolRegistryRow]) { for row in rows { let Some(venue) = Venue::from_source(&row.venue) else { continue; }; self.venue.insert(row.contract_id.clone(), venue.clone()); + let pair_complete = !row.token0.is_empty() && !row.token1.is_empty(); match venue { - Venue::Soroswap => { + Venue::Soroswap if pair_complete => { self.soroswap.register( row.contract_id.clone(), row.token0.clone(), row.token1.clone(), ); } - Venue::Sushiswap => { + Venue::Sushiswap if pair_complete => { self.sushiswap.register( row.contract_id.clone(), row.token0.clone(), row.token1.clone(), ); } + Venue::Soroswap | Venue::Sushiswap => {} Venue::Phoenix => match hex_decode32(&row.wasm_hash) { Some(hash) => self.phoenix.register_with_wasm( row.contract_id.clone(), @@ -219,6 +231,59 @@ mod tests { assert_eq!(loaded.pool_count(), reg.pool_count()); } + #[test] + fn a_persisted_pair_row_without_tokens_loads_its_venue_but_no_pair() { + // A hand-written or imported row that names a pair-backed venue but + // carries blank tokens. Registering it would make `contains()` true and + // price the pool against asset "" — it must stay pair-unresolved so + // `classify_amm_groups` records it in `unresolved_pools` (task 0290). + let rows: Vec = ["soroswap", "sushiswap"] + .iter() + .map(|venue| PoolRegistryRow { + contract_id: format!("CBLANK_{venue}"), + venue: (*venue).to_string(), + token0: String::new(), + token1: String::new(), + pool_type: 0, + wasm_hash: String::new(), + }) + .collect(); + + let mut loaded = Registries::new(); + loaded.load_pool_rows(&rows); + + assert_eq!( + loaded.venue.get("CBLANK_soroswap"), + Some(&Venue::Soroswap), + "the venue is known — only the pair is missing" + ); + assert_eq!( + loaded.venue.get("CBLANK_sushiswap"), + Some(&Venue::Sushiswap) + ); + assert!(!loaded.soroswap.contains("CBLANK_soroswap")); + assert!(!loaded.sushiswap.contains("CBLANK_sushiswap")); + assert_eq!(loaded.pool_count(), 0); + } + + #[test] + fn a_persisted_pair_row_missing_one_token_loads_no_pair_either() { + let rows = vec![PoolRegistryRow { + contract_id: "CHALFPAIR".into(), + venue: "sushiswap".into(), + token0: "CTOKEN0".into(), + token1: String::new(), + pool_type: 0, + wasm_hash: String::new(), + }]; + + let mut loaded = Registries::new(); + loaded.load_pool_rows(&rows); + + assert_eq!(loaded.venue.get("CHALFPAIR"), Some(&Venue::Sushiswap)); + assert!(!loaded.sushiswap.contains("CHALFPAIR")); + } + fn snapshot(reg: &Registries) -> HashMap { reg.to_pool_rows() .into_iter() From 6ff06a4f6db8ba3c575f2dbc559adcaca65bdd30 Mon Sep 17 00:00:00 2001 From: karczuRF Date: Mon, 21 Sep 2026 12:14:51 +0200 Subject: [PATCH 7/7] docs(lore-0290): fix the venue docs the review flagged - seed-pool-registry: the SushiSwap run pointed at a command the 0290 task file does not carry; the command is now inline, with its range and why one read covers every factory generation - init.sql: pool_registry's header listed three venues and called token0/token1 the Soroswap pair; both are now accurate - extractors-core: the Sushiswap doc linked TokenPair, which lives in a downstream crate and could not resolve --- docs/runbooks/seed-pool-registry.md | 28 +++++++++++++++++++++- packages/extractors-core/src/lib.rs | 8 ++++--- packages/prices-clickhouse/schema/init.sql | 12 ++++++---- 3 files changed, 40 insertions(+), 8 deletions(-) diff --git a/docs/runbooks/seed-pool-registry.md b/docs/runbooks/seed-pool-registry.md index 0024358f..8d709415 100644 --- a/docs/runbooks/seed-pool-registry.md +++ b/docs/runbooks/seed-pool-registry.md @@ -185,7 +185,7 @@ chunk on the shared box. As of 2026-09-17 every missing pool was created after ledger 63,000,000 (checked per venue over the whole Soroban era), so the catch-up only needs `63000000` to the tip. **Task 0290 is the exception:** SushiSwap V3's pools go back to ledger 60,147,305, so its run starts at -`60000000` — the exact command is in the 0290 task file. +`60000000` — the command is [below](#task-0290--sushiswap-v3s-wider-range). ```bash # On the prod host, under tmux: @@ -204,6 +204,32 @@ investigate before the write. Then drop `--dry-run` to write, and run the dry run once more: it must report `to_write=0`. +#### Task 0290 — SushiSwap V3's wider range + +SushiSwap V3 is the one venue whose pools predate ledger 63,000,000: the first +`pool_created` is at 60,147,305 and the live factory's pools start at ~61.49M, +so the 63M catch-up above misses them. Run it over its own range **once**, then +the 63M catch-up covers it like every other venue: + +```bash +# On the prod host, under tmux: +read -rs CH_PW +CLICKHOUSE_PASSWORD="$CH_PW" ~/events-backfill --discover-pools \ + --start 60000000 --end \ + --clickhouse-url http://localhost:8123 --dry-run +``` + +That is ~4.5M ledgers, so ~14 chunks at the default `--chunk-size 320000` and +2-4 s of `topics_xdr` parsing each — a couple of minutes, well inside a tmux +session. It reads the same factory events as the run above; only `--start` +differs. + +Expect `per_venue` to carry a `"sushiswap"` entry. The read has **no emitter +filter**, so it learns every generation's pools, not just the live factory's — +which is what this wider range is for: 99 SushiSwap pools have traded all-time +and three of them come from an earlier factory generation that is still trading. +Confirm with the same `FINAL` count below, then drop `--dry-run` to write. + ### Verify ```sql diff --git a/packages/extractors-core/src/lib.rs b/packages/extractors-core/src/lib.rs index c808c64d..9ed32be6 100644 --- a/packages/extractors-core/src/lib.rs +++ b/packages/extractors-core/src/lib.rs @@ -6,9 +6,11 @@ pub enum Venue { Aquarius, Phoenix, /// SushiSwap V3 — a Uniswap-v3-style concentrated-liquidity venue (task - /// 0290). Its pool `swap` carries signed `amount0`/`amount1`, the same - /// shape [`Venue::Soroswap`]'s CLMM pools use, so both decode through - /// [`TokenPair`]-backed extraction. + /// 0290). Its pool `swap` carries signed `amount0`/`amount1`, the CLMM + /// shape the `soroswap-extractor` crate already decodes, and it resolves + /// its tokens through that crate's pair registry exactly as + /// [`Venue::Soroswap`] does — a separate instance, so the two venues never + /// share a `contract_id`. Sushiswap, } diff --git a/packages/prices-clickhouse/schema/init.sql b/packages/prices-clickhouse/schema/init.sql index 97d07723..17b9a8b2 100644 --- a/packages/prices-clickhouse/schema/init.sql +++ b/packages/prices-clickhouse/schema/init.sql @@ -605,10 +605,14 @@ SETTINGS index_granularity = 8192; -- output of the in-window registry so a partial re-backfill (a mid-history -- window) or the live processor can LOAD it instead of re-deriving from Soroban -- activation (this inverts task 0069: registry-as-output, not required-input). --- venue = 'soroswap' | 'phoenix' | 'aquarius'. token0/token1 are the Soroswap --- pair tokens (needed because a Soroswap swap event omits them); pool_type / --- wasm_hash are Phoenix pool details; both default empty for venues that don't --- use them. ReplacingMergeTree(updated_at) on contract_id collapses re-runs; +-- venue = 'soroswap' | 'phoenix' | 'aquarius' | 'sushiswap' (task 0290). +-- token0/token1 are the pair tokens of the two pair-backed venues — Soroswap +-- (from `new_pair`) and SushiSwap V3 (from `pool_created`) — needed because +-- their swap events omit them; pool_type / wasm_hash are Phoenix pool details; +-- both default empty for venues that don't use them. A pair-backed row with +-- blank tokens does NOT resolve: it loads its venue only, and the pool is +-- reported in `prices.unresolved_pools` rather than priced against an empty +-- asset. ReplacingMergeTree(updated_at) on contract_id collapses re-runs; -- read with FINAL. -- --------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS prices.pool_registry (