Use @njit throughout

piker/brokers/kraken.py

@@ -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")
+

piker/fsp/__init__.py

@@ -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

piker/fsp/_momo.py

@@ -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,

piker/fsp/_volume.py

@@ -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

piker/ui/_graphics/_ohlc.py

@@ -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(