get positions working for kraken

gb_kraken_orders
Konstantine Tsafatinos 2021-10-29 15:51:41 -04:00 committed by Tyler Goodlet
parent 4a3515541d
commit ca8ef26ea5
4 changed files with 76 additions and 26 deletions

View File

@ -8,8 +8,8 @@ expires_at = 1616095326.355846
[kraken] [kraken]
key_descr = "api_0" key_descr = "api_0"
public_key = "" api_key = ""
private_key = "" secret = ""
[ib] [ib]
host = "127.0.0.1" host = "127.0.0.1"

View File

@ -138,6 +138,19 @@ class OHLC:
ticks: List[Any] = field(default_factory=list) ticks: List[Any] = field(default_factory=list)
def get_config() -> dict[str, Any]:
conf, path = config.load()
section = conf.get('kraken')
if section is None:
log.warning(f'No config section found for kraken in {path}')
return {}
return section
def get_kraken_signature( def get_kraken_signature(
urlpath: str, urlpath: str,
data: Dict[str, Any], data: Dict[str, Any],
@ -170,6 +183,7 @@ class Client:
'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)' 'krakenex/2.1.0 (+https://github.com/veox/python3-krakenex)'
}) })
self._pairs: list[str] = [] self._pairs: list[str] = []
self._name = ''
self._api_key = '' self._api_key = ''
self._secret = '' self._secret = ''
@ -258,7 +272,7 @@ class Client:
else: else:
positions[pair] /= asset_balance positions[pair] /= asset_balance
return positions return positions, vols
async def symbol_info( async def symbol_info(
self, self,
@ -373,16 +387,15 @@ class Client:
async def get_client() -> Client: async def get_client() -> Client:
client = Client() client = Client()
conf, path = config.load() ## TODO: maybe add conditional based on section
section = conf.get('kraken') section = get_config()
client._name = section['key_descr']
client._api_key = section['api_key'] client._api_key = section['api_key']
client._secret = section['secret'] client._secret = section['secret']
data = { ## TODO: Add a client attribute to hold this info
# add non-nonce and non-ofs vars #data = {
} # # add non-nonce and non-ofs vars
# positions = await client.get_positions(data) #}
# await tractor.breakpoint()
# at startup, load all symbols locally for fast search # at startup, load all symbols locally for fast search
await client.cache_symbols() await client.cache_symbols()
@ -390,6 +403,36 @@ async def get_client() -> Client:
yield client yield client
def pack_position(
acc: str,
symkey: str,
pos: float,
vol: float
) -> dict[str, Any]:
return BrokerdPosition(
broker='kraken',
account=acc,
symbol=symkey,
currency=symkey[-3:],
size=float(vol),
avg_price=float(pos),
)
def normalize_symbol(
ticker: str
) -> str:
symlen = len(ticker)
if symlen == 6:
return ticker.lower()
else:
for sym in ['XXBT', 'XXMR', 'ZEUR']:
if sym in ticker:
ticker = ticker.replace(sym, sym[1:])
return ticker.lower()
@tractor.context @tractor.context
async def trades_dialogue( async def trades_dialogue(
ctx: tractor.Context, ctx: tractor.Context,
@ -399,23 +442,28 @@ async def trades_dialogue(
# XXX: required to propagate ``tractor`` loglevel to piker logging # XXX: required to propagate ``tractor`` loglevel to piker logging
get_console_log(loglevel or tractor.current_actor().loglevel) get_console_log(loglevel or tractor.current_actor().loglevel)
# deliver positions to subscriber before anything else # Authenticated block
# positions = await _trio_run_client_method(method='positions') async with get_client() as client:
acc_name = 'kraken.' + client._name
positions, vols = await client.get_positions()
global _accounts2clients all_positions = []
positions = await client.get_positions() for ticker, pos in positions.items():
norm_sym = normalize_symbol(ticker)
if float(vols[ticker]) != 0:
msg = pack_position(acc_name, norm_sym, pos, vols[ticker])
all_positions.append(msg.dict())
#await tractor.breakpoint()
await tractor.breakpoint() await ctx.started((all_positions, (acc_name,)))
await trio.sleep_forever()
all_positions = {} # async with (
# ctx.open_stream() as ems_stream,
for pos in positions: #
msg = pack_position(pos)
all_positions[msg.symbol] = msg.dict()
await ctx.started(all_positions)
async def stream_messages(ws): async def stream_messages(ws):

View File

@ -489,7 +489,8 @@ async def open_brokerd_trades_dialogue(
finally: finally:
# parent context must have been closed # parent context must have been closed
# remove from cache so next client will respawn if needed # remove from cache so next client will respawn if needed
_router.relays.pop(broker) ## TODO: Maybe add a warning
_router.relays.pop(broker, None)
@tractor.context @tractor.context

View File

@ -389,7 +389,7 @@ async def handle_order_requests(
account = request_msg['account'] account = request_msg['account']
if account != 'paper': if account != 'paper':
log.error( log.error(
'On a paper account, only a `paper` selection is valid' 'This is a paper account, only a `paper` selection is valid'
) )
await ems_order_stream.send(BrokerdError( await ems_order_stream.send(BrokerdError(
oid=request_msg['oid'], oid=request_msg['oid'],
@ -463,7 +463,8 @@ async def trades_dialogue(
): ):
# TODO: load paper positions per broker from .toml config file # TODO: load paper positions per broker from .toml config file
# and pass as symbol to position data mapping: ``dict[str, dict]`` # and pass as symbol to position data mapping: ``dict[str, dict]``
await ctx.started(({}, ['paper'])) # await ctx.started(all_positions)
await ctx.started(({}, {'paper',}))
async with ( async with (
ctx.open_stream() as ems_stream, ctx.open_stream() as ems_stream,