Initial watchlist management cli

kivy_mainline_and_py3.8
K0nstantine 2018-03-21 00:52:06 -04:00
parent 397c27e05a
commit 7ada8a291e
1 changed files with 119 additions and 0 deletions

View File

@ -2,6 +2,11 @@
Console interface to broker client/daemons. Console interface to broker client/daemons.
""" """
from functools import partial from functools import partial
from importlib import import_module
from os import path, makedirs, stat
from collections import defaultdict
import json
import ast
import click import click
import trio import trio
@ -13,6 +18,8 @@ from .brokers import core, get_brokermod
log = get_logger('cli') log = get_logger('cli')
DEFAULT_BROKER = 'robinhood' DEFAULT_BROKER = 'robinhood'
_config_dir = click.get_app_dir('piker')
_watchlists_data_path = path.join(_config_dir, 'watchlists.json')
def run(main, loglevel='info'): def run(main, loglevel='info'):
log = get_console_log(loglevel) log = get_console_log(loglevel)
@ -135,3 +142,115 @@ def watch(loglevel, broker, rate, name):
rate = broker_limit rate = broker_limit
log.warn(f"Limiting {brokermod.__name__} query rate to {rate}/sec") log.warn(f"Limiting {brokermod.__name__} query rate to {rate}/sec")
trio.run(_async_main, name, watchlists[name], brokermod, rate) trio.run(_async_main, name, watchlists[name], brokermod, rate)
# broker_conf_path = os.path.join(
# click.get_app_dir('piker'), 'watchlists.json')
# from piker.testing import _quote_streamer as brokermod
trio.run(_async_main, name, watchlists[name], brokermod)
@cli.group()
@click.option('--loglevel', '-l', default='warning', help='Logging level')
@click.pass_context
def watchlists(ctx, loglevel):
"""Watchlists cl commands and operations
"""
# import pdb; pdb.set_trace()
get_console_log(loglevel) # activate console logging
ctx.obj = {}
if not path.isdir(_config_dir):
log.debug(f"Creating config dir {_config_dir}")
makedirs(_config_dir)
if path.isfile(_watchlists_data_path):
f = open(_watchlists_data_path, 'r')
if not stat(_watchlists_data_path).st_size == 0:
ctx.obj = json.load(f)
f.close()
else:
f = open(_watchlists_data_path, 'w')
f.close()
@watchlists.command(help='show watchlist')
@click.argument('name', nargs=1, required=False)
@click.pass_context
def show(ctx, name):
watchlist = ctx.obj
click.echo(colorize_json(
watchlist if name is None else watchlist[name]))
@watchlists.command(help='add a new watchlist')
@click.argument('name', nargs=1, required=True)
@click.pass_context
def new(ctx, name):
watchlist = ctx.obj
f = open(_watchlists_data_path, 'w')
watchlist.setdefault(name, [])
json.dump(watchlist, f)
f.close()
@watchlists.command(help='add ticker to watchlist')
@click.argument('name', nargs=1, required=True)
@click.argument('ticker_name', nargs=1, required=True)
@click.pass_context
def add(ctx, name, ticker_name):
watchlist = ctx.obj
f = open(_watchlists_data_path, 'w')
if name in watchlist:
watchlist[name].append(str(ticker_name).upper())
json.dump(watchlist, f)
f.close()
@watchlists.command(help='remove ticker from watchlist')
@click.argument('name', nargs=1, required=True)
@click.argument('ticker_name', nargs=1, required=True)
@click.pass_context
def remove(ctx, name, ticker_name):
watchlist = ctx.obj
f = open(_watchlists_data_path, 'w')
if name in watchlist:
watchlist[name].remove(str(ticker_name).upper())
json.dump(watchlist, f)
f.close()
@watchlists.command(help='delete watchlist')
@click.argument('name', nargs=1, required=True)
@click.pass_context
def delete(ctx, name):
watchlist = ctx.obj
f = open(_watchlists_data_path, 'w')
if name in watchlist:
del watchlist[name]
json.dump(watchlist, f)
f.close()
@watchlists.command(help='merge a watchlist from another user')
@click.argument('watchlist_to_merge', nargs=1, required=True)
@click.pass_context
def merge(ctx, watchlist_to_merge):
watchlist = ctx.obj
f = open(_watchlists_data_path, 'w')
merged_watchlist = defaultdict(list)
watchlist_to_merge = ast.literal_eval(watchlist_to_merge)
for d in (watchlist, watchlist_to_merge):
for key, value in d.items():
merged_watchlist[key].extend(value)
json.dump(merged_watchlist, f)
f.close()
print('merge these') #remember to convert to set
@watchlists.command(help='dump a text respresentation of a watchlist to console')
@click.argument('name', nargs=1, required=False)
@click.pass_context
def dump(ctx, name):
watchlist = ctx.obj
f = open(_watchlists_data_path, 'r')
print(json.dumps(watchlist))
f.close()