aiotrade-sdk 0.4.1

High-performance async trading API client for Python supporting BingX and Bybit exchanges with intelligent session and cache management

aiotrade

High-performance async trading API client for Python supporting BingX and Bybit exchanges with intelligent session and cache management.

Architecture

The library uses a sophisticated architecture for optimal performance:

Session Management

• Shared Session: SharedSessionManager creates a single aiohttp session with high-performance connection pooling
• Individual Sessions: Clients automatically create individual sessions if shared session isn't initialized
• Connection Pooling: Up to 2000 concurrent connections with smart distribution per host

Client Caching

• TTL Cache: BingxClientsCache and BybitClientsCache cache client instances with 10-minute lifetime
• Lock-Free: No blocking operations for maximum performance
• Lazy Cleanup: Expired entries removed on access, not proactively

Implemented methods

BybitClient methods (40):
    batch_cancel_order                     get_smp_group_id
    batch_place_order                      get_trade_behaviour_setting
    batch_set_collateral_coin              get_transaction_log
    cancel_all_orders                      get_transferable_amount
    cancel_order                           get_wallet_balance
    get_account_info                       manual_borrow
    get_account_instruments_info           manual_repay
    get_borrow_history                     manual_repay_without_asset_conversion
    get_closed_pnl                         place_order
    get_coin_greeks                        repay_liability
    get_collateral_info                    reset_mmp
    get_dcp_info                           set_collateral_coin
    get_fee_rate                           set_leverage
    get_instruments_info                   set_limit_price_behaviour
    get_kline                              set_margin_mode
    get_mmp_state                          set_mmp
    get_open_and_closed_orders             set_spot_hedging
    get_order_history                      set_trading_stop
    get_position_info                      switch_position_mode
    get_server_time                        upgrade_to_unified_account_pro

BingxClient methods (30):
    cancel_all_spot_open_orders                get_spot_trade_details
    cancel_all_swap_open_orders                get_swap_account_balance
    cancel_spot_batch_orders                   get_swap_contracts
    cancel_swap_batch_orders                   get_swap_klines
    change_swap_margin_type                    get_swap_leverage_and_available_positions
    close_swap_position                        get_swap_margin_type
    get_account_asset_overview                 get_swap_open_orders
    get_api_permissions                        get_swap_order_details
    get_server_time                            get_swap_order_history
    get_spot_account_assets                    get_swap_position_history
    get_spot_klines                            get_swap_position_mode
    get_spot_open_orders                       get_swap_positions
    get_spot_order_details                     place_swap_order
    get_spot_order_history                     set_swap_leverage
    get_spot_symbols_like                      set_swap_position_mode

Installation

poetry add aiotrade-sdk

Quick Start

Option 1: Shared Session (Recommended for Production)

from aiotrade import SharedSessionManager, BingxClient, BybitClient

# Initialize shared session at startup (once per application)
SharedSessionManager.setup(max_connections=2000)

# Create clients for different exchanges - they automatically use the shared session
bingx_client = BingxClient(api_key="bingx_key", api_secret="bingx_secret", demo=True)
bybit_client = BybitClient(api_key="bybit_key", api_secret="bybit_secret", testnet=True)

try:
    # Use clients for API calls
    bingx_assets = await bingx_client.get_spot_account_assets()
    bybit_tickers = await bybit_client.get_tickers(category="spot")
finally:
    # Close shared session at shutdown
    await SharedSessionManager.close()

Option 2: Individual Sessions

from aiotrade import BingxClient, BybitClient

# BingX client with individual session
async with BingxClient(api_key="your_key", api_secret="your_secret", demo=True) as client:
    assets = await client.get_spot_account_assets()
    print(f"BingX assets: {assets}")

# Bybit client with individual session
async with BybitClient(api_key="your_key", api_secret="your_secret", testnet=True) as client:
    tickers = await client.get_tickers(category="spot")
    print(f"Bybit tickers: {tickers}")

Option 3: Cached Clients

from aiotrade import BingxClientsCache, BybitClientsCache

# Get cached BingX client (creates new if doesn't exist)
bingx_client = BingxClientsCache.get_or_create(
    api_key="your_key",
    api_secret="your_secret",
    demo=True
)

# Get cached Bybit client
bybit_client = BybitClientsCache.get_or_create(
    api_key="your_key",
    api_secret="your_secret",
    testnet=True
)

# Use clients (session management is automatic)
async with bingx_client:
    assets = await bingx_client.get_spot_account_assets()

async with bybit_client:
    tickers = await bybit_client.get_tickers(category="spot")

# Same parameters return the same cached instance
cached_bingx = BingxClientsCache.get_or_create(
    api_key="your_key",
    api_secret="your_secret",
    demo=True
)
assert bingx_client is cached_bingx  # True

Session Behavior

Scenario	Session Type	When Used
SharedSessionManager.setup() called	Shared session	All clients
No shared session initialized	Individual session	Each client
Cached clients	Depends on initialization	Cached per credentials

Cache Features

• Automatic TTL: 10 minutes default, configurable
• Memory Safe: Prevents client accumulation
• High Performance: Lock-free operations
• Background Cleanup: Optional periodic cleanup task

# Configure cache lifetime for each exchange
BingxClientsCache.configure(lifetime_seconds=1800)  # 30 minutes
BybitClientsCache.configure(lifetime_seconds=1800)  # 30 minutes

# Start background cleanup
bingx_cleanup = BingxClientsCache.create_cleanup_task(interval_seconds=300)
bybit_cleanup = BybitClientsCache.create_cleanup_task(interval_seconds=300)

# Manual cleanup
bingx_removed = BingxClientsCache.cleanup_expired()
bybit_removed = BybitClientsCache.cleanup_expired()

API Methods

BingX Client Methods

from aiotrade import BingxClient

client = BingxClient(api_key="your_key", api_secret="your_secret", demo=True)

# Market data
server_time = await client.get_server_time()

# Spot trading
assets = await client.get_spot_account_assets()
tickers = await client.get_spot_tickers()

# Swap trading (perpetual futures)
await client.place_swap_order({
    "symbol": "BTC-USDT",
    "side": "BUY",
    "positionSide": "BOTH",
    "type": "MARKET",
    "quantity": 0.001
})

Bybit Client Methods

from aiotrade import BybitClient

client = BybitClient(api_key="your_key", api_secret="your_secret", testnet=True)

# Market data
server_time = await client.get_server_time()
tickers = await client.get_tickers(category="spot")
klines = await client.get_kline("BTCUSDT", "1h", category="linear")

# Trading
await client.place_order({
    "category": "linear",
    "symbol": "BTCUSDT",
    "side": "Buy",
    "orderType": "Market",
    "qty": "0.001"
})

Requirements

• Python >= 3.12
• aiohttp
• High-performance connection pooling for production use

Performance Tips

1. Use Shared Session for applications creating many clients
2. Enable Caching for repeated API credential usage
3. Configure Connection Limits based on your throughput needs
4. Use Background Cleanup for long-running applications