pip install lux-dex

pip install "lux-dex[all]"

pip install "lux-dex[dev]"

from lux_dex import Client

# Initialize client
client = Client(
    json_rpc_url="http://localhost:8080/rpc",
    ws_url="ws://localhost:8081"
)

# Check connectivity
info = client.get_info()
print(f"Connected to node: {info.node_id}")
print(f"Block height: {info.block_height}")

# Get order book
book = client.get_order_book("BTC-USD", depth=10)
print(f"Best bid: {book.best_bid.price} ({book.best_bid.size})")
print(f"Best ask: {book.best_ask.price} ({book.best_ask.size})")
print(f"Spread: {book.spread}")

from lux_dex import Client, OrderType, Side

client = Client("http://localhost:8080/rpc")

# Place a limit buy order
order = client.place_order(
    symbol="BTC-USD",
    order_type=OrderType.LIMIT,
    side=Side.BUY,
    price=50000.0,
    size=1.0,
    user_id="trader1"
)

print(f"Order ID: {order.order_id}")
print(f"Status: {order.status}")
print(f"Filled: {order.filled_size}/{order.size}")

import asyncio
from lux_dex import AsyncClient, OrderType, Side

async def main() -> None:
    async with AsyncClient("http://localhost:8080/rpc") as client:
        # Place order asynchronously
        order = await client.place_order(
            symbol="BTC-USD",
            order_type=OrderType.LIMIT,
            side=Side.BUY,
            price=50000.0,
            size=1.0
        )
        print(f"Order placed: {order.order_id}")

        # Stream order book updates
        async for book in client.stream_order_book("BTC-USD"):
            print(f"Bid: {book.best_bid}, Ask: {book.best_ask}")
            if book.spread > 100:
                break

asyncio.run(main())

from lux_dex import Client
from lux_dex.config import ClientConfig, RetryConfig

config = ClientConfig(
    json_rpc_url="https://api.lux.network/rpc",
    ws_url="wss://api.lux.network/ws",
    api_key="your-api-key",
    api_secret="your-api-secret",
    timeout=30.0,
    retry=RetryConfig(
        max_retries=3,
        backoff_base=1.0,
        backoff_max=30.0
    )
)

client = Client(config=config)

export LX_DEX_RPC_URL="https://api.lux.network/rpc"
export LX_DEX_WS_URL="wss://api.lux.network/ws"
export LX_DEX_API_KEY="your-api-key"
export LX_DEX_API_SECRET="your-api-secret"

from lux_dex import Client

# Automatically reads from environment
client = Client.from_env()

from lux_dex.types import (
    Order,
    OrderType,
    Side,
    TimeInForce,
    OrderStatus
)

# OrderType enum
OrderType.LIMIT       # Limit order at specific price
OrderType.MARKET      # Execute at best available price
OrderType.STOP        # Trigger at stop price
OrderType.STOP_LIMIT  # Stop that becomes limit
OrderType.ICEBERG     # Hidden quantity order
OrderType.PEG         # Pegged to best bid/ask

# Side enum
Side.BUY   # Buy order
Side.SELL  # Sell order

# TimeInForce enum
TimeInForce.GTC  # Good Till Cancelled
TimeInForce.IOC  # Immediate Or Cancel
TimeInForce.FOK  # Fill Or Kill
TimeInForce.DAY  # Day order

from lux_dex.types import (
    OrderBook,
    OrderBookLevel,
    Trade,
    Ticker,
    OHLCV
)

# Type aliases with full annotations
type Price = float
type Size = float
type Timestamp = int  # Unix milliseconds

from lux_dex import Client
from lux_dex.exceptions import (
    LXDexError,
    ConnectionError,
    AuthenticationError,
    InsufficientBalanceError,
    OrderNotFoundError,
    RateLimitError,
    ValidationError
)

client = Client("http://localhost:8080/rpc")

try:
    order = client.place_order(
        symbol="BTC-USD",
        order_type="limit",
        side="buy",
        price=50000.0,
        size=100.0
    )
except InsufficientBalanceError as e:
    print(f"Insufficient funds: need {e.required}, have {e.available}")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ValidationError as e:
    print(f"Invalid request: {e.field} - {e.message}")
except ConnectionError as e:
    print(f"Connection failed: {e}")
except LXDexError as e:
    print(f"DEX error [{e.code}]: {e.message}")

# Enable async in Jupyter
import nest_asyncio
nest_asyncio.apply()

from lux_dex import AsyncClient
import pandas as pd

async def analyze_order_book():
    async with AsyncClient("http://localhost:8080/rpc") as client:
        book = await client.get_order_book("BTC-USD", depth=20)

        # Convert to DataFrame
        df = book.to_dataframe()
        display(df.head(10))

        # Calculate metrics
        print(f"Total bid volume: {df[df['side'] == 'bid']['size'].sum():.2f}")
        print(f"Total ask volume: {df[df['side'] == 'ask']['size'].sum():.2f}")

await analyze_order_book()

client = Client(
    json_rpc_url="https://api.lux.network/rpc",
    ws_url="wss://api.lux.network/ws"
)

client = Client(
    json_rpc_url="https://testnet-api.lux.network/rpc",
    ws_url="wss://testnet-api.lux.network/ws"
)

client = Client(
    json_rpc_url="http://localhost:8080/rpc",
    ws_url="ws://localhost:8081"
)