luziadev

pip install luziadev
# or
uv add luziadev

pip install luziadev[websocket]
# or
uv add luziadev[websocket]

import asyncio
from luziadev import Luzia

async def main():
    async with Luzia("lz_your_api_key") as client:
        # List all supported exchanges
        exchanges = await client.exchanges.list()
        print(exchanges)
        # [Exchange(id='binance', name='Binance', status='operational', ...)]

        # Get a single ticker
        ticker = await client.tickers.get("binance", "BTC/USDT")
        print(f"BTC/USDT: ${ticker.last}")

        # Get multiple tickers
        result = await client.tickers.list_filtered(
            exchange="binance",
            symbols=["BTC/USDT", "ETH/USDT"],
        )
        for t in result.tickers:
            print(f"{t.symbol}: ${t.last}")

        # Get markets with filters
        markets = await client.markets.list("binance", quote="USDT")

        # Get historical OHLCV candles
        ohlcv = await client.history.get(
            "binance", "BTC/USDT",
            interval="1h",
            limit=24,
        )
        print(f"Last close: ${ohlcv.candles[-1].close}")

asyncio.run(main())

# Option 1: Async context manager (recommended)
async with Luzia("lz_your_api_key") as client:
    ticker = await client.tickers.get("binance", "BTC/USDT")

# Option 2: Manual lifecycle
client = Luzia("lz_your_api_key")
try:
    ticker = await client.tickers.get("binance", "BTC/USDT")
finally:
    await client.close()

from luziadev import Luzia, RetryOptions

client = Luzia(
    # Required: Your API key
    api_key="lz_xxxxx",

    # Optional: Custom base URL (default: https://api.luzia.dev)
    base_url="http://localhost:3000",

    # Optional: Request timeout in seconds (default: 30)
    timeout=10,

    # Optional: Retry configuration
    retry=RetryOptions(
        max_retries=3,            # Number of retry attempts (default: 3)
        initial_delay_ms=1000,    # Initial delay before first retry (default: 1000)
        max_delay_ms=30000,       # Maximum delay between retries (default: 30000)
        backoff_multiplier=2.0,   # Multiplier for exponential backoff (default: 2)
        jitter=True,              # Add random jitter to delays (default: True)
    ),
)

import asyncio
from luziadev import Luzia

async def main():
    client = Luzia("lz_your_api_key")
    ws = client.create_websocket()

    ws.on("connected", lambda data: print(f"Connected! Max subs: {data['limits']['maxSubscriptions']}"))
    ws.on("ticker", lambda data: print(f"{data['data']['symbol']}: ${data['data']['last']}"))
    ws.on("error", lambda data: print(f"Error: {data['message']}"))
    ws.on("reconnecting", lambda data: print(f"Reconnecting (attempt {data['attempt']})..."))

    await ws.connect()

    ws.subscribe(["ticker:binance:BTC-USDT", "ticker:binance:ETH-USDT"])

    # Keep running
    try:
        await asyncio.sleep(3600)
    finally:
        ws.disconnect()
        await client.close()

asyncio.run(main())

ws = client.create_websocket(
    auto_reconnect=True,              # Auto-reconnect on disconnect (default: True)
    max_reconnect_attempts=10,        # Max reconnect attempts (default: 10)
    reconnect_delay_ms=1000,          # Initial reconnect delay (default: 1000)
    max_reconnect_delay_ms=30000,     # Max reconnect delay (default: 30000)
    heartbeat_interval_ms=30000,      # Heartbeat ping interval (default: 30000)
)

from luziadev import Luzia, LuziaError

async with Luzia("lz_your_api_key") as client:
    try:
        ticker = await client.tickers.get("invalid", "BTC/USDT")
    except LuziaError as e:
        match e.code:
            case "auth":
                print("Invalid API key")
            case "rate_limit":
                print(f"Rate limited. Retry after {e.retry_after}s")
            case "not_found":
                print("Resource not found")
            case "validation":
                print(f"Invalid request parameters: {e.details}")
            case "server":
                print("Server error (exchange may be temporarily unavailable)")
            case "network":
                print(f"Network error: {e}")
            case "timeout":
                print(f"Request timed out after {e.timeout_ms}ms")

from luziadev import is_luzia_error, is_retryable_error

if is_luzia_error(error):
    print(f"Luzia error: {error}")

if is_retryable_error(error):
    print("This error can be retried")

ticker = await client.tickers.get("binance", "BTC/USDT")

info = client.rate_limit_info
if info:
    print(f"Requests remaining: {info.remaining}/{info.limit}")
    print(f"Resets at: {info.reset}")

    # Free tier also has daily limits
    if info.daily_limit:
        print(f"Daily remaining: {info.daily_remaining}/{info.daily_limit}")

from luziadev import (
    Exchange,
    Ticker,
    Market,
    MarketListResponse,
    OHLCVCandle,
    OHLCVResponse,
    TickerListResponse,
    RateLimitInfo,
    RetryOptions,
    RetryContext,
    LuziaError,
    ErrorCode,
    LuziaWebSocket,
)