WebSocket API v1 命令行客户端 (Python)

#!/usr/bin/env python

# Kraken WebSocket API
#
# 用法: ./krakenws.py feed/endpoint [参数]
# 示例: ./krakenws.py ticker XBT/USD
# 示例: ./krakenws.py spread XBT/USD XBT/EUR ETH/USD LTC/EUR
# 示例: ./krakenws.py book XBT/USD XBT/EUR 10
# 示例: ./krakenws.py openOrders
# 示例: ./krakenws.py ownTrades
# 示例: ./krakenws.py addOrder pair=XBT/EUR type=sell ordertype=limit price=7500 volume=0.125
# 示例: ./krakenws.py cancelOrder txid=OADMSD-7AGC3-IMB46A,OD6VRE-HCSPM-CKORER
#
# 对于账户管理和交易，必须在名为 WS_Token 的纯文本文件中提供有效的 WebSocket 身份验证令牌（来自 REST API GetWebSocketsToken 端点）。

import sys
import signal
from websocket import create_connection

def timeoutfunction(signalnumber, frame):
raise KeyboardInterrupt

signal.signal(signal.SIGALRM, timeoutfunction)

api_status = {"ping"}
api_public = {"trade", "book", "ticker", "spread", "ohlc"}
api_private = {"openOrders", "ownTrades", "balances"}
api_trading = {"addOrder", "editOrder", "cancelOrder", "cancelAll", "cancelAllOrdersAfter"}
api_domain_public = "wss://ws.kraken.com/"
api_domain_private = "wss://ws-auth.kraken.com/"
api_symbols = ""
api_number = 0

if len(sys.argv) < 2:
api_feed = "ping"
else:
api_feed = sys.argv[1]

if api_feed in api_status:
api_domain = api_domain_public
api_data = '{"event":"%(feed)s"}' % {"feed":api_feed}
signal.alarm(3)
elif api_feed in api_public:
if len(sys.argv) < 3:
print("用法: %s feed/endpoint [参数]" % sys.argv[0])
print("示例: %s ticker XBT/USD" % sys.argv[0])
sys.exit(1)
for count in range(2, len(sys.argv)):
if sys.argv[count].isdecimal() == True:
api_number = int(sys.argv[count])
else:
if len(api_symbols) == 0:
api_symbols += sys.argv[count].upper()
else:
api_symbols += '","' + sys.argv[count].upper()
if api_feed == 'book':
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s", "depth":%(depth)d}, "pair":["%(symbols)s"]}' % {"feed":api_feed, "symbols":api_symbols, "depth":api_number if api_number != 0 else 10}
elif api_feed == 'ohlc':
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s", "interval":%(interval)d}, "pair":["%(symbols)s"]}' % {"feed":api_feed, "symbols":api_symbols, "interval":api_number if api_number != 0 else 1}
else:
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s"}, "pair":["%(symbols)s"]}' % {"feed":api_feed, "symbols":api_symbols}
api_domain = api_domain_public
elif api_feed in api_private:
api_domain = api_domain_private
try:
api_token = open("WS_Token").read().strip()
except:
print("WebSocket 身份验证令牌缺失 (WS_Token)")
sys.exit(1)
if len(sys.argv) >= 3:
if api_feed == 'openOrders':
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s", "ratecounter":%(ratecounter)s, "token":"%(token)s"}}' % {"feed":api_feed, "ratecounter":sys.argv[2].split('=')[1], "token":api_token}
elif api_feed == 'ownTrades':
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s", "snapshot":%(snapshot)s, "token":"%(token)s"}}' % {"feed":api_feed, "snapshot":sys.argv[2].split('=')[1], "token":api_token}
else:
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s", "token":"%(token)s"}}' % {"feed":api_feed, "token":api_token}
else:
api_data = '{"event":"subscribe", "subscription":{"name":"%(feed)s", "token":"%(token)s"}}' % {"feed":api_feed, "token":api_token}
elif api_feed in api_trading:
api_domain = api_domain_private
try:
api_token = open("WS_Token").read().strip()
except:
print("WebSocket 身份验证令牌缺失 (WS_Token)")
sys.exit(1)
api_data = '{"event":"%(feed)s", "token":"%(token)s"' % {"feed":api_feed, "token":api_token}
for count in range(2, len(sys.argv)):
if sys.argv[count].split('=')[0] == 'txid':
api_data = api_data + ', "%(name)s":["%(value)s"]' % {"name":sys.argv[count].split('=')[0], "value":sys.argv[count].split('=')[1].replace(',', '","')}
elif sys.argv[count].split('=')[0] == 'reqid':
api_data = api_data + ', "%(name)s":%(value)s' % {"name":sys.argv[count].split('=')[0], "value":sys.argv[count].split('=')[1]}
elif sys.argv[count].split('=')[0] == 'timeout':
api_data = api_data + ', "%(name)s":%(value)s' % {"name":sys.argv[count].split('=')[0], "value":sys.argv[count].split('=')[1]}
else:
api_data = api_data + ', "%(name)s":"%(value)s"' % {"name":sys.argv[count].split('=')[0], "value":sys.argv[count].split('=')[1]}
api_data = api_data + '}'
signal.alarm(3)
else:
print("用法: %s feed/endpoint [参数]" % sys.argv[0])
print("示例: %s ticker XBT/USD" % sys.argv[0])
sys.exit(1)

try:
ws = create_connection(api_domain)
print("WebSocket -> 客户端: %s" % ws.recv())
except Exception as error:
print("WebSocket 连接失败 (%s)" % error)
sys.exit(1)

try:
print("客户端 -> WebSocket: %s" % api_data)
ws.send(api_data)
print("WebSocket -> 客户端: %s" % ws.recv())
except Exception as error:
print("WebSocket 订阅/请求失败 (%s)" % error)
ws.close()
sys.exit(1)

while True:
try:
print("WebSocket -> 客户端: %s" % ws.recv())
except KeyboardInterrupt:
ws.close()
sys.exit(0)
except Exception as error:
print("WebSocket 消息失败 (%s)" % error)
sys.exit(1)

sys.exit(1)