pikers
/
piker
3
0
Fork 0
Code Issues 2 Pull Requests 4 Packages Projects Releases Wiki Activity

Port `kraken` backend to `httpx`

go_httpx_binance
Tyler Goodlet 2024-05-20 11:09:10 -04:00
parent 129cf58d41
commit 95ace5acb8
2 changed files with 76 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

85
piker/brokers/kraken/api.py
View File

@ -27,8 +27,8 @@ from typing import (
) )
import time import time
import httpx
import pendulum import pendulum
import asks
import numpy as np import numpy as np
import urllib.parse import urllib.parse
import hashlib import hashlib
@ -60,6 +60,11 @@ log = get_logger('piker.brokers.kraken')
# <uri>/<version>/ # <uri>/<version>/
_url = 'https://api.kraken.com/0' _url = 'https://api.kraken.com/0'
_headers: dict[str, str] = {
'User-Agent': 'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
}
# TODO: this is the only backend providing this right? # TODO: this is the only backend providing this right?
# in which case we should drop it from the defaults and # in which case we should drop it from the defaults and
# instead make a custom fields descr in this module! # instead make a custom fields descr in this module!
@ -135,16 +140,15 @@ class Client:
def __init__( def __init__(
self, self,
config: dict[str, str], config: dict[str, str],
httpx_client: httpx.AsyncClient,
name: str = '', name: str = '',
api_key: str = '', api_key: str = '',
secret: str = '' secret: str = ''
) -> None: ) -> None:
self._sesh = asks.Session(connections=4)
self._sesh.base_location = _url self._sesh: httpx.AsyncClient = httpx_client
self._sesh.headers.update({
'User-Agent':
'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
})
self._name = name self._name = name
self._api_key = api_key self._api_key = api_key
self._secret = secret self._secret = secret
@ -167,9 +171,8 @@ class Client:
data: dict, data: dict,
) -> dict[str, Any]: ) -> dict[str, Any]:
resp = await self._sesh.post( resp = await self._sesh.post(
path=f'/public/{method}', url=f'/public/{method}',
json=data, json=data,
timeout=float('inf')
) )
return resproc(resp, log) return resproc(resp, log)
@ -180,18 +183,18 @@ class Client:
uri_path: str uri_path: str
) -> dict[str, Any]: ) -> dict[str, Any]:
headers = { headers = {
'Content-Type': 'Content-Type': 'application/x-www-form-urlencoded',
'application/x-www-form-urlencoded', 'API-Key': self._api_key,
'API-Key': 'API-Sign': get_kraken_signature(
self._api_key, uri_path,
'API-Sign': data,
get_kraken_signature(uri_path, data, self._secret) self._secret,
),
} }
resp = await self._sesh.post( resp = await self._sesh.post(
path=f'/private/{method}', url=f'/private/{method}',
data=data, data=data,
headers=headers, headers=headers,
timeout=float('inf')
) )
return resproc(resp, log) return resproc(resp, log)
@ -665,24 +668,36 @@ class Client:
@acm @acm
async def get_client() -> Client: async def get_client() -> Client:
conf = get_config() conf: dict[str, Any] = get_config()
if conf: async with httpx.AsyncClient(
client = Client( base_url=_url,
conf, headers=_headers,
# TODO: don't break these up and just do internal # TODO: is there a way to numerate this?
# conf lookups instead.. # https://www.python-httpx.org/advanced/clients/#why-use-a-client
name=conf['key_descr'], # connections=4
api_key=conf['api_key'], ) as trio_client:
secret=conf['secret'] if conf:
) client = Client(
else: conf,
client = Client({}) httpx_client=trio_client,
# at startup, load all symbols, and asset info in # TODO: don't break these up and just do internal
# batch requests. # conf lookups instead..
async with trio.open_nursery() as nurse: name=conf['key_descr'],
nurse.start_soon(client.get_assets) api_key=conf['api_key'],
await client.get_mkt_pairs() secret=conf['secret']
)
else:
client = Client(
conf={},
httpx_client=trio_client,
)
yield client # at startup, load all symbols, and asset info in
# batch requests.
async with trio.open_nursery() as nurse:
nurse.start_soon(client.get_assets)
await client.get_mkt_pairs()
yield client

44
piker/brokers/kraken/broker.py
View File

@ -612,18 +612,18 @@ async def open_trade_dialog(
# enter relay loop # enter relay loop
await handle_order_updates( await handle_order_updates(
client, client=client,
ws, ws=ws,
stream, ws_stream=stream,
ems_stream, ems_stream=ems_stream,
apiflows, apiflows=apiflows,
ids, ids=ids,
reqids2txids, reqids2txids=reqids2txids,
acnt, acnt=acnt,
api_trans, ledger=ledger,
acctid, acctid=acctid,
acc_name, acc_name=acc_name,
token, token=token,
) )
@ -639,7 +639,8 @@ async def handle_order_updates(
# transaction records which will be updated # transaction records which will be updated
# on new trade clearing events (aka order "fills") # on new trade clearing events (aka order "fills")
ledger_trans: dict[str, Transaction], ledger: TransactionLedger,
# ledger_trans: dict[str, Transaction],
acctid: str, acctid: str,
acc_name: str, acc_name: str,
token: str, token: str,
@ -699,7 +700,8 @@ async def handle_order_updates(
# if tid not in ledger_trans # if tid not in ledger_trans
} }
for tid, trade in trades.items(): for tid, trade in trades.items():
assert tid not in ledger_trans # assert tid not in ledger_trans
assert tid not in ledger
txid = trade['ordertxid'] txid = trade['ordertxid']
reqid = trade.get('userref') reqid = trade.get('userref')
@ -747,11 +749,17 @@ async def handle_order_updates(
client, client,
api_name_set='wsname', api_name_set='wsname',
) )
ppmsgs = trades2pps( ppmsgs: list[BrokerdPosition] = trades2pps(
acnt, acnt=acnt,
acctid, ledger=ledger,
new_trans, acctid=acctid,
new_trans=new_trans,
) )
# ppmsgs = trades2pps(
# acnt,
# acctid,
# new_trans,
# )
for pp_msg in ppmsgs: for pp_msg in ppmsgs:
await ems_stream.send(pp_msg) await ems_stream.send(pp_msg)