pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

Drop `marketstore` mod import from CLIs loader 

Means commenting out the `data.cli.ingest()` as it will be deleted in
the up coming #486 anyway.
poetry2nix
Tyler Goodlet 2023-08-16 12:15:49 -04:00
parent 7d84b5a3e5
commit 19a6f5c25c
2 changed files with 43 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
piker/cli/__init__.py
View File

@ -226,8 +226,6 @@ def services(config, tl, ports):
def _load_clis() -> None: def _load_clis() -> None:
from ..service import marketstore # noqa
from ..service import elastic
from ..data import cli # noqa from ..data import cli # noqa
from ..brokers import cli # noqa from ..brokers import cli # noqa
from ..ui import cli # noqa from ..ui import cli # noqa

86
piker/data/cli.py
View File

@ -22,14 +22,14 @@ import trio
import tractor import tractor
import click import click
from ..service.marketstore import ( # from ..service.marketstore import (
# get_client, # # get_client,
# stream_quotes, # # stream_quotes,
ingest_quote_stream, # ingest_quote_stream,
# _url, # # _url,
# _tick_tbk_ids, # # _tick_tbk_ids,
# mk_tbk, # # mk_tbk,
) # )
from ..cli import cli from ..cli import cli
from .. import watchlists as wl from .. import watchlists as wl
from ._util import ( from ._util import (
@ -212,44 +212,44 @@ def storage(
trio.run(main) trio.run(main)
@cli.command() # @cli.command()
@click.option('--test-file', '-t', help='Test quote stream file') # @click.option('--test-file', '-t', help='Test quote stream file')
@click.option('--tl', is_flag=True, help='Enable tractor logging') # @click.option('--tl', is_flag=True, help='Enable tractor logging')
@click.argument('name', nargs=1, required=True) # @click.argument('name', nargs=1, required=True)
@click.pass_obj # @click.pass_obj
def ingest(config, name, test_file, tl): # def ingest(config, name, test_file, tl):
''' # '''
Ingest real-time broker quotes and ticks to a marketstore instance. # Ingest real-time broker quotes and ticks to a marketstore instance.
''' # '''
# global opts # # global opts
loglevel = config['loglevel'] # loglevel = config['loglevel']
tractorloglevel = config['tractorloglevel'] # tractorloglevel = config['tractorloglevel']
# log = config['log'] # # log = config['log']
watchlist_from_file = wl.ensure_watchlists(config['wl_path']) # watchlist_from_file = wl.ensure_watchlists(config['wl_path'])
watchlists = wl.merge_watchlist(watchlist_from_file, wl._builtins) # watchlists = wl.merge_watchlist(watchlist_from_file, wl._builtins)
symbols = watchlists[name] # symbols = watchlists[name]
grouped_syms = {} # grouped_syms = {}
for sym in symbols: # for sym in symbols:
symbol, _, provider = sym.rpartition('.') # symbol, _, provider = sym.rpartition('.')
if provider not in grouped_syms: # if provider not in grouped_syms:
grouped_syms[provider] = [] # grouped_syms[provider] = []
grouped_syms[provider].append(symbol) # grouped_syms[provider].append(symbol)
async def entry_point(): # async def entry_point():
async with tractor.open_nursery() as n: # async with tractor.open_nursery() as n:
for provider, symbols in grouped_syms.items(): # for provider, symbols in grouped_syms.items():
await n.run_in_actor( # await n.run_in_actor(
ingest_quote_stream, # ingest_quote_stream,
name='ingest_marketstore', # name='ingest_marketstore',
symbols=symbols, # symbols=symbols,
brokername=provider, # brokername=provider,
tries=1, # tries=1,
actorloglevel=loglevel, # actorloglevel=loglevel,
loglevel=tractorloglevel # loglevel=tractorloglevel
) # )
tractor.run(entry_point) # tractor.run(entry_point)