exchange.Go() - 异步执行

功能说明

异步执行交易所方法，立即返回 AsyncTask 对象，不阻塞当前线程。支持并发执行多个API调用，显著提升交易速度。

适用场景：

✅ 跨所套利（双边并行下单）
✅ 网格策略（批量挂单）
✅ 批量建仓/平仓
✅ 并行获取行情数据
✅ 任何需要提升速度的场景

方法签名

python

exchange.Go(method_name, *args, **kwargs) -> AsyncTask

参数

参数	类型	必填	说明
method_name	str	✅	方法名（字符串），如 `"GetTicker"`, `"Buy"`, `"Sell"`
*args	any	❌	方法的位置参数
**kwargs	any	❌	方法的关键字参数

返回值

返回 AsyncTask 对象，调用 wait() 方法获取结果。

AsyncTask 对象

wait() 方法

等待任务完成并获取结果。

python

result, ok = task.wait(timeout=None)

参数：

timeout (int, 可选): 超时时间（毫秒），None 表示无限等待

返回值：

result: 方法执行结果（成功时）或 None（失败/超时时）
ok (bool): True 表示成功，False 表示失败/超时

is_done() 方法

检查任务是否完成（不阻塞）。

python

done = task.is_done()  # True/False

cancel() 方法

取消任务（仅在任务还未开始执行时有效）。

python

success = task.cancel()  # True/False

elapsed_ms() 方法

获取任务已执行时间（毫秒）。

python

elapsed = task.elapsed_ms()  # float

使用示例

示例1：并行获取行情数据

性能提升：串行 125ms → 并行 45ms（2.8倍）

python

# 异步调用（立即返回）
a = exchange.Go("GetTicker")
b = exchange.Go("GetDepth")
c = exchange.Go("GetAccount")

# 等待结果
ticker, ok1 = a.wait()
depth, ok2 = b.wait()
account, ok3 = c.wait()

if ok1:
    Log(f"价格: {ticker['Last']}")
if ok2:
    Log(f"深度: {len(depth['Asks'])} 档")
if ok3:
    Log(f"余额: {account['Balance']} USDT")

示例2：跨所套利（双边并行下单）⭐

性能提升：串行 148ms → 并行 77ms（1.9倍）

python

# 发现套利机会
ticker1 = exchanges[0].GetTicker()
ticker2 = exchanges[1].GetTicker()

price1 = ticker1['Last']  # Binance: 30100
price2 = ticker2['Last']  # Gate: 30150

if price1 < price2:
    # 并行下单（关键！）
    buy = exchanges[0].Go("Buy", price1, 0.01)
    sell = exchanges[1].Go("Sell", price2, 0.01)
    
    # 等待结果（3秒超时）
    buy_id, buy_ok = buy.wait(3000)
    sell_id, sell_ok = sell.wait(3000)
    
    # 检查结果（单边保护）
    if buy_ok and sell_ok:
        profit = (price2 - price1) * 0.01
        Log(f"✅ 套利成功 | 利润: {profit:.2f} USDT")
    else:
        # 单边失败，取消另一边
        if buy_ok and not sell_ok:
            Log("⚠️ 卖单失败，取消买单")
            exchanges[0].CancelOrder(buy_id)
        elif sell_ok and not buy_ok:
            Log("⚠️ 买单失败，取消卖单")
            exchanges[1].CancelOrder(sell_id)

示例3：网格批量挂单

性能提升：串行 770ms → 并行 82ms（9.4倍）

python

symbol = "BTC_USDT"
center_price = 30000
grid_count = 10
grid_spacing = 100
amount = 0.01

tasks = []

# 设置交易对
exchange.SetCurrency(symbol)

# 提交所有任务（不等待）
for i in range(1, grid_count + 1):
    # 买单（低于中心价）
    buy_price = center_price - grid_spacing * i
    task = exchange.Go("Buy", buy_price, amount)
    tasks.append(('buy', buy_price, task))
    
    # 卖单（高于中心价）
    sell_price = center_price + grid_spacing * i
    task = exchange.Go("Sell", sell_price, amount)
    tasks.append(('sell', sell_price, task))

# 等待所有完成
success = 0
for side, price, task in tasks:
    order_id, ok = task.wait(10000)  # 10秒超时
    if ok:
        success += 1
        Log(f"✅ {side} @ {price}: {order_id}")

Log(f"网格挂单完成: {success}/{len(tasks)}")

示例4：超时重试

python

# 提交任务
task = exchange.Go("Buy", 30100, 0.01)

# 先等1秒
order_id, ok = task.wait(1000)

if not ok:
    Log("⚠️ 下单超时（1秒），继续等待...")
    # 继续等待（无限）
    order_id, ok = task.wait()

if ok:
    Log(f"✅ 订单成功: {order_id}")
else:
    Log("❌ 订单失败")

示例5：检查任务状态（不阻塞）

python

import time

# 提交任务
task = exchange.Go("Buy", 30100, 0.01)

# 轮询检查（不阻塞）
for i in range(10):
    if task.is_done():
        order_id, ok = task.wait()
        if ok:
            Log(f"✅ 订单完成: {order_id} | 耗时: {task.elapsed_ms():.0f}ms")
        break
    else:
        Log(f"等待中... {i+1}s")
        time.sleep(1)
else:
    Log("⚠️ 超时，取消任务")
    task.cancel()

示例6：性能测试（计算执行时间）

测量单个任务执行时间

python

import time

# 测试单个订单
start = time.time()
task = exchange.Go("Buy", 30100, 0.01)
order_id, ok = task.wait()
end = time.time()

if ok:
    # 方式1：使用task.elapsed_ms()
    Log(f"✅ 订单成功 | 任务耗时: {task.elapsed_ms():.0f}ms")
    
    # 方式2：手动计算
    total_time = (end - start) * 1000
    Log(f"总耗时（含提交）: {total_time:.0f}ms")

对比串行 vs 并行性能

python

import time

# ========== 串行执行 ==========
Log("========== 串行执行 ==========")
start = time.time()

ticker = exchange.GetTicker()
depth = exchange.GetDepth()
account = exchange.GetAccount()

serial_time = (time.time() - start) * 1000
Log(f"串行耗时: {serial_time:.0f}ms")

# ========== 并行执行 ==========
Log("========== 并行执行 ==========")
start = time.time()

a = exchange.Go("GetTicker")
b = exchange.Go("GetDepth")
c = exchange.Go("GetAccount")

# 提交耗时
submit_time = (time.time() - start) * 1000
Log(f"提交3个任务耗时: {submit_time:.2f}ms")

ticker, ok1 = a.wait()
depth, ok2 = b.wait()
account, ok3 = c.wait()

parallel_time = (time.time() - start) * 1000
Log(f"并行总耗时: {parallel_time:.0f}ms")

# 性能对比
speedup = serial_time / parallel_time
Log(f"性能提升: {speedup:.1f}x (节省 {serial_time - parallel_time:.0f}ms)")

# 每个任务的实际耗时
Log(f"GetTicker 耗时: {a.elapsed_ms():.0f}ms")
Log(f"GetDepth 耗时: {b.elapsed_ms():.0f}ms")
Log(f"GetAccount 耗时: {c.elapsed_ms():.0f}ms")

输出示例：

plaintext

========== 串行执行 ==========
串行耗时: 125ms

========== 并行执行 ==========
提交3个任务耗时: 0.31ms
并行总耗时: 45ms
性能提升: 2.8x (节省 80ms)
GetTicker 耗时: 45ms
GetDepth 耗时: 38ms
GetAccount 耗时: 42ms

示例7：批量建仓多个币种

性能提升：串行 320ms → 并行 95ms（3.4倍）

python

symbols = ['BTC_USDT', 'ETH_USDT', 'SOL_USDT', 'DOGE_USDT']
amount_usdt = 100  # 每个币种100 USDT

# 先获取所有价格（并行）
price_tasks = []
for s in symbols:
    exchange.SetCurrency(s)
    price_tasks.append(exchange.Go("GetTicker"))

# 等待价格
tickers = {}
for s, task in zip(symbols, price_tasks):
    ticker, ok = task.wait()
    if ok:
        tickers[s] = ticker['Last']

# 准备下单（并行）
order_tasks = []
for s in symbols:
    if s in tickers:
        amount = amount_usdt / tickers[s]
        exchange.SetCurrency(s)
        task = exchange.Go("Buy", -1, amount)  # 市价单
        order_tasks.append((s, task))

# 等待所有订单完成
success = []
failed = []
for s, task in order_tasks:
    order_id, ok = task.wait(5000)
    if ok:
        success.append(s)
    else:
        failed.append(s)

Log(f"✅ 成功: {success}")
Log(f"❌ 失败: {failed}")

支持的方法

所有 exchange.* 方法都支持异步调用：

市场行情

GetTicker() - 获取行情
GetDepth() - 获取深度
GetTrades() - 获取成交
GetRecords(period, limit) - 获取K线

账户信息

GetAccount() - 获取账户
GetPosition() - 获取持仓
GetTradingFee() - 获取手续费

交易操作

Buy(price, amount) - 买入（使用当前交易对）
Sell(price, amount) - 卖出（使用当前交易对）
CancelOrder(order_id) - 取消订单
GetOrder(order_id) - 查询订单
GetOrders() - 获取未完成订单

其他方法

IO(api, method, params) - 底层API调用
SetCurrency(symbol) - 设置交易对
SetContractType(type) - 设置合约类型
SetMarginLevel(level) - 设置杠杆

性能对比

场景	串行耗时	并行耗时	提升
获取3个行情数据	125ms	45ms	2.8x
跨所套利（2单）	148ms	77ms	1.9x
网格挂单（10单）	770ms	82ms	9.4x
批量建仓（4单）	320ms	95ms	3.4x

注意事项

✅ 优势

性能提升显著：2-10倍速度提升
完全兼容：不影响现有同步调用
使用简单：只需添加 Go() 和 wait()
线程安全：内置线程池管理
自动清理：线程池自动回收

⚠️ 注意

并发数量：默认线程池最大10个worker，超过会排队
超时处理：建议设置合理的超时时间
错误处理：检查 ok 值，失败时 result 为 None
单边风险：套利场景务必做单边保护
资源占用：每个Exchange实例有独立线程池

🚫 不适用场景

只有1个操作时（无并发优势）
需要严格顺序执行的逻辑
超过50个并发任务（建议分批）

常见问题

Q1: Go() 和直接调用有什么区别？

python

# 直接调用（同步，阻塞）
ticker = exchange.GetTicker()  # 等待45ms

# Go()异步调用（不阻塞）
task = exchange.Go("GetTicker")  # 立即返回 (~0.005ms)
ticker, ok = task.wait()          # 等待45ms

区别：Go()允许多个任务并发执行

Q2: 多次调用Go()会创建多个线程池吗？

不会。每个 Exchange 实例只有一个线程池（懒加载），所有 Go() 调用共享。

Q3: 超时后任务会被取消吗？

不会。超时只是 wait() 返回 (None, False)，任务仍在后台执行。如需取消，调用 cancel()。

Q4: 如何设置更大的并发数？

python

# 在首次调用Go()之前修改
exchange._thread_pool = None  # 重置
from concurrent.futures import ThreadPoolExecutor
exchange._thread_pool = ThreadPoolExecutor(max_workers=20)

Q5: Go() 支持 exchanges[0] 和 exchanges[1] 吗？

完全支持！每个交易所实例都有独立的线程池。

python

buy = exchanges[0].Go("Buy", 30100, 0.01)
sell = exchanges[1].Go("Sell", 30150, 0.01)

更新日志

版本	日期	说明
1.0.0	2025-12-13	首次发布，支持所有exchange方法异步调用

📌 推荐用法：任何需要并发执行的场景，都用 Go() + wait() 替代直接调用！

exchange.Go() - 异步执行

功能说明

方法签名

参数

返回值

AsyncTask 对象

wait() 方法

is_done() 方法

cancel() 方法

elapsed_ms() 方法

使用示例

示例1：并行获取行情数据

示例2：跨所套利（双边并行下单）⭐

示例3：网格批量挂单

示例4：超时重试

示例5：检查任务状态（不阻塞）

示例6：性能测试（计算执行时间）

示例7：批量建仓多个币种

支持的方法

市场行情

账户信息

交易操作

其他方法

性能对比

注意事项

✅ 优势

⚠️ 注意

🚫 不适用场景

常见问题

Q1: Go() 和直接调用有什么区别？

Q2: 多次调用Go()会创建多个线程池吗？

Q3: 超时后任务会被取消吗？

Q4: 如何设置更大的并发数？

Q5: Go() 支持 exchanges[0] 和 exchanges[1] 吗？

相关文档

更新日志

On this page