Token Resolution¶
Unified token resolution for addresses, decimals, and symbol lookups across all chains.
Symbol-based token references are deprecated
Token symbols are metadata, not stable asset identity — the same ticker
resolves to a different contract on every chain and can be spoofed.
Passing a bare symbol to TokenResolver, MarketSnapshot, or Intent
construction emits a SymbolTokenResolutionWarning (a FutureWarning),
once per external callsite.
Symbols keep working for the remainder of the 2.x line and are rejected
in Almanak SDK 3.0.0 with SymbolTokenResolutionError. Prefer a
chain-specific contract address or a
CAIP-19 asset identifier.
Usage¶
from almanak.framework.data.tokens import get_token_resolver
resolver = get_token_resolver()
# Preferred: resolve by address — stable asset identity
token = resolver.resolve("0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "arbitrum")
print(token.symbol, token.decimals) # USDC 6
# Preferred: resolve by CAIP-19 asset identifier
token = resolver.resolve_caip19("eip155:42161/erc20:0xaf88d065e77c8cC2239327C5EDb3A432268e5831")
# Deprecated: resolve by symbol — warns today, raises in 3.0.0
token = resolver.resolve("USDC", "arbitrum")
# Convenience methods (also symbol-deprecated when given a bare symbol)
decimals = resolver.get_decimals("arbitrum", "0xaf88d065e77c8cC2239327C5EDb3A432268e5831")
address = resolver.get_address("arbitrum", "USDC")
# For DEX swaps (auto-wraps native tokens: ETH->WETH, etc.)
token = resolver.resolve_for_swap("ETH", "arbitrum")
To check whether a value already carries address-based identity before handing it to the SDK:
from almanak.framework.data.tokens.deprecation import is_address_based_token_reference
is_address_based_token_reference("USDC", "arbitrum") # False - deprecated symbol
is_address_based_token_reference("0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "arbitrum") # True
get_token_resolver
¶
get_token_resolver(
gateway_client: Any | None = None,
cache_file: str | None = None,
gateway_channel: Channel | None = None,
) -> TokenResolver
Get the singleton TokenResolver instance.
This is the recommended entry point for token resolution.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
gateway_client
|
Any | None
|
DEPRECATED - Use gateway_channel instead. |
None
|
cache_file
|
str | None
|
Optional path to cache file. Only used on first call. |
None
|
gateway_channel
|
Channel | None
|
Optional gRPC channel to gateway for on-chain lookups. If None, only static resolution is available. On-chain discovery gracefully falls back to static resolution if the gateway becomes unavailable. |
None
|
返回:
| 类型 | 描述 |
|---|---|
TokenResolver
|
The singleton TokenResolver instance |
Example
from almanak.framework.data.tokens import get_token_resolver
# Static resolution only
resolver = get_token_resolver()
usdc = resolver.resolve("USDC", "arbitrum")
# With gateway for on-chain discovery
import grpc
channel = grpc.insecure_channel("localhost:50051")
resolver = get_token_resolver(gateway_channel=channel)
TokenResolver
¶
TokenResolver(
gateway_client: Any | None = None,
cache_file: str | None = None,
gateway_channel: Channel | None = None,
)
Unified token resolver with multi-layer caching.
This class provides the main API for token resolution in the Almanak framework. It implements a singleton pattern for thread-safe global access.
Resolution Order
- Memory cache - fastest, O(1)
- Disk cache - loads from JSON, promotes to memory
- Static registry - DEFAULT_TOKENS from defaults.py
- Gateway on-chain lookup - queries ERC20 contracts (if gateway_client provided)
Thread Safety
Uses threading.RLock for all operations. Safe for concurrent access.
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
gateway_client |
Optional gateway client for on-chain lookups |
Example
Initialize the TokenResolver.
NOTE: Prefer using get_instance() for singleton access.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
gateway_client
|
Any | None
|
DEPRECATED - Use gateway_channel instead. Kept for backward compatibility. |
None
|
cache_file
|
str | None
|
Optional path to cache file. Defaults to ~/.almanak/token_cache.json |
None
|
gateway_channel
|
Channel | None
|
Optional gRPC channel to gateway for on-chain lookups. If None, only static resolution is available. On-chain discovery will gracefully fall back to static resolution if the gateway becomes unavailable. |
None
|
get_instance
classmethod
¶
get_instance(
gateway_client: Any | None = None,
cache_file: str | None = None,
gateway_channel: Channel | None = None,
) -> TokenResolver
Get the singleton TokenResolver instance.
This is the recommended way to get a TokenResolver. The first call creates the instance, subsequent calls return the same instance.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
gateway_client
|
Any | None
|
DEPRECATED - Use gateway_channel instead. |
None
|
cache_file
|
str | None
|
Optional path to cache file. Only used on first call. |
None
|
gateway_channel
|
Channel | None
|
Optional gRPC channel to gateway for on-chain lookups. Only used on first call when creating instance. Pass a grpc.Channel connected to the gateway server. |
None
|
返回:
| 类型 | 描述 |
|---|---|
TokenResolver
|
The singleton TokenResolver instance |
Example
reset_instance
classmethod
¶
Reset the singleton instance. Primarily for testing.
scoped_metadata
¶
Supply gateway-discovered address metadata for one synchronous compile.
Only unresolved addresses enter the scope. Symbols, shared caches and trust flags are unchanged. All resolver instances in this async context see the scope, including connector code using the singleton. Gateway lookups are disabled until the context exits to prevent self-RPC deadlocks.
resolve_caip19
¶
resolve_caip19(
caip19: str,
*,
log_errors: bool = True,
skip_gateway: bool = False,
) -> ResolvedToken
Resolve a CAIP-19 asset id to a fully-resolved token.
Parses the id, resolves the chain via its CAIP-2 part, then resolves the
asset through the normal cascade so decimals / symbol come back
populated (CAIP-19 carries identity only, not decimals). slip44
assets resolve the chain's native token; erc20 / token (SPL)
assets resolve by address. VIB-5175.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
caip19
|
str
|
A CAIP-19 asset id, e.g. |
必需 |
log_errors
|
bool
|
Forwarded to |
True
|
skip_gateway
|
bool
|
Forwarded to |
False
|
返回:
| 类型 | 描述 |
|---|---|
ResolvedToken
|
ResolvedToken with full metadata. |
引发:
| 类型 | 描述 |
|---|---|
ValueError
|
|
TokenResolutionError
|
the asset cannot be resolved on its chain. |
resolve
¶
resolve(
token: str,
chain: str,
*,
log_errors: bool = True,
skip_gateway: bool = False,
) -> ResolvedToken
Resolve a token by address or CAIP-19 asset id on a chain.
Bare symbols remain available with SymbolTokenResolutionWarning in
SDK 2.x and raise SymbolTokenResolutionError in SDK 3.0.0+.
This is the main resolution method. It checks: 1. Memory cache 2. Disk cache 3. Static registry 4. Gateway on-chain lookup (if token is an address and gateway available)
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
str
|
Token address or CAIP-19 identity. Bare symbols are deprecated. |
必需 |
chain
|
str
|
Chain name or Chain enum |
必需 |
log_errors
|
bool
|
If False, suppress warning logs on resolution failure (default True). Use False for best-effort lookups where failures are expected and handled. |
True
|
skip_gateway
|
bool
|
If True, skip the slow gateway on-chain lookup and fail fast after cache + static registry. Use for cosmetic/best-effort lookups where a 30s gateway timeout is unacceptable. |
False
|
返回:
| 类型 | 描述 |
|---|---|
ResolvedToken
|
ResolvedToken with full metadata |
引发:
| 类型 | 描述 |
|---|---|
TokenNotFoundError
|
If token cannot be resolved |
InvalidTokenAddressError
|
If address format is invalid |
TokenResolutionError
|
For other resolution errors |
is_gateway_connected
¶
Check if gateway is connected and available for on-chain lookups.
This method checks if a gateway channel is configured and appears to be connected. Note that the actual availability is verified lazily - the gateway might become unavailable between this check and actual use.
返回:
| 类型 | 描述 |
|---|---|
bool
|
True if gateway channel is configured and appears available, |
bool
|
False otherwise. |
set_gateway_channel
¶
Set or update the gateway channel.
This allows changing the gateway connection after initialization. Useful for reconnection scenarios or testing.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
channel
|
Channel | None
|
gRPC channel to gateway, or None to disable gateway |
必需 |
resolve_pair
¶
Resolve a pair of tokens for a swap operation.
Convenience method for resolving both tokens in a trading pair.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token_in
|
str
|
Input token symbol or address |
必需 |
token_out
|
str
|
Output token symbol or address |
必需 |
chain
|
str
|
Chain name or Chain enum |
必需 |
返回:
| 类型 | 描述 |
|---|---|
tuple[ResolvedToken, ResolvedToken]
|
Tuple of (resolved_token_in, resolved_token_out) |
引发:
| 类型 | 描述 |
|---|---|
TokenNotFoundError
|
If either token cannot be resolved |
TokenResolutionError
|
For other resolution errors |
get_decimals
¶
Get the decimals for a token on a specific chain.
Convenience method that extracts just the decimals from resolution. NEVER defaults to 18 - always raises TokenNotFoundError if unknown.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
chain
|
str
|
Chain name or Chain enum |
必需 |
token
|
str
|
Token address or CAIP-19 identity. Bare symbols are deprecated. |
必需 |
返回:
| 类型 | 描述 |
|---|---|
int
|
Number of decimal places |
引发:
| 类型 | 描述 |
|---|---|
TokenNotFoundError
|
If token cannot be resolved |
known_static_tokens_by_chain
¶
Return a read-only snapshot of static token metadata by chain/address.
This is intentionally static-only: it exposes the JSON-backed token
catalogue plus connector-published synthetic metadata that was registered
into the resolver at construction time. Gateway-discovered and manually
registered runtime tokens remain available through resolve().
get_address
¶
Get the address for a token reference on a specific chain.
Convenience method that extracts just the address from resolution. Passing a bare symbol is deprecated in SDK 2.x and rejected in 3.0.0+.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
chain
|
str
|
Chain name or Chain enum |
必需 |
symbol
|
str
|
Address or CAIP-19 identity. Bare symbols are deprecated. |
必需 |
返回:
| 类型 | 描述 |
|---|---|
str
|
Contract address |
引发:
| 类型 | 描述 |
|---|---|
TokenNotFoundError
|
If token cannot be resolved |
resolve_for_swap
¶
Resolve a token for swap operations, auto-wrapping native tokens.
This method resolves a token and if it's a native token (ETH, MATIC, AVAX, BNB), automatically returns the wrapped version instead (WETH, WMATIC, WAVAX, WBNB). This is because most DEX protocols cannot swap native tokens directly.
For non-native tokens, this behaves identically to resolve().
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
str
|
Token address or CAIP-19 identity. Bare symbols are deprecated. |
必需 |
chain
|
str
|
Chain name or Chain enum |
必需 |
返回:
| 类型 | 描述 |
|---|---|
ResolvedToken
|
ResolvedToken - wrapped version if native, original otherwise |
引发:
| 类型 | 描述 |
|---|---|
TokenNotFoundError
|
If token or wrapped version cannot be resolved |
InvalidTokenAddressError
|
If address format is invalid |
TokenResolutionError
|
For other resolution errors |
resolve_for_protocol
¶
Resolve a token with protocol-specific handling.
This method provides a hook for future protocol-specific token resolution. Currently, it simply delegates to resolve_for_swap() for DEX protocols and to resolve() for other protocols.
This allows for future expansion where specific protocols might have unique token requirements (e.g., protocol-specific wrapped tokens, canonical bridge tokens, etc.).
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
str
|
Token address or CAIP-19 identity. Bare symbols are deprecated. |
必需 |
chain
|
str
|
Chain name or Chain enum |
必需 |
protocol
|
str
|
Protocol identifier (e.g., "uniswap_v3", "aave_v3") |
必需 |
返回:
| 类型 | 描述 |
|---|---|
ResolvedToken
|
ResolvedToken with appropriate protocol handling |
引发:
| 类型 | 描述 |
|---|---|
TokenNotFoundError
|
If token cannot be resolved |
TokenResolutionError
|
For other resolution errors |
Example
# DEX protocols get auto-wrapped native tokens
token = resolver.resolve_for_protocol(
"eip155:42161/slip44:60",
"arbitrum",
"uniswap_v3",
)
assert token.symbol == "WETH"
# Lending protocols get the original token
token = resolver.resolve_for_protocol(
"eip155:1/slip44:60",
"ethereum",
"aave_v3",
)
assert token.symbol == "ETH"
clear_negative_cache
¶
Drop all negative-cache entries. Useful in tests and after a large registry refresh.
register
¶
Register a token explicitly at runtime.
This allows adding custom tokens that aren't in the static registry. Registered tokens are stored in the cache.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
ResolvedToken
|
ResolvedToken to register |
必需 |
register_token
¶
register_token(
symbol: str,
chain: str,
address: str,
decimals: int,
*,
name: str | None = None,
coingecko_id: str | None = None,
is_stablecoin: bool = False,
) -> ResolvedToken
Register a custom token by its basic properties.
Convenience wrapper around register() for strategy authors who need to register protocol-specific tokens (e.g., Pendle PT/YT, LP tokens) that aren't in the static registry.
After registration, the token is resolvable via resolve(), get_address(), and get_decimals() within the same process.
Note: This registers tokens in the local resolver only. Gateway-backed lookups (e.g., MarketSnapshot.balance() by symbol) require the gateway to also know the token. For balance queries on custom tokens, use the token address directly: market.balance("0x...").
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
symbol
|
str
|
Token symbol (e.g., "PT-wstETH-25JUN2026") |
必需 |
chain
|
str
|
Chain name or Chain enum |
必需 |
address
|
str
|
Token contract address |
必需 |
decimals
|
int
|
Token decimal places |
必需 |
name
|
str | None
|
Optional human-readable name |
None
|
coingecko_id
|
str | None
|
Optional CoinGecko ID for price fetching |
None
|
is_stablecoin
|
bool
|
Whether this is a stablecoin (default False) |
False
|
返回:
| 类型 | 描述 |
|---|---|
ResolvedToken
|
The registered ResolvedToken (can be used immediately) |
引发:
| 类型 | 描述 |
|---|---|
InvalidTokenAddressError
|
If address format is invalid |
TokenResolutionError
|
If chain is not recognized |
stats
¶
Get resolver performance statistics.
返回:
| 类型 | 描述 |
|---|---|
dict[str, int]
|
Dict with cache_hits, static_hits, gateway_lookups, errors |
cache_stats
¶
Get cache performance statistics.
返回:
| 类型 | 描述 |
|---|---|
dict[str, int]
|
Dict with memory_hits, disk_hits, misses, evictions |
ResolvedToken
dataclass
¶
ResolvedToken(
symbol: str,
address: str,
decimals: int,
chain: str,
chain_id: int,
name: str | None = None,
coingecko_id: str | None = None,
is_stablecoin: bool = False,
peg_class: PegClass | None = None,
is_native: bool = False,
is_wrapped_native: bool = False,
canonical_symbol: str | None = None,
bridge_type: BridgeType = BridgeType.NATIVE,
source: str = "static",
is_verified: bool = True,
resolved_at: datetime | None = None,
)
Fully resolved token with all metadata for a specific chain.
This is a frozen (immutable) dataclass representing a token that has been fully resolved with all its metadata. It's designed for caching and thread-safe access.
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
symbol |
str
|
Token symbol (e.g., "ETH", "USDC", "WBTC") |
address |
str
|
Contract address on the resolved chain in resolver output/display form |
decimals |
int
|
Token decimal places |
chain |
str
|
Canonical lowercase chain name where this token is resolved |
chain_id |
int
|
Numeric chain ID for the resolved chain |
name |
str | None
|
Human-readable token name (e.g., "Ether", "USD Coin") |
coingecko_id |
str | None
|
CoinGecko API identifier for price fetching |
is_stablecoin |
bool
|
Whether this token is a stablecoin |
is_native |
bool
|
Whether this is the native gas token (ETH, MATIC, AVAX, etc.) |
is_wrapped_native |
bool
|
Whether this is wrapped native (WETH, WMATIC, WAVAX, etc.) |
canonical_symbol |
str | None
|
Canonical symbol for cross-chain identification (e.g., "USDC" for both USDC and USDC.e) |
bridge_type |
BridgeType
|
Bridge status of the token |
source |
str
|
Where the token metadata came from ("static", "on_chain", "cache") |
is_verified |
bool
|
Whether the token metadata has been verified |
resolved_at |
datetime | None
|
Timestamp when the token was resolved |
Example
resolved_usdc = ResolvedToken(
symbol="USDC",
address="0xaf88d065e77c8cC2239327C5EDb3A432268e5831",
decimals=6,
chain="arbitrum",
chain_id=42161,
name="USD Coin",
coingecko_id="usd-coin",
is_stablecoin=True,
is_native=False,
is_wrapped_native=False,
canonical_symbol="USDC",
bridge_type=BridgeType.NATIVE,
source="static",
is_verified=True,
resolved_at=datetime.now(),
)
BridgeType
¶
Bases: Enum
Token bridge status indicating origin of the token on a chain.
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
NATIVE |
Token is native to this chain (e.g., ETH on Ethereum, USDC native on Arbitrum) |
|
BRIDGED |
Token was bridged from another chain (e.g., USDC.e on Arbitrum) |
|
CANONICAL |
Token is the canonical/official bridge representation for cross-chain transfers |
Example
Exceptions¶
TokenResolutionError
¶
Bases: Exception
Base exception for token resolution errors.
This is the base class for all token resolution-related exceptions. It provides structured error information including the token identifier, chain, reason for failure, and actionable suggestions.
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
token |
The token identifier that failed to resolve (symbol or address) |
|
chain |
The chain where resolution was attempted |
|
reason |
Explanation of why resolution failed |
|
suggestions |
List of actionable suggestions to fix the issue |
Example
Initialize the exception.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
str
|
The token identifier that failed to resolve |
必需 |
chain
|
str
|
The chain where resolution was attempted |
必需 |
reason
|
str
|
Explanation of why resolution failed |
必需 |
suggestions
|
list[str] | None
|
List of actionable suggestions to fix the issue |
None
|
TokenNotFoundError
¶
TokenNotFoundError(
token: str,
chain: str,
reason: str = "Token not found in any registry",
suggestions: list[str] | None = None,
)
Bases: TokenResolutionError
Raised when a token is not found in any registry.
This exception is raised when: - Token symbol is not in the static registry - Token symbol is not in the cache - Token address (if provided) doesn't match any known token - Gateway on-chain lookup (if enabled) also fails
Example
Initialize the exception.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
str
|
The token identifier that was not found |
必需 |
chain
|
str
|
The chain where the token was searched |
必需 |
reason
|
str
|
Explanation of why the token wasn't found |
'Token not found in any registry'
|
suggestions
|
list[str] | None
|
List of actionable suggestions |
None
|
AmbiguousTokenError
¶
AmbiguousTokenError(
token: str,
chain: str,
reason: str = "Multiple tokens match the identifier",
matching_addresses: list[str] | None = None,
suggestions: list[str] | None = None,
)
Bases: TokenResolutionError
Raised when multiple tokens match the given identifier.
This exception is raised when: - A symbol matches multiple tokens on the same chain (e.g., multiple USDC variants) - Bridged tokens create ambiguity (USDC vs USDC.e) - Multiple protocols have deployed tokens with the same symbol
属性:
| 名称 | 类型 | 描述 |
|---|---|---|
matching_addresses |
List of addresses that match the token identifier |
Example
raise AmbiguousTokenError(
token="USDC",
chain="arbitrum",
reason="Multiple USDC variants found on Arbitrum",
matching_addresses=[
"0xaf88d065e77c8cC2239327C5EDb3A432268e5831", # Native USDC
"0xFF970A61A04b1cA14834A43f5dE4533eBDDB5CC8", # USDC.e (bridged)
],
suggestions=[
"Use 'USDC' for native USDC: 0xaf88d065e77c8cC2239327C5EDb3A432268e5831",
"Use 'USDC.e' for bridged USDC: 0xFF970A61A04b1cA14834A43f5dE4533eBDDB5CC8",
"Or specify the full contract address",
],
)
Initialize the exception.
参数:
| 名称 | 类型 | 描述 | 默认 |
|---|---|---|---|
token
|
str
|
The ambiguous token identifier |
必需 |
chain
|
str
|
The chain where ambiguity occurred |
必需 |
reason
|
str
|
Explanation of the ambiguity |
'Multiple tokens match the identifier'
|
matching_addresses
|
list[str] | None
|
List of addresses that match |
None
|
suggestions
|
list[str] | None
|
List of actionable suggestions |
None
|
Symbol deprecation¶
Raised (3.0.0+) or warned (2.x) when a bare symbol is used where stable asset
identity is required. SYMBOL_TOKEN_REMOVAL_VERSION is the release that flips
the warning into an error.
SymbolTokenResolutionError
¶
Bases: TokenResolutionError
Raised when a token symbol is used after the SDK 3.0.0 removal boundary.
SymbolTokenResolutionWarning
¶
Bases: FutureWarning
Warn that a symbol is being used where stable token identity is required.