Base Infrastructure¶

Shared base classes and utilities used by all connectors.

BaseReceiptParser¶

almanak.framework.connectors.base.BaseReceiptParser ¶

BaseReceiptParser(
    registry: EventRegistry | None = None,
    known_topics: set[str] | None = None,
)

Bases: ABC

Abstract base class for receipt parsers using template method pattern.

This class implements the common receipt parsing flow and provides hook methods for protocol-specific customization. Subclasses must implement: - _decode_log_data(): Protocol-specific log decoding - _create_event(): Create protocol-specific event object - _build_result(): Build protocol-specific result object

The template method parse_receipt() handles: - Transaction status validation - Log iteration and filtering - Event creation and collection - Error handling

Attributes:

Name	Type	Description
`registry`		EventRegistry for topic lookups (optional)
`known_topics`		Set of known topic signatures (optional)
`SUPPORTED_EXTRACTIONS`	`frozenset[str]`	Class-level frozenset declaring which extraction fields this parser supports (e.g., {"swap_amounts", "position_id"}). Used by ResultEnricher to warn when expected fields are unsupported.

Example

from almanak.framework.connectors.base import BaseReceiptParser

class MyProtocolParser(BaseReceiptParser[MyEvent, MyResult]): ... def init(self): ... super().init(registry=my_registry) ... ... def _decode_log_data(self, event_name, topics, data, contract_address): ... # Decode protocol-specific log data ... if event_name == "Swap": ... return { ... "amount_in": HexDecoder.decode_uint256(data, 0), ... "amount_out": HexDecoder.decode_uint256(data, 32), ... } ... return {} ... ... def _create_event(self, event_name, log_index, tx_hash, ...): ... # Create protocol-specific event object ... return MyEvent(...) ... ... def _build_result(self, events, receipt, **kwargs): ... # Build protocol-specific result ... return MyResult(success=True, events=events)

Initialize the base parser.

Parameters:

Name	Type	Description	Default
`registry`	`EventRegistry \| None`	EventRegistry for topic lookups (optional)	`None`
`known_topics`	`set[str] \| None`	Set of known topic signatures (optional, used if no registry)	`None`

parse_receipt ¶

parse_receipt(receipt: dict[str, Any], **kwargs) -> TResult

Parse a transaction receipt (template method).

This is the main entry point that implements the template method pattern. It handles common parsing logic and calls hook methods for protocol-specific customization.

Flow: 1. Validate transaction status 2. Extract transaction metadata 3. Iterate through logs 4. For each log: a. Check if event is known b. Decode log data (via _decode_log_data hook) c. Create event object (via _create_event hook) 5. Build final result (via _build_result hook)

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict from web3.py containing: - transactionHash: Transaction hash (bytes or hex string) - blockNumber: Block number (int) - status: Transaction status (1=success, 0=reverted) - logs: List of log dicts	required
`**kwargs`		Additional protocol-specific parameters	`{}`

Returns:

Type	Description
`TResult`	Protocol-specific result object

Raises:

Type	Description
`Exception`	If parsing fails critically (caught and returned in result)

EventRegistry¶

almanak.framework.connectors.base.EventRegistry ¶

EventRegistry(
    event_topics: dict[str, str],
    event_type_map: dict[str, Any],
)

Registry for managing event topic mappings.

This class manages the mapping between event topic signatures (keccak256 hashes) and event names/types. Provides fast lookup and validation methods.

Attributes:

Name	Type	Description
`event_topics`		Mapping from event name to topic signature
`topic_to_event`	`dict[str, str]`	Reverse mapping from topic signature to event name
`event_type_map`		Mapping from event name to enum type
`known_topics`		Set of all known topic signatures (for fast lookup)

Example

from enum import Enum from almanak.framework.connectors.base import EventRegistry

class MyEventType(Enum): ... SWAP = "SWAP" ... MINT = "MINT"

EVENT_TOPICS = { ... "Swap": "0xc42079...", ... "Mint": "0x7a5308...", ... }

EVENT_NAME_TO_TYPE = { ... "Swap": MyEventType.SWAP, ... "Mint": MyEventType.MINT, ... }

registry = EventRegistry(EVENT_TOPICS, EVENT_NAME_TO_TYPE) registry.get_event_name("0xc42079...") 'Swap' registry.get_event_type("Swap")

Initialize the event registry.

Parameters:

Name	Type	Description	Default
`event_topics`	`dict[str, str]`	Mapping from event name to topic signature Example: {"Swap": "0xc42079...", "Mint": "0x7a5308..."}	required
`event_type_map`	`dict[str, Any]`	Mapping from event name to enum type Example: {"Swap": MyEventType.SWAP, "Mint": MyEventType.MINT}	required

get_event_name ¶

get_event_name(topic: str) -> str | None

Get event name from topic signature.

Parameters:

Name	Type	Description	Default
`topic`	`str`	Event topic signature (keccak256 hash)	required

Returns:

Type	Description
`str \| None`	Event name or None if topic is unknown

Example

registry.get_event_name("0xc42079...") 'Swap'

get_event_type ¶

get_event_type(event_name: str) -> Any | None

Get event type enum from event name.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Event name (e.g., "Swap")	required

Returns:

Type	Description
`Any \| None`	Event type enum or None if name is unknown

Example

registry.get_event_type("Swap")

get_event_type_from_topic ¶

get_event_type_from_topic(topic: str) -> Any | None

Get event type enum directly from topic signature.

Convenience method that combines get_event_name() and get_event_type().

Parameters:

Name	Type	Description	Default
`topic`	`str`	Event topic signature (keccak256 hash)	required

Returns:

Type	Description
`Any \| None`	Event type enum or None if topic is unknown

Example

registry.get_event_type_from_topic("0xc42079...")

is_known_event ¶

is_known_event(topic: str) -> bool

Check if a topic is a known event.

Parameters:

Name	Type	Description	Default
`topic`	`str`	Event topic signature to check	required

Returns:

Type	Description
`bool`	True if topic is in the registry

Example

registry.is_known_event("0xc42079...") True registry.is_known_event("0xunknown...") False

get_topic_signature ¶

get_topic_signature(event_name: str) -> str | None

Get topic signature from event name.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Event name (e.g., "Swap")	required

Returns:

Type	Description
`str \| None`	Topic signature or None if name is unknown

Example

registry.get_topic_signature("Swap") '0xc42079...'

len ¶

__len__() -> int

Get number of registered events.

contains ¶

__contains__(topic: str) -> bool

Check if topic is in registry (allows 'in' operator).

repr ¶

__repr__() -> str

Get string representation of registry.

HexDecoder¶

almanak.framework.connectors.base.HexDecoder ¶

Static utilities for decoding hex-encoded event data.

This class provides methods for decoding common EVM types from hex strings, including: - Addresses (20 bytes) - Unsigned integers: uint256, uint160, uint128 - Signed integers: int256, int128, int24 - Dynamic arrays (for batch events) - Raw bytes32 values

All methods are static and handle both bytes and string inputs.

normalize_hex `staticmethod` ¶

normalize_hex(value: Any) -> str

Normalize a hex value to a string without '0x' prefix.

Parameters:

Name	Type	Description	Default
`value`	`Any`	Bytes or string value to normalize	required

Returns:

Type	Description
`str`	Hex string without '0x' prefix

topic_to_address `staticmethod` ¶

topic_to_address(topic: Any) -> str

Convert a log topic to an Ethereum address.

Topics are 32 bytes, but addresses are only 20 bytes. This extracts the last 20 bytes as the address.

Parameters:

Name	Type	Description	Default
`topic`	`Any`	Topic value (bytes or hex string)	required

Returns:

Type	Description
`str`	Lowercase address with '0x' prefix, or empty string if topic is empty

Example

HexDecoder.topic_to_address( ... "0x000000000000000000000000a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" ... ) '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'

topic_to_bytes32 `staticmethod` ¶

topic_to_bytes32(topic: Any) -> str

Convert a log topic to a bytes32 hex string.

Parameters:

Name	Type	Description	Default
`topic`	`Any`	Topic value (bytes or hex string)	required

Returns:

Type	Description
`str`	Full 32-byte hex string with '0x' prefix

Example

HexDecoder.topic_to_bytes32(b'\x00' * 31 + b'\x01') '0x0000000000000000000000000000000000000000000000000000000000000001'

decode_uint256 `staticmethod` ¶

decode_uint256(hex_str: str, offset: int = 0) -> int

Decode an unsigned 256-bit integer from hex string.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`int`	Decoded unsigned integer value

Example

HexDecoder.decode_uint256("0x00000000000000000000000000000000000000000000000000000000000003e8") 1000

decode_uint160 `staticmethod` ¶

decode_uint160(hex_str: str, offset: int = 0) -> int

Decode an unsigned 160-bit integer from hex string.

Used for Uniswap V3 sqrtPriceX96 which is uint160.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`int`	Decoded unsigned integer value

decode_uint128 `staticmethod` ¶

decode_uint128(hex_str: str, offset: int = 0) -> int

Decode an unsigned 128-bit integer from hex string.

Used for Uniswap V3 liquidity which is uint128.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`int`	Decoded unsigned integer value

decode_int256 `staticmethod` ¶

decode_int256(hex_str: str, offset: int = 0) -> int

Decode a signed 256-bit integer from hex string.

Handles two's complement representation for negative numbers.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`int`	Decoded signed integer value (can be negative)

Example

Positive value¶

HexDecoder.decode_int256("0x00000000000000000000000000000000000000000000000000000000000003e8") 1000

Negative value (two's complement)¶

HexDecoder.decode_int256("0xfffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffc18") -1000

decode_int128 `staticmethod` ¶

decode_int128(hex_str: str, offset: int = 0) -> int

Decode a signed 128-bit integer from hex string.

Used by Curve for token amounts. Handles two's complement.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`int`	Decoded signed integer value (can be negative)

decode_int24 `staticmethod` ¶

decode_int24(hex_str: str, offset: int = 0) -> int

Decode a signed 24-bit integer from hex string.

Used for Uniswap V3 ticks. Value is stored in 256-bit slot but only uses 24 bits. Handles two's complement.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`int`	Decoded signed integer value (can be negative)

Example

Positive tick¶

HexDecoder.decode_int24("0x0000000000000000000000000000000000000000000000000000000000000064") 100

Negative tick¶

HexDecoder.decode_int24("0xffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff9c") -100

decode_dynamic_array `staticmethod` ¶

decode_dynamic_array(
    hex_str: str, offset: int = 0
) -> list[int]

Decode a dynamic array from hex string.

Dynamic arrays in EVM events are encoded as: - Offset to array data (32 bytes) - Array length (32 bytes) - Array elements (32 bytes each)

Used by TraderJoe V2 for bin IDs and Polymarket for batch transfers.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading array offset	`0`

Returns:

Type	Description
`list[int]`	List of decoded uint256 values

Example

Array [1, 2, 3] encoded in event data¶

data = "0x" + "0" * 64 + "0" * 62 + "03" + "0" * 62 + "01" + "0" * 62 + "02" + "0" * 62 + "03" HexDecoder.decode_dynamic_array(data, 0) [1, 2, 3]

decode_address_from_data `staticmethod` ¶

decode_address_from_data(
    hex_str: str, offset: int = 0
) -> str

Decode an address from event data (not indexed topic).

Unlike indexed addresses which are in topics, non-indexed addresses appear in the data section as 32-byte values with leading zeros.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to decode (with or without '0x')	required
`offset`	`int`	Byte offset to start reading from	`0`

Returns:

Type	Description
`str`	Lowercase address with '0x' prefix

Example

HexDecoder.decode_address_from_data( ... "0x000000000000000000000000a0b86991c6218b36c1d19d4a2e9eb0ce3606eb48" ... ) '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'

split_into_chunks `staticmethod` ¶

split_into_chunks(
    hex_str: str, chunk_size: int = 64
) -> list[str]

Split hex string into chunks of specified size.

Useful for parsing event data with multiple parameters.

Parameters:

Name	Type	Description	Default
`hex_str`	`str`	Hex string to split (with or without '0x')	required
`chunk_size`	`int`	Size of each chunk in hex characters (default 64 = 32 bytes)	`64`

Returns:

Type	Description
`list[str]`	List of hex chunks

Example

data = "0x" + "0" * 64 + "1" * 64 chunks = HexDecoder.split_into_chunks(data) len(chunks) 2

Base Infrastructure¶

BaseReceiptParser¶

almanak.framework.connectors.base.BaseReceiptParser ¶

parse_receipt ¶

EventRegistry¶

almanak.framework.connectors.base.EventRegistry ¶

get_event_name ¶

get_event_type ¶

get_event_type_from_topic ¶

is_known_event ¶

get_topic_signature ¶

__len__ ¶

__contains__ ¶

__repr__ ¶

HexDecoder¶

almanak.framework.connectors.base.HexDecoder ¶

normalize_hex staticmethod ¶

topic_to_address staticmethod ¶

topic_to_bytes32 staticmethod ¶

decode_uint256 staticmethod ¶

decode_uint160 staticmethod ¶

decode_uint128 staticmethod ¶

decode_int256 staticmethod ¶

Positive value¶

Negative value (two's complement)¶

decode_int128 staticmethod ¶

decode_int24 staticmethod ¶

Positive tick¶

Negative tick¶

decode_dynamic_array staticmethod ¶

Array [1, 2, 3] encoded in event data¶

decode_address_from_data staticmethod ¶

split_into_chunks staticmethod ¶

len ¶

contains ¶

repr ¶

normalize_hex `staticmethod` ¶

topic_to_address `staticmethod` ¶

topic_to_bytes32 `staticmethod` ¶

decode_uint256 `staticmethod` ¶

decode_uint160 `staticmethod` ¶

decode_uint128 `staticmethod` ¶

decode_int256 `staticmethod` ¶

decode_int128 `staticmethod` ¶

decode_int24 `staticmethod` ¶

decode_dynamic_array `staticmethod` ¶

decode_address_from_data `staticmethod` ¶

split_into_chunks `staticmethod` ¶