Compare commits

...

4 Commits

Author SHA1 Message Date
Tyler Goodlet 513646c802 Use @njit throughout 2021-01-11 08:40:13 -05:00
Tyler Goodlet 4bd42b52c9 Add support for Trades ep 2021-01-11 08:34:59 -05:00
Guillermo Rodriguez e56d065dbc
Add fill_bars function to kraken 2021-01-04 08:30:05 -03:00
Guillermo Rodriguez b1fd986a3a
Change kraken.stream_quotes to use tractor.stream api 2021-01-02 12:35:38 -03:00
5 changed files with 282 additions and 151 deletions

View File

@ -1,5 +1,5 @@
# piker: trading gear for hackers
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of piker0)
# Copyright (C) Tyler Goodlet (in stewardship for piker0)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
@ -15,11 +15,15 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""
Kraken backend.
kraken buttz.
Get da crypto bois pampin da btcccccssssss (and or da tezos)
"""
from contextlib import asynccontextmanager
from dataclasses import dataclass, asdict, field
from typing import List, Dict, Any, Tuple, Optional
from typing import List, Dict, Any, Tuple, Optional, AsyncIterator
import json
import time
@ -28,6 +32,8 @@ from trio_websocket._impl import ConnectionClosed, DisconnectionTimeout
import arrow
import asks
import numpy as np
from numba import njit, float64
from numba import from_dtype
import trio
import tractor
@ -44,8 +50,9 @@ log = get_logger(__name__)
# <uri>/<version>/
_url = 'https://api.kraken.com/0'
_url: str = 'https://api.kraken.com/0'
_epoch_s: int = 1499000000
# Broker specific ohlc schema which includes a vwap field
_ohlc_dtype = [
@ -60,13 +67,49 @@ _ohlc_dtype = [
('bar_wap', float),
]
_trade_dtype = [
# ('index', int),
('price', float),
('volume', float),
('time', int),
('is_bid', bool),
('is_limit', bool),
('exchange', 'U16'),
]
# UI components allow this to be declared such that additional
# (historical) fields can be exposed.
ohlc_dtype = np.dtype(_ohlc_dtype)
trade_dtype = np.dtype(_trade_dtype)
trade_ndtype = from_dtype(trade_dtype)
_show_wap_in_history = True
# @njit
def json2np(
trades: list,
out: np.ndarray,
# _dtype=_trade_dtype,
) -> np.ndarray:
for i, trade in enumerate(trades):
price, volume, t, direction, typ, misc = trade
time_ns = float64(t) * 1e9
is_bid = {'s': False, 'b': True}[direction]
is_trade = True # do we care if it's a market vs. limti?
out[i] = (
float64(price),
float64(volume),
time_ns,
is_bid,
is_trade,
'kraken',
)
return out
class Client:
def __init__(self) -> None:
@ -100,19 +143,57 @@ class Client:
true_pair_key, data = next(iter(resp['result'].items()))
return data
async def trades(
self,
symbol: str = 'XBTUSD',
# UTC 2017-07-02 12:53:20
since: int = 0, # this is a special value indicating epoch of symbol
as_np: bool = True,
) -> dict:
# UTC 2017-07-02 12:53:20 is oldest seconds value
# since_s = max(_epoch_s, int(since))
# pick a timestamp 1H ago
since_ns = time.time_ns() - 60*60*1e9
json = await self._public(
'Trades',
data={
'pair': symbol,
'since': since_ns,
},
)
res = json['result']
last_ns = res.pop('last')
trades = next(iter(res.values()))
out = np.zeros(1000, dtype=trade_ndtype)
array = json2np(trades, out)
return array
async def bars(
self,
symbol: str = 'XBTUSD',
# UTC 2017-07-02 12:53:20
since: int = None,
count: int = 720, # <- max allowed per query
as_np: bool = True,
) -> dict:
if since is None:
"""Retreive OHLC bars.
Note only a max of 720 candles for each sampling interval
can be retreived. To acquire longer term history use .`trades()`
above. See here:
https://support.kraken.com/hc/en-us/articles/218198197-How-to-retrieve-historical-time-and-sales-trading-history-using-the-REST-API-Trades-endpoint-
We're mostly just keeping this method for bookkeeping
but we can probably just remove it eventually.
"""
# member 720 is farthest back they'll go
since = arrow.utcnow().floor('minute').shift(
minutes=-count).timestamp
# UTC 2017-07-02 12:53:20 is oldest seconds value
since = str(max(1499000000, since))
minutes=-720).timestamp
json = await self._public(
'OHLC',
data={
@ -129,12 +210,14 @@ class Client:
first = bars[0]
last_nz_vwap = first[-3]
if last_nz_vwap == 0:
# use close if vwap is zero
last_nz_vwap = first[-4]
# convert all fields to native types
for i, bar in enumerate(bars):
# normalize weird zero-ed vwap values..cmon kraken..
# indicates vwap didn't change since last bar
vwap = float(bar.pop(-3))
@ -290,60 +373,109 @@ def make_sub(pairs: List[str], data: Dict[str, Any]) -> Dict[str, str]:
}
# @tractor.msg.pub
async def stream_quotes(
# get_topics: Callable,
shm_token: Tuple[str, str, List[tuple]],
symbols: List[str] = ['XBTUSD', 'XMRUSD'],
# These are the symbols not expected by the ws api
# they are looked up inside this routine.
sub_type: str = 'ohlc',
loglevel: str = None,
# compat with eventual ``tractor.msg.pub``
topics: Optional[List[str]] = None,
async def fill_bars(
first_bars,
shm,
client,
symbol: str,
count: int = 75
) -> None:
"""Subscribe for ohlc stream of quotes for ``pairs``.
``pairs`` must be formatted <crypto_symbol>/<fiat_symbol>.
"""
# async with get_client() as client:
next_dt = first_bars[0][1]
i = 0
while i < count:
# timestamp in seconds?
since = arrow.get(next_dt).floor(
'minute').shift(minutes=-(2*720)).timestamp
try:
bars_array = await client.bars(symbol=symbol, since=since)
# push to shared mem
shm.push(bars_array, prepend=True)
i += 1
next_dt = bars_array[0][1]
except BaseException as e:
log.exception(e)
await tractor.breakpoint()
_local_buffer_writers = {}
@asynccontextmanager
async def activate_writer(key: str) -> (bool, trio.Nursery):
try:
writer_already_exists = _local_buffer_writers.get(key, False)
if not writer_already_exists:
_local_buffer_writers[key] = True
async with trio.open_nursery() as n:
yield writer_already_exists, n
else:
yield writer_already_exists, None
finally:
_local_buffer_writers.pop(key, None)
@tractor.stream
async def stream_quotes(
ctx: tractor.Context,
symbols: List[str],
shm_token: Tuple[str, str, List[tuple]],
loglevel: str = None,
# compat for @tractor.msg.pub
topics: Any = None
) -> AsyncIterator[Dict[str, Any]]:
# XXX: required to propagate ``tractor`` loglevel to piker logging
get_console_log(loglevel or tractor.current_actor().loglevel)
ws_pairs = {}
async with activate_writer(
shm_token['shm_name']
) as (writer_already_exists, ln):
async with get_client() as client:
# keep client cached for real-time section
for sym in symbols:
ws_pairs[sym] = (await client.symbol_info(sym))['wsname']
# maybe load historical ohlcv in to shared mem
# check if shm has already been created by previous
# feed initialization
writer_exists = get_shm_token(shm_token['shm_name'])
symbol = symbols[0]
if not writer_exists:
if not writer_already_exists:
shm = attach_shm_array(
token=shm_token,
# we are writer
readonly=False,
)
bars = await client.bars(symbol=symbol)
trades = await client.trades(symbol=symbol)
await tractor.breakpoint()
shm.push(bars)
shm_token = shm.token
ln.start_soon(fill_bars, bars, shm, client, symbol)
#
times = shm.array['time']
delay_s = times[-1] - times[times != times[-1]][-1]
subscribe_ohlc_for_increment(shm, delay_s)
yield shm_token, not writer_exists
# pass back token, and bool, signalling if we're the writer
await ctx.send_yield((shm_token, not writer_already_exists))
while True:
try:
async with trio_websocket.open_websocket_url(
'wss://ws.kraken.com/',
'wss://ws.kraken.com',
) as ws:
# XXX: setup subs
@ -379,7 +511,7 @@ async def stream_quotes(
topic, quote = normalize(ohlc_last)
# packetize as {topic: quote}
yield {topic: quote}
await ctx.send_yield({topic: quote})
# keep start of last interval for volume tracking
last_interval_start = ohlc_last.etime
@ -416,7 +548,7 @@ async def stream_quotes(
# if we are the lone tick writer start writing
# the buffer with appropriate trade data
if not writer_exists:
if not writer_already_exists:
# update last entry
# benchmarked in the 4-5 us range
o, high, low, v = shm.array[-1][
@ -453,7 +585,8 @@ async def stream_quotes(
# XXX: format required by ``tractor.msg.pub``
# requires a ``Dict[topic: str, quote: dict]``
yield {topic: quote}
await ctx.send_yield({topic: quote})
except (ConnectionClosed, DisconnectionTimeout):
log.exception("Good job kraken...reconnecting")

View File

@ -1,5 +1,5 @@
# piker: trading gear for hackers
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of piker0)
# Copyright (C) Tyler Goodlet (in stewardship for piker0)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by

View File

@ -20,7 +20,7 @@ Momentum bby.
from typing import AsyncIterator, Optional
import numpy as np
from numba import jit, float64, optional, int64
from numba import njit, float64, optional, int64
from ..data._normalize import iterticks
@ -29,13 +29,12 @@ from ..data._normalize import iterticks
# - how to handle non-plottable values
# - composition of fsps / implicit chaining
@jit(
@njit(
float64[:](
float64[:],
optional(float64),
optional(float64)
),
nopython=True,
nogil=True
)
def ema(
@ -94,7 +93,7 @@ def ema(
return s
# @jit(
# @njit(
# float64[:](
# float64[:],
# int64,

View File

@ -1,5 +1,5 @@
# piker: trading gear for hackers
# Copyright (C) 2018-present Tyler Goodlet (in stewardship of piker0)
# Copyright (C) Tyler Goodlet (in stewardship for piker0)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by

View File

@ -21,7 +21,7 @@ from typing import List, Optional, Tuple
import numpy as np
import pyqtgraph as pg
from numba import jit, float64, int64 # , optional
from numba import njit, float64, int64 # , optional
from PyQt5 import QtCore, QtGui, QtWidgets
from PyQt5.QtCore import QLineF, QPointF
# from numba import types as ntypes
@ -70,7 +70,7 @@ def lines_from_ohlc(row: np.ndarray, w: float) -> Tuple[QLineF]:
return [hl, o, c]
@jit(
@njit(
# TODO: for now need to construct this manually for readonly arrays, see
# https://github.com/numba/numba/issues/4511
# ntypes.Tuple((float64[:], float64[:], float64[:]))(
@ -78,7 +78,6 @@ def lines_from_ohlc(row: np.ndarray, w: float) -> Tuple[QLineF]:
# int64,
# optional(float64),
# ),
nopython=True,
nogil=True
)
def path_arrays_from_ohlc(