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

Get order cancellation working

basic_alerts
Tyler Goodlet 2021-01-07 22:08:25 -05:00
parent dba8457be9
commit 39e4953a6a
2 changed files with 132 additions and 139 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

227
piker/_ems.py
View File

@ -18,12 +18,11 @@
In suit parlance: "Execution management systems" In suit parlance: "Execution management systems"
""" """
import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import ( from typing import (
AsyncIterator, Dict, Callable, Tuple, AsyncIterator, Dict, Callable, Tuple,
Any,
) )
# import uuid
import trio import trio
from trio_typing import TaskStatus from trio_typing import TaskStatus
@ -71,11 +70,6 @@ class OrderBook:
symbol: 'Symbol', symbol: 'Symbol',
price: float price: float
) -> str: ) -> str:
# XXX: should make this an explicit attr
# it's assigned inside ``.add_plot()``
# lc = self.view.linked_charts
# uid = str(uuid.uuid4())
cmd = { cmd = {
'msg': 'alert', 'msg': 'alert',
'price': price, 'price': price,
@ -86,17 +80,22 @@ class OrderBook:
self._sent_orders[uuid] = cmd self._sent_orders[uuid] = cmd
self._to_ems.send_nowait(cmd) self._to_ems.send_nowait(cmd)
async def buy(self, price: float) -> str: def buy(self, price: float) -> str:
... ...
async def sell(self, price: float) -> str: def sell(self, price: float) -> str:
... ...
async def cancel(self, oid: str) -> bool: def cancel(self, uuid: str) -> bool:
"""Cancel an order (or alert) from the EMS. """Cancel an order (or alert) from the EMS.
""" """
... cmd = {
'msg': 'cancel',
'oid': uuid,
}
self._sent_orders[uuid] = cmd
self._to_ems.send_nowait(cmd)
# higher level operations # higher level operations
@ -138,9 +137,7 @@ async def send_order_cmds():
"pushed" from the parent to the EMS actor. "pushed" from the parent to the EMS actor.
""" """
global _from_order_book global _from_order_book
# book = get_orders()
async for cmd in _from_order_book: async for cmd in _from_order_book:
@ -148,27 +145,6 @@ async def send_order_cmds():
log.info(f'sending order cmd: {cmd}') log.info(f'sending order cmd: {cmd}')
yield cmd yield cmd
# lc = order['chart']
# symbol = order['symol']
# msg = order['msg']
# price = order['price']
# oid = order['oid']
# TODO
# oid = str(uuid.uuid4())
# cmd = {
# 'price': price,
# 'action': 'alert',
# 'symbol': symbol.key,
# 'brokers': symbol.brokers,
# 'msg': msg,
# 'price': price,
# 'oid': oid,
# }
# book._sent_orders[oid] = cmd
# TODO: numba all of this # TODO: numba all of this
def mk_check(trigger_price, known_last) -> Callable[[float, float], bool]: def mk_check(trigger_price, known_last) -> Callable[[float, float], bool]:
@ -210,13 +186,13 @@ class _ExecBook:
# levels which have an executable action (eg. alert, order, signal) # levels which have an executable action (eg. alert, order, signal)
orders: Dict[ orders: Dict[
Tuple[str, str], Tuple[str, str],
Tuple[ Dict[
# predicates str, # uuid
Callable[[float], bool], Tuple[
Callable[[float], bool], # predicate
# actions str, # name
Callable[[float], Dict[str, Any]], dict, # cmd / msg type
]
] ]
] = field(default_factory=dict) ] = field(default_factory=dict)
@ -273,6 +249,7 @@ async def exec_orders(
# XXX: optimize this for speed # XXX: optimize this for speed
############################## ##############################
start = time.time()
for sym, quote in quotes.items(): for sym, quote in quotes.items():
execs = book.orders.get((broker, sym)) execs = book.orders.get((broker, sym))
@ -289,7 +266,7 @@ async def exec_orders(
if not execs: if not execs:
continue continue
for oid, pred, name, cmd in tuple(execs): for oid, (pred, name, cmd) in tuple(execs.items()):
# push trigger msg back to parent as an "alert" # push trigger msg back to parent as an "alert"
# (mocking for eg. a "fill") # (mocking for eg. a "fill")
@ -299,25 +276,19 @@ async def exec_orders(
cmd['index'] = feed.shm._last.value - 1 cmd['index'] = feed.shm._last.value - 1
# current shm array index # current shm array index
cmd['trigger_price'] = price cmd['trigger_price'] = price
cmd['msg'] = 'executed'
await ctx.send_yield(cmd) await ctx.send_yield(cmd)
# await ctx.send_yield({
# 'type': 'alert',
# 'price': price,
# # current shm array index
# 'index': feed.shm._last.value - 1,
# 'name': name,
# 'oid': oid,
# })
print( print(
f"GOT ALERT FOR {exec_price} @ \n{tick}\n") f"GOT ALERT FOR {exec_price} @ \n{tick}\n")
print(f'removing pred for {oid}') print(f'removing pred for {oid}')
execs.remove((oid, pred, name, cmd)) pred, name, cmd = execs.pop(oid)
print(f'execs are {execs}') print(f'execs are {execs}')
print(f'execs scan took: {time.time() - start}')
# feed teardown # feed teardown
@ -333,6 +304,8 @@ async def stream_and_route(ctx, ui_name):
actor = tractor.current_actor() actor = tractor.current_actor()
book = get_book() book = get_book()
_active_execs: Dict[str, (str, str)] = {}
# new router entry point # new router entry point
async with tractor.wait_for_actor(ui_name) as portal: async with tractor.wait_for_actor(ui_name) as portal:
@ -341,52 +314,63 @@ async def stream_and_route(ctx, ui_name):
async for cmd in await portal.run(send_order_cmds): async for cmd in await portal.run(send_order_cmds):
log.info(f'{cmd} received in {actor.uid}')
msg = cmd['msg'] msg = cmd['msg']
if msg == 'cancel':
# TODO:
pass
trigger_price = cmd['price']
sym = cmd['symbol']
brokers = cmd['brokers']
oid = cmd['oid'] oid = cmd['oid']
if msg == 'alert': if msg == 'cancel':
log.info(f'Alert {cmd} received in {actor.uid}') # destroy exec
pred, name, cmd = book.orders[_active_execs[oid]].pop(oid)
broker = brokers[0] # ack-cmdond that order is live
last = book.lasts.get((broker, sym)) await ctx.send_yield({'msg': 'cancelled', 'oid': oid})
if last is None: # spawn new brokerd feed task continue
quote = await n.start( elif msg in ('alert', 'buy', 'sell',):
exec_orders,
ctx,
# TODO: eventually support N-brokers
broker,
sym,
trigger_price,
)
print(f"received first quote {quote}")
last = book.lasts[(broker, sym)] trigger_price = cmd['price']
print(f'Known last is {last}') sym = cmd['symbol']
brokers = cmd['brokers']
# Auto-gen scanner predicate: broker = brokers[0]
# we automatically figure out what the alert check condition last = book.lasts.get((broker, sym))
# should be based on the current first price received from the
# feed, instead of being like every other shitty tina platform
# that makes the user choose the predicate operator.
pred, name = mk_check(trigger_price, last)
# create list of executions on first entry if last is None: # spawn new brokerd feed task
book.orders.setdefault((broker, sym), []).append(
(oid, pred, name, cmd)
)
# ack-respond that order is live quote = await n.start(
await ctx.send_yield({'msg': 'ack', 'oid': oid}) exec_orders,
ctx,
# TODO: eventually support N-brokers
broker,
sym,
trigger_price,
)
print(f"received first quote {quote}")
last = book.lasts[(broker, sym)]
print(f'Known last is {last}')
# Auto-gen scanner predicate:
# we automatically figure out what the alert check
# condition should be based on the current first
# price received from the feed, instead of being
# like every other shitty tina platform that makes
# the user choose the predicate operator.
pred, name = mk_check(trigger_price, last)
# create list of executions on first entry
book.orders.setdefault(
(broker, sym), {})[oid] = (pred, name, cmd)
# reverse lookup for cancellations
_active_execs[oid] = (broker, sym)
# ack-cmdond that order is live
await ctx.send_yield({
'msg': 'active',
'oid': oid
})
# continue and wait on next order cmd # continue and wait on next order cmd
@ -410,7 +394,7 @@ async def spawn_router_stream_alerts(
portal = await n.start_actor( portal = await n.start_actor(
subactor_name, subactor_name,
rpc_module_paths=[__name__], enable_modules=[__name__],
) )
stream = await portal.run( stream = await portal.run(
stream_and_route, stream_and_route,
@ -424,51 +408,52 @@ async def spawn_router_stream_alerts(
# begin the trigger-alert stream # begin the trigger-alert stream
# this is where we receive **back** messages # this is where we receive **back** messages
# about executions **from** the EMS actor # about executions **from** the EMS actor
async for alert in stream: async for msg in stream:
# delete the line from view # delete the line from view
oid = alert['oid'] oid = msg['oid']
msg_type = alert['msg'] resp = msg['msg']
if msg_type == 'ack': if resp in ('active',):
print(f"order accepted: {alert}") print(f"order accepted: {msg}")
# show line label once order is live # show line label once order is live
order_mode.lines.commit_line(oid) order_mode.lines.commit_line(oid)
continue continue
order_mode.arrows.add( elif resp in ('cancelled',):
oid,
alert['index'],
alert['price'],
pointing='up' if alert['name'] == 'up' else 'down'
)
# print(f'_lines: {_lines}') # delete level from view
print(f'deleting line with oid: {oid}') order_mode.lines.remove_line(uuid=oid)
print(f'deleting line with oid: {oid}')
# delete level from view elif resp in ('executed',):
order_mode.lines.remove_line(uuid=oid)
# chart._vb._lines_editor order_mode.lines.remove_line(uuid=oid)
# _lines.pop(oid).delete() print(f'deleting line with oid: {oid}')
# TODO: this in another task? order_mode.arrows.add(
# not sure if this will ever be a bottleneck, oid,
# we probably could do graphics stuff first tho? msg['index'],
msg['price'],
pointing='up' if msg['name'] == 'up' else 'down'
)
# XXX: linux only for now # DESKTOP NOTIFICATIONS
result = await trio.run_process( #
[ # TODO: this in another task?
'notify-send', # not sure if this will ever be a bottleneck,
'-u', 'normal', # we probably could do graphics stuff first tho?
'-t', '10000',
'piker',
f'alert: {alert}',
],
)
log.runtime(result)
# do we need this? # XXX: linux only for now
# await _from_ems.put(alert) result = await trio.run_process(
[
'notify-send',
'-u', 'normal',
'-t', '10000',
'piker',
f'alert: {msg}',
],
)
log.runtime(result)

44
piker/ui/_interaction.py
View File

@ -308,6 +308,7 @@ class LineEditor:
""" """
line = self._order_lines[uuid] line = self._order_lines[uuid]
line.oid = uuid
line.label.show() line.label.show()
# TODO: other flashy things to indicate the order is active # TODO: other flashy things to indicate the order is active
@ -316,6 +317,13 @@ class LineEditor:
return line return line
def lines_under_cursor(self):
"""Get the line(s) under the cursor position.
"""
# Delete any hoverable under the cursor
return self.chart._cursor._hovered
def remove_line( def remove_line(
self, self,
line: LevelLine = None, line: LevelLine = None,
@ -328,21 +336,17 @@ class LineEditor:
""" """
if line: if line:
# If line is passed delete it uuid = line.oid
line.delete()
elif uuid: # try to look up line from our registry
# try to look up line from our registry line = self._order_lines.pop(uuid)
self._order_lines.pop(uuid).delete()
else: # if hovered remove from cursor set
# Delete any hoverable under the cursor hovered = self.chart._cursor._hovered
cursor = self.chart._cursor if line in hovered:
hovered.remove(line)
for item in cursor._hovered: line.delete()
# hovered items must also offer
# a ``.delete()`` method
item.delete()
@dataclass @dataclass
@ -392,10 +396,15 @@ class OrderMode:
"""Major mode for placing orders on a chart view. """Major mode for placing orders on a chart view.
""" """
chart: 'ChartPlotWidget' chart: 'ChartPlotWidget' # type: ignore # noqa
book: OrderBook book: OrderBook
lines: LineEditor lines: LineEditor
arrows: ArrowEditor arrows: ArrowEditor
_arrow_colors = {
'alert': 'alert_yellow',
'buy': 'buy_green',
'sell': 'sell_red',
}
key_map: Dict[str, Callable] = field(default_factory=dict) key_map: Dict[str, Callable] = field(default_factory=dict)
@ -664,7 +673,6 @@ class ChartView(ViewBox):
price=y price=y
) )
def keyReleaseEvent(self, ev): def keyReleaseEvent(self, ev):
""" """
Key release to normally to trigger release of input mode Key release to normally to trigger release of input mode
@ -686,7 +694,6 @@ class ChartView(ViewBox):
if text == 'a': if text == 'a':
# draw "staged" line under cursor position # draw "staged" line under cursor position
# self._lines_editor.unstage_line()
self.mode.lines.unstage_line() self.mode.lines.unstage_line()
def keyPressEvent(self, ev): def keyPressEvent(self, ev):
@ -732,13 +739,14 @@ class ChartView(ViewBox):
elif text == 'a': elif text == 'a':
# add a line at the current cursor # add a line at the current cursor
# self._lines_editor.stage_line()
self.mode.lines.stage_line() self.mode.lines.stage_line()
elif text == 'd': elif text == 'd':
# delete any lines under the cursor # delete any lines under the cursor
# self._lines_editor.remove_line() mode = self.mode
self.mode.lines.remove_line() for line in mode.lines.lines_under_cursor():
mode.book.cancel(uuid=line.oid)
# XXX: Leaving this for light reference purposes, there # XXX: Leaving this for light reference purposes, there
# seems to be some work to at least gawk at for history mgmt. # seems to be some work to at least gawk at for history mgmt.