PriceLevels
Entry point for Titan's pAMM price levels. Wraps a snapshot source and exposes quote helpers, mirroring the structure of PropAmmRouter.
Usage
from propamm.prices import PriceLevels, PriceLevelsWsSource
# one-shot HTTP snapshot (default)
prices = PriceLevels()
snapshot = await prices.get_price_levels()
# live stream
prices = PriceLevels(
source=PriceLevelsWsSource(url="wss://eu.rpc.titanbuilder.xyz/ws/pamm_price_levels"),
)
snapshot = await prices.get_price_levels()
await prices.close()Constructor
PriceLevels(
source: PriceLevelsSource | None = None,
*,
rpc_url: str | None = None,
)source (optional)
Source the snapshot is pulled from. Defaults to a PriceLevelsRpcSource. Pass a PriceLevelsWsSource for a live stream.
rpc_url (optional)
Titan JSON-RPC endpoint used by the quote helpers. Ignored when source is a PriceLevelsRpcSource (its URL is reused). Set this when pairing a PriceLevelsWsSource with a private or regional deployment so quotes go to the same host:
prices = PriceLevels(
source=PriceLevelsWsSource(url="wss://us.rpc.titanbuilder.xyz/ws/pamm_price_levels"),
rpc_url="https://us.rpc.titanbuilder.xyz",
)Methods
get_price_levels()
async def get_price_levels() -> PriceLevelsSnapshotLatest price-level snapshot from the configured source.
Returns PriceLevelsSnapshot.
get_quote()
async def get_quote(token_in: str, token_out: str, amount_in: int) -> TitanQuoteBest quote across all pAMMs for a size (titan_getPammQuote). Served from Titan's latest snapshot over HTTP, skipping the on-chain eth_call that router.quote runs.
Returns TitanQuote.
get_quote_venue()
async def get_quote_venue(venue: str, token_in: str, token_out: str, amount_in: int) -> TitanQuoteQuote from a specific venue for a size (titan_getPammQuoteVenue).
Returns TitanQuote.
close()
async def close() -> NoneTear down the snapshot source. Closes the stream socket when the source is a PriceLevelsWsSource; no-op for PriceLevelsRpcSource.