commit
014bd58db4
|
|
@ -50,7 +50,7 @@ __brokers__: list[str] = [
|
|||
'binance',
|
||||
'ib',
|
||||
'kraken',
|
||||
'kucoin'
|
||||
'kucoin',
|
||||
|
||||
# broken but used to work
|
||||
# 'questrade',
|
||||
|
|
@ -71,7 +71,7 @@ def get_brokermod(brokername: str) -> ModuleType:
|
|||
Return the imported broker module by name.
|
||||
|
||||
'''
|
||||
module = import_module('.' + brokername, 'piker.brokers')
|
||||
module: ModuleType = import_module('.' + brokername, 'piker.brokers')
|
||||
# we only allow monkeying because it's for internal keying
|
||||
module.name = module.__name__.split('.')[-1]
|
||||
return module
|
||||
|
|
|
|||
|
|
@ -18,10 +18,11 @@
|
|||
Handy cross-broker utils.
|
||||
|
||||
"""
|
||||
from __future__ import annotations
|
||||
from functools import partial
|
||||
|
||||
import json
|
||||
import asks
|
||||
import httpx
|
||||
import logging
|
||||
|
||||
from ..log import (
|
||||
|
|
@ -60,11 +61,11 @@ class NoData(BrokerError):
|
|||
def __init__(
|
||||
self,
|
||||
*args,
|
||||
info: dict,
|
||||
info: dict|None = None,
|
||||
|
||||
) -> None:
|
||||
super().__init__(*args)
|
||||
self.info: dict = info
|
||||
self.info: dict|None = info
|
||||
|
||||
# when raised, machinery can check if the backend
|
||||
# set a "frame size" for doing datetime calcs.
|
||||
|
|
@ -90,16 +91,18 @@ class DataThrottle(BrokerError):
|
|||
|
||||
|
||||
def resproc(
|
||||
resp: asks.response_objects.Response,
|
||||
resp: httpx.Response,
|
||||
log: logging.Logger,
|
||||
return_json: bool = True,
|
||||
log_resp: bool = False,
|
||||
|
||||
) -> asks.response_objects.Response:
|
||||
"""Process response and return its json content.
|
||||
) -> httpx.Response:
|
||||
'''
|
||||
Process response and return its json content.
|
||||
|
||||
Raise the appropriate error on non-200 OK responses.
|
||||
"""
|
||||
|
||||
'''
|
||||
if not resp.status_code == 200:
|
||||
raise BrokerError(resp.body)
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -25,14 +25,13 @@ from __future__ import annotations
|
|||
from collections import ChainMap
|
||||
from contextlib import (
|
||||
asynccontextmanager as acm,
|
||||
AsyncExitStack,
|
||||
)
|
||||
from datetime import datetime
|
||||
from pprint import pformat
|
||||
from typing import (
|
||||
Any,
|
||||
Callable,
|
||||
Hashable,
|
||||
Sequence,
|
||||
Type,
|
||||
)
|
||||
import hmac
|
||||
|
|
@ -43,8 +42,7 @@ import trio
|
|||
from pendulum import (
|
||||
now,
|
||||
)
|
||||
import asks
|
||||
from rapidfuzz import process as fuzzy
|
||||
import httpx
|
||||
import numpy as np
|
||||
|
||||
from piker import config
|
||||
|
|
@ -54,6 +52,7 @@ from piker.clearing._messages import (
|
|||
from piker.accounting import (
|
||||
Asset,
|
||||
digits_to_dec,
|
||||
MktPair,
|
||||
)
|
||||
from piker.types import Struct
|
||||
from piker.data import (
|
||||
|
|
@ -69,7 +68,6 @@ from .venues import (
|
|||
PAIRTYPES,
|
||||
Pair,
|
||||
MarketType,
|
||||
|
||||
_spot_url,
|
||||
_futes_url,
|
||||
_testnet_futes_url,
|
||||
|
|
@ -79,19 +77,18 @@ from .venues import (
|
|||
log = get_logger('piker.brokers.binance')
|
||||
|
||||
|
||||
def get_config() -> dict:
|
||||
|
||||
def get_config() -> dict[str, Any]:
|
||||
conf: dict
|
||||
path: Path
|
||||
conf, path = config.load(
|
||||
conf_name='brokers',
|
||||
touch_if_dne=True,
|
||||
)
|
||||
|
||||
section = conf.get('binance')
|
||||
|
||||
section: dict = conf.get('binance')
|
||||
if not section:
|
||||
log.warning(f'No config section found for binance in {path}')
|
||||
log.warning(
|
||||
f'No config section found for binance in {path}'
|
||||
)
|
||||
return {}
|
||||
|
||||
return section
|
||||
|
|
@ -147,7 +144,7 @@ def binance_timestamp(
|
|||
|
||||
class Client:
|
||||
'''
|
||||
Async ReST API client using ``trio`` + ``asks`` B)
|
||||
Async ReST API client using `trio` + `httpx` B)
|
||||
|
||||
Supports all of the spot, margin and futures endpoints depending
|
||||
on method.
|
||||
|
|
@ -156,10 +153,17 @@ class Client:
|
|||
def __init__(
|
||||
self,
|
||||
|
||||
venue_sessions: dict[
|
||||
str, # venue key
|
||||
tuple[httpx.AsyncClient, str] # session, eps path
|
||||
],
|
||||
conf: dict[str, Any],
|
||||
# TODO: change this to `Client.[mkt_]venue: MarketType`?
|
||||
mkt_mode: MarketType = 'spot',
|
||||
|
||||
) -> None:
|
||||
self.conf = conf
|
||||
|
||||
# build out pair info tables for each market type
|
||||
# and wrap in a chain-map view for search / query.
|
||||
self._spot_pairs: dict[str, Pair] = {} # spot info table
|
||||
|
|
@ -186,44 +190,13 @@ class Client:
|
|||
# market symbols for use by search. See `.exch_info()`.
|
||||
self._pairs: ChainMap[str, Pair] = ChainMap()
|
||||
|
||||
# spot EPs sesh
|
||||
self._sesh = asks.Session(connections=4)
|
||||
self._sesh.base_location: str = _spot_url
|
||||
# spot testnet
|
||||
self._test_sesh: asks.Session = asks.Session(connections=4)
|
||||
self._test_sesh.base_location: str = _testnet_spot_url
|
||||
|
||||
# margin and extended spot endpoints session.
|
||||
self._sapi_sesh = asks.Session(connections=4)
|
||||
self._sapi_sesh.base_location: str = _spot_url
|
||||
|
||||
# futes EPs sesh
|
||||
self._fapi_sesh = asks.Session(connections=4)
|
||||
self._fapi_sesh.base_location: str = _futes_url
|
||||
# futes testnet
|
||||
self._test_fapi_sesh: asks.Session = asks.Session(connections=4)
|
||||
self._test_fapi_sesh.base_location: str = _testnet_futes_url
|
||||
|
||||
# global client "venue selection" mode.
|
||||
# set this when you want to switch venues and not have to
|
||||
# specify the venue for the next request.
|
||||
self.mkt_mode: MarketType = mkt_mode
|
||||
|
||||
# per 8
|
||||
self.venue_sesh: dict[
|
||||
str, # venue key
|
||||
tuple[asks.Session, str] # session, eps path
|
||||
] = {
|
||||
'spot': (self._sesh, '/api/v3/'),
|
||||
'spot_testnet': (self._test_sesh, '/fapi/v1/'),
|
||||
|
||||
'margin': (self._sapi_sesh, '/sapi/v1/'),
|
||||
|
||||
'usdtm_futes': (self._fapi_sesh, '/fapi/v1/'),
|
||||
'usdtm_futes_testnet': (self._test_fapi_sesh, '/fapi/v1/'),
|
||||
|
||||
# 'futes_coin': self._dapi, # TODO
|
||||
}
|
||||
# per-mkt-venue API client table
|
||||
self.venue_sesh = venue_sessions
|
||||
|
||||
# lookup for going from `.mkt_mode: str` to the config
|
||||
# subsection `key: str`
|
||||
|
|
@ -238,40 +211,6 @@ class Client:
|
|||
'futes': ['usdtm_futes'],
|
||||
}
|
||||
|
||||
# for creating API keys see,
|
||||
# https://www.binance.com/en/support/faq/how-to-create-api-keys-on-binance-360002502072
|
||||
self.conf: dict = get_config()
|
||||
|
||||
for key, subconf in self.conf.items():
|
||||
if api_key := subconf.get('api_key', ''):
|
||||
venue_keys: list[str] = self.confkey2venuekeys[key]
|
||||
|
||||
venue_key: str
|
||||
sesh: asks.Session
|
||||
for venue_key in venue_keys:
|
||||
sesh, _ = self.venue_sesh[venue_key]
|
||||
|
||||
api_key_header: dict = {
|
||||
# taken from official:
|
||||
# https://github.com/binance/binance-futures-connector-python/blob/main/binance/api.py#L47
|
||||
"Content-Type": "application/json;charset=utf-8",
|
||||
|
||||
# TODO: prolly should just always query and copy
|
||||
# in the real latest ver?
|
||||
"User-Agent": "binance-connector/6.1.6smbz6",
|
||||
"X-MBX-APIKEY": api_key,
|
||||
}
|
||||
sesh.headers.update(api_key_header)
|
||||
|
||||
# if `.use_tesnet = true` in the config then
|
||||
# also add headers for the testnet session which
|
||||
# will be used for all order control
|
||||
if subconf.get('use_testnet', False):
|
||||
testnet_sesh, _ = self.venue_sesh[
|
||||
venue_key + '_testnet'
|
||||
]
|
||||
testnet_sesh.headers.update(api_key_header)
|
||||
|
||||
def _mk_sig(
|
||||
self,
|
||||
data: dict,
|
||||
|
|
@ -290,7 +229,6 @@ class Client:
|
|||
'to define the creds for auth-ed endpoints!?'
|
||||
)
|
||||
|
||||
|
||||
# XXX: Info on security and authentification
|
||||
# https://binance-docs.github.io/apidocs/#endpoint-security-type
|
||||
if not (api_secret := subconf.get('api_secret')):
|
||||
|
|
@ -330,8 +268,9 @@ class Client:
|
|||
- /fapi/v3/ USD-M FUTURES, or
|
||||
- /api/v3/ SPOT/MARGIN
|
||||
|
||||
account/market endpoint request depending on either passed in `venue: str`
|
||||
or the current setting `.mkt_mode: str` setting, default `'spot'`.
|
||||
account/market endpoint request depending on either passed in
|
||||
`venue: str` or the current setting `.mkt_mode: str` setting,
|
||||
default `'spot'`.
|
||||
|
||||
|
||||
Docs per venue API:
|
||||
|
|
@ -360,9 +299,6 @@ class Client:
|
|||
venue=venue_key,
|
||||
)
|
||||
|
||||
sesh: asks.Session
|
||||
path: str
|
||||
|
||||
# Check if we're configured to route order requests to the
|
||||
# venue equivalent's testnet.
|
||||
use_testnet: bool = False
|
||||
|
|
@ -387,11 +323,12 @@ class Client:
|
|||
# ctl machinery B)
|
||||
venue_key += '_testnet'
|
||||
|
||||
sesh, path = self.venue_sesh[venue_key]
|
||||
|
||||
meth: Callable = getattr(sesh, method)
|
||||
client: httpx.AsyncClient
|
||||
path: str
|
||||
client, path = self.venue_sesh[venue_key]
|
||||
meth: Callable = getattr(client, method)
|
||||
resp = await meth(
|
||||
path=path + endpoint,
|
||||
url=path + endpoint,
|
||||
params=params,
|
||||
timeout=float('inf'),
|
||||
)
|
||||
|
|
@ -433,7 +370,15 @@ class Client:
|
|||
item['filters'] = filters
|
||||
|
||||
pair_type: Type = PAIRTYPES[venue]
|
||||
try:
|
||||
pair: Pair = pair_type(**item)
|
||||
except Exception as e:
|
||||
e.add_note(
|
||||
"\nDon't panic, prolly stupid binance changed their symbology schema again..\n"
|
||||
'Check out their API docs here:\n\n'
|
||||
'https://binance-docs.github.io/apidocs/spot/en/#exchange-information'
|
||||
)
|
||||
raise
|
||||
pair_table[pair.symbol.upper()] = pair
|
||||
|
||||
# update an additional top-level-cross-venue-table
|
||||
|
|
@ -528,7 +473,9 @@ class Client:
|
|||
|
||||
'''
|
||||
pair_table: dict[str, Pair] = self._venue2pairs[
|
||||
venue or self.mkt_mode
|
||||
venue
|
||||
or
|
||||
self.mkt_mode
|
||||
]
|
||||
if (
|
||||
expiry
|
||||
|
|
@ -547,9 +494,9 @@ class Client:
|
|||
venues: list[str] = [venue]
|
||||
|
||||
# batch per-venue download of all exchange infos
|
||||
async with trio.open_nursery() as rn:
|
||||
async with trio.open_nursery() as tn:
|
||||
for ven in venues:
|
||||
rn.start_soon(
|
||||
tn.start_soon(
|
||||
self._cache_pairs,
|
||||
ven,
|
||||
)
|
||||
|
|
@ -602,11 +549,11 @@ class Client:
|
|||
|
||||
) -> dict[str, Any]:
|
||||
|
||||
fq_pairs: dict = await self.exch_info()
|
||||
fq_pairs: dict[str, Pair] = await self.exch_info()
|
||||
|
||||
# TODO: cache this list like we were in
|
||||
# `open_symbol_search()`?
|
||||
keys: list[str] = list(fq_pairs)
|
||||
# keys: list[str] = list(fq_pairs)
|
||||
|
||||
return match_from_pairs(
|
||||
pairs=fq_pairs,
|
||||
|
|
@ -614,9 +561,19 @@ class Client:
|
|||
score_cutoff=50,
|
||||
)
|
||||
|
||||
def pair2venuekey(
|
||||
self,
|
||||
pair: Pair,
|
||||
) -> str:
|
||||
return {
|
||||
'USDTM': 'usdtm_futes',
|
||||
# 'COINM': 'coin_futes',
|
||||
# ^-TODO-^ bc someone might want it..?
|
||||
}[pair.venue]
|
||||
|
||||
async def bars(
|
||||
self,
|
||||
symbol: str,
|
||||
mkt: MktPair,
|
||||
|
||||
start_dt: datetime | None = None,
|
||||
end_dt: datetime | None = None,
|
||||
|
|
@ -646,16 +603,20 @@ class Client:
|
|||
start_time = binance_timestamp(start_dt)
|
||||
end_time = binance_timestamp(end_dt)
|
||||
|
||||
bs_pair: Pair = self._pairs[mkt.bs_fqme.upper()]
|
||||
|
||||
# https://binance-docs.github.io/apidocs/spot/en/#kline-candlestick-data
|
||||
bars = await self._api(
|
||||
'klines',
|
||||
params={
|
||||
'symbol': symbol.upper(),
|
||||
# NOTE: always query using their native symbology!
|
||||
'symbol': mkt.bs_mktid.upper(),
|
||||
'interval': '1m',
|
||||
'startTime': start_time,
|
||||
'endTime': end_time,
|
||||
'limit': limit
|
||||
},
|
||||
venue=self.pair2venuekey(bs_pair),
|
||||
allow_testnet=False,
|
||||
)
|
||||
new_bars: list[tuple] = []
|
||||
|
|
@ -972,17 +933,148 @@ class Client:
|
|||
await self.close_listen_key(key)
|
||||
|
||||
|
||||
@acm
|
||||
async def get_client() -> Client:
|
||||
_venue_urls: dict[str, str] = {
|
||||
'spot': (
|
||||
_spot_url,
|
||||
'/api/v3/',
|
||||
),
|
||||
'spot_testnet': (
|
||||
_testnet_spot_url,
|
||||
'/fapi/v1/'
|
||||
),
|
||||
# margin and extended spot endpoints session.
|
||||
# TODO: did this ever get implemented fully?
|
||||
# 'margin': (
|
||||
# _spot_url,
|
||||
# '/sapi/v1/'
|
||||
# ),
|
||||
|
||||
client = Client()
|
||||
await client.exch_info()
|
||||
log.info(
|
||||
f'{client} in {client.mkt_mode} mode: caching exchange infos..\n'
|
||||
'Cached multi-market pairs:\n'
|
||||
f'spot: {len(client._spot_pairs)}\n'
|
||||
f'usdtm_futes: {len(client._ufutes_pairs)}\n'
|
||||
f'Total: {len(client._pairs)}\n'
|
||||
'usdtm_futes': (
|
||||
_futes_url,
|
||||
'/fapi/v1/',
|
||||
),
|
||||
|
||||
'usdtm_futes_testnet': (
|
||||
_testnet_futes_url,
|
||||
'/fapi/v1/',
|
||||
),
|
||||
|
||||
# TODO: for anyone who actually needs it ;P
|
||||
# 'coin_futes': ()
|
||||
}
|
||||
|
||||
|
||||
def init_api_keys(
|
||||
client: Client,
|
||||
conf: dict[str, Any],
|
||||
) -> None:
|
||||
'''
|
||||
Set up per-venue API keys each http client according to the user's
|
||||
`brokers.conf`.
|
||||
|
||||
For ex, to use spot-testnet and live usdt futures APIs:
|
||||
|
||||
```toml
|
||||
[binance]
|
||||
# spot test net
|
||||
spot.use_testnet = true
|
||||
spot.api_key = '<spot_api_key_from_binance_account>'
|
||||
spot.api_secret = '<spot_api_key_password>'
|
||||
|
||||
# futes live
|
||||
futes.use_testnet = false
|
||||
accounts.usdtm = 'futes'
|
||||
futes.api_key = '<futes_api_key_from_binance>'
|
||||
futes.api_secret = '<futes_api_key_password>''
|
||||
|
||||
# if uncommented will use the built-in paper engine and not
|
||||
# connect to `binance` API servers for order ctl.
|
||||
# accounts.paper = 'paper'
|
||||
```
|
||||
|
||||
'''
|
||||
for key, subconf in conf.items():
|
||||
if api_key := subconf.get('api_key', ''):
|
||||
venue_keys: list[str] = client.confkey2venuekeys[key]
|
||||
|
||||
venue_key: str
|
||||
client: httpx.AsyncClient
|
||||
for venue_key in venue_keys:
|
||||
client, _ = client.venue_sesh[venue_key]
|
||||
|
||||
api_key_header: dict = {
|
||||
# taken from official:
|
||||
# https://github.com/binance/binance-futures-connector-python/blob/main/binance/api.py#L47
|
||||
"Content-Type": "application/json;charset=utf-8",
|
||||
|
||||
# TODO: prolly should just always query and copy
|
||||
# in the real latest ver?
|
||||
"User-Agent": "binance-connector/6.1.6smbz6",
|
||||
"X-MBX-APIKEY": api_key,
|
||||
}
|
||||
client.headers.update(api_key_header)
|
||||
|
||||
# if `.use_tesnet = true` in the config then
|
||||
# also add headers for the testnet session which
|
||||
# will be used for all order control
|
||||
if subconf.get('use_testnet', False):
|
||||
testnet_sesh, _ = client.venue_sesh[
|
||||
venue_key + '_testnet'
|
||||
]
|
||||
testnet_sesh.headers.update(api_key_header)
|
||||
|
||||
|
||||
@acm
|
||||
async def get_client(
|
||||
mkt_mode: MarketType = 'spot',
|
||||
) -> Client:
|
||||
'''
|
||||
Construct an single `piker` client which composes multiple underlying venue
|
||||
specific API clients both for live and test networks.
|
||||
|
||||
'''
|
||||
venue_sessions: dict[
|
||||
str, # venue key
|
||||
tuple[httpx.AsyncClient, str] # session, eps path
|
||||
] = {}
|
||||
async with AsyncExitStack() as client_stack:
|
||||
for name, (base_url, path) in _venue_urls.items():
|
||||
api: httpx.AsyncClient = await client_stack.enter_async_context(
|
||||
httpx.AsyncClient(
|
||||
base_url=base_url,
|
||||
# headers={},
|
||||
|
||||
# TODO: is there a way to numerate this?
|
||||
# https://www.python-httpx.org/advanced/clients/#why-use-a-client
|
||||
# connections=4
|
||||
)
|
||||
)
|
||||
venue_sessions[name] = (
|
||||
api,
|
||||
path,
|
||||
)
|
||||
|
||||
conf: dict[str, Any] = get_config()
|
||||
# for creating API keys see,
|
||||
# https://www.binance.com/en/support/faq/how-to-create-api-keys-on-binance-360002502072
|
||||
client = Client(
|
||||
venue_sessions=venue_sessions,
|
||||
conf=conf,
|
||||
mkt_mode=mkt_mode,
|
||||
)
|
||||
init_api_keys(
|
||||
client=client,
|
||||
conf=conf,
|
||||
)
|
||||
fq_pairs: dict[str, Pair] = await client.exch_info()
|
||||
assert fq_pairs
|
||||
log.info(
|
||||
f'Loaded multi-venue `Client` in mkt_mode={client.mkt_mode!r}\n\n'
|
||||
f'Symbology Summary:\n'
|
||||
f'------ - ------\n'
|
||||
f'spot: {len(client._spot_pairs)}\n'
|
||||
f'usdtm_futes: {len(client._ufutes_pairs)}\n'
|
||||
'------ - ------\n'
|
||||
f'total: {len(client._pairs)}\n'
|
||||
)
|
||||
yield client
|
||||
|
|
|
|||
|
|
@ -264,15 +264,20 @@ async def open_trade_dialog(
|
|||
# do a open_symcache() call.. though maybe we can hide
|
||||
# this in a new async version of open_account()?
|
||||
async with open_cached_client('binance') as client:
|
||||
subconf: dict = client.conf[venue_name]
|
||||
use_testnet = subconf.get('use_testnet', False)
|
||||
subconf: dict|None = client.conf.get(venue_name)
|
||||
|
||||
# XXX: if no futes.api_key or spot.api_key has been set we
|
||||
# always fall back to the paper engine!
|
||||
if not subconf.get('api_key'):
|
||||
if (
|
||||
not subconf
|
||||
or
|
||||
not subconf.get('api_key')
|
||||
):
|
||||
await ctx.started('paper')
|
||||
return
|
||||
|
||||
use_testnet: bool = subconf.get('use_testnet', False)
|
||||
|
||||
async with (
|
||||
open_cached_client('binance') as client,
|
||||
):
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ import tractor
|
|||
|
||||
from piker.brokers import (
|
||||
open_cached_client,
|
||||
NoData,
|
||||
)
|
||||
from piker._cacheables import (
|
||||
async_lifo_cache,
|
||||
|
|
@ -252,24 +253,30 @@ async def open_history_client(
|
|||
else:
|
||||
client.mkt_mode = 'spot'
|
||||
|
||||
# NOTE: always query using their native symbology!
|
||||
mktid: str = mkt.bs_mktid
|
||||
array = await client.bars(
|
||||
mktid,
|
||||
array: np.ndarray = await client.bars(
|
||||
mkt=mkt,
|
||||
start_dt=start_dt,
|
||||
end_dt=end_dt,
|
||||
)
|
||||
if array.size == 0:
|
||||
raise NoData(
|
||||
f'No frame for {start_dt} -> {end_dt}\n'
|
||||
)
|
||||
|
||||
times = array['time']
|
||||
if (
|
||||
end_dt is None
|
||||
):
|
||||
inow = round(time.time())
|
||||
if not times.any():
|
||||
raise ValueError(
|
||||
'Bad frame with null-times?\n\n'
|
||||
f'{times}'
|
||||
)
|
||||
|
||||
if end_dt is None:
|
||||
inow: int = round(time.time())
|
||||
if (inow - times[-1]) > 60:
|
||||
await tractor.pause()
|
||||
|
||||
start_dt = from_timestamp(times[0])
|
||||
end_dt = from_timestamp(times[-1])
|
||||
|
||||
return array, start_dt, end_dt
|
||||
|
||||
yield get_ohlc, {'erlangs': 3, 'rate': 3}
|
||||
|
|
|
|||
|
|
@ -137,10 +137,12 @@ class SpotPair(Pair, frozen=True):
|
|||
quoteOrderQtyMarketAllowed: bool
|
||||
isSpotTradingAllowed: bool
|
||||
isMarginTradingAllowed: bool
|
||||
otoAllowed: bool
|
||||
|
||||
defaultSelfTradePreventionMode: str
|
||||
allowedSelfTradePreventionModes: list[str]
|
||||
permissions: list[str]
|
||||
permissionSets: list[list[str]]
|
||||
|
||||
# NOTE: see `.data._symcache.SymbologyCache.load()` for why
|
||||
ns_path: str = 'piker.brokers.binance:SpotPair'
|
||||
|
|
|
|||
|
|
@ -27,8 +27,8 @@ from typing import (
|
|||
)
|
||||
import time
|
||||
|
||||
import httpx
|
||||
import pendulum
|
||||
import asks
|
||||
import numpy as np
|
||||
import urllib.parse
|
||||
import hashlib
|
||||
|
|
@ -60,6 +60,11 @@ log = get_logger('piker.brokers.kraken')
|
|||
|
||||
# <uri>/<version>/
|
||||
_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?
|
||||
# in which case we should drop it from the defaults and
|
||||
# instead make a custom fields descr in this module!
|
||||
|
|
@ -135,16 +140,15 @@ class Client:
|
|||
def __init__(
|
||||
self,
|
||||
config: dict[str, str],
|
||||
httpx_client: httpx.AsyncClient,
|
||||
|
||||
name: str = '',
|
||||
api_key: str = '',
|
||||
secret: str = ''
|
||||
) -> None:
|
||||
self._sesh = asks.Session(connections=4)
|
||||
self._sesh.base_location = _url
|
||||
self._sesh.headers.update({
|
||||
'User-Agent':
|
||||
'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
|
||||
})
|
||||
|
||||
self._sesh: httpx.AsyncClient = httpx_client
|
||||
|
||||
self._name = name
|
||||
self._api_key = api_key
|
||||
self._secret = secret
|
||||
|
|
@ -166,10 +170,9 @@ class Client:
|
|||
method: str,
|
||||
data: dict,
|
||||
) -> dict[str, Any]:
|
||||
resp = await self._sesh.post(
|
||||
path=f'/public/{method}',
|
||||
resp: httpx.Response = await self._sesh.post(
|
||||
url=f'/public/{method}',
|
||||
json=data,
|
||||
timeout=float('inf')
|
||||
)
|
||||
return resproc(resp, log)
|
||||
|
||||
|
|
@ -180,18 +183,18 @@ class Client:
|
|||
uri_path: str
|
||||
) -> dict[str, Any]:
|
||||
headers = {
|
||||
'Content-Type':
|
||||
'application/x-www-form-urlencoded',
|
||||
'API-Key':
|
||||
self._api_key,
|
||||
'API-Sign':
|
||||
get_kraken_signature(uri_path, data, self._secret)
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
'API-Key': self._api_key,
|
||||
'API-Sign': get_kraken_signature(
|
||||
uri_path,
|
||||
data,
|
||||
self._secret,
|
||||
),
|
||||
}
|
||||
resp = await self._sesh.post(
|
||||
path=f'/private/{method}',
|
||||
resp: httpx.Response = await self._sesh.post(
|
||||
url=f'/private/{method}',
|
||||
data=data,
|
||||
headers=headers,
|
||||
timeout=float('inf')
|
||||
)
|
||||
return resproc(resp, log)
|
||||
|
||||
|
|
@ -665,10 +668,19 @@ class Client:
|
|||
@acm
|
||||
async def get_client() -> Client:
|
||||
|
||||
conf = get_config()
|
||||
conf: dict[str, Any] = get_config()
|
||||
async with httpx.AsyncClient(
|
||||
base_url=_url,
|
||||
headers=_headers,
|
||||
|
||||
# TODO: is there a way to numerate this?
|
||||
# https://www.python-httpx.org/advanced/clients/#why-use-a-client
|
||||
# connections=4
|
||||
) as trio_client:
|
||||
if conf:
|
||||
client = Client(
|
||||
conf,
|
||||
httpx_client=trio_client,
|
||||
|
||||
# TODO: don't break these up and just do internal
|
||||
# conf lookups instead..
|
||||
|
|
@ -677,7 +689,10 @@ async def get_client() -> Client:
|
|||
secret=conf['secret']
|
||||
)
|
||||
else:
|
||||
client = Client({})
|
||||
client = Client(
|
||||
conf={},
|
||||
httpx_client=trio_client,
|
||||
)
|
||||
|
||||
# at startup, load all symbols, and asset info in
|
||||
# batch requests.
|
||||
|
|
|
|||
|
|
@ -612,18 +612,18 @@ async def open_trade_dialog(
|
|||
|
||||
# enter relay loop
|
||||
await handle_order_updates(
|
||||
client,
|
||||
ws,
|
||||
stream,
|
||||
ems_stream,
|
||||
apiflows,
|
||||
ids,
|
||||
reqids2txids,
|
||||
acnt,
|
||||
api_trans,
|
||||
acctid,
|
||||
acc_name,
|
||||
token,
|
||||
client=client,
|
||||
ws=ws,
|
||||
ws_stream=stream,
|
||||
ems_stream=ems_stream,
|
||||
apiflows=apiflows,
|
||||
ids=ids,
|
||||
reqids2txids=reqids2txids,
|
||||
acnt=acnt,
|
||||
ledger=ledger,
|
||||
acctid=acctid,
|
||||
acc_name=acc_name,
|
||||
token=token,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -639,7 +639,8 @@ async def handle_order_updates(
|
|||
|
||||
# transaction records which will be updated
|
||||
# on new trade clearing events (aka order "fills")
|
||||
ledger_trans: dict[str, Transaction],
|
||||
ledger: TransactionLedger,
|
||||
# ledger_trans: dict[str, Transaction],
|
||||
acctid: str,
|
||||
acc_name: str,
|
||||
token: str,
|
||||
|
|
@ -699,7 +700,8 @@ async def handle_order_updates(
|
|||
# if tid not in ledger_trans
|
||||
}
|
||||
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']
|
||||
reqid = trade.get('userref')
|
||||
|
||||
|
|
@ -747,11 +749,17 @@ async def handle_order_updates(
|
|||
client,
|
||||
api_name_set='wsname',
|
||||
)
|
||||
ppmsgs = trades2pps(
|
||||
acnt,
|
||||
acctid,
|
||||
new_trans,
|
||||
ppmsgs: list[BrokerdPosition] = trades2pps(
|
||||
acnt=acnt,
|
||||
ledger=ledger,
|
||||
acctid=acctid,
|
||||
new_trans=new_trans,
|
||||
)
|
||||
# ppmsgs = trades2pps(
|
||||
# acnt,
|
||||
# acctid,
|
||||
# new_trans,
|
||||
# )
|
||||
for pp_msg in ppmsgs:
|
||||
await ems_stream.send(pp_msg)
|
||||
|
||||
|
|
|
|||
|
|
@ -16,10 +16,9 @@
|
|||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
'''
|
||||
Kucoin broker backend
|
||||
Kucoin cex API backend.
|
||||
|
||||
'''
|
||||
|
||||
from contextlib import (
|
||||
asynccontextmanager as acm,
|
||||
aclosing,
|
||||
|
|
@ -42,7 +41,7 @@ import wsproto
|
|||
from uuid import uuid4
|
||||
|
||||
from trio_typing import TaskStatus
|
||||
import asks
|
||||
import httpx
|
||||
from bidict import bidict
|
||||
import numpy as np
|
||||
import pendulum
|
||||
|
|
@ -212,7 +211,11 @@ def get_config() -> BrokerConfig | None:
|
|||
|
||||
class Client:
|
||||
|
||||
def __init__(self) -> None:
|
||||
def __init__(
|
||||
self,
|
||||
httpx_client: httpx.AsyncClient,
|
||||
) -> None:
|
||||
self._http: httpx.AsyncClient = httpx_client
|
||||
self._config: BrokerConfig|None = get_config()
|
||||
self._pairs: dict[str, KucoinMktPair] = {}
|
||||
self._fqmes2mktids: bidict[str, str] = bidict()
|
||||
|
|
@ -227,18 +230,24 @@ class Client:
|
|||
|
||||
) -> dict[str, str | bytes]:
|
||||
'''
|
||||
Generate authenticated request headers
|
||||
Generate authenticated request headers:
|
||||
|
||||
https://docs.kucoin.com/#authentication
|
||||
https://www.kucoin.com/docs/basic-info/connection-method/authentication/creating-a-request
|
||||
https://www.kucoin.com/docs/basic-info/connection-method/authentication/signing-a-message
|
||||
|
||||
'''
|
||||
|
||||
if not self._config:
|
||||
raise ValueError(
|
||||
'No config found when trying to send authenticated request')
|
||||
'No config found when trying to send authenticated request'
|
||||
)
|
||||
|
||||
str_to_sign = (
|
||||
str(int(time.time() * 1000))
|
||||
+ action + f'/api/{api}/{endpoint.lstrip("/")}'
|
||||
+
|
||||
action
|
||||
+
|
||||
f'/api/{api}/{endpoint.lstrip("/")}'
|
||||
)
|
||||
|
||||
signature = base64.b64encode(
|
||||
|
|
@ -249,6 +258,7 @@ class Client:
|
|||
).digest()
|
||||
)
|
||||
|
||||
# TODO: can we cache this between calls?
|
||||
passphrase = base64.b64encode(
|
||||
hmac.new(
|
||||
self._config.key_secret.encode('utf-8'),
|
||||
|
|
@ -270,8 +280,10 @@ class Client:
|
|||
self,
|
||||
action: Literal['POST', 'GET'],
|
||||
endpoint: str,
|
||||
|
||||
api: str = 'v2',
|
||||
headers: dict = {},
|
||||
|
||||
) -> Any:
|
||||
'''
|
||||
Generic request wrapper for Kucoin API
|
||||
|
|
@ -284,14 +296,19 @@ class Client:
|
|||
api,
|
||||
)
|
||||
|
||||
api_url = f'https://api.kucoin.com/api/{api}/{endpoint}'
|
||||
|
||||
res = await asks.request(action, api_url, headers=headers)
|
||||
|
||||
json = res.json()
|
||||
if 'data' in json:
|
||||
return json['data']
|
||||
req_meth: Callable = getattr(
|
||||
self._http,
|
||||
action.lower(),
|
||||
)
|
||||
res = await req_meth(
|
||||
url=f'/{api}/{endpoint}',
|
||||
headers=headers,
|
||||
)
|
||||
json: dict = res.json()
|
||||
if (data := json.get('data')) is not None:
|
||||
return data
|
||||
else:
|
||||
api_url: str = self._http.base_url
|
||||
log.error(
|
||||
f'Error making request to {api_url} ->\n'
|
||||
f'{pformat(res)}'
|
||||
|
|
@ -349,8 +366,8 @@ class Client:
|
|||
currencies: dict[str, Currency] = {}
|
||||
entries: list[dict] = await self._request(
|
||||
'GET',
|
||||
api='v1',
|
||||
endpoint='currencies',
|
||||
api='v1',
|
||||
)
|
||||
for entry in entries:
|
||||
curr = Currency(**entry).copy()
|
||||
|
|
@ -366,7 +383,10 @@ class Client:
|
|||
dict[str, KucoinMktPair],
|
||||
bidict[str, KucoinMktPair],
|
||||
]:
|
||||
entries = await self._request('GET', 'symbols')
|
||||
entries = await self._request(
|
||||
'GET',
|
||||
endpoint='symbols',
|
||||
)
|
||||
log.info(f' {len(entries)} Kucoin market pairs fetched')
|
||||
|
||||
pairs: dict[str, KucoinMktPair] = {}
|
||||
|
|
@ -567,10 +587,18 @@ def fqme_to_kucoin_sym(
|
|||
|
||||
@acm
|
||||
async def get_client() -> AsyncGenerator[Client, None]:
|
||||
client = Client()
|
||||
'''
|
||||
Load an API `Client` preconfigured from user settings
|
||||
|
||||
async with trio.open_nursery() as n:
|
||||
n.start_soon(client.get_mkt_pairs)
|
||||
'''
|
||||
async with (
|
||||
httpx.AsyncClient(
|
||||
base_url=f'https://api.kucoin.com/api',
|
||||
) as trio_client,
|
||||
):
|
||||
client = Client(httpx_client=trio_client)
|
||||
async with trio.open_nursery() as tn:
|
||||
tn.start_soon(client.get_mkt_pairs)
|
||||
await client.get_currencies()
|
||||
|
||||
yield client
|
||||
|
|
|
|||
Loading…
Reference in New Issue