Sync Client API

The WeexSyncClient provides a synchronous interface that internally manages the async operations, making it easier to integrate with existing synchronous code or for developers new to async programming.

class weex_client.sync.WeexSyncClient(config, *, timeout=30.0, max_retries=3, backoff_base=0.5, backoff_max=8.0)[source]

Bases: object

Synchronous wrapper for WeexAsyncClient.

Provides a blocking interface while maintaining the benefits of the async implementation internally.

__init__(config, *, timeout=30.0, max_retries=3, backoff_base=0.5, backoff_max=8.0)[source]

Initialize synchronous client wrapper.

Parameters:

config (weex_client.config.WeexConfig) – WeexConfig instance
timeout (weex_client.types.TypeAliasType) – Request timeout in seconds
max_retries (int) – Maximum retry attempts
backoff_base (float) – Base backoff delay for retries
backoff_max (float) – Maximum backoff delay

__enter__()[source]: Context manager entry.

__exit__(exc_type, exc, tb)[source]: Context manager exit.

get_account_balance()[source]: GET /capi/v2/account/assets - Get account balance.

get_all_positions()[source]: GET /capi/v2/account/position/allPosition - Get all positions.

get_position(symbol)[source]: GET /capi/v2/account/position/singlePosition - Get single position.

place_order(order_request)[source]: POST /capi/v2/order/placeOrder - Place a new order.

cancel_order(order_id=None, client_oid=None)[source]: POST /capi/v2/order/cancel_order - Cancel an order.

get_all_tickers()[source]: GET /capi/v2/market/tickers - Get all tickers.

get_ticker(symbol)[source]: GET /capi/v2/market/ticker - Get ticker for specific symbol.

get_order_book(symbol, limit=15)[source]: GET /capi/v2/market/depth - Get order book.

get_trades(symbol, limit=100)[source]: GET /capi/v2/market/trades - Get recent trades.

get_contracts(symbol=None)[source]: GET /capi/v2/market/contracts - Get contract information.

get_market_overview(symbol, *, order_book_limit=15, trades_limit=100, return_exceptions=False)[source]

Get market overview with concurrent requests.

This method showcases the power of the async implementation while providing a synchronous interface.

get_multiple_positions(symbols, *, return_exceptions=False)[source]

Get positions for multiple symbols concurrently.

Returns results in the same order as input symbols.

close()[source]: Close the client and cleanup resources.

get()[source]: Legacy method for backward compatibility.

Account Management

WeexSyncClient.get_account_balance()[source]: GET /capi/v2/account/assets - Get account balance.

WeexSyncClient.get_all_positions()[source]: GET /capi/v2/account/position/allPosition - Get all positions.

WeexSyncClient.get_position(symbol)[source]: GET /capi/v2/account/position/singlePosition - Get single position.

WeexSyncClient.get_multiple_positions(symbols, *, return_exceptions=False)[source]

Get positions for multiple symbols concurrently.

Returns results in the same order as input symbols.

Order Management

WeexSyncClient.place_order(order_request)[source]: POST /capi/v2/order/placeOrder - Place a new order.

WeexSyncClient.cancel_order(order_id=None, client_oid=None)[source]: POST /capi/v2/order/cancel_order - Cancel an order.

Market Data

WeexSyncClient.get_ticker(symbol)[source]: GET /capi/v2/market/ticker - Get ticker for specific symbol.

WeexSyncClient.get_all_tickers()[source]: GET /capi/v2/market/tickers - Get all tickers.

WeexSyncClient.get_order_book(symbol, limit=15)[source]: GET /capi/v2/market/depth - Get order book.

WeexSyncClient.get_trades(symbol, limit=100)[source]: GET /capi/v2/market/trades - Get recent trades.

WeexSyncClient.get_contracts(symbol=None)[source]: GET /capi/v2/market/contracts - Get contract information.

Context Manager

The sync client also implements context manager protocol for resource management:

from weex_client import WeexSyncClient, WeexConfig

def sync_trading_example():
    config = WeexConfig.from_env()

    with WeexSyncClient(config) as client:
        # Client resources are automatically managed
        balance = client.get_account_balance()
        print(f"Balance: {balance}")

        # Client is automatically closed when exiting context

Quick Creation Function

weex_client.sync.create_sync_client(api_key, secret_key, passphrase, environment='production', *, timeout=30.0, max_retries=3, backoff_base=0.5, backoff_max=8.0)[source]

Create a pre-configured synchronous Weex client.

This is a convenience function for quick client creation without needing to create a WeexConfig instance first.

Parameters:

api_key (str) – Your Weex API key
secret_key (str) – Your Weex secret key
passphrase (str) – Your Weex API passphrase
environment (str) – Environment (‘production’, ‘sandbox’, ‘test’)
timeout (weex_client.types.TypeAliasType) – Request timeout in seconds
max_retries (int) – Maximum retry attempts
backoff_base (float) – Base backoff delay for retries
backoff_max (float) – Maximum backoff delay

Returns:

Configured WeexSyncClient instance

When to Use Sync vs Async

Use WeexSyncClient when:

Integrating with existing sync code - No need to refactor existing applications
Simple scripts - One-off operations or quick checks
Learning phase - Easier for developers new to async programming
Data analysis - Working with pandas, numpy, or other sync libraries

Use WeexAsyncClient when:

High performance needed - Multiple concurrent API calls
Real-time applications - WebSocket streaming or live trading
I/O bound operations - Multiple API calls that can run in parallel
Modern Python applications - Using FastAPI, asyncio, etc.

Basic Usage Examples

Simple Balance Check

from weex_client import WeexSyncClient, WeexConfig

def check_balance():
    config = WeexConfig.from_env()

    with WeexSyncClient(config) as client:
        balance_data = client.get_account_balance()
        available_balance = float(balance_data["data"]["balance"])
        print(f"Available balance: ${available_balance:.2f}")

        return available_balance

Safe Order Placement

from weex_client import WeexSyncClient, WeexConfig
from weex_client.models import PlaceOrderRequest
import time

def place_safe_order(symbol: str, order_type: str = "1"):
    """Place order with 1% risk management (sync version)"""
    config = WeexConfig.from_env()

    with WeexSyncClient(config) as client:
        # Get current balance
        balance_data = client.get_account_balance()
        available_balance = float(balance_data["data"]["balance"])

        # Calculate 1% position size
        max_risk_amount = available_balance * 0.01
        ticker_data = client.get_ticker(symbol)
        current_price = float(ticker_data["data"]["last"])

        # Calculate safe position size
        safe_size = max_risk_amount / current_price

        # Place order with safety
        order = PlaceOrderRequest(
            symbol=symbol,
            client_oid=f"safe_sync_order_{int(time.time())}",
            size=str(safe_size),
            type=order_type,
            order_type="1",  # Market order
            match_price="1"   # Use current price
        )

        result = client.place_order(order)
        print(f"✅ Safe order placed: {result}")
        return result

Market Data Analysis

import pandas as pd
from weex_client import WeexSyncClient, WeexConfig

def analyze_markets(symbols: list[str]):
    """Analyze multiple markets using pandas"""
    config = WeexConfig.from_env()

    with WeexSyncClient(config) as client:
        market_data = []

        for symbol in symbols:
            ticker_data = client.get_ticker(symbol)
            position_data = client.get_position(symbol)

            market_data.append({
                "symbol": symbol,
                "price": float(ticker_data["data"]["last"]),
                "volume": float(ticker_data["data"]["volume"]),
                "position_size": float(position_data["data"].get("size", 0))
            })

        # Convert to pandas DataFrame for analysis
        df = pd.DataFrame(market_data)

        # Calculate additional metrics
        df["position_value"] = df["price"] * df["position_size"]

        print("📊 Market Analysis:")
        print(df.to_string())

        return df

Thread Management

The sync client manages background threads automatically:

from weex_client import create_sync_client

def quick_trading():
    """Quick trading with automatic thread management"""

    # Creates and manages client automatically
    client = create_sync_client()

    try:
        balance = client.get_account_balance()
        print(f"Balance: {balance}")

        # Client is automatically cleaned up
    finally:
        # No need to manually close - handled automatically
        pass

Performance Considerations

Thread Pool Management

The sync client uses a thread pool to run async operations:

# The default thread pool is optimized for most use cases
# You can customize it if needed:

from weex_client.sync import WeexSyncClient
import concurrent.futures

def custom_thread_example():
    config = WeexConfig.from_env()

    # Custom thread pool for specific requirements
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        with WeexSyncClient(config, executor=executor) as client:
            # Operations will use the custom thread pool
            result = client.get_account_balance()
            print(f"Result: {result}")

Blocking Operations

Remember that sync client operations are blocking:

import time
from weex_client import WeexSyncClient, WeexConfig

def timing_example():
    config = WeexConfig.from_env()

    with WeexSyncClient(config) as client:
        start_time = time.time()

        # Each operation blocks until complete
        balance = client.get_account_balance()
        ticker = client.get_ticker("BTCUSDT")
        position = client.get_position("BTCUSDT")

        end_time = time.time()
        print(f"Total time: {end_time - start_time:.2f}s")

        # For better performance, consider using async client
        # for multiple concurrent operations

Integration Examples

Django Integration

# views.py
from django.http import JsonResponse
from weex_client import create_sync_client

def account_balance_view(request):
    """Django view using sync client"""
    client = create_sync_client()

    try:
        balance_data = client.get_account_balance()
        return JsonResponse({
            "balance": balance_data["data"]["balance"],
            "status": "success"
        })
    except Exception as e:
        return JsonResponse({
            "error": str(e),
            "status": "error"
        }, status=500)

Flask Integration

# app.py
from flask import Flask, jsonify
from weex_client import create_sync_client

app = Flask(__name__)

@app.route('/api/balance')
def get_balance():
    """Flask route using sync client"""
    client = create_sync_client()

    try:
        balance_data = client.get_account_balance()
        return jsonify({
            "balance": balance_data["data"]["balance"]
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/ticker/<symbol>')
def get_ticker(symbol):
    """Get ticker data for a symbol"""
    client = create_sync_client()

    try:
        ticker_data = client.get_ticker(symbol)
        return jsonify(ticker_data["data"])
    except Exception as e:
        return jsonify({"error": str(e)}), 500

Data Analysis with Jupyter

# In Jupyter notebook
import pandas as pd
import matplotlib.pyplot as plt
from weex_client import create_sync_client

def market_analysis_notebook():
    """Analyze markets in Jupyter"""
    client = create_sync_client()

    symbols = ["BTCUSDT", "ETHUSDT", "ADAUSDT"]

    # Collect data
    market_data = []
    for symbol in symbols:
        ticker = client.get_ticker(symbol)
        market_data.append({
            "symbol": symbol,
            "price": float(ticker["data"]["last"]),
            "volume_24h": float(ticker["data"]["volume"])
        })

    # Create DataFrame
    df = pd.DataFrame(market_data)

    # Visualize
    plt.figure(figsize=(10, 6))
    plt.bar(df["symbol"], df["price"])
    plt.title("Current Prices")
    plt.ylabel("Price (USD)")
    plt.show()

    return df

Script Integration

# trading_script.py
#!/usr/bin/env python3

import sys
import argparse
from weex_client import create_sync_client

def main():
    parser = argparse.ArgumentParser(description="Simple trading CLI")
    parser.add_argument("command", choices=["balance", "ticker", "position"])
    parser.add_argument("--symbol", default="BTCUSDT")

    args = parser.parse_args()

    client = create_sync_client()

    try:
        if args.command == "balance":
            balance = client.get_account_balance()
            print(f"Balance: ${balance['data']['balance']}")

        elif args.command == "ticker":
            ticker = client.get_ticker(args.symbol)
            print(f"{args.symbol}: ${ticker['data']['last']}")

        elif args.command == "position":
            position = client.get_position(args.symbol)
            size = position['data'].get('size', 0)
            print(f"{args.symbol} position: {size}")

    except Exception as e:
        print(f"Error: {e}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Patterns

Error Handling

from weex_client import create_sync_client
from weex_client.exceptions import WEEXError

def safe_api_call(func, *args, **kwargs):
    """Safe wrapper for sync API calls"""
    client = create_sync_client()

    try:
        return func(client, *args, **kwargs)
    except WEEXError as e:
        print(f"API Error: {e}")
        return None
    except Exception as e:
        print(f"Unexpected error: {e}")
        return None
    finally:
        # Client is automatically cleaned up
        pass

# Usage
def get_balance_safe():
    return safe_api_call(lambda client: client.get_account_balance())

Batch Processing

def process_multiple_symbols(symbols: list[str]):
    """Process multiple symbols sequentially"""
    client = create_sync_client()

    results = {}

    try:
        for symbol in symbols:
            print(f"Processing {symbol}...")

            ticker = client.get_ticker(symbol)
            position = client.get_position(symbol)

            results[symbol] = {
                "ticker": ticker,
                "position": position
            }

    except Exception as e:
        print(f"Error processing symbols: {e}")

    return results

# Note: For better performance with many symbols,
# consider using the async client instead

Configuration

The sync client accepts the same configuration as the async client:

from weex_client import WeexSyncClient, WeexConfig

def custom_config_example():
    config = WeexConfig(
        api_key="your_key",
        secret_key="your_secret",
        passphrase="your_passphrase",
        environment="development"
    )

    with WeexSyncClient(config) as client:
        # Client uses custom configuration
        balance = client.get_account_balance()
        print(f"Balance: {balance}")