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

.data.history: update to new naming 

-> `._source.def_iohlcv_fields`
-> `.storage.StorageClient`
basic_buy_bot
Tyler Goodlet 2023-05-31 17:56:32 -04:00
parent bf21d2e329
commit af64152640
1 changed files with 9 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
piker/data/history.py
View File

@ -48,7 +48,7 @@ from ._sharedmem import (
ShmArray, ShmArray,
_secs_in_day, _secs_in_day,
) )
from ._source import base_iohlc_dtype from ._source import def_iohlcv_fields
from ._sampling import ( from ._sampling import (
open_sample_stream, open_sample_stream,
) )
@ -57,7 +57,7 @@ from ..brokers._util import (
) )
if TYPE_CHECKING: if TYPE_CHECKING:
from ..service.marketstore import Storage from ..service.marketstore import StorageClient
from .feed import _FeedsBus from .feed import _FeedsBus
@ -87,7 +87,7 @@ async def start_backfill(
feed_is_live: trio.Event, feed_is_live: trio.Event,
last_tsdb_dt: datetime | None = None, last_tsdb_dt: datetime | None = None,
storage: Storage | None = None, storage: StorageClient | None = None,
write_tsdb: bool = True, write_tsdb: bool = True,
tsdb_is_up: bool = False, tsdb_is_up: bool = False,
@ -177,6 +177,7 @@ async def start_backfill(
# based on the sample step size, maybe load a certain amount history # based on the sample step size, maybe load a certain amount history
if last_tsdb_dt is None: if last_tsdb_dt is None:
if step_size_s not in (1, 60): if step_size_s not in (1, 60):
raise ValueError( raise ValueError(
'`piker` only needs to support 1m and 1s sampling ' '`piker` only needs to support 1m and 1s sampling '
@ -388,7 +389,7 @@ async def tsdb_backfill(
mod: ModuleType, mod: ModuleType,
storemod: ModuleType, storemod: ModuleType,
bus: _FeedsBus, bus: _FeedsBus,
storage: Storage, storage: StorageClient,
mkt: MktPair, mkt: MktPair,
shms: dict[int, ShmArray], shms: dict[int, ShmArray],
sampler_stream: tractor.MsgStream, sampler_stream: tractor.MsgStream,
@ -406,6 +407,7 @@ async def tsdb_backfill(
fqme: str = mkt.fqme fqme: str = mkt.fqme
# start history anal and load missing new data via backend. # start history anal and load missing new data via backend.
timeframe: int
for timeframe, shm in shms.items(): for timeframe, shm in shms.items():
# loads a (large) frame of data from the tsdb depending # loads a (large) frame of data from the tsdb depending
# on the db's query size limit. # on the db's query size limit.
@ -527,7 +529,7 @@ async def tsdb_backfill(
len(tsdb_history) len(tsdb_history)
): ):
# load the first (smaller) bit of history originally loaded # load the first (smaller) bit of history originally loaded
# above from ``Storage.load()``. # above from ``StorageClient.load()``.
to_push = tsdb_history[-prepend_start:] to_push = tsdb_history[-prepend_start:]
shm.push( shm.push(
to_push, to_push,
@ -645,7 +647,7 @@ async def manage_history(
key=f'piker.{service}[{uuid[:16]}].{fqme}.hist', key=f'piker.{service}[{uuid[:16]}].{fqme}.hist',
# use any broker defined ohlc dtype: # use any broker defined ohlc dtype:
dtype=getattr(mod, '_ohlc_dtype', base_iohlc_dtype), dtype=getattr(mod, '_ohlc_dtype', def_iohlcv_fields),
# we expect the sub-actor to write # we expect the sub-actor to write
readonly=False, readonly=False,
@ -662,7 +664,7 @@ async def manage_history(
key=f'piker.{service}[{uuid[:16]}].{fqme}.rt', key=f'piker.{service}[{uuid[:16]}].{fqme}.rt',
# use any broker defined ohlc dtype: # use any broker defined ohlc dtype:
dtype=getattr(mod, '_ohlc_dtype', base_iohlc_dtype), dtype=getattr(mod, '_ohlc_dtype', def_iohlcv_fields),
# we expect the sub-actor to write # we expect the sub-actor to write
readonly=False, readonly=False,