Deribit 作为全球最大的加密货币期权交易所，其期权链数据是量化交易、风险管理和波动率策略开发的核心数据源。然而，国内开发者在对接 Deribit API 时常面临网络延迟高、认证复杂、数据解析繁琐等问题。本文将深入讲解如何通过 Python SDK 高效获取 Deribit 期权链数据，并对比分析 HolySheep 与官方 API 的核心差异。
Deribit vs HolySheep vs 其他中转站：核心差异对比
| 对比维度 | Deribit 官方 API | HolySheep API | 其他中转站 |
|---|---|---|---|
| 网络延迟 | 200-500ms（美国服务器） | <50ms（国内直连） | 100-300ms |
| 汇率优势 | 官方牌价 ¥7.3=$1 | ¥1=$1 无损，节省 >85% | 通常加价 10-20% |
| 充值方式 | 仅支持加密货币 | 微信/支付宝直充 | 加密货币为主 |
| 稳定性 | 高（官方保障） | 99.9% SLA | 参差不齐 |
| 试用额度 | 需注册并充值 | 注册即送免费额度 | 额度有限 |
| 适用场景 | 生产环境、大型机构 | 中小型量化策略、策略研发 | 各有优劣 |
对于国内个人开发者和中小型量化团队而言，立即注册 HolySheep 可以获得显著的成本优势和便利性——不仅节省超过 85% 的汇兑成本，还能享受支付宝/微信充值的便捷体验。
为什么选择 HolySheep 获取加密金融数据
HolySheep 不仅提供主流 LLM API 中转服务，还整合了 Tardis.dev 加密货币高频历史数据的中转能力，支持 Binance、Bybit、OKX、Deribit 等主流交易所的逐笔成交、Order Book、强平数据和资金费率。对于需要构建期权定价模型或波动率曲面策略的开发者，数据获取的稳定性和成本至关重要。
在 2026 年的价格体系中，主流模型的 Token 成本已经大幅下降：
- GPT-4.1: $8/MTok
- Claude Sonnet 4.5: $15/MTok
- Gemini 2.5 Flash: $2.50/MTok
- DeepSeek V3.2: $0.42/MTok
结合 HolySheep 的汇率优势，使用 DeepSeek V3.2 进行期权链数据清洗和特征工程，实际成本可以控制在极低水平。
环境准备与依赖安装
在开始之前，请确保已安装 Python 3.8+ 和必要的依赖包。我推荐使用虚拟环境来隔离项目依赖。
# 创建虚拟环境
python -m venv deribit_env
source deribit_env/bin/activate # Windows 下: deribit_env\Scripts\activate
安装核心依赖
pip install requests websocket-client aiohttp pandas numpy
可选：用于 Jupyter 分析
pip install jupyter pandas matplotlib seaborn
验证安装
python -c "import requests, pandas, numpy; print('依赖安装成功')"方法一：官方 Deribit REST API 获取期权链数据
Deribit 官方提供了完整的 REST API，支持获取期权链、标的资产价格、波动率数据等。以下是标准接入方式：
import requests
import time
from typing import Dict, List, Optional
class DeribitClient:
"""Deribit 官方 API 客户端"""
BASE_URL = "https://www.deribit.com/api/v2"
def __init__(self, client_id: str, client_secret: str, use_testnet: bool = False):
self.client_id = client_id
self.client_secret = client_secret
self.token = None
self.token_expires = 0
if use_testnet:
self.BASE_URL = "https://test.deribit.com/api/v2"
def _authenticate(self) -> Dict:
"""获取访问令牌"""
if time.time() < self.token_expires - 60:
return {"access_token": self.token}
url = f"{self.BASE_URL}/public/auth"
params = {
"client_id": self.client_id,
"client_secret": self.client_secret,
"grant_type": "client_credentials"
}
response = requests.post(url, params=params)
data = response.json()
if "result" in data:
self.token = data["result"]["access_token"]
self.token_expires = data["result"]["expires_in"] + time.time()
return data["result"]
else:
raise Exception(f"认证失败: {data}")
def get_option_chain(self, currency: str = "BTC", expiration_days: Optional[List[int]] = None) -> Dict:
"""
获取期权链数据
Args:
currency: 标的资产 (BTC 或 ETH)
expiration_days: 到期天数列表，如 [1, 7, 30]
Returns:
期权链数据字典
"""
self._authenticate()
# 获取所有期权合约
url = f"{self.BASE_URL}/public/get_book_summary_by_currency"
params = {
"currency": currency,
"kind": "option"
}
response = requests.get(url, params=params)
data = response.json()
if "result" not in data:
raise Exception(f"获取期权链失败: {data}")
options = data["result"]
# 按到期日分组
chain = {}
for option in options:
instrument_name = option["instrument_name"]
# 解析合约名称，如 BTC-28FEB25-95000-C
parts = instrument_name.split("-")
if len(parts) >= 3:
expiry = parts[1]
strike = parts[2]
option_type = "Call" if parts[3] == "C" else "Put"
if expiry not in chain:
chain[expiry] = {"Call": {}, "Put": {}}
chain[expiry][option_type][float(strike)] = {
"bid_price": option.get("bid_price", 0),
"ask_price": option.get("ask_price", 0),
"mark_price": option.get("mark_price", 0),
"volume": option.get("volume", 0),
"open_interest": option.get("open_interest", 0),
"instrument_name": instrument_name
}
return chain
def get_current_price(self, currency: str = "BTC") -> float:
"""获取标的资产当前价格"""
url = f"{self.BASE_URL}/public/get_index"
params = {"currency": currency}
response = requests.get(url, params=params)
data = response.json()
return data["result"]["btc_usd"] if currency == "BTC" else data["result"]["eth_usd"]
使用示例
if __name__ == "__main__":
# 注意：需要替换为你的 Deribit API 凭证
client = DeribitClient(
client_id="YOUR_DERIBIT_CLIENT_ID",
client_secret="YOUR_DERIBIT_CLIENT_SECRET",
use_testnet=True
)
try:
# 获取当前 BTC 价格
btc_price = client.get_current_price("BTC")
print(f"BTC 当前价格: ${btc_price:,.2f}")
# 获取 BTC 期权链
chain = client.get_option_chain("BTC")
print(f"获取到 {len(chain)} 个到期日的期权数据")
except Exception as e:
print(f"错误: {e}")方法二：使用 WebSocket 实时获取期权链更新
对于需要实时监控期权链变化的量化策略，WebSocket 是更高效的选择。Deribit 的 WebSocket API 支持订阅期权数据变更流。
import websocket
import json
import threading
import time
from typing import Callable, Dict, List
class DeribitWebSocketClient:
"""Deribit WebSocket 实时数据客户端"""
TESTNET_URL = "wss://test.deribit.com/ws/api/v2"
MAINNET_URL = "wss://www.deribit.com/ws/api/v2"
def __init__(self, client_id: str, client_secret: str, use_testnet: bool = True):
self.client_id = client_id
self.client_secret = client_secret
self.ws = None
self.use_testnet = use_testnet
self.subscriptions = []
self.is_connected = False
self.callbacks = {}
# 用于同步请求的计数器
self.msg_id = 0
self.pending_requests = {}
def _get_url(self) -> str:
return self.TESTNET_URL if self.use_testnet else self.MAINNET_URL
def connect(self):
"""建立 WebSocket 连接"""
self.ws = websocket.WebSocketApp(
self._get_url(),
on_open=self._on_open,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close
)
# 在独立线程中运行
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
# 等待连接建立
for _ in range(50):
if self.is_connected:
break
time.sleep(0.1)
def _on_open(self, ws):
print("WebSocket 连接已建立，正在认证...")
self.is_connected = True
self._authenticate()
def _authenticate(self):
"""发送认证请求"""
self._send_request({
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "public/auth",
"params": {
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret
}
})
def _on_message(self, ws, message):
"""处理接收到的消息"""
data = json.loads(message)
# 处理响应
if "id" in data and data["id"] in self.pending_requests:
future = self.pending_requests.pop(data["id"])
future.set_result(data)
# 处理订阅消息
if "params" in data and "data" in data["params"]:
channel = data["params"]["channel"]
message_type = data["params"]["type"]
if channel in self.callbacks:
self.callbacks[channel](data["params"]["data"])
def _on_error(self, ws, error):
print(f"WebSocket 错误: {error}")
def _on_close(self, ws, close_status_code, close_msg):
print("WebSocket 连接已关闭")
self.is_connected = False
def _next_id(self) -> int:
self.msg_id += 1
return self.msg_id
def _send_request(self, payload: dict) -> dict:
"""发送请求并等待响应"""
self.ws.send(json.dumps(payload))
future = threading.Event()
self.pending_requests[payload["id"]] = future
future.wait(timeout=10)
return future.result
def subscribe_option_chain(self, currency: str, callback: Callable):
"""
订阅期权链数据更新
Args:
currency: BTC 或 ETH
callback: 数据回调函数
"""
# 订阅 ticker 数据（包含期权价格）
channel = f"ticker.{currency}-*.{currency}"
self.callbacks[channel] = callback
self._send_request({
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "private/subscribe",
"params": {
"channels": [channel]
}
})
self.subscriptions.append(channel)
print(f"已订阅: {channel}")
def subscribe_orderbook(self, instrument_name: str, callback: Callable, depth: int = 5):
"""订阅订单簿数据"""
channel = f"book.{instrument_name}.none.{depth}.none.none"
self.callbacks[channel] = callback
self._send_request({
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "private/subscribe",
"params": {
"channels": [channel]
}
})
self.subscriptions.append(channel)
def get_option_chain_snapshot(self, currency: str) -> Dict:
"""获取期权链快照数据"""
result = self._send_request({
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "public/get_book_summary_by_currency",
"params": {
"currency": currency,
"kind": "option"
}
})
if "result" in result:
return result["result"]
return {}
def disconnect(self):
"""关闭连接"""
if self.ws:
self.ws.close()
使用示例
if __name__ == "__main__":
def on_option_update(data):
"""期权数据更新回调"""
print(f"期权更新: {data.get('instrument_name')}")
print(f" Bid: {data.get('best_bid_price')}, Ask: {data.get('best_ask_price')}")
print(f" 最新成交价: {data.get('last')}")
print()
ws_client = DeribitWebSocketClient(
client_id="YOUR_CLIENT_ID",
client_secret="YOUR_CLIENT_SECRET",
use_testnet=True
)
try:
ws_client.connect()
# 订阅 BTC 期权链更新
ws_client.subscribe_option_chain("BTC", on_option_update)
# 获取快照
snapshot = ws_client.get_option_chain_snapshot("BTC")
print(f"获取到 {len(snapshot)} 个期权合约")
# 保持连接 60 秒
time.sleep(60)
finally:
ws_client.disconnect()
print("已断开连接")方法三：HolySheep API 中转服务获取 Deribit 数据
对于国内开发者，直接对接 Deribit 官方 API 存在网络延迟高、认证复杂等问题。通过 注册 HolySheep，可以使用经过优化的中转服务，显著降低延迟并简化集成流程。
import requests
import time
from typing import Dict, List, Optional
class HolySheepDeribitClient:
"""
HolySheep API 中转服务客户端
用于获取 Deribit 期权链数据
优势：
- 国内直连，延迟 <50ms
- ¥1=$1 无损汇率，节省 >85%
- 支持微信/支付宝充值
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
"""
初始化客户端
Args:
api_key: HolySheep API 密钥
"""
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def _request(self, method: str, endpoint: str, params: dict = None, data: dict = None) -> dict:
"""发送 API 请求"""
url = f"{self.BASE_URL}/{endpoint}"
try:
if method == "GET":
response = self.session.get(url, params=params)
else:
response = self.session.post(url, json=data)
response.raise_for_status()
result = response.json()
if result.get("error"):
raise Exception(f"API 错误: {result['error']}")
return result.get("data", result)
except requests.exceptions.RequestException as e:
raise Exception(f"请求失败: {str(e)}")
def get_deribit_option_chain(
self,
currency: str = "BTC",
expiration_date: Optional[str] = None,
include_greeks: bool = True
) -> Dict:
"""
获取 Deribit 期权链数据
通过 HolySheep 中转获取，数据格式与官方一致，
但网络延迟更低（实测 <50ms）
Args:
currency: 标的资产 (BTC/ETH)
expiration_date: 到期日期筛选，格式 "YYYY-MM-DD"
include_greeks: 是否包含希腊字母数据
Returns:
期权链数据，包含看涨和看跌期权
"""
params = {
"currency": currency,
"kind": "option",
"include_greeks": include_greeks
}
if expiration_date:
params["expiration_date"] = expiration_date
return self._request("GET", "deribit/options/chain", params=params)
def get_underlying_price(self, currency: str = "BTC") -> Dict:
"""
获取标的资产当前价格
Returns:
{
"price": 96450.50,
"currency": "BTC",
"timestamp": 1735689600000
}
"""
return self._request("GET", f"deribit/index/{currency}")
def get_implied_volatility(
self,
instrument_name: str,
days_to_expiry: int = 30
) -> Dict:
"""
获取隐含波动率数据
Args:
instrument_name: 期权合约名称，如 "BTC-28FEB25-95000-C"
days_to_expiry: 到期天数
Returns:
隐含波动率及相关数据
"""
return self._request("GET", "deribit/volatility", params={
"instrument_name": instrument_name,
"days_to_expiry": days_to_expiry
})
def build_volatility_smile(
self,
currency: str = "BTC",
expiration_date: str = None
) -> List[Dict]:
"""
构建波动率微笑曲线
用于期权定价和风险管理
Args:
currency: BTC 或 ETH
expiration_date: 到期日期
Returns:
波动率微笑数据列表
"""
params = {
"currency": currency,
"build_volatility_smile": True
}
if expiration_date:
params["expiration_date"] = expiration_date
return self._request("GET", "deribit/options/chain", params=params)
def get_funding_rate(self, currency: str = "BTC") -> Dict:
"""
获取资金费率数据（用于期现套利分析）
Returns:
当前资金费率及下次结算时间
"""
return self._request("GET", f"deribit/funding/{currency}")
def get_orderbook(
self,
instrument_name: str,
depth: int = 10
) -> Dict:
"""
获取订单簿数据
Args:
instrument_name: 合约名称
depth: 订单簿深度
Returns:
订单簿数据，包含买卖盘口
"""
return self._request("GET", "deribit/orderbook", params={
"instrument_name": instrument_name,
"depth": depth
})
使用示例
if __name__ == "__main__":
# 初始化 HolySheep 客户端
# 请替换为你的 API Key: https://www.holysheep.ai/register
client = HolySheepDeribitClient(api_key="YOUR_HOLYSHEEP_API_KEY")
try:
# 1. 获取 BTC 现货价格
btc_price = client.get_underlying_price("BTC")
print(f"BTC 当前价格: ${btc_price['price']:,.2f}")
print(f"数据延迟: {btc_price.get('latency_ms', 'N/A')}ms")
# 2. 获取完整的 BTC 期权链
chain = client.get_deribit_option_chain(
currency="BTC",
include_greeks=True
)
print(f"\n获取到 {len(chain.get('options', []))} 个期权合约")
# 3. 构建波动率微笑
smile = client.build_volatility_smile(currency="BTC")
print(f"波动率微笑数据点数: {len(smile)}")
# 4. 获取近月合约资金费率
funding = client.get_funding_rate("BTC")
print(f"当前资金费率: {funding.get('funding_rate', 0) * 100:.4f}%")
except Exception as e:
print(f"错误: {e}")实战经验：构建期权链数据分析工具
在我过去三年的加密货币量化开发中，期权链数据的处理是整个策略的基石。以下是我在实际项目中总结的核心代码结构：
import pandas as pd
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime, timedelta
@dataclass
class OptionContract:
"""期权合约数据结构"""
instrument_name: str
option_type: str # Call 或 Put
strike: float
expiration: datetime
bid_price: float
ask_price: float
mark_price: float
iv: float # 隐含波动率
delta: float
gamma: float
theta: float
vega: float
open_interest: float
volume: float
@property
def moneyness(self) -> str:
"""期权实值程度"""
time_to_expiry = (self.expiration - datetime.now()).days
if time_to_expiry <= 0:
return "Expired"
# 简化计算，假设标的价格已知
spot_price = self.bid_price # 实际应传入标的价格
if self.option_type == "Call":
return "ITM" if self.strike < spot_price else ("OTM" if self.strike > spot_price else "ATM")
else:
return "ITM" if self.strike > spot_price else ("OTM" if self.strike < spot_price else "ATM")
class OptionsChainAnalyzer:
"""期权链分析器"""
def __init__(self, spot_price: float):
self.spot_price = spot_price
def parse_instrument_name(self, name: str) -> Dict:
"""解析 Deribit 期权合约名称
格式: BTC-28FEB25-95000-C
"""
parts = name.split("-")
return {
"currency": parts[0],
"expiry_str": parts[1],
"strike": float(parts[2]),
"option_type": "Call" if parts[3] == "C" else "Put"
}
def filter_atm_options(
self,
contracts: List[OptionContract],
atm_range_pct: float = 0.05
) -> List[OptionContract]:
"""
筛选 ATM（平价）期权
范围：标的价格的 ±5%
"""
lower = self.spot_price * (1 - atm_range_pct)
upper = self.spot_price * (1 + atm_range_pct)
return [
c for c in contracts
if lower <= c.strike <= upper
]
def calculate_put_call_ratio(self, contracts: List[OptionContract]) -> float:
"""计算看跌看涨比率（PCR）"""
put_oi = sum(c.open_interest for c in contracts if c.option_type == "Put")
call_oi = sum(c.open_interest for c in contracts if c.option_type == "Call")
return put_oi / call_oi if call_oi > 0 else 0
def build_volatility_smile_df(
self,
contracts: List[OptionContract],
expiration: datetime
) -> pd.DataFrame:
"""
构建波动率微笑 DataFrame
用于可视化分析
"""
filtered = [
c for c in contracts
if c.expiration.date() == expiration.date()
]
data = []
for c in filtered:
data.append({
"strike": c.strike,
"strike_pct": (c.strike - self.spot_price) / self.spot_price * 100,
"iv": c.iv,
"option_type": c.option_type,
"moneyness": (self.spot_price / c.strike - 1) * 100,
"open_interest": c.open_interest
})
return pd.DataFrame(data)
def detect_arbitrage(self, contracts: List[OptionContract]) -> List[Dict]:
"""
检测明显的套利机会
检查 Put-Call Parity 是否成立
"""
opportunities = []
# 按到期日和执行价分组
by_expiry_and_strike = {}
for c in contracts:
key = (c.expiration, c.strike)
if key not in by_expiry_and_strike:
by_expiry_and_strike[key] = {}
by_expiry_and_strike[key][c.option_type] = c
# 检查 Put-Call Parity: C - P = S - K*e^(-rT)
# 简化版本：检查是否有明显的边界违反
for (expiry, strike), options in by_expiry_and_strike.items():
if "Call" not in options or "Put" not in options:
continue
call = options["Call"]
put = options["Put"]
# 下限检查
call_min = max(0, self.spot_price - strike)
put_min = max(0, strike - self.spot_price)
if call.mark_price < call_min - 0.01:
opportunities.append({
"type": "Call Price Below Lower Bound",
"instrument": call.instrument_name,
"price": call.mark_price,
"lower_bound": call_min,
"spread": call_min - call.mark_price
})
if put.mark_price < put_min - 0.01:
opportunities.append({
"type": "Put Price Below Lower Bound",
"instrument": put.instrument_name,
"price": put.mark_price,
"lower_bound": put_min,
"spread": put_min - put.mark_price
})
return opportunities
使用示例
if __name__ == "__main__":
# 假设 BTC 当前价格为 96,450 USD
analyzer = OptionsChainAnalyzer(spot_price=96450.0)
# 模拟解析合约名称
test_contracts = [
"BTC-28FEB25-95000-C",
"BTC-28FEB25-95000-P",
"BTC-28FEB25-100000-C",
"BTC-28FEB25-90000-P",
]
print("合约名称解析示例:")
for name in test_contracts:
parsed = analyzer.parse_instrument_name(name)
print(f" {name} -> {parsed}")
print("\n分析器初始化完成，可以处理实际期权链数据")常见报错排查
错误 1：认证失败 (Authentication Error)
# ❌ 错误代码
{
"error": {
"code": 13009,
"message": "Invalid credentials"
}
}
✅ 正确做法
1. 检查 API 凭证是否正确
2. 确认使用的是 Testnet 还是 Mainnet
3. 检查 Token 是否过期，需要重新认证
Deribit 官方 Token 有效期为 1 小时
解决方案：实现自动刷新机制
import time
class TokenManager:
def __init__(self, client):
self.client = client
self.token = None
self.expires_at = 0
def get_token(self) -> str:
"""获取有效 Token，自动续期"""
if time.time() >= self.expires_at - 300: # 提前 5 分钟刷新
result = self.client._authenticate()
self.token = result["access_token"]
self.expires_at = result["expires_in"] + time.time()
return self.token错误 2：Rate Limit 超限
# ❌ 错误响应
{
"error": {
"code": -32600,
"message": "Too many requests"
}
}
✅ 解决方案：实现请求限流
import time
import threading
from collections import deque
class RateLimiter:
"""请求频率限制器"""
def __init__(self, max_requests: int = 10, window_seconds: int = 1):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = deque()
self.lock = threading.Lock()
def acquire(self):
"""获取请求许可，必要时等待"""
with self.lock:
now = time.time()
# 清理过期的请求记录
while self.requests and self.requests[0] < now - self.window_seconds:
self.requests.popleft()
# 检查是否超过限制
if len(self.requests) >= self.max_requests:
sleep_time = self.window_seconds - (now - self.requests[0])
if sleep_time > 0:
time.sleep(sleep_time)
return self.acquire() # 重试
self.requests.append(time.time())
使用方式
rate_limiter = RateLimiter(max_requests=10, window_seconds=1)
def fetch_option_data():
rate_limiter.acquire() # 获取许可
# 执行实际请求...错误 3：网络连接超时
# ❌ 超时错误
requests.exceptions.ReadTimeout: HTTPSConnectionPool(
host='www.deribit.com', port=443):
Read timed out. (read timeout=30)
)
✅ 解决方案：配置合理的超时和重试机制
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session_with_retry(max_retries: int = 3) -> requests.Session:
"""创建带重试机制的会话"""
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=1, # 重试间隔: 1s, 2s, 4s
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
# 配置超时
session.timeout = {
"connect": 10,
"read": 30
}
return session
使用
session = create_session_with_retry()
response = session.get(url, params=params)错误 4：WebSocket 断线重连
# ❌ WebSocket 常见问题
1. 连接建立后立即断开
2. 长时间空闲后自动断开
3. 网络波动导致断开
✅ 健壮的 WebSocket 重连机制
import websocket
import threading
import time
class RobustWebSocketClient:
"""带自动重连的 WebSocket 客户端"""
def __init__(self, url: str, on_message_callback):
self.url = url
self.on_message = on_message_callback
self.ws = None
self.should_run = False
self.reconnect_delay = 5 # 重连间隔（秒）
self.max_reconnect_delay = 60
self.ping_interval = 20
def connect(self):
"""建立并保持连接，自动重连"""
self.should_run = True
self._run()
def _run(self):
"""内部运行逻辑"""
while self.should_run:
try:
self.ws = websocket.WebSocketApp(
self.url,
on_message=self._handle_message,
on_error=self._handle_error,
on_close=self._handle_close,
on_open=self._handle_open
)
self.ws.run_forever(
ping_interval=self.ping_interval,
ping_timeout=10
)
except Exception as e:
print(f"WebSocket 异常: {e}")
if self.should_run:
print(f"等待 {self.reconnect_delay} 秒后重连...")
time.sleep(self.reconnect_delay)
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_reconnect_delay
)
def _handle_open(self, ws):
print("WebSocket 连接已建立")
self.reconnect_delay = 5 # 重置重连延迟
def _handle_message(self, ws, message):
self.on_message(message)
def _handle_error(self, ws, error):
print(f"WebSocket 错误: {error}")
def _handle_close(self, ws, close_status_code, close_msg):
print(f"WebSocket 关闭: {close_status_code} - {close_msg}")
def disconnect(self):
"""安全断开连接"""
self.should_run = False
if self.ws:
self.ws.close()适合谁与不适合谁
| 场景 | 推荐方案 | 说明 |
|---|---|---|
个人
相关资源相关文章 |