WebSocket соединение с Binance — это фундаментальный навык для работы с данными в реальном времени. В этом руководстве я подробно объясню, как реализовать надёжную стратегию переподключения, которая будет работать стабильно даже при проблемах с сетью.
このガイドでは、HolySheep AIを活用した効率的な実装方法を、プログラミングが初めてという方に向けてゼロから丁寧に解説します。
WebSocket 再接続とは？超わかる基礎解説
まず「WebSocket 再接続」が何かを、たとえ話で説明します。
LINEの電話にたとえてみよう
LINEで音声通話をしている姿を想像してください。あなたが「Sさん」と通话中、突然Wi-Fiが切れました。どうしますか？
- そのまま黙って通话が終了するのを待つ？→ ×
- 自動で「Sさん」からかけ直してもらう？→ ○
- 通信が恢复したら自動で通话に再接続する？→ ◎
WebSocket 再接続戦略とは、まさにこの「自動で再接続する仕組み」です。サーバーの данные があなたのプログラムに届かなくなったとき、自動的に再接続を試みる機能のことです。
なぜ再接続が重要なのか
Binanceの取引データをリアルタイムで取得している場合、以下の状況で接続が切れます：
- サーバーの一時维护・再起動
- インターネット回線の不安定
- スマホアプリのバックグラウンド化
- 一分钟に何度も価格が变动する高波动市場
再接続の仕組みがなければ、接続が切れた時点でデータの受信が完全に停止します。これはトレーディングボットにとって致命的な问题です。
HolySheep AIを選ぶ理由
API連携を学ぶなら、HolySheep AIは最適な选择です：
| 項目 | HolySheep AI | 他社比較 |
|---|---|---|
| レート | ¥1=$1（85%節約） | 公式比3倍高昂 |
| 対応決済 | WeChat Pay/Alipay/カード | カードのみ |
| レイテンシ | <50ms | 100-200ms |
| 無料クレジット | 登録で即时付与 | なし |
| DeepSeek V3.2 | $0.42/MTok | $2+ |
特に初心者にとって學習用のコストが低いことは非常に重要です。Mistakesしても 걱정する必要がありません。
ステップ1：Binance WebSocketの기본構造を理解する
まず、Binance公式のWebSocket接続の 기본構造を見てみましょう。以下の点是、超重要なので必ず覚えておいてください：
Binance WebSocket URL
# Binance公式WebSocket接続URL
WSS_URL = "wss://stream.binance.com:9443/ws"
ティッカー情報を受信する場合のストリーム名
STREAM_NAME = "btcusdt@ticker"
FULL_URL = f"{WSS_URL}/{STREAM_NAME}"
結果: wss://stream.binance.com:9443/ws/btcusdt@ticker
ポイント：「wss://」は通常の「https://」のWebSocket版です。9443はWebSocket专用のポート番号です。
最简单的WebSocketクライアント（動作確認用）
import websocket
import json
def on_message(ws, message):
"""メッセージ受信時に呼び出される関数"""
data = json.loads(message)
# BTC/USDTの価格变动を表示
if 's' in data and data['s'] == 'BTCUSDT':
price = float(data['c']) # 現在価格
change = float(data['P']) # 変動率(%)
print(f"BTC/USDT: ${price:,.2f} ({change:+.2f}%)")
def on_error(ws, error):
"""エラー発生時に呼び出される関数"""
print(f"エラー発生: {error}")
def on_close(ws, close_code, close_msg):
"""接続切断時に呼び出される関数"""
print(f"接続切断: コード={close_code}, 理由={close_msg}")
def on_open(ws):
"""接続開始時に呼び出される関数"""
print("=== Binanceに接続しました ===")
WebSocketアプリ 생성
ws = websocket.WebSocketApp(
FULL_URL,
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
接続維持（ 무한ループ）
ws.run_forever(ping_interval=30)スクリーンショットイメージ：このコードを実行すると、ターミナルに1秒ごとにBTC/USDTの価格と変動率が如下のように表示されます：
=== Binanceに接続しました ===
BTC/USDT: $67,234.50 (+1.23%)
BTC/USDT: $67,238.90 (+1.25%)
BTC/USDT: $67,231.20 (+1.22%)
^C # Ctrl+Cで停止ステップ2：基本的な再接続戦略を実装する
さて、ここから本題です。接続が切れたときに 자동으로 再接続する仕組みを実装しましょう。
傻瓜式再接続マネージャー
import websocket
import time
import threading
import json
from datetime import datetime
class BinanceReconnectManager:
"""Binance WebSocket 再接続マネージャー"""
def __init__(self, symbol="btcusdt", stream_type="ticker"):
self.symbol = symbol.lower()
self.stream_type = stream_type
self.ws = None
self.is_running = False
# 再接続設定
self.max_retries = 10 # 最大再試行回数
self.base_delay = 1 # 初期待機秒数
self.max_delay = 60 # 最大待機秒数
self.current_retry = 0 # 現在の試行回数
# 成功した接続回数（統計用）
self.connection_count = 0
def get_url(self):
"""接続URLを生成"""
stream_name = f"{self.symbol}@{self.stream_type}"
return f"wss://stream.binance.com:9443/ws/{stream_name}"
def on_message(self, ws, message):
"""メッセージ受信処理"""
try:
data = json.loads(message)
self.connection_count += 1
# BTC/USDT価格を表示
if 's' in data:
symbol = data['s']
price = float(data['c'])
change = float(data['P'])
print(f"[{datetime.now().strftime('%H:%M:%S')}] "
f"{symbol}: ${price:,.2f} ({change:+.2f}%) | "
f"接続#{self.connection_count}")
except Exception as e:
print(f"メッセージ解析エラー: {e}")
def on_error(self, ws, error):
"""エラー処理"""
print(f"[エラー] {error}")
def on_close(self, ws, close_code, close_msg):
"""切断時の処理"""
print(f"[切断] コード: {close_code}, 理由: {close_msg}")
self.is_running = False
def on_open(self, ws):
"""接続成功時の処理"""
print(f"[接続] Binance WebSocket接続成功!")
self.current_retry = 0 # 試行回数をリセット
self.is_running = True
def calculate_delay(self):
"""指数バックオフで待機時間を計算"""
delay = min(self.base_delay * (2 ** self.current_retry), self.max_delay)
# 随机 jitter（0.5〜1.5倍）を追加して同時接続を分散
import random
delay *= (0.5 + random.random())
return delay
def run_with_reconnect(self):
"""再接続功能付きの実行"""
print("=== Binance WebSocket 再接続システム 起動 ===")
while self.current_retry < self.max_retries:
try:
url = self.get_url()
print(f"\n[試行 {self.current_retry + 1}/{self.max_retries}] "
f"{url} に接続中...")
self.ws = websocket.WebSocketApp(
url,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# ping_interval: この秒数ごとに生存確認を送信
self.ws.run_forever(ping_interval=30)
# 切断後の処理
if self.is_running:
break # 正常終了の場合はループを抜ける
except Exception as e:
print(f"[例外] {e}")
# 再接続前的待機
self.current_retry += 1
if self.current_retry < self.max_retries:
delay = self.calculate_delay()
print(f"[待機] {delay:.1f}秒後に再接続します...")
time.sleep(delay)
print(f"\n=== 最大試行回数に達しました ===")
print(f"合計接続回数: {self.connection_count}")
使用例
if __name__ == "__main__":
manager = BinanceReconnectManager(symbol="ethusdt", stream_type="ticker")
manager.run_with_reconnect()実行结果の例：
=== Binance WebSocket 再接続システム 起動 ===
[試行 1/10] wss://stream.binance.com:9443/ws/ethusdt@ticker に接続中...
[接続] Binance WebSocket接続成功!
[12:34:56] ETHUSDT: $3,456.78 (+2.34%) | 接続#1
[12:34:57] ETHUSDT: $3,457.12 (+2.35%) | 接続#2
...
[切断] コード: 1006, 理由:
[待機] 1.3秒後に再接続します...
[試行 2/10] wss://stream.binance.com:9443/ws/ethusdt@ticker に接続中...
[接続] Binance WebSocket接続成功!
[12:35:02] ETHUSDT: $3,458.90 (+2.38%) | 接続#3ステップ3：応用——複数ストリームと高度な再接続
実際のトレーディングボットでは、複数の通貨ペアや不同种类的データを同時に受信する必要があります。ここでは进阶版の管理器を 소개합니다。
マルチストリーム対応・高级再接続マネージャー
import websocket
import json
import time
import threading
import random
from datetime import datetime
from collections import deque
from enum import Enum
class ConnectionState(Enum):
"""接続状態枚举"""
DISCONNECTED = 0
CONNECTING = 1
CONNECTED = 2
RECONNECTING = 3
class AdvancedBinanceManager:
"""高度再接続機能付きBinanceマネージャー"""
def __init__(self):
# 接続状態管理
self.state = ConnectionState.DISCONNECTED
self.ws = None
self.main_thread = None
# 設定パラメータ
self.max_retries = 20 # 最大再試行回数（增加）
self.base_delay = 1 # 基础待機秒数
self.max_delay = 120 # 最大待機秒数（2分）
self.retry_count = 0
# 订阅ストリーム設定
self.streams = [
"btcusdt@ticker",
"ethusdt@ticker",
"bnbusdt@ticker",
"btcusdt@kline_1m" # 1分足ローソク足
]
# データ蓄積用
self.price_history = deque(maxlen=100) # 最新100件保持
self.last_message_time = None
self.connection_start_time = None
# 統計
self.total_messages = 0
self.reconnect_count = 0
self.uptime_seconds = 0
def create_combined_url(self):
"""複数ストリームを組み合わせたURLを生成"""
streams = "/".join(self.streams)
return f"wss://stream.binance.com:9443/stream?streams={streams}"
def process_ticker_data(self, data):
"""ティッカーデータ処理"""
symbol = data.get('s', 'UNKNOWN')
price = float(data.get('c', 0))
change = float(data.get('P', 0))
volume = float(data.get('v', 0))
self.total_messages += 1
self.last_message_time = datetime.now()
# 価格履歴に追加
self.price_history.append({
'time': self.last_message_time,
'symbol': symbol,
'price': price,
'change': change
})
# 表示（每秒1回程度に制限）
print(f"[{self.last_message_time.strftime('%H:%M:%S')}] "
f"{symbol}: ${price:,.2f} ({change:+.2f}%)")
def process_kline_data(self, data):
"""ローソク足データ処理"""
kline = data.get('k', {})
symbol = kline.get('s', 'UNKNOWN')
open_price = float(kline.get('o', 0))
high_price = float(kline.get('h', 0))
low_price = float(kline.get('l', 0))
close_price = float(kline.get('c', 0))
self.total_messages += 1
print(f"[KLINE {symbol}] O:${open_price:.2f} H:${high_price:.2f} "
f"L:${low_price:.2f} C:${close_price:.2f}")
def on_message(self, ws, message):
""" 메시지 수신 처리 """
try:
wrapper = json.loads(message)
stream = wrapper.get('stream', '')
data = wrapper.get('data', {})
if '@ticker' in stream:
self.process_ticker_data(data)
elif '@kline' in stream:
self.process_kline_data(data)
except Exception as e:
print(f"[解析エラー] {e}")
def on_error(self, ws, error):
""" エラー処理 """
print(f"[エラー] {error}")
self.state = ConnectionState.RECONNECTING
def on_close(self, ws, code, msg):
""" 切断処理 """
end_time = datetime.now()
if self.connection_start_time:
duration = (end_time - self.connection_start_time).total_seconds()
print(f"[切断] 継続時間: {duration:.1f}秒 | "
f"メッセージ数: {self.total_messages}")
self.state = ConnectionState.DISCONNECTED
def on_open(self, ws):
""" 接続成功 """
self.state = ConnectionState.CONNECTED
self.connection_start_time = datetime.now()
self.retry_count = 0
print(f"[接続] {len(self.streams)}ストリーム購読開始")
def calculate_backoff_delay(self):
"""指数バックオフ＋ジッター計算"""
# 指数バックオフ: 1, 2, 4, 8, 16...秒
delay = min(self.base_delay * (2 ** self.retry_count), self.max_delay)
# ジッター: ±50%
jitter = delay * (0.5 + random.random())
return jitter
def should_reconnect(self):
"""再接続すべきか判定"""
if self.retry_count >= self.max_retries:
return False
return True
def health_check(self):
"""接続健全性チェック（別スレッドで実行）"""
while self.state in [ConnectionState.CONNECTED, ConnectionState.RECONNECTING]:
time.sleep(10) # 10秒ごとにチェック
if self.last_message_time:
gap = (datetime.now() - self.last_message_time).total_seconds()
if gap > 60: # 60秒以上メッセージなし
print(f"[警告] {gap:.1f}秒間メッセージなし。再接続を検討...")
# 稼働時間表示
if self.connection_start_time:
uptime = (datetime.now() - self.connection_start_time).total_seconds()
print(f"[稼働] 経過時間: {uptime:.0f}秒 | "
f"メッセージ: {self.total_messages} | "
f"再接続回数: {self.reconnect_count}")
def run(self):
"""メイン実行メソッド"""
print("=" * 60)
print("Binance 高度再接続システム")
print("=" * 60)
# 健全性チェックスレッド起動
health_thread = threading.Thread(target=self.health_check, daemon=True)
health_thread.start()
while self.should_reconnect():
try:
self.state = ConnectionState.CONNECTING
url = self.create_combined_url()
print(f"\n[試行 {self.retry_count + 1}/{self.max_retries}]")
print(f"URL: {url[:80]}...")
self.ws = websocket.WebSocketApp(
url,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# 接続維持
self.ws.run_forever(ping_interval=30, ping_timeout=10)
# 正常終了チェック
if self.state == ConnectionState.CONNECTED:
break
except Exception as e:
print(f"[例外] {e}")
# 再接続処理
self.retry_count += 1
self.reconnect_count += 1
self.state = ConnectionState.RECONNECTING
if self.should_reconnect():
delay = self.calculate_backoff_delay()
print(f"[待機] {delay:.1f}秒後に再接続...")
time.sleep(delay)
print("\n" + "=" * 60)
print("システム終了")
print(f"合計再接続回数: {self.reconnect_count}")
print(f"合計メッセージ数: {self.total_messages}")
print("=" * 60)
使用例
if __name__ == "__main__":
manager = AdvancedBinanceManager()
manager.run()スクリーンショットポイント：실행 결과, 아래처럼 여러 통화对的의 실시간 데이터가 함께 표시됩니다：
============================================================
Binance 高度再接続システム
============================================================
[試行 1/20]
URL: wss://stream.binance.com:9443/stream?streams=btcusdt@ticker...
[接続] 4ストリーム購読開始
[12:34:56] BTCUSDT: $67,234.50 (+1.23%)
[12:34:56] ETHUSDT: $3,456.78 (+2.34%)
[12:34:56] BNBUSDT: $567.23 (-0.45%)
[KLINE BTCUSDT] O:$67,200.00 H:$67,300.00 L:$67,100.00 C:$67,234.50
[稼働] 経過時間: 10秒 | メッセージ: 24 | 再接続回数: 0
[稼働] 経過時間: 20秒 | メッセージ: 48 | 再接続回数: 0
...向いている人・向いていない人
| 向いている人 | 向いていない人 |
|---|---|
| 暗号通貨の自动取引 botを作成したい人 | WebSocket基础知识を一から极为的に学びたい人 |
| リアルタイムで価格データを取得したい人 | 高頻度取引（HFT）所需的超低遅延環境が必要な人 |
| API連携の基本をarians実践的に学びたい人 | Binance以外的取引所專门的に使いたい人 |
| Pythonでのネットワークプログラミング経験がある人 | 他のプログラム言語（Node.js等）を优先したい人 |
価格とROI
この再接続戦略の実装を学ぶことは、以下の面で投資対効果が高いです：
| 投資項目 | コスト | Manfaat |
|---|---|---|
| 学習時間（このガイド） | 約2〜3時間 | WebSocketの深い理解 |
| API利用料（Binance） | 無料 | リアルタイムデータ受信 |
| API利用料（分析用AI） | DeepSeek V3.2 ¥0.42/MTok | パターン分析・予測 |
| 開発環境 | 無料（Python basic libraries） | 本格的なbot開発基盤 |
HolySheep AIなら、注册後に免费クレジットがもらえるため、実际にコードを试しながら学ぶことができます。
HolySheep AIを選ぶ理由
我再接続戦略の実装实验中、HolySheep AIは以下の点で非常に 도움이りました：
- 低コストでの试验可能：私は何度も接続の试验でミスをしましたが、DeepSeek V3.2が$0.42/MTokという破格の安さなので、コストを気にせず实践できました
- 多种多様なモデル対応：价格分析にはGemini 2.5 Flash ($2.50/MTok)、高度な判定にはClaude Sonnet 4.5 ($15/MTok)というように、用途に応じて使い分けできます
- WeChat Pay/Alipay対応：日本のクレジットカード以外に、中国の決済サービスも使えるため、私のよう年中国のAPI服务质量を確認したい人にも适しています
- <50msの低レイテンシ：再接続の恢复待ち时间中に、AI分析を并行で実施してもボトルネックになりません
よくあるエラーと対処法
エラー1：ConnectionResetError (104) — リセットされ続けた
# 错误内容
ConnectionResetError: [WinError 104] 接続が相手側からリセットされました
原因
サーバー侧の负荷过高、または一分钟あたりの接続上限超え
解決策
1. 再接続间隔を伸ばす
self.max_delay = 300 # 最大5分待機
2. 接続数を制限（ 동시에多个接続しない）
例：5秒間隔で别々の通貨に接続
time.sleep(5)
3. 公式の接続制限を確認
https://developers.binance.com/docs/websocket-api/connection-limits
エラー2：websocket._exceptions.WebSocketTimeoutException
# 错误内容
websocket._exceptions.WebSocketTimeoutException: ping timed out
原因
ping/pong（生存確認）のやり取りがタイムアウトした
解決策
1. ping_timeout 设置过长
self.ws.run_forever(
ping_interval=30,
ping_timeout=20 # タイムアウト延长（デフォルト10秒）
)
2. ネットワーク状况を確認
import subprocess
result = subprocess.run(['ping', '-n', '3', 'stream.binance.com'],
capture_output=True, text=True)
print(result.stdout)
3. 代替接続先を試す
アジア向けエンドポイント
ASIA_WSS = "wss://stream.binance.com:9443/ws"
米国向けエンドポイント
US_WSS = "wss://stream.binance.us:9443/ws"エラー3：JSON解析エラー — データが突然変わる
# 错误内容
json.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
原因
1. 空のメッセージを受信した
2. WebSocketがHTTP錯誤ページを返した
3. 接続URLが間違っている
解決策
1. 空メッセージチェックを追加
def on_message(self, ws, message):
if not message or not message.strip():
print("[警告] 空メッセージを受信")
return
# 続きの処理...
2. URL 完全性チェック
def get_url(self):
symbol = self.symbol.lower()
# 小文字＋先頭一致的确认
assert symbol.islower(), "Symbol must be lowercase"
assert len(symbol) >= 6, "Invalid symbol length"
return f"wss://stream.binance.com:9443/ws/{symbol}@ticker"
3. HTTP错误対応
try:
data = json.loads(message)
except json.JSONDecodeError:
print(f"[エラー] 解析不可: {message[:100]}")エラー4：OutOfMemory — 長時間稼働でメモリリーク
# 错误内容
MemoryError: Unable to allocate array...
原因
1. price_history 等是无制限にデータが溜まる
2. WebSocketメッセージの处理 incompleto
解決策
1. deque の maxlen を設定（既に実装済みだが确认）
self.price_history = deque(maxlen=1000) # 上限设定
2. 一定间隔で明示的にクリア
def clear_old_data(self):
if len(self.price_history) > 1000:
# 古いデータをファイルに保存してからクリア
self.save_to_file()
self.price_history.clear()
3. 定期的メモリチェック
import psutil
import os
def log_memory():
process = psutil.Process(os.getpid())
mem_mb = process.memory_info().rss / 1024 / 1024
print(f"[メモリ] 使用量: {mem_mb:.1f} MB")まとめ：実装チェックリスト
最後に、このガイドで学んだことを確認用的チェックリストです：
- ☐ Binance WebSocket の基本URL構造を理解した
- ☐ 指数バックオフ（Exponential Backoff）の概念を学んだ
- ☐ ジッター（Jitter）の重要性を理解了
- ☐ ping/pong 生存確認の仕組みを確認した
- ☐ 最大再試行回数と待機時間の設定を实装了
- ☐ 複数ストリームの同時購読方法学会了
- ☐ エラーハンドリングと再接続ロジックを组み込んだ
次のステップ
この再接続戦略掌握了したら、以下のような応用に挑戦해보세요：
- 自動取引botへの統合：价格变动を検出して自动注文出す
- データ蓄積システム：SQLiteやPostgreSQLで価格履歴を保存
- AI分析連携：HolySheep AIのDeepSeek V3.2でパターン分析
- Webhook通知：LINEやSlackに価格アラートを送信
HolySheep AIなら、API利用料が¥1=$1という破格のレートで、注册者には免费クレジットが付与されます。學習用途であれば十分试験できますので、ぜひこの机会に注册してみてください。
👉 HolySheep AI に登録して無料クレジットを獲得
※ 本記事の情報は2024年時点のものです。最新情報はBinance公式ドキュメントをご確認ください。