feat(simulation): migrate the price level stream to Titan V2

crates/tycho-simulation/CLAUDE.md

@@ -52,17 +52,30 @@ for any protocol indexed by Tycho.
   `tvl`, `component` (pair ids and components) and `quote_tokens`. The `BookFeed*` aliases fill the
   snapshot types into the feed readers
 - **`price_level_stream/`**: Titan pAMM price level stream — `PriceLevelStreamBuilder` turns the
-  Titan WebSocket's per-pair quote-ladder frames directly into `Update`s (no indexer feed
+  Titan WebSocket's book frames directly into `Update`s (no indexer feed
   round-trip); `PriceLevelStreamState` quotes by interpolating the ladder and refuses to quote
   once its frame is one slot old (`quotable_until`, monotonic, never serialized, ignored by
-  `eq`; `without_quote_guard` on the builder turns it off for slow quoters). Frames are best
+  `eq`; `without_quote_guard` on the builder turns it off for slow quoters). A frame carries one
+  flat `books` list — one book per venue and pair, priced in human units as float
+  `price`/`cumVolume` levels over symbol-named tokens — so `convert.rs` derives the integer
+  ladders the states serve: `resolve_book_tokens` resolves the two symbols and `ladders_for_book`
+  turns the bids into the `token1 -> token2` ladder and the asks into the reverse, converting the
+  sides independently so a rejected one never costs the other. `tracker.rs` groups the books per
+  venue and feeds the freshness machinery below unchanged. Resolution is fail-closed and never
+  trusts the stream: a symbol prices only through `config.rs`'s `SYMBOL_BINDINGS` (or a caller's
+  `with_symbol_bindings` override, which wins) **and** only if the bound address is in the
+  consumer's token map, which also supplies the decimals — so a token calling itself `WBTC`
+  neither captures the bound asset's quotes nor keeps it from being served, and an unbound symbol
+  stays unpriced. Frames are best
   effort, so `tracker.rs` never removes a component because a frame omits it. It gives every
   component its own deadline and emits `removed_pairs` only when the component's data is
   `stale_after` old; the next accepted frame carrying the component re-adds it. Frames are
-  accepted only if their wire `timestamp` is younger than `stale_after`, not more than one slot
+  accepted only if their stream `ts` is younger than `stale_after`, not more than one slot
   in the future, not older than the newest accepted one (equal allowed), and their block neither
   regresses nor jumps more than one block per elapsed slot plus 2; the block frontier lives in
-  `ServingState::Serving`, so it exists only while something is served. The windows and their
+  `ServingState::Serving`, so it exists only while something is served. The frame's beacon `slot`
+  carries no freshness decision, so a frame without it is judged like one carrying it. The windows
+  and their
   defaults are listed under `# Freshness contract` in `price_level_stream/mod.rs`; every window
   is a multiple of `SLOT` in `mod.rs`; the `stale_after` setter caps its value at
   `MAX_STALE_AFTER` (one hour). `build()` is an `async_stream` loop that selects over frames and
@@ -71,14 +84,26 @@ for any protocol indexed by Tycho.
   when no frame parses within the idle timeout; pings, unparsable text and the consumer's own
   pauses between polls do not count. `telemetry.rs` emits `price_level_stream_*` metrics through
   the `metrics` facade, with label values as enums there and a frame-age histogram at
-  acceptance. Per-venue series start at zero, and no label carries a wire value except the
-  address of an auto-detected venue. Components are identified as `pricelevelstream:{pamm}`. A
+  acceptance; `skipped_books_total{reason}` counts the books left unpriced
+  (`unbound_symbol`, `unadmitted_token`, `one_asset_pair`) and `rejected_sides_total{reason}` the
+  sides dropped whole (`unsupported_level`, `non_finite_amount`, `zero_amount`,
+  `conflicting_knots`). Per-venue series and every one of those reasons start at zero, and no
+  label carries a streamed value except the address of an auto-detected venue — a streamed symbol
+  never becomes a label, only a DEBUG log field. Components are identified as
+  `pricelevelstream:{pamm}`, where `{pamm}` names the venue's **canonical public identity**: the
+  address a swap goes to. `config.rs`'s `PAMM_ADDRESS_ALIASES` normalizes every boundary onto it —
+  incoming book `quoteAddress`es as well as the addresses handed to `add_pamm`/`deny_pamm`, either
+  of which may name either address (today's one entry is Kipseli, streamed under `0xc9a9…` and
+  executed at `0x71e7…`). Component ids, venue names, gas costs, denial and the `pamm_address`
+  attribute therefore always carry the public identity, never the streamed one; `maker` is
+  recorded nowhere. A
   new builder serves nothing: `with_known_pamms` registers the known-good venues and denies
   known-unexecutable ones, `add_pamm` registers individual ones, `deny_pamm` excludes one
   (dropping any registration and blocking auto-detection), and opt-in auto-detection
   additionally serves unknown venues under their address (`pricelevelstream:{0xaddress}`), at
-  most 64 per process. Precedence: between `add_pamm` and `deny_pamm` for the same address the
-  later call wins; `with_known_pamms` defaults never override either, regardless of call order.
+  most 64 per process. Precedence: between `add_pamm` and `deny_pamm` for the same venue the
+  later call wins, whichever of its addresses each named; `with_known_pamms` defaults never
+  override either, regardless of call order.
   By default `build` emits every venue under `fallback:{pamm}`, so tycho-execution routes their
   swaps through `TychoFallbackRouter` (retry on a solver-named fallback pool when the venue
   reverts); `without_fallback_router` keeps every venue on the direct `pricelevelstream:` path.

crates/tycho-simulation/src/price_level_stream/config.rs

@@ -1,4 +1,4 @@
-use std::str::FromStr;
+use std::{collections::HashMap, str::FromStr, sync::LazyLock};
 
 use num_bigint::BigUint;
 use tycho_common::Bytes;
@@ -23,7 +23,9 @@ pub struct PriceLevelStreamConfig {
     /// Bare pAMM name (e.g. `fermiswap`); the emitted components carry
     /// `pricelevelstream:{protocol}` as their protocol system.
     pub protocol: String,
-    /// The pAMM venue address under which Titan streams its quotes.
+    /// The venue's canonical public identity: the address its swaps are sent to, which the
+    /// emitted components carry. Either of a venue's addresses may be given here — see
+    /// [`PAMM_ADDRESS_ALIASES`].
     pub address: Bytes,
     /// Constant per-swap gas cost estimate reported for every quote of this pAMM.
     pub gas_cost: BigUint,
@@ -68,11 +70,10 @@ pub const DEFAULT_AUTO_DETECTED_GAS_COST: u64 = 335_000;
 /// [`add_pamm`](super::stream::PriceLevelStreamBuilder::add_pamm) call for one of these
 /// addresses overrides the corresponding entry.
 ///
-/// Only the venues' router addresses are registered — the keys the price level stream has been
-/// observed to use — because the streamed key doubles as the execution target
-/// ([`PAMM_ADDRESS_ATTRIBUTE`](super::stream::PAMM_ADDRESS_ATTRIBUTE)): unlike the state-override
-/// stream, which also publishes frames under non-executable oracle aliases, an entry here must
-/// be an address a swap can be sent to.
+/// Only the venues' router addresses are registered, because an entry here doubles as the
+/// execution target ([`PAMM_ADDRESS_ATTRIBUTE`](super::stream::PAMM_ADDRESS_ATTRIBUTE)): it must
+/// be an address a swap can be sent to. A venue whose streamed key is not that address is
+/// bridged by [`PAMM_ADDRESS_ALIASES`] rather than registered under the streamed key.
 pub fn default_served_pamms() -> Vec<PriceLevelStreamConfig> {
     // The venues' `IPropAMM::swap` gas, calibrated by replaying real fills on the live venues at
     // fresh-oracle blocks via `debug_traceCall`, plus a small headroom. Deliberately excludes
@@ -82,8 +83,9 @@ pub fn default_served_pamms() -> Vec<PriceLevelStreamConfig> {
         // The FermiSwapper router. Measured ~177k-182k (2026-08-18).
         ("fermiswap", "0x5979458912f80b96d30d4220af8e2e4925a33320", 185_000u64),
         // The KipseliPropAMMWrapper router. Measured ~308k-329k (2026-08-18). Titan's venue docs
-        // list a newer Kipseli router (0x342b8458…), but the stream still keys Kipseli
-        // quotes by this address and the newer one has no activity.
+        // list a newer Kipseli router (0x342b8458…), which has no activity. The stream keys
+        // Kipseli's books by 0xc9a9… instead, which `PAMM_ADDRESS_ALIASES` maps onto this
+        // address.
         ("kipseli", "0x71e790dd841c8a9061487cb3e78c288e75ce0b3d", 335_000u64),
         // The Metric router (unverified; identified via its pools' pricing reads of the Metric
         // oracle 0x28d9cced…). Measured ~225k (2026-08-18).
@@ -122,3 +124,107 @@ pub fn default_denied_pamms() -> Vec<Bytes> {
     // by an arbitrary caller.
     Vec::new()
 }
+
+/// The streamed identities Titan streams some venues under, each paired with that venue's canonical
+/// public identity — the address a swap is sent to.
+///
+/// A streamed `quoteAddress` is a join key on Titan's streams, not necessarily an executable
+/// address. Where the two differ, the public identity is the executable one: it is what component
+/// ids, venue names, and the
+/// [`PAMM_ADDRESS_ATTRIBUTE`](super::stream::PAMM_ADDRESS_ATTRIBUTE) carry, and what
+/// [`add_pamm`](super::stream::PriceLevelStreamBuilder::add_pamm) and
+/// [`deny_pamm`](super::stream::PriceLevelStreamBuilder::deny_pamm) are expected to name. Either
+/// address may be passed to those two, since both normalize through this table.
+///
+/// Venues absent from this table stream under their public identity, so they need no entry.
+pub const PAMM_ADDRESS_ALIASES: &[(&str, &str)] = &[
+    // Kipseli streams both its books and its overrides under 0xc9a9…, while its swaps go to the
+    // KipseliPropAMMWrapper router (0x71e7…) — the alias Titan's public taker docs publish, and
+    // the address `default_served_pamms` registers it under.
+    ("0xc9a956b5196dfe110debe8781857bc8b97d32091", "0x71e790dd841c8a9061487cb3e78c288e75ce0b3d"),
+];
+
+/// [`PAMM_ADDRESS_ALIASES`] as a lookup, built once per process.
+static CANONICAL_PAMM_ADDRESSES: LazyLock<HashMap<Bytes, Bytes>> = LazyLock::new(|| {
+    PAMM_ADDRESS_ALIASES
+        .iter()
+        .map(|(streamed, public)| {
+            (
+                Bytes::from_str(streamed).expect("hardcoded streamed pAMM address must parse"),
+                Bytes::from_str(public).expect("hardcoded public pAMM address must parse"),
+            )
+        })
+        .collect()
+});
+
+/// The canonical public identity of the venue `address` names: the aliased address from
+/// [`PAMM_ADDRESS_ALIASES`], or `address` itself when no alias names it.
+///
+/// Idempotent — a public identity canonicalizes to itself — so normalizing an already
+/// canonical address is harmless.
+pub(super) fn canonical_pamm_address(address: &Bytes) -> Bytes {
+    CANONICAL_PAMM_ADDRESSES
+        .get(address)
+        .unwrap_or(address)
+        .clone()
+}
+
+/// The token symbols the price level stream quotes in, bound to the Ethereum addresses of the
+/// assets they name.
+///
+/// A streamed book names its tokens by symbol only, and a symbol is not an identity: resolving
+/// one against the consumer's token metadata by name would let any token that calls itself
+/// `WBTC` capture the bound asset's quotes. A symbol is therefore only ever priced if this table
+/// — or a caller's override — binds it to an address, and that address is additionally present
+/// in the consumer's own token metadata.
+///
+/// Symbols match exactly, case included, so a symbol differing in case stays unbound instead of
+/// being guessed at. A streamed symbol absent here is never priced: `NVDAx`, quoted by TaurusFi
+/// in the 2026-10-08 capture, is deliberately unbound until its address is independently
+/// resolved, since inferring it from the symbol is exactly what this table exists to prevent.
+pub const SYMBOL_BINDINGS: &[(&str, &str)] = &[
+    ("WETH", "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"),
+    ("USDC", "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"),
+    ("USDT", "0xdac17f958d2ee523a2206206994597c13d831ec7"),
+    ("WBTC", "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599"),
+    ("cbBTC", "0xcbb7c0000ab88b473b1f5afd9ef808440eed33bf"),
+];
+
+/// The symbol bindings one stream resolves streamed book symbols through: [`SYMBOL_BINDINGS`]
+/// with the caller's overrides layered on top.
+///
+/// Internal: callers pass their overrides as a plain map to
+/// [`with_symbol_bindings`](super::stream::PriceLevelStreamBuilder::with_symbol_bindings), which
+/// layers them here.
+#[derive(Debug, Clone)]
+pub(super) struct SymbolBindings(HashMap<String, Bytes>);
+
+impl SymbolBindings {
+    /// The trusted defaults with `overrides` layered over them: an override replaces the default
+    /// binding of its symbol and binds symbols the defaults do not carry.
+    pub(super) fn with_overrides(overrides: HashMap<String, Bytes>) -> Self {
+        let mut bindings: HashMap<String, Bytes> = SYMBOL_BINDINGS
+            .iter()
+            .map(|(symbol, address)| {
+                (
+                    (*symbol).to_string(),
+                    Bytes::from_str(address).expect("hardcoded token address must parse"),
+                )
+            })
+            .collect();
+        bindings.extend(overrides);
+        Self(bindings)
+    }
+
+    /// The address bound to `symbol`, or `None` if no binding names it. Matching is exact.
+    pub(super) fn address(&self, symbol: &str) -> Option<&Bytes> {
+        self.0.get(symbol)
+    }
+}
+
+impl Default for SymbolBindings {
+    /// The trusted defaults, without overrides.
+    fn default() -> Self {
+        Self::with_overrides(HashMap::new())
+    }
+}