weex-client 0.2.2

Modern async-first Weex API client for Python 3.14+

Weex Client

Modern async-first Weex API client for Python 3.14+ with comprehensive WebSocket support.

Features

• Async-first design built on httpx and websockets
• Modern Python 3.14+ with strict typing and pattern matching
• Comprehensive REST API coverage for trading, account, and market data
• Real-time WebSocket streaming with automatic reconnection
• Structured logging with contextual information
• Pydantic models for request/response validation
• Sync wrapper for compatibility with existing code

Quick Start

Installation

pip install weex-client

Basic Usage

import asyncio
from weex_client import WeexAsyncClient, WeexConfig

async def main():
    # Initialize client with configuration
    config = WeexConfig.from_env()
    
    async with WeexAsyncClient(config) as client:
        # Get account balance
        balance = await client.get_account_balance()
        print(f"Balance: {balance}")
        
        # Get all positions
        positions = await client.get_all_positions()
        print(f"Positions: {positions}")

if __name__ == "__main__":
    asyncio.run(main())

WebSocket Streaming

import asyncio
from weex_client import WeexConfig
from weex_client.websocket import WeexWebSocketClient

async def stream_tickers():
    config = WeexConfig.from_env()
    
    async with WeexWebSocketClient(config) as ws:
        # Subscribe to ticker updates
        await ws.subscribe_tickers(["cmt_btcusdt", "cmt_ethusdt"])
        
        # Stream real-time updates
        async for message in ws.stream_messages():
            print(f"Ticker update: {message}")

if __name__ == "__main__":
    asyncio.run(stream_tickers())

Sync Wrapper (for legacy code)

from weex_client import WeexConfig, WeexSyncClient

# Synchronous interface
config = WeexConfig.from_env()
client = WeexSyncClient(config)

# Use synchronous methods
balance = client.get_account_balance()
positions = client.get_all_positions()

Configuration

Set environment variables:

export WEEX_API_KEY="your_api_key"
export WEEX_SECRET_KEY="your_secret_key" 
export WEEX_PASSPHRASE="your_passphrase"

Or create configuration programmatically:

from weex_client import WeexConfig

config = WeexConfig(
    api_key="your_api_key",
    secret_key="your_secret_key", 
    passphrase="your_passphrase",
    environment="development",  # development, staging, or production
    timeout=30.0,
    max_retries=3
)

Python 3.14+ Features

This library leverages modern Python features:

• Pattern matching for robust error handling
• TaskGroup for concurrent operations
• Self type for improved type hints
• Async generators for data streaming
• Strict typing with runtime validation

Configuration File

Create a .env file from the example:

cp .env.example .env

Edit .env with your credentials:

# Your Weex API Key (starts with 'weex_')
WEEX_API_KEY=your_actual_api_key_here

# Your Weex Secret Key 
WEEX_SECRET_KEY=your_actual_secret_key_here

# Your Weex API Passphrase
WEEX_PASSPHRASE=your_actual_passphrase_here

# Environment: development, staging, or production
WEEX_ENVIRONMENT=development

# Optional: Custom timeout settings (seconds)
API_TIMEOUT=30

# Optional: Enable debug logging
DEBUG=false

Available Methods

Contract API (Futures Trading)

# Get position data
position_data = await client.get_contract_data(
    "/capi/v2/account/position/singlePosition", 
    "?symbol=cmt_btcusdt"
)

# Place order
order_data = await client.post_contract_data("/capi/v2/order/placeOrder", {
    "symbol": "cmt_btcusdt",
    "client_oid": "unique_order_id",
    "size": "0.01",
    "type": "1",
    "order_type": "0",
    "match_price": "1",
    "price": "80000"
})

Error Handling

The client provides comprehensive error handling with Python 3.14+ pattern matching:

from weex_client.exceptions import (
    WEEXError,
    WEEXRateLimitError,
    WEEXAuthenticationError,
    WEEXSystemError
)

try:
    result = await client.get_contract_data("/endpoint", "?params=data")
except WEEXError as e:
    match e:
        case WEEXAuthenticationError(code=code):
            print(f"Authentication failed: {code}")
        case WEEXRateLimitError(retry_after=delay):
            print(f"Rate limited, retry after {delay}s")
        case WEEXSystemError():
            print("System error, please try again later")
        case _:
            print(f"Unexpected error: {e}")

Testing

The project includes comprehensive test coverage:

# Run all tests
uv run pytest

# Run with markers
uv run pytest -m unit          # Unit tests only
uv run pytest -m integration   # Integration tests only
uv run pytest -m websocket     # WebSocket tests only
uv run pytest -m "not slow"    # Skip slow tests

# Run with coverage report
uv run pytest --cov=weex_client --cov-report=html

Documentation

See the /docs directory for comprehensive documentation:

# Build documentation locally (requires dev dependencies)
cd docs && make html

Project Structure

weex_client/
├── weex_client/           # Main package
│   ├── __init__.py        # Package exports
│   ├── client.py          # Async REST client
│   ├── sync.py            # Sync wrapper
│   ├── websocket.py       # WebSocket client
│   ├── config.py          # Configuration management
│   ├── models.py          # Pydantic models
│   ├── types.py           # Type definitions
│   ├── auth.py            # Authentication utilities
│   ├── exceptions.py      # Error handling
│   └── utils.py           # Utility functions
├── tests/                 # Test suite
├── examples/              # Usage examples
├── docs/                  # Documentation source
├── pyproject.toml         # Project configuration
├── .env.example          # Environment template
└── README.md              # This file

Development

# Clone repository
git clone https://github.com/xsa-dev/weex-client.git
cd weex_client

# Setup development environment (requires Python 3.14+)
uv sync --dev

# Run tests
uv run pytest

# Run specific test
uv run pytest tests/test_client.py::TestWeexAsyncClient::test_method_name

# Run with coverage
uv run pytest --cov=weex_client

# Run E2E tests
python run_e2e_tests.py all

# Code quality checks
uv run ruff check weex_client
uv run black weex_client
uv run mypy weex_client

# Run all quality checks in sequence
uv run ruff check weex_client && uv run black weex_client && uv run mypy weex_client

License

MIT License. See LICENSE file for details.

Contributing

Contributions are welcome! Please see our Contributing Guide for details.

Changelog

See CHANGELOG.md for version history.

---

About This Library

This library was AI-generated using Claude Code to provide a modern, async-first Weex API client for Python 3.14+.

Feedback and Issues

If you encounter any bugs, have feature requests, or need additional API endpoints:

1. Check existing issues on GitHub Issues
2. Create a new issue describing problem you're experiencing:
   • The
   • Expected vs actual behavior
   • Steps to reproduce (if applicable)
   • Python version and environment details

Development Roadmap

This library is currently in active development. However, due to the nature of AI-generated projects, new feature development is prioritized based on community demand:

• Feature requests with clear use cases and user demand will be prioritized
• Critical bug fixes will be addressed promptly
• New API endpoint coverage depends on user requirements

Your feedback matters! Creating issues helps us understand what features are most needed and drives the future development of this library.

For the best experience, please:

• Search existing issues before creating new ones
• Provide detailed descriptions of bugs or feature requests
• Include code examples when reporting issues

---

Support

• GitHub Issues: https://github.com/xsa-dev/weex-client/issues
• PyPI Package: https://pypi.org/project/weex-client/

---