Khoảng 3 giờ sáng ngày 15/03/2024, hệ thống giao dịch của tôi đột nhiên ngừng hoạt động. Logs hiển thị ConnectionError: timeout after 30000ms — WebSocket OKX đã bị ngắt kết nối mà không có cơ chế tự phục hồi. Kết quả? Bỏ lỡ 3 lệnh giao dịch quan trọng với tổng thiệt hại khoảng 2,400 USD. Đó là bài học đắt giá nhất tôi từng có về việc không xử lý đúng cách cơ chế reconnect WebSocket.
WebSocket là gì và Tại sao cần cơ chế Reconnect?
WebSocket là giao thức kết nối liên tục hai chiều, cho phép server gửi dữ liệu đến client mà không cần client yêu cầu. Trong bối cảnh API OKX, WebSocket được sử dụng để:
- Nhận dữ liệu thị trường real-time (ticker, orderbook, trades)
- Nhận thông báo về trạng thái lệnh
- Cập nhật số dư tài khoản tức thì
- Webhooks cho các sự kiện quan trọng
Vấn đề: Kết nối WebSocket có thể bị ngắt bất cứ lúc nào do:
- Mất kết nối mạng (timeout)
- Server OKX restart hoặc maintenance
- Load balancer ngắt kết nối idle
- Token xác thực hết hạn (401 Unauthorized)
- CGI gateway timeout
Kiến trúc Kết nối OKX WebSocket
Cấu trúc Endpoint
# OKX WebSocket Endpoint chính thức
WSS_ENDPOINT = "wss://ws.okx.com:8443/ws/v5/public"
Endpoint cho private channel (cần xác thực)
WSS_PRIVATE_ENDPOINT = "wss://ws.okx.com:8443/ws/v5/private"
Các cổng alternative
WSS_ENDPOINT_V2 = "wss://ws.okx.com:8443/ws/v5/public" # Same endpoint nhưng khác region
Mô hình subscription message
# Python - OKX WebSocket Client với Exponential Backoff Reconnect
import websocket
import threading
import time
import json
import hashlib
import hmac
from typing import Callable, Optional
from datetime import datetime
class OKXWebSocketClient:
def __init__(
self,
api_key: str,
api_secret: str,
passphrase: str,
testnet: bool = False
):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.testnet = testnet
# Cấu hình reconnect
self.max_reconnect_attempts = 10
self.base_delay = 1 # Giây
self.max_delay = 60 # Giây
self.ping_interval = 20 # OKX yêu cầu ping mỗi 20-30 giây
self.ping_timeout = 5
self.ws = None
self.is_running = False
self.subscriptions = {}
self.reconnect_count = 0
def _get_signature(self, timestamp: str) -> tuple[str, str]:
"""Tạo chữ ký xác thực cho private channel"""
message = timestamp + "GET" + "/users/self/verify"
mac = hmac.new(
self.api_secret.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
sign = mac.digest()
sign_b64 = sign.decode('utf-8') if isinstance(sign, bytes) else sign
return sign_b64, self.passphrase
def _on_message(self, ws, message):
"""Xử lý tin nhắn nhận được"""
try:
data = json.loads(message)
# Xử lý event login thành công
if data.get('event') == 'login':
if data.get('code') == '0':
print(f"[{datetime.now()}] ✓ Login OKX WebSocket thành công")
# Resubscribe các channel đã đăng ký
self._resubscribe_all()
else:
print(f"[{datetime.now()}] ✗ Login thất bại: {data.get('msg')}")
return
# Xử lý subscription response
if data.get('event') == 'subscribe':
if data.get('code') == '0':
print(f"[{datetime.now()}] ✓ Subscribe {data.get('arg', {}).get('channel')} thành công")
else:
print(f"[{datetime.now()}] ✗ Subscribe lỗi: {data.get('msg')}")
return
# Xử lý dữ liệu thị trường
if 'data' in data:
channel = data.get('arg', {}).get('channel')
for item in data['data']:
self._process_data(channel, item)
except json.JSONDecodeError as e:
print(f"Lỗi parse JSON: {e}")
except Exception as e:
print(f"Lỗi xử lý message: {e}")
def _on_error(self, ws, error):
"""Xử lý lỗi WebSocket"""
error_type = type(error).__name__
print(f"[{datetime.now()}] ⚠ Lỗi WebSocket: {error_type} - {error}")
# Phân loại lỗi để xử lý phù hợp
if "timeout" in str(error).lower():
print("→ Lỗi timeout - cần reconnect")
elif "401" in str(error) or "unauthorized" in str(error).lower():
print("→ Lỗi xác thực - kiểm tra API key")
elif "connection" in str(error).lower():
print("→ Lỗi kết nối - kiểm tra mạng")
def _on_close(self, ws, close_status_code, close_msg):
"""Xử lý khi WebSocket đóng"""
print(f"[{datetime.now()}] WebSocket đóng: code={close_status_code}, msg={close_msg}")
self.is_running = False
if close_status_code == 1000:
print("→ Đóng bình thường (normal closure)")
else:
print("→ Đóng bất thường - sẽ reconnect")
self._schedule_reconnect()
def _on_open(self, ws):
"""Xử lý khi WebSocket mở"""
print(f"[{datetime.now()}] ✓ Kết nối WebSocket OKX thành công")
self.is_running = True
self.reconnect_count = 0
# Login nếu là private channel
if hasattr(self, 'api_key') and self.api_key:
self._login()
def _login(self):
"""Đăng nhập vào private channel"""
timestamp = str(time.time())
sign, passphrase = self._get_signature(timestamp)
login_data = {
"op": "login",
"args": [{
"apiKey": self.api_key,
"passphrase": passphrase,
"timestamp": timestamp,
"sign": sign
}]
}
self.ws.send(json.dumps(login_data))
def _resubscribe_all(self):
"""Đăng ký lại tất cả subscription đã lưu"""
for channel, instId in self.subscriptions.items():
self.subscribe(channel, instId)
def subscribe(self, channel: str, instId: str = None):
"""Đăng ký một channel"""
arg = {"channel": channel, "instId": instId or "BTC-USDT-SWAP"}
sub_data = {
"op": "subscribe",
"args": [arg]
}
# Lưu lại subscription
self.subscriptions[channel] = instId
if self.ws and self.is_running:
self.ws.send(json.dumps(sub_data))
print(f"[{datetime.now()}] Đã gửi subscribe: {channel}")
def _process_data(self, channel: str, data: dict):
"""Xử lý dữ liệu - implement theo nhu cầu"""
# Override this method in subclass
pass
def _schedule_reconnect(self):
"""Lên lịch reconnect với Exponential Backoff"""
if self.reconnect_count >= self.max_reconnect_attempts:
print(f"[{datetime.now()}] Đã đạt max reconnect attempts ({self.max_reconnect_attempts})")
return
# Tính delay với exponential backoff
delay = min(
self.base_delay * (2 ** self.reconnect_count),
self.max_delay
)
# Thêm jitter ngẫu nhiên ±20%
jitter = delay * 0.2 * (2 * (time.time() % 1) - 1)
actual_delay = delay + jitter
self.reconnect_count += 1
print(f"[{datetime.now()}] Reconnect lần {self.reconnect_count}/{self.max_reconnect_attempts} sau {actual_delay:.2f}s")
threading.Timer(actual_delay, self.connect).start()
def connect(self):
"""Kết nối WebSocket"""
self.ws = websocket.WebSocketApp(
self.WSS_PRIVATE_ENDPOINT,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
# Chạy WebSocket với ping/pong tự động
self.ws.run_forever(
ping_interval=self.ping_interval,
ping_timeout=self.ping_timeout,
reconnect=False # Disable default reconnect - dùng custom logic
)
def start(self):
"""Bắt đầu WebSocket client"""
self.is_running = True
self.connect()
def stop(self):
"""Dừng WebSocket client"""
self.is_running = False
if self.ws:
self.ws.close(code=1000, reason="Client stopped")
=== SỬ DỤNG ===
if __name__ == "__main__":
client = OKXWebSocketClient(
api_key="your_api_key",
api_secret="your_api_secret",
passphrase="your_passphrase"
)
# Đăng ký các channel
client.subscribe("tickers", "BTC-USDT-SWAP")
client.subscribe("orders", "BTC-USDT-SWAP")
# Chạy client
client.start()
Chiến lược Exponential Backoff - Best Practice
Exponential Backoff là chiến lược tăng thời gian chờ theo cấp số nhân sau mỗi lần thử kết nối lại, kết hợp với jitter ngẫu nhiên để tránh thundering herd problem.
"""
OKX WebSocket Reconnect với Circuit Breaker Pattern
"""
import time
import random
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CircuitState(Enum):
CLOSED = "closed" # Bình thường
OPEN = "open" # Ngắt mạch - không reconnect
HALF_OPEN = "half_open" # Thử reconnect
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Số lần thất bại để mở circuit
success_threshold: int = 2 # Số lần thành công để đóng circuit
timeout: float = 30.0 # Giây trước khi thử HALF_OPEN
max_retry: int = 10
class CircuitBreaker:
"""Circuit Breaker để tránh reconnect liên tục khi server có vấn đề"""
def __init__(self, config: CircuitBreakerConfig = None):
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.retry_count = 0
def can_attempt(self) -> bool:
"""Kiểm tra có được phép thử reconnect không"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self._is_timeout_elapsed():
self.state = CircuitState.HALF_OPEN
logger.info("Circuit breaker chuyển sang HALF_OPEN")
return True
return False
# HALF_OPEN - cho phép 1 request thử
return True
def _is_timeout_elapsed(self) -> bool:
"""Kiểm tra đã hết timeout chưa"""
if self.last_failure_time is None:
return True
return (time.time() - self.last_failure_time) >= self.config.timeout
def record_success(self):
"""Ghi nhận thành công"""
self.failure_count = 0
self.retry_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
logger.info("Circuit breaker đã đóng - kết nối ổn định")
elif self.state == CircuitState.OPEN:
self.state = CircuitState.CLOSED
def record_failure(self):
"""Ghi nhận thất bại"""
self.failure_count += 1
self.last_failure_time = time.time()
self.retry_count += 1
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
logger.warning("HALF_OPEN thất bại - Circuit breaker mở lại")
elif (self.failure_count >= self.config.failure_threshold and
self.retry_count >= self.config.max_retry):
self.state = CircuitState.OPEN
logger.error(f"Circuit breaker MỞ - Quá nhiều thất bại liên tục")
def get_delay(self) -> float:
"""Tính toán thời gian chờ với Exponential Backoff + Jitter"""
# Base delay: 1s
base_delay = 1.0
# Max delay: 60s
max_delay = 60.0
# Max backoff exponent: 2^6 = 64
max_exponent = 6
# Tính exponential delay
exponent = min(self.retry_count, max_exponent)
delay = base_delay * (2 ** exponent)
# Thêm jitter (±25%)
jitter_range = delay * 0.25
jitter = random.uniform(-jitter_range, jitter_range)
final_delay = min(delay + jitter, max_delay)
return max(0.1, final_delay) # Tối thiểu 0.1s
class OKXReconnectManager:
"""Manager toàn diện cho OKX WebSocket reconnection"""
def __init__(self):
self.circuit_breaker = CircuitBreaker()
self.subscriptions = set()
self.heartbeat_interval = 25 # OKX khuyến nghị 20-30s
self.last_heartbeat: Optional[float] = None
def should_reconnect(self, error: Exception) -> tuple[bool, str]:
"""
Quyết định có nên reconnect không
Returns: (should_reconnect, reason)
"""
error_msg = str(error).lower()
# Lỗi nghiêm trọng - không reconnect
if "401" in error_msg or "unauthorized" in error_msg:
return False, "Lỗi xác thực - cần kiểm tra API key"
if "403" in error_msg or "forbidden" in error_msg:
return False, "Bị cấm truy cập - kiểm tra quyền API"
# Lỗi có thể khắc phục bằng reconnect
if any(keyword in error_msg for keyword in
["timeout", "connection", "network", "reset", "refused"]):
if not self.circuit_breaker.can_attempt():
return False, "Circuit breaker OPEN - đợi timeout"
return True, "Lỗi kết nối - sẽ reconnect"
# Lỗi không xác định - thử reconnect
return True, "Lỗi không xác định - thử reconnect"
def get_reconnect_delay(self) -> float:
"""Lấy thời gian chờ trước khi reconnect"""
return self.circuit_breaker.get_delay()
def on_connect_success(self):
"""Xử lý khi kết nối thành công"""
self.circuit_breaker.record_success()
self.last_heartbeat = time.time()
def on_connect_failure(self, error: Exception):
"""Xử lý khi kết nối thất bại"""
self.circuit_breaker.record_failure()
logger.warning(f"Kết nối thất bại: {error}")
def is_healthy(self) -> bool:
"""Kiểm tra trạng thái kết nối"""
if self.circuit_breaker.state == CircuitState.OPEN:
return False
if self.last_heartbeat:
# Không nhận heartbeat trong 2 chu kỳ = unhealthy
time_since_heartbeat = time.time() - self.last_heartbeat
if time_since_heartbeat > self.heartbeat_interval * 2:
return False
return True
=== Ví dụ sử dụng trong main loop ===
def main():
reconnect_mgr = OKXReconnectManager()
while True:
if not reconnect_mgr.should_reconnect(Exception("timeout"))[0]:
print("Không thể reconnect - kiểm tra trạng thái")
time.sleep(5)
continue
delay = reconnect_mgr.get_reconnect_delay()
print(f"Chờ {delay:.2f}s trước khi reconnect...")
time.sleep(delay)
try:
# Thử kết nối OKX WebSocket ở đây
print("Đang kết nối...")
# ws.connect(...)
reconnect_mgr.on_connect_success()
break
except Exception as e:
reconnect_mgr.on_connect_failure(e)
if __name__ == "__main__":
main()
Các Loại Lỗi Phổ Biến và Mã Xử Lý
Dưới đây là bảng tổng hợp các lỗi OKX WebSocket thường gặp với mã lỗi và cách xử lý:
| Mã Lỗi | Mô Tả | Nguyên Nhân | Cách Xử Lý |
|---|---|---|---|
| 401 | Unauthorized | API key sai, hết hạn, hoặc thiếu signature | Kiểm tra lại API key, tạo key mới |
| 28001 | System error | Server OKX bảo trì hoặc quá tải | Chờ 30s-5p, exponential backoff |
| 30001 | Illegal parameter | Tham số subscription không đúng | Kiểm tra format channel/instId |
| 30002 | Illegal request | Request vượt quá giới hạn | Giảm số lượng subscription |
| 30003 | Channel not found | Channel không tồn tại | Kiểm tra tên channel trong docs |
| 30004 | Login required | Channel private chưa login | Gọi login trước khi subscribe |
| 30005 | Login failed | Chữ ký xác thực sai | Kiểm tra thuật toán HMAC |
| 30007 | Too many connections | Vượt quá số kết nối cho phép | Đóng kết nối cũ, thử lại |
| 1001 | Disconnect | Server ngắt kết nối | Auto reconnect với backoff |
| 1005 | No login | Chưa login trước khi subscribe private | Implement login sequence |
Chiến lược Subscription Quản Lý State
"""
OKX WebSocket - Subscription Manager với State Recovery
"""
import json
import time
from typing import Dict, Set, List, Any
from dataclasses import dataclass, field
from enum import Enum
import sqlite3
class SubscriptionType(Enum):
PUBLIC = "public"
PRIVATE = "private"
@dataclass
class Subscription:
channel: str
inst_id: str = ""
sub_type: SubscriptionType = SubscriptionType.PUBLIC
subscribed_at: float = field(default_factory=time.time)
last_data_at: float = field(default_factory=time.time)
data_count: int = 0
class SubscriptionManager:
"""Quản lý subscription với persistence và recovery"""
def __init__(self, db_path: str = "okx_subscriptions.db"):
self.subscriptions: Dict[str, Subscription] = {}
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Khởi tạo database SQLite để lưu state"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS subscriptions (
key TEXT PRIMARY KEY,
channel TEXT,
inst_id TEXT,
sub_type TEXT,
subscribed_at REAL,
last_data_at REAL,
data_count INTEGER
)
""")
conn.commit()
conn.close()
def _make_key(self, channel: str, inst_id: str = "") -> str:
return f"{channel}:{inst_id}"
def subscribe(self, channel: str, inst_id: str = "",
sub_type: SubscriptionType = SubscriptionType.PUBLIC) -> dict:
"""Đăng ký subscription mới"""
key = self._make_key(channel, inst_id)
sub = Subscription(
channel=channel,
inst_id=inst_id,
sub_type=sub_type
)
self.subscriptions[key] = sub
# Lưu vào database
self._save_subscription(key, sub)
# Trả về message subscription
return {
"op": "subscribe",
"args": [{
"channel": channel,
"instId": inst_id
}]
}
def unsubscribe(self, channel: str, inst_id: str = ""):
"""Hủy subscription"""
key = self._make_key(channel, inst_id)
if key in self.subscriptions:
del self.subscriptions[key]
self._delete_subscription(key)
return {"op": "unsubscribe", "args": [{"channel": channel, "instId": inst_id}]}
return None
def update_data_received(self, channel: str, inst_id: str = ""):
"""Cập nhật timestamp khi nhận được dữ liệu"""
key = self._make_key(channel, inst_id)
if key in self.subscriptions:
sub = self.subscriptions[key]
sub.last_data_at = time.time()
sub.data_count += 1
self._save_subscription(key, sub)
def get_stale_subscriptions(self, timeout: float = 120) -> List[Subscription]:
"""Lấy các subscription không có dữ liệu trong timeout"""
stale = []
now = time.time()
for sub in self.subscriptions.values():
if now - sub.last_data_at > timeout:
stale.append(sub)
return stale
def resubscribe_all(self) -> List[dict]:
"""Trả về tất cả subscription để resubscribe"""
messages = []
for sub in self.subscriptions.values():
messages.append({
"op": "subscribe",
"args": [{
"channel": sub.channel,
"instId": sub.inst_id
}]
})
return messages
def get_subscription_status(self) -> Dict[str, Any]:
"""Lấy trạng thái tất cả subscriptions"""
total = len(self.subscriptions)
active = sum(1 for s in self.subscriptions.values()
if time.time() - s.last_data_at < 120)
return {
"total": total,
"active": active,
"stale": total - active,
"subscriptions": {
key: {
"channel": sub.channel,
"inst_id": sub.inst_id,
"data_count": sub.data_count,
"last_data": sub.last_data_at
}
for key, sub in self.subscriptions.items()
}
}
def _save_subscription(self, key: str, sub: Subscription):
"""Lưu subscription vào database"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT OR REPLACE INTO subscriptions
(key, channel, inst_id, sub_type, subscribed_at, last_data_at, data_count)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (key, sub.channel, sub.inst_id, sub.sub_type.value,
sub.subscribed_at, sub.last_data_at, sub.data_count))
conn.commit()
conn.close()
def _delete_subscription(self, key: str):
"""Xóa subscription khỏi database"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("DELETE FROM subscriptions WHERE key = ?", (key,))
conn.commit()
conn.close()
def load_from_db(self):
"""Load subscriptions từ database khi restart"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("SELECT * FROM subscriptions")
for row in cursor.fetchall():
key, channel, inst_id, sub_type, sub_at, last_data, count = row
sub = Subscription(
channel=channel,
inst_id=inst_id,
sub_type=SubscriptionType(sub_type),
subscribed_at=sub_at,
last_data_at=last_data,
data_count=count
)
self.subscriptions[key] = sub
conn.close()
print(f"Đã load {len(self.subscriptions)} subscriptions từ database")
=== Integration với WebSocket Client ===
class ResilientOKXClient:
"""OKX Client với đầy đủ cơ chế resilience"""
def __init__(self, api_key: str, api_secret: str, passphrase: str):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.sub_manager = SubscriptionManager()
self.reconnect_mgr = OKXReconnectManager()
self.ws = None
# Load subscriptions cũ từ database
self.sub_manager.load_from_db()
def on_message(self, ws, message):
"""Xử lý message với cập nhật state"""
data = json.loads(message)
# Cập nhật last_data_at cho subscription
if 'arg' in data:
channel = data['arg'].get('channel')
inst_id = data['arg'].get('instId', '')
self.sub_manager.update_data_received(channel, inst_id)
# Xử lý business logic ở đây
# ...
def on_open(self, ws):
"""Khi kết nối mở - resubscribe tất cả"""
self.reconnect_mgr.on_connect_success()
# Resubscribe tất cả subscriptions cũ
for msg in self.sub_manager.resubscribe_all():
ws.send(json.dumps(msg))
print(f"Resubscribed: {msg}")
def on_close(self, ws, code, reason):
"""Khi kết nối đóng"""
should_reconnect, reason_text = self.reconnect_mgr.should_reconnect(
Exception(reason)
)
if should_reconnect:
delay = self.reconnect_mgr.get_reconnect_delay()
print(f"Sẽ reconnect sau {delay:.2f}s: {reason_text}")
# schedule reconnect...
else:
print(f"Không reconnect: {reason_text}")
def check_stale_subscriptions(self):
"""Kiểm tra và resubscribe các channel không hoạt động"""
stale = self.sub_manager.get_stale_subscriptions(timeout=120)
if stale:
print(f"Tìm thấy {len(stale)} subscriptions không hoạt động")
# Resubscribe
for sub in stale:
self.ws.send(json.dumps({
"op": "subscribe",
"args": [{"channel": sub.channel, "instId": sub.inst_id}]
}))
Best Practices từ Kinh Nghiệm Thực Chiến
1. Ping/Pong Heartbeat
OKX yêu cầu client gửi ping mỗi 20-30 giây để duy trì kết nối. Nếu không ping, server sẽ tự đóng connection sau ~60s.
# Cấu hình WebSocketApp với ping/pong
ws = websocket.WebSocketApp(
url,
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
Chạy với ping tự động
ws.run_forever(
ping_interval=20, # Ping mỗi 20 giây
ping_timeout=5, # Timeout cho pong response
keep_running=True
)
Hoặc ping thủ công trong on_open
def on_open(ws):
def send_ping():
while ws.sock and ws.sock.connected:
ws.ping()
time.sleep(20)
threading.Thread(target=send_ping, daemon=True).start()
2. Connection Pooling
OKX giới hạn số kết nối WebSocket đồng thời. Với tài khoản thường là 25 kết nối, VIP có thể lên đến 100.
- Dùng chung 1 connection cho nhiều subscriptions
- Group các channel cùng loại vào 1 subscription request
- Đóng connection khi không cần thiết
3. Graceful Shutdown
import signal
import sys
class GracefulShutdown:
def __init__(self, ws_app):
self.ws_app = ws_app
signal.signal(signal.SIGINT, self.handle_shutdown)
signal.signal(signal.SIGTERM, self.handle_shutdown)
def handle_shutdown(self, signum, frame):
print("Nhận tín hiệu shutdown - đóng connection...")
self.ws_app.close(code=1000, reason="Graceful shutdown")
sys.exit(0)
Lỗi thường gặp và cách khắc phục
1. Lỗi "Connection timeout after 30000ms"
Nguyên nhân: Firewall chặn kết nối, proxy không hỗ trợ WSS, hoặc mạng không ổn định.
# Cách khắc phục:
1. Kiểm tra proxy
import os
os.environ['HTTP_PROXY'] = 'http://proxy.example.com:8080'
os.environ['HTTPS_PROXY'] = 'http://proxy.example.com:8080'
2. Tăng timeout
ws.run_forever(
ping_interval=20,
ping_timeout=10, # Tăng t�