ওয়েবসকেট API v1 কমান্ড লাইন ক্লায়েন্ট (পাইথন)

#!/usr/bin/env python

# ক্র্যাকেন ওয়েবসকেট API
#
# ব্যবহার: ./krakenws.py ফিড/এন্ডপয়েন্ট [প্যারামিটার]
# উদাহরণ: ./krakenws.py টিকার XBT/USD
# উদাহরণ: ./krakenws.py স্প্রেড XBT/USD XBT/EUR ETH/USD LTC/EUR
# উদাহরণ: ./krakenws.py বুক XBT/USD XBT/EUR 10
# উদাহরণ: ./krakenws.py ওপেনঅর্ডার
# উদাহরণ: ./krakenws.py ওনট্রেডস
# উদাহরণ: ./krakenws.py অ্যাডঅর্ডার pair=XBT/EUR type=sell ordertype=limit price=7500 volume=0.125
# উদাহরণ: ./krakenws.py ক্যানসেলঅর্ডার txid=OADMSD-7AGC3-IMB46A,OD6VRE-HCSPM-CKORER
#
# অ্যাকাউন্ট ম্যানেজমেন্ট এবং ট্রেডিংয়ের জন্য, একটি বৈধ ওয়েবসকেট প্রমাণীকরণ টোকেন (REST API GetWebSocketsToken এন্ডপয়েন্ট থেকে) WS_Token নামের একটি প্লেইন টেক্সট ফাইলে সরবরাহ করতে হবে।

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 ফিড/এন্ডপয়েন্ট [প্যারামিটার]" % sys.argv[0])
print("উদাহরণ: %s টিকার 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("ওয়েবসকেট প্রমাণীকরণ টোকেন অনুপস্থিত (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("ওয়েবসকেট প্রমাণীকরণ টোকেন অনুপস্থিত (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 ফিড/এন্ডপয়েন্ট [প্যারামিটার]" % sys.argv[0])
print("উদাহরণ: %s টিকার XBT/USD" % sys.argv[0])
sys.exit(1)

try:
ws = create_connection(api_domain)
print("ওয়েবসকেট -> ক্লায়েন্ট: %s" % ws.recv())
except Exception as error:
print("ওয়েবসকেট সংযোগ ব্যর্থ হয়েছে (%s)" % error)
sys.exit(1)

try:
print("ক্লায়েন্ট -> ওয়েবসকেট: %s" % api_data)
ws.send(api_data)
print("ওয়েবসকেট -> ক্লায়েন্ট: %s" % ws.recv())
except Exception as error:
print("ওয়েবসকেট সাবস্ক্রিপশন/অনুরোধ ব্যর্থ হয়েছে (%s)" % error)
ws.close()
sys.exit(1)

while True:
try:
print("ওয়েবসকেট -> ক্লায়েন্ট: %s" % ws.recv())
except KeyboardInterrupt:
ws.close()
sys.exit(0)
except Exception as error:
print("ওয়েবসকেট বার্তা ব্যর্থ হয়েছে (%s)" % error)
sys.exit(1)

sys.exit(1)