Execution

The execution pipeline compiles intents, signs transactions, simulates, and submits them on-chain.

GatewayExecutionOrchestrator

The primary orchestrator used when running with the gateway sidecar.

almanak.framework.execution.GatewayExecutionOrchestrator

GatewayExecutionOrchestrator(
    client: GatewayClient,
    chain: str = "arbitrum",
    wallet_address: str | None = None,
    timeout: float = 120.0,
    max_gas_price_gwei: int = 0,
)

ExecutionOrchestrator that executes through the gateway.

This implementation routes all execution requests to the gateway sidecar, which has access to private keys and can sign/submit transactions.

The interface is intentionally simpler than the full ExecutionOrchestrator since the complex signing and submission logic lives in the gateway.

Example

from almanak.framework.gateway_client import GatewayClient from almanak.framework.execution.gateway_orchestrator import GatewayExecutionOrchestrator

with GatewayClient() as client: orchestrator = GatewayExecutionOrchestrator( client=client, chain="arbitrum", wallet_address="0x1234...", ) result = await orchestrator.execute(action_bundle) print(f"Execution success: {result.success}")

Initialize gateway-backed execution orchestrator.

Parameters:

Name	Type	Description	Default
client	GatewayClient	Connected GatewayClient instance	required
chain	str	Chain name for execution	'arbitrum'
wallet_address	str | None	Wallet address for signing	None
timeout	float	RPC timeout in seconds (default 120s for tx confirmation)	120.0
max_gas_price_gwei	int	Gas price cap in gwei (0 = use gateway default). Passed to the gateway so the ExecutionOrchestrator enforces the cap.	0

chain property

chain: str

Get the chain name.

wallet_address property

wallet_address: str | None

Get the wallet address.

compile_intent async

compile_intent(
    intent: Any, wallet_address: str | None = None
) -> dict[str, Any]

Compile an intent into an action bundle through gateway.

Parameters:

Name	Type	Description	Default
intent	Any	Intent object to compile	required
wallet_address	str | None	Override wallet address	None

Returns:

Type	Description
dict[str, Any]	Action bundle as dictionary

Raises:

Type	Description
CompilationError	If compilation fails

execute async

execute(
    action_bundle: Any,
    context: Any | None = None,
    strategy_id: str = "",
    intent_id: str = "",
    dry_run: bool = False,
    simulation_enabled: bool = True,
    wallet_address: str | None = None,
) -> GatewayExecutionResult

Execute an action bundle through gateway.

Parameters:

Name	Type	Description	Default
action_bundle	Any	Action bundle to execute (object or dict)	required
context	Any | None	Optional ExecutionContext for interface compatibility with ExecutionOrchestrator. If provided, extracts strategy_id, intent_id, dry_run, simulation_enabled, and wallet_address from it.	None
strategy_id	str	Strategy identifier for tracking	''
intent_id	str	Intent identifier for tracking	''
dry_run	bool	If True, simulate only without submitting	False
simulation_enabled	bool	If True, run simulation before execution	True
wallet_address	str | None	Override wallet address	None

Returns:

Type	Description
GatewayExecutionResult	GatewayExecutionResult with tx hashes and receipts

Raises:

Type	Description
ExecutionError	If execution fails

get_transaction_status async

get_transaction_status(
    tx_hash: str, chain: str | None = None
) -> dict[str, Any]

Get transaction status from gateway.

Parameters:

Name	Type	Description	Default
tx_hash	str	Transaction hash to check	required
chain	str | None	Chain to query (defaults to orchestrator chain)	None

Returns:

Type	Description
dict[str, Any]	Status dictionary with status, confirmations, block_number

ExecutionOrchestrator

almanak.framework.execution.ExecutionOrchestrator

ExecutionOrchestrator(
    signer: Signer,
    submitter: Submitter,
    simulator: Simulator,
    chain: str = "arbitrum",
    rpc_url: str | None = None,
    risk_guard: RiskGuard | None = None,
    event_callback: EventCallback | None = None,
    gas_buffer_multiplier: float | None = None,
    tx_timeout_seconds: float = 120.0,
    session_store: ExecutionSessionStore | None = None,
    tx_risk_config: TransactionRiskConfig | None = None,
)

Orchestrates the full transaction execution flow.

The ExecutionOrchestrator coordinates: - RiskGuard validation - Transaction simulation (optional) - Nonce assignment - Transaction signing - Transaction submission - Receipt polling and parsing

Events are emitted at each step for observability.

Example

orchestrator = ExecutionOrchestrator( signer=signer, submitter=submitter, simulator=simulator, chain="arbitrum", )

result = await orchestrator.execute(action_bundle) if result.success: print(f"All transactions confirmed: {result.transaction_results}") else: print(f"Execution failed at {result.error_phase}: {result.error}")

Initialize the orchestrator.

Parameters:

Name	Type	Description	Default
signer	Signer	Signer implementation for transaction signing	required
submitter	Submitter	Submitter implementation for transaction submission	required
simulator	Simulator	Simulator implementation for pre-execution simulation	required
chain	str	Target blockchain network	'arbitrum'
rpc_url	str | None	RPC URL for nonce queries (optional if submitter provides)	None
risk_guard	RiskGuard | None	RiskGuard for validation (uses default if not provided)	None
event_callback	EventCallback | None	Optional callback for execution events	None
gas_buffer_multiplier	float | None	Gas buffer multiplier (uses chain default if not provided)	None
tx_timeout_seconds	float	Timeout for transaction confirmation	120.0
session_store	ExecutionSessionStore | None	Optional ExecutionSessionStore for crash recovery checkpoints	None
tx_risk_config	TransactionRiskConfig | None	Transaction risk configuration (uses default if not provided)	None

execute async

execute(
    action_bundle: ActionBundle,
    context: ExecutionContext | None = None,
) -> ExecutionResult

Execute an ActionBundle through the full execution pipeline.

This method: 1. Validates transactions via RiskGuard 2. Simulates transactions (if enabled) 3. Assigns sequential nonces 4. Signs all transactions 5. Submits transactions 6. Polls for and parses receipts

Parameters:

Name	Type	Description	Default
action_bundle	ActionBundle	The ActionBundle to execute	required
context	ExecutionContext | None	Optional execution context	None

Returns:

Type	Description
ExecutionResult	ExecutionResult with complete execution details

get_current_nonce async

get_current_nonce(address: str | None = None) -> int

Get the current nonce for an address.

Parameters:

Name	Type	Description	Default
address	str | None	Address to query (defaults to signer address)	None

Returns:

Type	Description
int	Current nonce for the address

get_gas_price async

get_gas_price() -> dict[str, int]

Get current gas prices from the network.

Returns:

Type	Description
dict[str, int]	Dict with max_fee_per_gas and max_priority_fee_per_gas

set_event_callback

set_event_callback(callback: EventCallback | None) -> None

Set the event callback.

Parameters:

Name	Type	Description	Default
callback	EventCallback | None	Callback function for execution events	required

ExecutionResult

almanak.framework.execution.ExecutionResult dataclass

ExecutionResult(
    success: bool,
    phase: ExecutionPhase,
    transaction_results: list[TransactionResult] = list(),
    simulation_result: SimulationResult | None = None,
    total_gas_used: int = 0,
    total_gas_cost_wei: int = 0,
    error: str | None = None,
    error_phase: ExecutionPhase | None = None,
    started_at: datetime = (lambda: datetime.now(UTC))(),
    completed_at: datetime | None = None,
    correlation_id: str = "",
    gas_warnings: list[str] = list(),
    position_id: int | None = None,
    swap_amounts: SwapAmounts | None = None,
    lp_close_data: LPCloseData | None = None,
    bin_ids: list[int] | None = None,
    extracted_data: dict[str, Any] = dict(),
    extraction_warnings: list[str] = list(),
)

Complete result of an execution attempt.

Attributes:

Name	Type	Description
success	bool	Whether all transactions succeeded
phase	ExecutionPhase	Phase where execution completed or failed
transaction_results	list[TransactionResult]	Results for each transaction
simulation_result	SimulationResult | None	Simulation result (if simulation was run)
total_gas_used	int	Sum of gas used across all transactions
total_gas_cost_wei	int	Sum of gas costs across all transactions
error	str | None	Error message if failed
error_phase	ExecutionPhase | None	Phase where error occurred
started_at	datetime	When execution started
completed_at	datetime | None	When execution completed
correlation_id	str	Unique identifier for this execution
position_id	int | None	LP position ID for LP_OPEN intents (NFT tokenId)
swap_amounts	SwapAmounts | None	Swap execution data for SWAP intents
lp_close_data	LPCloseData | None	LP close data for LP_CLOSE intents
bin_ids	list[int] | None	TraderJoe V2 bin IDs for LP positions
extracted_data	dict[str, Any]	Flexible dict for protocol-specific extracted data
extraction_warnings	list[str]	Non-fatal warnings from extraction process

effective_price property

effective_price: Decimal | None

Convenience accessor for swap effective price.

slippage_bps property

slippage_bps: int | None

Convenience accessor for swap slippage in basis points.

__post_init__

__post_init__() -> None

Generate correlation_id if not provided.

get_extracted

get_extracted(
    key: str,
    expected_type: type | None = None,
    default: Any = None,
) -> Any

Get extracted data with optional type checking.

Provides safe access to protocol-specific extracted data with optional type validation.

Parameters:

Name	Type	Description	Default
key	str	Data key to retrieve from extracted_data	required
expected_type	type | None	Optional type to validate against	None
default	Any	Default value if key not found or wrong type	None

Returns:

Type	Description
Any	Extracted value or default

Example

tick_lower = result.get_extracted("tick_lower", int, 0) liquidity = result.get_extracted("liquidity")

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

ExecutionContext

almanak.framework.execution.ExecutionContext dataclass

ExecutionContext(
    strategy_id: str = "unknown",
    intent_id: str = "",
    chain: str = "arbitrum",
    wallet_address: str = "",
    correlation_id: str = "",
    session_id: str = "",
    simulation_enabled: bool = False,
    intent_description: str = "",
    dry_run: bool = False,
)

Context for the current execution.

Attributes:

Name	Type	Description
strategy_id	str	Strategy identifier
intent_id	str	Intent identifier (for session tracking)
chain	str	Blockchain network
wallet_address	str	Address executing transactions
correlation_id	str	Unique identifier for this execution
session_id	str	Execution session identifier (for crash recovery)
simulation_enabled	bool	Whether to simulate before execution
dry_run	bool	If True, don't actually submit transactions

__post_init__

__post_init__() -> None

Generate correlation_id if not provided.

Result Enrichment

After successful execution, ResultEnricher automatically extracts data from transaction receipts (position IDs, swap amounts, etc.) and attaches it to the result.

ResultEnricher

almanak.framework.execution.ResultEnricher

ResultEnricher(
    parser_registry: ReceiptParserRegistry | None = None,
)

Enriches ExecutionResult with intent-specific extracted data.

This component implements the "Framework Orchestrates, Protocols Execute" pattern. It determines WHAT to extract based on intent type, and delegates HOW to extract to protocol-specific parsers.

Key Design Principles: 1. Fail-Safe: Extraction errors are logged but never crash execution 2. Type-Safe: Core fields are strongly typed 3. Extensible: New protocols can be added without framework changes 4. Zero Cognitive Load: Data "just appears" on result

Example

enricher = ResultEnricher()

In StrategyRunner after execution:

result = await orchestrator.execute(bundle) if result.success: result = enricher.enrich(result, intent, context)

Strategy callback receives enriched result:

strategy.on_intent_executed(intent, success=True, result=result)

Strategy can use result.position_id directly!

Initialize the ResultEnricher.

Parameters:

Name	Type	Description	Default
parser_registry	ReceiptParserRegistry | None	Registry for protocol parsers. If not provided, uses the default global registry.	None

enrich

enrich(
    result: ExecutionResult,
    intent: Any,
    context: ExecutionContext,
) -> ExecutionResult

Enrich execution result with intent-specific extracted data.

This method extracts relevant data from transaction receipts based on the intent type and attaches it to the ExecutionResult.

IMPORTANT: This method NEVER raises exceptions. All errors are logged as warnings and added to result.extraction_warnings.

Parameters:

Name	Type	Description	Default
result	ExecutionResult	Raw execution result from orchestrator	required
intent	Any	The intent that was executed	required
context	ExecutionContext	Execution context with chain info	required

Returns:

Type	Description
ExecutionResult	Enriched ExecutionResult (same instance, mutated)

Example

result = enricher.enrich(result, intent, context)

result.position_id is now populated (if LP_OPEN)

result.swap_amounts is now populated (if SWAP)

SwapAmounts

almanak.framework.execution.SwapAmounts dataclass

SwapAmounts(
    amount_in: int,
    amount_out: int,
    amount_in_decimal: Decimal,
    amount_out_decimal: Decimal,
    effective_price: Decimal,
    slippage_bps: int | None = None,
    token_in: str | None = None,
    token_out: str | None = None,
)

Extracted swap execution data.

Represents the token amounts exchanged in a swap transaction. All fields are immutable (frozen=True) for safety.

Attributes:

Name	Type	Description
amount_in	int	Raw input amount (in token's smallest unit)
amount_out	int	Raw output amount (in token's smallest unit)
amount_in_decimal	Decimal	Human-readable input amount
amount_out_decimal	Decimal	Human-readable output amount
effective_price	Decimal	Actual execution price (out/in)
slippage_bps	int | None	Actual slippage in basis points (None if unknown)
token_in	str | None	Input token address or symbol
token_out	str | None	Output token address or symbol

Example

if result.swap_amounts: price = result.swap_amounts.effective_price slippage = result.swap_amounts.slippage_bps

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

LPCloseData

almanak.framework.execution.LPCloseData dataclass

LPCloseData(
    amount0_collected: int,
    amount1_collected: int,
    fees0: int = 0,
    fees1: int = 0,
    liquidity_removed: int | None = None,
)

Extracted LP close execution data.

Represents the amounts collected when closing an LP position, including principal and fees.

Attributes:

Name	Type	Description
amount0_collected	int	Total amount of token0 collected (principal + fees)
amount1_collected	int	Total amount of token1 collected (principal + fees)
fees0	int	Fees earned in token0 (if separately tracked)
fees1	int	Fees earned in token1 (if separately tracked)
liquidity_removed	int | None	Amount of liquidity removed (if available)

Example

if result.lp_close_data: total_0 = result.lp_close_data.amount0_collected fees_0 = result.lp_close_data.fees0

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary for serialization.

Signers

LocalKeySigner

almanak.framework.execution.LocalKeySigner

LocalKeySigner(private_key: str)

Bases: Signer

Signs transactions using a local private key.

This signer uses the eth-account library to sign transactions locally. The private key is stored in memory and derived to an Ethereum account at initialization time.

SECURITY CONTRACT

• Private keys are NEVER logged, printed, or included in error messages
• Private keys are NEVER exposed through any method or property
• Transaction fields are validated before signing
• The wallet address is derived from the key at initialization

Supported transaction types

• EIP-1559 (Type 2): Modern fee market transactions
• Legacy (Type 0): Pre-EIP-1559 transactions

Attributes:

Name	Type	Description
address	str	The checksummed Ethereum address derived from the private key

Example

Initialize with private key

signer = LocalKeySigner(private_key="0x...")

Create unsigned transaction

tx = UnsignedTransaction( to="0x1234...", value=0, data="0xa9059cbb...", chain_id=42161, gas_limit=100000, nonce=5, max_fee_per_gas=100_000_000, max_priority_fee_per_gas=1_000_000, )

Sign transaction

signed = await signer.sign(tx, chain="arbitrum") print(f"Signed tx hash: {signed.tx_hash}")

Initialize the signer with a private key.

Parameters:

Name	Type	Description	Default
private_key	str	Hex-encoded private key (with or without 0x prefix)	required

Raises:

Type	Description
SigningError	If the private key is invalid

Example

signer = LocalKeySigner(private_key="0xabc123...")

address property

address: str

Return the wallet address associated with this signer.

The address is derived from the private key at initialization and is cached for efficiency.

Returns:

Type	Description
str	Checksummed Ethereum address (0x-prefixed, 42 characters)

Example

signer = LocalKeySigner(private_key="0x...") print(signer.address) # 0x71C7656EC7ab88b098defB751B7401B5f6d8976F

sign async

sign(
    tx: UnsignedTransaction, chain: str
) -> SignedTransaction

Sign a transaction with the local private key.

This method validates the transaction fields, builds the appropriate transaction dictionary based on transaction type (EIP-1559 or legacy), and signs it using the eth-account library.

Parameters:

Name	Type	Description	Default
tx	UnsignedTransaction	Unsigned transaction to sign	required
chain	str	Chain name (e.g., "arbitrum", "ethereum") for logging/validation	required

Returns:

Type	Description
SignedTransaction	SignedTransaction containing the raw signed bytes and transaction hash

Raises:

Type	Description
SigningError	If signing fails due to invalid fields or key issues
ValueError	If transaction fields are malformed

Example

tx = UnsignedTransaction( to="0x1234...", value=1_000_000_000_000_000_000, # 1 ETH data="0x", chain_id=1, gas_limit=21000, nonce=0, max_fee_per_gas=30_000_000_000, max_priority_fee_per_gas=1_000_000_000, ) signed = await signer.sign(tx, chain="ethereum")

__repr__

__repr__() -> str

Return string representation without exposing private key.

__str__

__str__() -> str

Return string representation without exposing private key.

Simulators

DirectSimulator

almanak.framework.execution.DirectSimulator

DirectSimulator(name: str = 'direct')

Bases: Simulator

Pass-through simulator that skips actual simulation.

This simulator returns a successful SimulationResult without performing any actual simulation. It is designed for environments where simulation is not needed, such as local fork testing or trusted execution contexts.

The key differentiator from other simulators: - Returns simulated=False to indicate no simulation was performed - Always returns success=True (assumes transactions are valid) - Logs that simulation was skipped for observability

This is NOT a stub or shortcut - it is a legitimate implementation for production use cases where pre-execution simulation adds no value or where the latency cost is unacceptable.

Future Alternatives: - TenderlySimulator: Full simulation via Tenderly API - LocalSimulator: Simulation via local node eth_call - FlashbotsSimulator: Simulation via Flashbots bundle simulation

Attributes:

Name	Type	Description
name	str	Identifier for this simulator (for logging and metrics)

Example

simulator = DirectSimulator()

Simulate a single transaction

result = await simulator.simulate([tx], chain="arbitrum")

if result.success: # Proceed to signing and submission ... else: # Handle simulation failure (won't happen with DirectSimulator) ...

Initialize the DirectSimulator.

Parameters:

Name	Type	Description	Default
name	str	Identifier for this simulator instance (default: "direct")	'direct'

name property

name: str

Return the simulator name.

simulate async

simulate(
    txs: list[UnsignedTransaction],
    chain: str,
    state_overrides: dict[str, Any] | None = None,
) -> SimulationResult

Return a pass-through simulation result without actual simulation.

This method logs that simulation was skipped and returns a successful SimulationResult with simulated=False to indicate no actual simulation was performed.

Parameters:

Name	Type	Description	Default
txs	list[UnsignedTransaction]	List of unsigned transactions to "simulate"	required
chain	str	Chain name (logged but not used for simulation)	required
state_overrides	dict[str, Any] | None	Ignored - DirectSimulator doesn't perform simulation	None

Returns:

Type	Description
SimulationResult	SimulationResult with:
SimulationResult	success=True (assumes transactions are valid)
SimulationResult	simulated=False (indicates no simulation was performed)
SimulationResult	Empty gas_estimates, warnings, state_changes, logs

Note

This method never raises exceptions for valid input. Infrastructure failures are not possible since no external calls are made.

TenderlySimulator

almanak.framework.execution.TenderlySimulator

TenderlySimulator(
    account_slug: str,
    project_slug: str,
    access_key: str,
    timeout_seconds: float = 10.0,
    name: str = "tenderly",
)

Bases: Simulator

Transaction simulation via Tenderly REST API.

This simulator uses Tenderly's simulate-bundle endpoint to simulate transactions before submission. It provides accurate gas estimates, pre-flight validation, and supports SAFE wallet simulations via state overrides.

Key Advantages over Alchemy

• No transaction bundle limit (Alchemy limited to 3)
• State override support for SAFE wallet ETH balance
• Support for more chains (9+ vs 4 for Alchemy)
• Detailed simulation dashboard URLs

Attributes:

Name	Type	Description
account_slug		Tenderly account identifier
project_slug		Tenderly project identifier
timeout_seconds		Request timeout

Example

simulator = TenderlySimulator( account_slug="my-account", project_slug="my-project", access_key="xxx", )

Basic simulation

result = await simulator.simulate([tx], chain="arbitrum")

SAFE wallet simulation with ETH balance override

result = await simulator.simulate( [tx], chain="arbitrum", state_overrides={"0xSafeAddress": {"balance": hex(10 * 10**18)}}, )

Initialize the TenderlySimulator.

Parameters:

Name	Type	Description	Default
account_slug	str	Tenderly account slug	required
project_slug	str	Tenderly project slug	required
access_key	str	Tenderly API access key	required
timeout_seconds	float	Request timeout (default 10s)	10.0
name	str	Simulator name for logging (default "tenderly")	'tenderly'

Raises:

Type	Description
ValueError	If any required parameter is missing

name property

name: str

Return the simulator name.

supports_chain

supports_chain(chain: str) -> bool

Check if this simulator supports a given chain.

Parameters:

Name	Type	Description	Default
chain	str	Chain name (lowercase)	required

Returns:

Type	Description
bool	True if Tenderly supports this chain

simulate async

simulate(
    txs: list[UnsignedTransaction],
    chain: str,
    state_overrides: dict[str, Any] | None = None,
) -> SimulationResult

Simulate transactions via Tenderly API.

This method simulates the execution of one or more transactions using Tenderly's bundle simulation endpoint. It returns gas estimates and validates that transactions will succeed.

Parameters:

Name	Type	Description	Default
txs	list[UnsignedTransaction]	List of unsigned transactions to simulate	required
chain	str	Chain name (e.g., "arbitrum")	required
state_overrides	dict[str, Any] | None	Optional state overrides for SAFE wallets Format: {"0xAddress": {"balance": "0xHexWei"}}	None

Returns:

Type	Description
SimulationResult	SimulationResult with gas estimates and success status

Raises:

Type	Description
SimulationError	For infrastructure failures (not tx failures)

Receipt Parsing

ReceiptParserRegistry

almanak.framework.execution.ReceiptParserRegistry

ReceiptParserRegistry()

Registry for protocol receipt parsers.

The registry provides lazy loading of parser classes and supports both built-in parsers and custom parser registration.

Built-in parsers are loaded from the connectors package when first requested. Custom parsers can be registered at any time.

Example

registry = ReceiptParserRegistry()

Get a built-in parser

spark_parser = registry.get("spark")

Register a custom parser

registry.register("my_protocol", MyProtocolReceiptParser)

Check available protocols

protocols = registry.list_protocols()

Initialize the registry.

get

get(protocol: str, **kwargs: Any) -> ReceiptParser

Get a receipt parser for a protocol.

Parameters:

Name	Type	Description	Default
protocol	str	Protocol name (e.g., "spark", "lido", "ethena", "pancakeswap_v3")	required
**kwargs	Any	Additional arguments to pass to parser constructor	{}

Returns:

Type	Description
ReceiptParser	Receipt parser instance

Raises:

Type	Description
ValueError	If protocol is not registered

register

register(
    protocol: str, parser_class: type[ReceiptParser]
) -> None

Register a custom receipt parser.

Parameters:

Name	Type	Description	Default
protocol	str	Protocol name	required
parser_class	type[ReceiptParser]	Parser class (not instance)	required

Raises:

Type	Description
TypeError	If parser_class is not a class

unregister

unregister(protocol: str) -> bool

Unregister a custom receipt parser.

Parameters:

Name	Type	Description	Default
protocol	str	Protocol name	required

Returns:

Type	Description
bool	True if parser was unregistered, False if not found

list_protocols

list_protocols() -> list[str]

List all available protocol names.

Returns:

Type	Description
list[str]	List of registered protocol names

is_registered

is_registered(protocol: str) -> bool

Check if a protocol is registered.

Parameters:

Name	Type	Description	Default
protocol	str	Protocol name	required

Returns:

Type	Description
bool	True if protocol is registered

clear_cache

clear_cache() -> None

Clear the parser instance cache.

Useful for testing or when parser configuration changes.

Exceptions

almanak.framework.execution.ExecutionError

Bases: Exception

Base exception for execution layer errors.

All execution-related exceptions inherit from this class to allow broad exception handling when needed.

almanak.framework.execution.SimulationError

SimulationError(reason: str, recoverable: bool = True)

Bases: ExecutionError

Raised when transaction simulation fails.

This is distinct from a transaction that simulates successfully but would revert - that returns SimulationResult with success=False.

This exception is for infrastructure failures: - Simulation service unavailable - Network timeout - Invalid simulation parameters

Attributes:

Name	Type	Description
reason		Human-readable explanation of the failure
recoverable		Whether the error is transient and can be retried

almanak.framework.execution.SigningError

SigningError(reason: str, tx_hash: str | None = None)

Bases: ExecutionError

Raised when transaction signing fails.

This exception should be raised when: - Private key is invalid or missing - Transaction fields are malformed - Signing algorithm encounters an error

Note: Never include sensitive key material in error messages.

Attributes:

Name	Type	Description
reason		Human-readable explanation of the failure
tx_hash		Optional hash of the transaction that failed (if available)