Websocket 在 ~2 小时后获得连接柜，尽管 ping 了

Question

Program从gdax websocket获取数据并将其存储到sqlite表中。 该程序在前 2-2,5 小时内运行良好，之后我从调试器收到连接关闭错误。女巫很奇怪，因为我相信程序能够在没有 debbuger 帮助的情况下捕获此错误。

这是我代码的精简版本：

import time
import datetime
import json
import sqlite3
import urllib3
import timeit
import urllib.request
import requests
from websocket import create_connection
conn=sqlite3.connect(":memory:")
c = conn.cursor()  
c.execute("CREATE TABLE `Ticker` (  `ID`    INTEGER PRIMARY KEY AUTOINCREMENT,  `Type`  TEXT,   `Sequence`  INTEGER,    `Product_id`    TEXT,   `Price` INTEGER,    `Open_24h`  INTEGER,    `Volume_24h`    INTEGER,    `Low_24h`   INTEGER,    `High_24h`  INTEGER,    `Volume_30d`    INTEGER,    `Best_bid`  INTEGER,    `Best_ask`  INTEGER,    `side`  TEXT,   `time`  TEXT,   `trade_id`  INTEGER,    `last_size` INTEGER)")
class Websocket():
def __init__(self, wsurl="wss://ws-feed.gdax.com", ws=None, agi1=2, ping_start=0,produkty=['ETH-EUR', 'LTC-EUR', 'BTC-EUR'], newdict={}):
self.wsurl = wsurl
self.ws = None
self.agi1=agi1
self.ping_start=ping_start
self.produkty=produkty
self.newdict=newdict
print(self.wsurl)
print(self.produkty)
def KonektorWebsocketTicker(self):
x=0
b=['type', 'sequence', 'product_id', 'price', 'open_24h', 'volume_24h', 'low_24h', 'high_24h', 'volume_30d', 'best_bid', 'best_ask', 'side', 'time', 'trade_id', 'last_size']
i=0
dane=json.loads(self.ws.recv())
typtranzakcji=dane.get('type',None)      
if typtranzakcji=='error':                                          #obsługa błędu... mam nadzieje
print(json.dumps(dane, indent=4, sort_keys=True))
if typtranzakcji=='ticker':
while i<len(b):
a = eval("dane.get('" + b[i] + "', None)")
self.newdict[b[i]]=a
i=i+1
c.execute("INSERT INTO Ticker(type, sequence, product_id, price, open_24h, volume_24h, low_24h, high_24h, volume_30d, best_bid, best_ask, side, time, trade_id, last_size) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (self.newdict['type'], self.newdict['sequence'], self.newdict['product_id'], self.newdict['price'], self.newdict['open_24h'], self.newdict['volume_24h'], self.newdict['low_24h'], self.newdict['high_24h'], self.newdict['volume_30d'], self.newdict['best_bid'], self.newdict['best_ask'], self.newdict['side'], self.newdict['time'], self.newdict['trade_id'], self.newdict['last_size']))
def Polacz(self):
self.ws=create_connection(self.wsurl)
if self.agi1 == 0:
kanaly =None
elif self.agi1 == 1:
kanaly= "heartbeat"
elif self.agi1 == 2:
kanaly=["ticker"]
elif self.agi1 == 3:
kanaly="level2"
else:
print('ERROR:WRONG ARGUMENT! (arg1)=', self.agi1)
print(kanaly)
if kanaly is None:
self.ws.send(json.dumps({'type': 'subscribe', 'product_ids': self.produkty}))
else:
self.ws.send(json.dumps({'type': 'subscribe', 'product_ids': self.produkty, 'channels': [{"name": kanaly, 'product_ids': self.produkty,}]}))    #wysłanie subskrybcji
while True:
try:
if (time.time() - self.ping_start) >= 20:
self.ws.ping("ping")
print('%s ping' % time.ctime())
self.ping_start = time.time()
except ValueError as e:
print('{} Error :{}'.format(time.ctime(), e))
except Exception as e:
print('{} Error :{}'.format(time.ctime(), e))
else:
print('%s  no ping' % time.ctime())
dane=json.loads(self.ws.recv())

if self.agi1==2:
self.KonektorWebsocketTicker()
conn.commit()
time.sleep(1)
class Autotrader():
a='1'
b='1'
if a=="1":
produkty=["BTC-EUR"]
if b=='1':
webs=Websocket(produkty=produkty)
webs.Polacz()  

有人可以向我解释这里的问题在哪里，因为 debbuger 不是很有帮助，随机更改代码并等待 +2 小时以获取有关错误的信息不是解决此问题的好方法

Answer 1

您打算在哪个交易所进行交易？ 通常github上有一些不错的库，用于较大的交换:)

我相信问题是"self.ws.recv(("，它会停止执行，直到收到消息。这会阻止您访问"self.ws.ping("ping"("，它使您的 websocket 保持活动状态。

由于我正在研究类似的东西，我还没有特定的解决方案，但是引用线程，非阻塞websocket客户端的话，似乎keepalive应该在单独的线程中运行，这样它就不会被阻止：

"总会有一个线程专门用于侦听套接字(在这种情况下，主线程进入run_forever内部的循环等待消息(。如果你想做一些其他的事情，你需要另一个线程。