From fd0acd21fb27998438e6794abee934030af26094 Mon Sep 17 00:00:00 2001 From: Konstantine Tsafatinos Date: Sun, 6 Mar 2022 15:16:42 -0500 Subject: [PATCH] refactory based on github comments, change doc string style --- piker/brokers/kraken.py | 150 ++++++++++++++++++++++++---------------- 1 file changed, 91 insertions(+), 59 deletions(-) diff --git a/piker/brokers/kraken.py b/piker/brokers/kraken.py index 6e9e55b3..90b40b2a 100644 --- a/piker/brokers/kraken.py +++ b/piker/brokers/kraken.py @@ -14,13 +14,13 @@ # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . -""" +''' Kraken backend. -""" +''' from contextlib import asynccontextmanager from dataclasses import asdict, field -from typing import List, Dict, Any, Tuple, Optional, AsyncIterator +from typing import Dict, List, Tuple, Any, Optional, AsyncIterator import time from trio_typing import TaskStatus @@ -34,6 +34,10 @@ from pydantic.dataclasses import dataclass from pydantic import BaseModel import wsproto from itertools import count +import urllib.parse +import hashlib +import hmac +import base64 from .. import config from .._cacheables import open_cached_client @@ -47,11 +51,6 @@ from ..clearing._messages import ( BrokerdFill, ) -import urllib.parse -import hashlib -import hmac -import base64 - log = get_logger(__name__) @@ -120,7 +119,10 @@ class Pair(BaseModel): class Trade(BaseModel): - """Trade class that helps parse and validate ownTrades stream""" + ''' + Trade class that helps parse and validate ownTrades stream + + ''' reqid: str # kraken order transaction id action: str # buy or sell price: str # price of asset @@ -130,11 +132,13 @@ class Trade(BaseModel): @dataclass class OHLC: - """Description of the flattened OHLC quote format. + ''' + Description of the flattened OHLC quote format. For schema details see: https://docs.kraken.com/websockets/#message-ohlc - """ + + ''' chan_id: int # internal kraken id chan_name: str # eg. ohlc-1 (name-interval) pair: str # fx pair @@ -179,16 +183,24 @@ def get_kraken_signature( class InvalidKey(ValueError): - """EAPI:Invalid key - This error is returned when the API key used for the call is - either expired or disabled, please review the API key in your - Settings -> API tab of account management or generate a new one - and update your application.""" + ''' + EAPI:Invalid key + This error is returned when the API key used for the call is + either expired or disabled, please review the API key in your + Settings -> API tab of account management or generate a new one + and update your application. + + ''' class Client: - def __init__(self) -> None: + def __init__( + self, + name: str = '', + api_key: str = '', + secret: str = '' + ) -> None: self._sesh = asks.Session(connections=4) self._sesh.base_location = _url self._sesh.headers.update({ @@ -196,9 +208,9 @@ class Client: 'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)' }) self._pairs: list[str] = [] - self._name = '' - self._api_key = '' - self._secret = '' + self._name = name + self._api_key = api_key + self._secret = secret @property def pairs(self) -> Dict[str, Any]: @@ -244,26 +256,26 @@ class Client: ) return resproc(resp, log) - async def kraken_endpoint( + async def endpoint( self, method: str, data: Dict[str, Any] ) -> Dict[str, Any]: uri_path = f'/0/private/{method}' data['nonce'] = str(int(1000*time.time())) - resp = await self._private(method, data, uri_path) - return resp + return await self._private(method, data, uri_path) async def get_positions( self, data: Dict[str, Any] = {} - ) -> Dict[str, Any]: + ) -> (Dict[str, Any], Dict[str, Any]): data['ofs'] = 0 positions = {} vols = {} # Grab all trade history + # https://docs.kraken.com/rest/#operation/getTradeHistory while True: - resp = await self.kraken_endpoint('TradesHistory', data) + resp = await self.endpoint('TradesHistory', data) # grab the first 50 trades if data['ofs'] == 0: trades = resp['result']['trades'] @@ -281,22 +293,28 @@ class Client: # increment the offset counter data['ofs'] += 50 # To avoid exceeding API rate limit in case of a lot of trades - time.sleep(1) + await trio.sleep(1) # make sure you grabbed all the trades assert count == len(trades.values()) # positions - ## TODO: Make sure to add option to include fees in positions calc + # TODO: Make sure to add option to include fees in positions calc for trade in trades.values(): sign = -1 if trade['type'] == 'sell' else 1 + # This catch is for populating the dict with new values + # as the plus assigment will fail if there no value + # tied to the key try: positions[trade['pair']] += sign * float(trade['cost']) vols[trade['pair']] += sign * float(trade['vol']) except KeyError: positions[trade['pair']] = sign * float(trade['cost']) vols[trade['pair']] = sign * float(trade['vol']) - + # This cycles through the summed trades of an asset and then + # normalizes the price with the current volume of the asset + # you are holding. If you have no more of the asset, the balance + # is 0, then it sets the position to 0. for pair in positions.keys(): asset_balance = vols[pair] if asset_balance == 0: @@ -316,9 +334,10 @@ class Client: # account: str, reqid: int = None, ) -> int: - """Place an order and return integer request id provided by client. + ''' + Place an order and return integer request id provided by client. - """ + ''' # Build order data for kraken api data = { "userref": reqid, @@ -330,20 +349,18 @@ class Client: # set to True test AddOrder call without a real submission "validate": False } - resp = await self.kraken_endpoint('AddOrder', data) - return resp + return await self.endpoint('AddOrder', data) async def submit_cancel( self, reqid: str, ) -> None: - """Send cancel request for order id ``reqid``. + ''' + Send cancel request for order id ``reqid``. - """ + ''' # txid is a transaction id given by kraken - data = {"txid": reqid} - resp = await self.kraken_endpoint('CancelOrder', data) - return resp + return await self.endpoint('CancelOrder', {"txid": reqid}) async def symbol_info( self, @@ -457,12 +474,13 @@ class Client: @asynccontextmanager async def get_client() -> Client: - client = Client() section = get_config() - client._name = section['key_descr'] - client._api_key = section['api_key'] - client._secret = section['secret'] + client = Client( + name=section['key_descr'], + api_key=section['api_key'], + secret=section['secret'] + ) # at startup, load all symbols locally for fast search await client.cache_symbols() @@ -490,6 +508,8 @@ def pack_position( def normalize_symbol( ticker: str ) -> str: + # This is to convert symbol names from what kraken + # uses to the traditional 3x3 pair symbol syntax symlen = len(ticker) if symlen == 6: return ticker.lower() @@ -501,12 +521,13 @@ def normalize_symbol( def make_auth_sub(data: Dict[str, Any]) -> Dict[str, str]: - """Create a request subscription packet dict. + ''' + Create a request subscription packet dict. ## TODO: point to the auth urls https://docs.kraken.com/websockets/#message-subscribe - """ + ''' # eg. specific logic for this in kraken's sync client: # https://github.com/krakenfx/kraken-wsclient-py/blob/master/kraken_wsclient_py/kraken_wsclient_py.py#L188 return { @@ -519,9 +540,6 @@ async def handle_order_requests( client: Client, ems_order_stream: tractor.MsgStream, - #ws: NoBsWs, - #token: str, - #userref_oid_map: dict, ) -> None: @@ -575,8 +593,8 @@ async def handle_order_requests( ).dict() ) else: - ## TODO: handle multiple cancels - ## txid is an array of strings + # TODO: handle multiple cancels + # txid is an array of strings reqid = resp['result']['txid'][0] # deliver ack that order has been submitted to broker routing await ems_order_stream.send( @@ -608,7 +626,7 @@ async def handle_order_requests( assert resp['error'] == [] assert resp['result']['count'] == 1 - ## TODO: Change this code using .get + # TODO: Change this code using .get try: pending = resp['result']['pending'] # Check to make sure the cancellation is NOT pending, @@ -626,6 +644,15 @@ async def handle_order_requests( ) except AssertionError: log.error(f'Order cancel was not successful') + await ems_order_stream.send( + BrokerdError( + oid=order.oid, + reqid=temp_id, + symbol=order.symbol, + reason="Failed order cancel", + broker_details=resp + ).dict() + ) else: log.error(f'Unknown order command: {request_msg}') @@ -669,7 +696,6 @@ async def trades_dialogue( # XXX: do we need to ack the unsub? # await ws.recv_msg() - # Authenticated block async with get_client() as client: acc_name = 'kraken.' + client._name @@ -687,7 +713,7 @@ async def trades_dialogue( # Get websocket token for authenticated data stream # Assert that a token was actually received - resp = await client.kraken_endpoint('GetWebSocketsToken', {}) + resp = await client.endpoint('GetWebSocketsToken', {}) assert resp['error'] == [] token = resp['result']['token'] @@ -717,7 +743,7 @@ async def trades_dialogue( status='executed', filled=float(trade.size), reason='Order filled by kraken', - # remaining='' ## TODO: not sure what to do here. + # remaining='' # TODO: not sure what to do here. broker_details={ 'name': 'kraken', 'broker_time': trade.broker_time @@ -734,7 +760,7 @@ async def trades_dialogue( action=trade.action, size=float(trade.size), price=float(trade.price), - ## TODO: maybe capture more msg data i.e fees? + # TODO: maybe capture more msg data i.e fees? broker_details={'name': 'kraken'}, broker_time=float(trade.broker_time) ) @@ -846,6 +872,8 @@ async def process_trade_msgs( try: # check that we are on the ownTrades stream and that msgs are # arriving in sequence with kraken + # For clarification the kraken ws api docs for this stream: + # https://docs.kraken.com/websockets/#message-ownTrades assert msg[1] == 'ownTrades' assert msg[2]['sequence'] > sequence_counter sequence_counter += 1 @@ -894,11 +922,12 @@ def normalize( def make_sub(pairs: List[str], data: Dict[str, Any]) -> Dict[str, str]: - """Create a request subscription packet dict. + ''' + Create a request subscription packet dict. https://docs.kraken.com/websockets/#message-subscribe - """ + ''' # eg. specific logic for this in kraken's sync client: # https://github.com/krakenfx/kraken-wsclient-py/blob/master/kraken_wsclient_py/kraken_wsclient_py.py#L188 return { @@ -916,8 +945,9 @@ async def backfill_bars( task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED, ) -> None: - """Fill historical bars into shared mem / storage afap. - """ + ''' + Fill historical bars into shared mem / storage afap. + ''' with trio.CancelScope() as cs: async with open_cached_client('kraken') as client: bars = await client.bars(symbol=sym) @@ -939,10 +969,12 @@ async def stream_quotes( task_status: TaskStatus[Tuple[Dict, Dict]] = trio.TASK_STATUS_IGNORED, ) -> None: - """Subscribe for ohlc stream of quotes for ``pairs``. + ''' + Subscribe for ohlc stream of quotes for ``pairs``. ``pairs`` must be formatted /. - """ + + ''' # XXX: required to propagate ``tractor`` loglevel to piker logging get_console_log(loglevel or tractor.current_actor().loglevel)