Python SDK Reference

The ZeroQuant Python SDK provides async-first vault operations, order management, TEE attestation, and exchange routing.

Installation

pip install zeroquant

Quick Start

import asyncio
from zeroquant import ZeroQuantClient, VaultConfig

async def main():
    client = ZeroQuantClient(
        provider="https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY",
        config=VaultConfig(
            owner="0xYourAddress...",
            permission_manager="0xPermissionManager...",
            factory_address="0xFactory...",
        ),
        private_key="0xYourPrivateKey...",
    )

    # Connect to existing vault
    await client.connect_vault("0xYourVault...")

    # Execute transaction
    result = await client.execute(
        target="0xTokenContract...",
        value=0,
        data=b"...",
    )
    print(f"TX: {result.tx_hash}")

asyncio.run(main())

ZeroQuantClient

The main client class for vault operations.

Constructor

ZeroQuantClient(
    provider: str | AsyncWeb3,
    config: VaultConfig,
    private_key: Optional[str] = None,
    session_key: Optional[str] = None,
    delegation_proof: Optional[str] = None,
    api_key: Optional[str] = None,
    api_url: Optional[str] = None,
)

Authentication Modes:

Mode	Parameters	Description
Wallet	private_key	Direct wallet signing
Delegated	session_key + delegation_proof	Delegated session
API	api_key + api_url	API-based auth
Read-only	None	Query only

Methods

create_vault(salt: int) -> str

Creates a new vault with deterministic address.

vault_address = await client.create_vault(salt=12345)

connect_vault(vault_address: str) -> None

Connects to an existing vault.

await client.connect_vault("0x...")

execute(target: str, value: int, data: bytes) -> TransactionResult

Executes a single operation through the vault.

result = await client.execute(
    target="0xUniswapRouter...",
    value=0,
    data=swap_calldata,
)

execute_batch(targets, values, calldatas) -> TransactionResult

Executes multiple operations atomically.

result = await client.execute_batch(
    targets=["0xToken1...", "0xToken2..."],
    values=[0, 0],
    calldatas=[approve_data, swap_data],
)

estimate_execute(target, value, data) -> GasEstimate

Estimates gas for an execute operation.

estimate = await client.estimate_execute(target, value, data)
print(f"Gas: {estimate.gas_limit}, Cost: {estimate.estimated_cost_eth} ETH")

get_balance() -> int

Gets vault ETH balance in wei.

get_owner() -> str

Gets vault owner address.

Session Management

create_session(params: CreateSessionParams) -> TransactionResult

Creates an agent session with permissions.

from zeroquant import CreateSessionParams

result = await client.create_session(CreateSessionParams(
    agent="0xAgentAddress...",
    expires_at=int(time.time()) + 86400,  # 24 hours
    daily_limit_usd=10000,
    per_tx_limit_usd=1000,
    max_position_size_pct=25,
    max_slippage_bps=100,  # 1%
    max_leverage=5,
    allowed_operations=["0x38ed1739"],  # swapExactTokensForTokens
    allowed_protocols=["0xUniswapRouter..."],
))

has_active_session(agent: str) -> bool

Checks if agent has an active session.

get_session(agent: str) -> Optional[AgentSession]

Gets session details.

revoke_session(agent: str) -> TransactionResult

Revokes an agent session.

Event Subscription

on_operation(callback, poll_interval) -> Callable[[], None]

Subscribes to vault operation events.

async def handle_operation(event):
    print(f"Operation: {event['tx_hash']}")

stop = await client.on_operation(handle_operation, poll_interval=2.0)

# Later: stop()

Models

VaultConfig

from zeroquant import VaultConfig

config = VaultConfig(
    owner="0x...",                    # Vault owner address
    permission_manager="0x...",       # PermissionManager contract
    factory_address="0x...",          # UVAFactory contract
    swap_intent_address="0x...",      # Optional SwapIntent contract
)

TransactionResult

class TransactionResult:
    tx_hash: str           # Transaction hash
    block_number: int      # Block number
    gas_used: int          # Gas used
    status: int            # 1=success, 0=failure
    logs: List[dict]       # Event logs

GasEstimate

class GasEstimate:
    gas_limit: int                    # Estimated gas limit
    gas_price: int                    # Gas price in wei
    max_fee_per_gas: Optional[int]    # EIP-1559
    max_priority_fee_per_gas: Optional[int]
    estimated_cost_wei: int
    estimated_cost_eth: str

AgentSession

class AgentSession:
    agent: str
    expires_at: int
    daily_limit_usd: int
    per_tx_limit_usd: int
    max_position_size_pct: int
    max_slippage_bps: int
    max_leverage: int
    allowed_operations: List[str]
    allowed_protocols: List[str]
    daily_spent_usd: int
    last_reset_timestamp: int

Order Management

OrderManager

Full-featured order manager with fill tracking.

from zeroquant import (
    OrderManager,
    CreateOrderParams,
    OrderSide,
    OrderType,
    TimeInForce,
)

manager = OrderManager()

# Create order
order = manager.create_order(
    vault_address="0xVault...",
    params=CreateOrderParams(
        base_token="0xWETH...",
        quote_token="0xUSDC...",
        side=OrderSide.BUY,
        type=OrderType.LIMIT,
        amount=1_000_000_000_000_000_000,  # 1 ETH
        price=2500_000_000,  # $2500 (6 decimals)
        time_in_force=TimeInForce.GTC,
    ),
)

# Update status
manager.update_order_status(order.id, OrderStatus.SUBMITTED)
manager.update_order_status(order.id, OrderStatus.OPEN)

# Record fill
fill = manager.record_fill(order.id, {
    "amount": 500_000_000_000_000_000,  # 0.5 ETH
    "price": 2498_000_000,
    "fees": {"total": 1_250_000, "protocol": 1_000_000, "network": 250_000},
    "tx_hash": "0x...",
})

# Query orders
open_orders = manager.get_open_orders(vault_address="0xVault...")
filled_orders = manager.get_filled_orders(limit=50)

# Statistics
stats = manager.get_order_stats()
print(f"Total: {stats.total_orders}, Filled: {stats.filled_orders}")

Order Types

class OrderSide(Enum):
    BUY = "buy"
    SELL = "sell"

class OrderType(Enum):
    MARKET = "market"
    LIMIT = "limit"
    STOP_LOSS = "stop_loss"
    TAKE_PROFIT = "take_profit"

class TimeInForce(Enum):
    GTC = "gtc"  # Good Till Cancelled
    IOC = "ioc"  # Immediate Or Cancel
    FOK = "fok"  # Fill Or Kill

class OrderStatus(Enum):
    PENDING = "pending"
    SUBMITTED = "submitted"
    OPEN = "open"
    PARTIAL_FILL = "partial_fill"
    FILLED = "filled"
    CANCELLED = "cancelled"
    FAILED = "failed"

Event Subscriptions

# Subscribe to specific order
unsubscribe = manager.on_order_update(order.id, lambda o: print(o.status))

# Subscribe to all orders
unsubscribe = manager.on_all_orders(lambda o: print(f"{o.id}: {o.status}"))

# Subscribe to fills
unsubscribe = manager.on_fill(lambda f: print(f"Fill: {f.amount} @ {f.price}"))

TEE Integration

AMDSEVClient

Client for AMD SEV-SNP attestation.

from web3 import Web3
from zeroquant import AMDSEVClient, TEEPlatform

w3 = Web3(Web3.HTTPProvider("https://eth-mainnet..."))

client = AMDSEVClient(
    w3=w3,
    verifier_address="0xAMDSEVVerifier...",
    permission_manager_address="0xTEEPermissionManager...",
    account="0xYourAccount...",
)

# Check attestation
has_valid = await client.has_valid_attestation(agent_address)

# Get attestation details
attestation = await client.get_attestation(agent_address)
print(f"Platform: {attestation.platform}")
print(f"Expires: {attestation.expires_at}")

# Create report data binding
report_data = client.create_report_data(agent_address)

# Submit attestation
result = await client.submit_attestation(report_hex, report_data)

# Upgrade session with TEE
await client.upgrade_session_with_tee(agent_address)

# Get effective limits
limits = await client.get_effective_limits(agent_address)
print(f"Daily Limit: ${limits.daily_limit}")
print(f"Using TEE Limits: {limits.using_tee_limits}")

IntelTDXClient

Client for Intel TDX attestation.

from zeroquant import IntelTDXClient

client = IntelTDXClient(
    w3=w3,
    verifier_address="0xIntelTDXVerifier...",
    permission_manager_address="0xTEEPermissionManager...",
    account="0xYourAccount...",
)

# Parse TDX quote
quote = client.parse_tdx_quote(quote_hex)
print(f"TD Report: {quote.td_report.mr_td.hex()}")

# Check/submit attestation
has_valid = await client.has_valid_attestation(agent_address)

Exchange Routing

ExchangeRouter

Route orders to centralized exchanges.

from zeroquant import (
    create_router,
    ExchangeCredentials,
    RouteOrderRequest,
    SupportedExchange,
)

# Create router with exchange credentials
router = create_router({
    SupportedExchange.BINANCE: ExchangeCredentials(
        api_key="...",
        api_secret="...",
    ),
    SupportedExchange.COINBASE: ExchangeCredentials(
        api_key="...",
        api_secret="...",
        passphrase="...",
    ),
})

# Route order
result = await router.route_order(RouteOrderRequest(
    exchange=SupportedExchange.BINANCE,
    symbol="ETHUSDT",
    side="buy",
    type="limit",
    amount=1.0,
    price=2500.0,
))

if result.success:
    print(f"Order ID: {result.exchange_order_id}")

# Get router stats
stats = router.get_stats()
print(f"Orders Routed: {stats.total_orders}")
print(f"Success Rate: {stats.success_rate * 100}%")

Supported Exchanges

Exchange	Spot	Futures	Margin
Binance	Yes	Yes	Yes
Coinbase	Yes	No	No
Kraken	Yes	Yes	Yes

Position Tracking

PositionTracker

Track positions across protocols.

from zeroquant import PositionTracker, Position, PositionSide

tracker = PositionTracker()

# Add position
position = Position(
    id="pos_1",
    symbol="ETH-USD",
    side=PositionSide.LONG,
    size=1.5,
    entry_price=2500.0,
    current_price=2600.0,
    unrealized_pnl=150.0,
    realized_pnl=0.0,
    margin_used=500.0,
    leverage=5.0,
)

tracker.add_position(position)

# Get snapshot
snapshot = tracker.get_snapshot()
print(f"Total Positions: {snapshot.total_positions}")
print(f"Total Unrealized PnL: ${snapshot.total_unrealized_pnl}")
print(f"Total Margin: ${snapshot.total_margin_used}")

Exceptions

from zeroquant import (
    ZeroQuantError,      # Base exception
    NotConnectedError,   # No vault connected
    ReadOnlyError,       # Write in read-only mode
    ValidationError,     # Input validation failed
    TransactionError,    # Transaction failed
    ContractError,       # Contract call failed
)

try:
    await client.execute(target, value, data)
except TransactionError as e:
    print(f"Failed: {e}")
    print(f"TX Hash: {e.tx_hash}")
except ContractError as e:
    print(f"Contract error: {e}")
    print(f"Revert reason: {e.revert_reason}")

Retry Utilities

from zeroquant import RetryConfig, retry_async

config = RetryConfig(
    max_retries=3,
    initial_delay=1.0,
    max_delay=30.0,
    exponential_base=2.0,
)

@retry_async(config)
async def fetch_with_retry():
    return await client.get_balance()

LangChain Integration

from zeroquant.langchain import ZeroQuantTools

tools = ZeroQuantTools(client)

# Available tools for LangChain agents:
# - get_vault_balance
# - execute_swap
# - check_session
# - get_positions

Type Reference

Enums

class TEEPlatform(Enum):
    AMD_SEV_SNP = 1
    INTEL_TDX = 2

class AttestationStatus(Enum):
    NONE = 0
    PENDING = 1
    VERIFIED = 2
    EXPIRED = 3
    REVOKED = 4

Full Model List

Category	Models
Core	VaultConfig, TransactionResult, GasEstimate
Sessions	AgentSession, CreateSessionParams
Orders	Order, OrderFill, OrderFees, CreateOrderParams, OrderStats
TEE	AMDSEVReport, TEEAttestation, TrustedMeasurement, EffectiveLimits
Positions	Position, PositionUpdate, PositionSnapshot
Exchange	ExchangeCredentials, RouteOrderRequest, RouteOrderResponse

Next Steps

• TypeScript SDK - TypeScript equivalent
• TEE Integration - Detailed TEE guide
• Order Management - Advanced order features
• Exchange Routing - CEX integration