Token Resolution¶

Unified token resolution for addresses, decimals, and symbol lookups across all chains.

Usage¶

from almanak.framework.data.tokens import get_token_resolver

resolver = get_token_resolver()

# Resolve by symbol
token = resolver.resolve("USDC", "arbitrum")
print(token.address, token.decimals)  # 0xaf88... 6

# Resolve by address
token = resolver.resolve("0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "arbitrum")

# Convenience methods
decimals = resolver.get_decimals("arbitrum", "USDC")
address = resolver.get_address("arbitrum", "USDC")

# For DEX swaps (auto-wraps native tokens: ETH->WETH, etc.)
token = resolver.resolve_for_swap("ETH", "arbitrum")

get_token_resolver¶

almanak.framework.data.tokens.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.

Parameters:

Name	Type	Description	Default
`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`

Returns:

Type	Description
`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¶

almanak.framework.data.tokens.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.

Attributes:

Name	Type	Description
`gateway_client`		Optional gateway client for on-chain lookups

Example

resolver = TokenResolver.get_instance()

Resolve by symbol¶

token = resolver.resolve("USDC", "arbitrum")

Resolve by address¶

token = resolver.resolve("0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "arbitrum")

Register a custom token¶

resolver.register(my_custom_token)

Initialize the TokenResolver.

NOTE: Prefer using get_instance() for singleton access.

Parameters:

Name	Type	Description	Default
`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.

Parameters:

Name	Type	Description	Default
`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`

Returns:

Type	Description
`TokenResolver`	The singleton TokenResolver instance

Example

Without gateway (static resolution only)¶

resolver = TokenResolver.get_instance() token = resolver.resolve("USDC", "arbitrum")

With gateway (enables on-chain discovery)¶

import grpc channel = grpc.insecure_channel("localhost:50051") resolver = TokenResolver.get_instance(gateway_channel=channel)

reset_instance `classmethod` ¶

reset_instance() -> None

Reset the singleton instance. Primarily for testing.

resolve ¶

resolve(token: str, chain: str | Chain) -> ResolvedToken

Resolve a token by symbol or address on a specific chain.

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)

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol (e.g., "USDC") or address (e.g., "0x...")	required
`chain`	`str \| Chain`	Chain name or Chain enum	required

Returns:

Type	Description
`ResolvedToken`	ResolvedToken with full metadata

Raises:

Type	Description
`TokenNotFoundError`	If token cannot be resolved
`InvalidTokenAddressError`	If address format is invalid
`TokenResolutionError`	For other resolution errors

Example

By symbol¶

usdc = resolver.resolve("USDC", "arbitrum")

By address¶

token = resolver.resolve("0xaf88d065e77c8cC2239327C5EDb3A432268e5831", "arbitrum")

is_gateway_connected ¶

is_gateway_connected() -> bool

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.

Returns:

Type	Description
`bool`	True if gateway channel is configured and appears available,
`bool`	False otherwise.

Example

resolver = get_token_resolver(gateway_channel=channel) if resolver.is_gateway_connected(): print("Gateway available for on-chain token discovery") else: print("Static resolution only")

set_gateway_channel ¶

set_gateway_channel(channel: Channel | None) -> None

Set or update the gateway channel.

This allows changing the gateway connection after initialization. Useful for reconnection scenarios or testing.

Parameters:

Name	Type	Description	Default
`channel`	`Channel \| None`	gRPC channel to gateway, or None to disable gateway	required

Example

import grpc resolver = get_token_resolver()

Connect to gateway later¶

channel = grpc.insecure_channel("localhost:50051") resolver.set_gateway_channel(channel)

Disconnect from gateway¶

resolver.set_gateway_channel(None)

resolve_pair ¶

resolve_pair(
    token_in: str, token_out: str, chain: str | Chain
) -> tuple[ResolvedToken, ResolvedToken]

Resolve a pair of tokens for a swap operation.

Convenience method for resolving both tokens in a trading pair.

Parameters:

Name	Type	Description	Default
`token_in`	`str`	Input token symbol or address	required
`token_out`	`str`	Output token symbol or address	required
`chain`	`str \| Chain`	Chain name or Chain enum	required

Returns:

Type	Description
`tuple[ResolvedToken, ResolvedToken]`	Tuple of (resolved_token_in, resolved_token_out)

Raises:

Type	Description
`TokenNotFoundError`	If either token cannot be resolved
`TokenResolutionError`	For other resolution errors

Example

usdc, weth = resolver.resolve_pair("USDC", "WETH", "arbitrum")

get_decimals ¶

get_decimals(chain: str | Chain, token: str) -> int

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.

Parameters:

Name	Type	Description	Default
`chain`	`str \| Chain`	Chain name or Chain enum	required
`token`	`str`	Token symbol or address	required

Returns:

Type	Description
`int`	Number of decimal places

Raises:

Type	Description
`TokenNotFoundError`	If token cannot be resolved

Example

decimals = resolver.get_decimals("arbitrum", "USDC")

Returns 6¶

get_address ¶

get_address(chain: str | Chain, symbol: str) -> str

Get the address for a token symbol on a specific chain.

Convenience method that extracts just the address from resolution.

Parameters:

Name	Type	Description	Default
`chain`	`str \| Chain`	Chain name or Chain enum	required
`symbol`	`str`	Token symbol (e.g., "USDC")	required

Returns:

Type	Description
`str`	Contract address

Raises:

Type	Description
`TokenNotFoundError`	If token cannot be resolved

Example

address = resolver.get_address("arbitrum", "USDC")

Returns "0xaf88d065e77c8cC2239327C5EDb3A432268e5831"¶

resolve_for_swap ¶

resolve_for_swap(
    token: str, chain: str | Chain
) -> ResolvedToken

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().

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol (e.g., "ETH", "USDC") or address	required
`chain`	`str \| Chain`	Chain name or Chain enum	required

Returns:

Type	Description
`ResolvedToken`	ResolvedToken - wrapped version if native, original otherwise

Raises:

Type	Description
`TokenNotFoundError`	If token or wrapped version cannot be resolved
`InvalidTokenAddressError`	If address format is invalid
`TokenResolutionError`	For other resolution errors

Example

ETH on Arbitrum returns WETH¶

token = resolver.resolve_for_swap("ETH", "arbitrum") assert token.symbol == "WETH"

USDC returns USDC (not native)¶

token = resolver.resolve_for_swap("USDC", "arbitrum") assert token.symbol == "USDC"

resolve_for_protocol ¶

resolve_for_protocol(
    token: str, chain: str | Chain, protocol: str
) -> ResolvedToken

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.).

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol or address	required
`chain`	`str \| Chain`	Chain name or Chain enum	required
`protocol`	`str`	Protocol identifier (e.g., "uniswap_v3", "aave_v3")	required

Returns:

Type	Description
`ResolvedToken`	ResolvedToken with appropriate protocol handling

Raises:

Type	Description
`TokenNotFoundError`	If token cannot be resolved
`TokenResolutionError`	For other resolution errors

Example

DEX protocols get auto-wrapped native tokens¶

token = resolver.resolve_for_protocol("ETH", "arbitrum", "uniswap_v3") assert token.symbol == "WETH"

Lending protocols get the original token¶

token = resolver.resolve_for_protocol("ETH", "ethereum", "aave_v3") assert token.symbol == "ETH"

register ¶

register(token: ResolvedToken) -> None

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.

Parameters:

Name	Type	Description	Default
`token`	`ResolvedToken`	ResolvedToken to register	required

Example

custom_token = ResolvedToken( symbol="CUSTOM", address="0x...", decimals=18, chain=Chain.ARBITRUM, chain_id=42161, name="Custom Token", ) resolver.register(custom_token)

stats ¶

stats() -> dict[str, int]

Get resolver performance statistics.

Returns:

Type	Description
`dict[str, int]`	Dict with cache_hits, static_hits, gateway_lookups, errors

cache_stats ¶

cache_stats() -> dict[str, int]

Get cache performance statistics.

Returns:

Type	Description
`dict[str, int]`	Dict with memory_hits, disk_hits, misses, evictions

ResolvedToken¶

almanak.framework.data.tokens.ResolvedToken `dataclass` ¶

ResolvedToken(
    symbol: str,
    address: str,
    decimals: int,
    chain: Chain,
    chain_id: int,
    name: str | None = None,
    coingecko_id: str | None = None,
    is_stablecoin: bool = False,
    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.

Attributes:

Name	Type	Description
`symbol`	`str`	Token symbol (e.g., "ETH", "USDC", "WBTC")
`address`	`str`	Contract address on the resolved chain
`decimals`	`int`	Token decimal places
`chain`	`Chain`	Chain enum value where this token is resolved
`chain_id`	`int`	Numeric chain ID for the resolved chain
`name`	`str \| None`	Human-readable token name (e.g., "Ethereum", "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=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(), )

__post_init__ ¶

__post_init__() -> None

Validate resolved token data.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> ResolvedToken

Create ResolvedToken from dictionary.

BridgeType¶

almanak.framework.data.tokens.BridgeType ¶

Bases: Enum

Token bridge status indicating origin of the token on a chain.

Attributes:

Name	Type	Description
`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

Native USDC on Arbitrum (issued by Circle directly)¶

native_usdc = ResolvedToken(..., bridge_type=BridgeType.NATIVE)

Bridged USDC.e on Arbitrum (bridged from Ethereum)¶

bridged_usdc = ResolvedToken(..., bridge_type=BridgeType.BRIDGED)

Exceptions¶

almanak.framework.data.tokens.TokenResolutionError ¶

TokenResolutionError(
    token: str,
    chain: str,
    reason: str,
    suggestions: list[str] | None = None,
)

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.

Attributes:

Name	Type	Description
`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

raise TokenResolutionError( token="USDC", chain="unknown_chain", reason="Chain 'unknown_chain' is not supported", suggestions=["Use a supported chain: ethereum, arbitrum, base, optimism"], )

Initialize the exception.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token identifier that failed to resolve	required
`chain`	`str`	The chain where resolution was attempted	required
`reason`	`str`	Explanation of why resolution failed	required
`suggestions`	`list[str] \| None`	List of actionable suggestions to fix the issue	`None`

repr ¶

__repr__() -> str

Return a detailed representation of the exception.

almanak.framework.data.tokens.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

raise TokenNotFoundError( token="UNKNOWNTOKEN", chain="arbitrum", reason="Token not in static registry or cache", suggestions=[ "Check spelling - did you mean 'UNI' or 'LINK'?", "If using an address, ensure it's a valid ERC20 contract", "Use register() to add custom tokens to the resolver", ], )

Initialize the exception.

Parameters:

Name	Type	Description	Default
`token`	`str`	The token identifier that was not found	required
`chain`	`str`	The chain where the token was searched	required
`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`

almanak.framework.data.tokens.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

Attributes:

Name	Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token`	`str`	The ambiguous token identifier	required
`chain`	`str`	The chain where ambiguity occurred	required
`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`

Token Resolution¶

Usage¶

get_token_resolver¶

almanak.framework.data.tokens.get_token_resolver ¶

Static resolution only¶

With gateway for on-chain discovery¶

TokenResolver¶

almanak.framework.data.tokens.TokenResolver ¶

Resolve by symbol¶

Resolve by address¶

Register a custom token¶

get_instance classmethod ¶

Without gateway (static resolution only)¶

With gateway (enables on-chain discovery)¶

reset_instance classmethod ¶

resolve ¶

By symbol¶

By address¶

is_gateway_connected ¶

set_gateway_channel ¶

Connect to gateway later¶

Disconnect from gateway¶

resolve_pair ¶

get_decimals ¶

Returns 6¶

get_address ¶

Returns "0xaf88d065e77c8cC2239327C5EDb3A432268e5831"¶

resolve_for_swap ¶

ETH on Arbitrum returns WETH¶

USDC returns USDC (not native)¶

resolve_for_protocol ¶

DEX protocols get auto-wrapped native tokens¶

Lending protocols get the original token¶

register ¶

stats ¶

cache_stats ¶

ResolvedToken¶

almanak.framework.data.tokens.ResolvedToken dataclass ¶

__post_init__ ¶

to_dict ¶

from_dict classmethod ¶

BridgeType¶

almanak.framework.data.tokens.BridgeType ¶

Native USDC on Arbitrum (issued by Circle directly)¶

Bridged USDC.e on Arbitrum (bridged from Ethereum)¶

Exceptions¶

almanak.framework.data.tokens.TokenResolutionError ¶

__repr__ ¶

almanak.framework.data.tokens.TokenNotFoundError ¶

almanak.framework.data.tokens.AmbiguousTokenError ¶

get_instance `classmethod` ¶

reset_instance `classmethod` ¶

almanak.framework.data.tokens.ResolvedToken `dataclass` ¶

from_dict `classmethod` ¶

repr ¶