diff --git a/piker/_ems.py b/piker/_ems.py
new file mode 100644
index 00000000..8ede8f31
--- /dev/null
+++ b/piker/_ems.py
@@ -0,0 +1,459 @@
+# piker: trading gear for hackers
+# 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
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+"""
+In suit parlance: "Execution management systems"
+
+"""
+import time
+from dataclasses import dataclass, field
+from typing import (
+    AsyncIterator, Dict, Callable, Tuple,
+)
+
+import trio
+from trio_typing import TaskStatus
+import tractor
+
+from . import data
+from .log import get_logger
+from .data._source import Symbol
+
+
+log = get_logger(__name__)
+
+# setup local ui event streaming channels for request/resp
+# streamging with EMS daemon
+_to_ems, _from_order_book = trio.open_memory_channel(100)
+
+
+@dataclass
+class OrderBook:
+    """Buy-side (client-side ?) order book ctl and tracking.
+
+    A style similar to "model-view" is used here where this api is
+    provided as a supervised control for an EMS actor which does all the
+    hard/fast work of talking to brokers/exchanges to conduct
+    executions.
+
+    Currently, mostly for keeping local state to match the EMS and use
+    received events to trigger graphics updates.
+
+    """
+    _sent_orders: Dict[str, dict] = field(default_factory=dict)
+    _confirmed_orders: Dict[str, dict] = field(default_factory=dict)
+
+    _to_ems: trio.abc.SendChannel = _to_ems
+    _from_order_book: trio.abc.ReceiveChannel = _from_order_book
+
+    def on_fill(self, uuid: str) -> None:
+        cmd = self._sent_orders[uuid]
+        log.info(f"Order executed: {cmd}")
+        self._confirmed_orders[uuid] = cmd
+
+    def alert(
+        self,
+        uuid: str,
+        symbol: 'Symbol',
+        price: float
+    ) -> str:
+        cmd = {
+            'msg': 'alert',
+            'price': price,
+            'symbol': symbol.key,
+            'brokers': symbol.brokers,
+            'oid': uuid,
+        }
+        self._sent_orders[uuid] = cmd
+        self._to_ems.send_nowait(cmd)
+
+    def buy(self, price: float) -> str:
+        ...
+
+    def sell(self, price: float) -> str:
+        ...
+
+    def cancel(self, uuid: str) -> bool:
+        """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
+
+    async def transmit_to_broker(self, price: float) -> str:
+        ...
+
+    async def modify(self, oid: str, price) -> bool:
+        ...
+
+
+_orders: OrderBook = None
+
+
+def get_orders(emsd_uid: Tuple[str, str] = None) -> OrderBook:
+
+    if emsd_uid is not None:
+        # TODO: read in target emsd's active book on startup
+        pass
+
+    global _orders
+
+    if _orders is None:
+        _orders = OrderBook()
+
+    return _orders
+
+
+# TODO: make this a ``tractor.msg.pub``
+async def send_order_cmds():
+    """Order streaming task: deliver orders transmitted from UI
+    to downstream consumers.
+
+    This is run in the UI actor (usually the one running Qt).
+    The UI simply delivers order messages to the above ``_to_ems``
+    send channel (from sync code using ``.send_nowait()``), these values
+    are pulled from the channel here and send to any consumer(s).
+
+    This effectively makes order messages look like they're being
+    "pushed" from the parent to the EMS actor.
+
+    """
+    global _from_order_book
+
+    async for cmd in _from_order_book:
+
+        # send msg over IPC / wire
+        log.info(f'sending order cmd: {cmd}')
+        yield cmd
+
+
+# TODO: numba all of this
+def mk_check(trigger_price, known_last) -> Callable[[float, float], bool]:
+    """Create a predicate for given ``exec_price`` based on last known
+    price, ``known_last``.
+
+    This is an automatic alert level thunk generator based on where the
+    current last known value is and where the specified value of
+    interest is; pick an appropriate comparison operator based on
+    avoiding the case where the a predicate returns true immediately.
+
+    """
+    # str compares:
+    # https://stackoverflow.com/questions/46708708/compare-strings-in-numba-compiled-function
+
+    if trigger_price >= known_last:
+
+        def check_gt(price: float) -> bool:
+            return price >= trigger_price
+
+        return check_gt, 'down'
+
+    elif trigger_price <= known_last:
+
+        def check_lt(price: float) -> bool:
+            return price <= trigger_price
+
+        return check_lt, 'up'
+
+
+@dataclass
+class _ExecBook:
+    """EMS-side execution book.
+
+    Contains conditions for executions (aka "orders").
+    A singleton instance is created per EMS actor (for now).
+
+    """
+    # levels which have an executable action (eg. alert, order, signal)
+    orders: Dict[
+        Tuple[str, str],
+        Dict[
+            str,  # uuid
+            Tuple[
+                Callable[[float], bool],  # predicate
+                str,  # name
+                dict,  # cmd / msg type
+            ]
+        ]
+    ] = field(default_factory=dict)
+
+    # tracks most recent values per symbol each from data feed
+    lasts: Dict[
+        Tuple[str, str],
+        float
+    ] = field(default_factory=dict)
+
+
+_book = None
+
+
+def get_book() -> _ExecBook:
+    global _book
+
+    if _book is None:
+        _book = _ExecBook()
+
+    return _book
+
+
+async def exec_orders(
+    ctx: tractor.Context,
+    broker: str,
+    symbol: str,
+    exec_price: float,
+    task_status: TaskStatus[dict] = trio.TASK_STATUS_IGNORED,
+) -> AsyncIterator[dict]:
+
+    async with data.open_feed(
+        broker,
+        [symbol],
+        loglevel='info',
+    ) as feed:
+
+        # TODO: get initial price
+
+        first_quote = await feed.receive()
+
+        book = get_book()
+        book.lasts[(broker, symbol)] = first_quote[symbol]['last']
+
+        task_status.started(first_quote)
+
+        # shield this field so the remote brokerd does not get cancelled
+        stream = feed.stream
+
+        with stream.shield():
+            async for quotes in stream:
+
+                ##############################
+                # begin price actions sequence
+                # XXX: optimize this for speed
+                ##############################
+
+                start = time.time()
+                for sym, quote in quotes.items():
+
+                    execs = book.orders.get((broker, sym))
+
+                    for tick in quote.get('ticks', ()):
+                        price = tick.get('price')
+                        if price < 0:
+                            # lel, fuck you ib
+                            continue
+
+                        # update to keep new cmds informed
+                        book.lasts[(broker, symbol)] = price
+
+                        if not execs:
+                            continue
+
+                        for oid, (pred, name, cmd) in tuple(execs.items()):
+
+                            # push trigger msg back to parent as an "alert"
+                            # (mocking for eg. a "fill")
+                            if pred(price):
+
+                                cmd['name'] = name
+                                cmd['index'] = feed.shm._last.value - 1
+                                # current shm array index
+                                cmd['trigger_price'] = price
+                                cmd['msg'] = 'executed'
+
+                                await ctx.send_yield(cmd)
+
+                                print(
+                                    f"GOT ALERT FOR {exec_price} @ \n{tick}\n")
+
+                                print(f'removing pred for {oid}')
+                                pred, name, cmd = execs.pop(oid)
+
+                                print(f'execs are {execs}')
+
+                print(f'execs scan took: {time.time() - start}')
+        # feed teardown
+
+
+@tractor.stream
+async def stream_and_route(ctx, ui_name):
+    """Order router (sub)actor entrypoint.
+
+    This is the daemon (child) side routine which starts an EMS
+    runtime per broker/feed and and begins streaming back alerts
+    from executions back to subscribers.
+
+    """
+    actor = tractor.current_actor()
+    book = get_book()
+
+    _active_execs: Dict[str, (str, str)] = {}
+
+    # new router entry point
+    async with tractor.wait_for_actor(ui_name) as portal:
+
+        # spawn one task per broker feed
+        async with trio.open_nursery() as n:
+
+            async for cmd in await portal.run(send_order_cmds):
+
+                log.info(f'{cmd} received in {actor.uid}')
+                msg = cmd['msg']
+                oid = cmd['oid']
+
+                if msg == 'cancel':
+                    # destroy exec
+                    pred, name, cmd = book.orders[_active_execs[oid]].pop(oid)
+
+                    # ack-cmdond that order is live
+                    await ctx.send_yield({'msg': 'cancelled', 'oid': oid})
+
+                    continue
+
+                elif msg in ('alert', 'buy', 'sell',):
+
+                    trigger_price = cmd['price']
+                    sym = cmd['symbol']
+                    brokers = cmd['brokers']
+
+                    broker = brokers[0]
+                    last = book.lasts.get((broker, sym))
+
+                    if last is None:  # spawn new brokerd feed task
+
+                        quote = await n.start(
+                            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
+
+
+async def spawn_router_stream_alerts(
+    order_mode,
+    symbol: Symbol,
+    # lines: 'LinesEditor',
+    task_status: TaskStatus[str] = trio.TASK_STATUS_IGNORED,
+) -> None:
+    """Spawn an EMS daemon and begin sending orders and receiving
+    alerts.
+
+    """
+
+    actor = tractor.current_actor()
+    subactor_name = 'emsd'
+
+    # TODO: add ``maybe_spawn_emsd()`` for this
+    async with tractor.open_nursery() as n:
+
+        portal = await n.start_actor(
+            subactor_name,
+            enable_modules=[__name__],
+        )
+        stream = await portal.run(
+            stream_and_route,
+            ui_name=actor.name
+        )
+
+        async with tractor.wait_for_actor(subactor_name):
+            # let parent task continue
+            task_status.started(_to_ems)
+
+        # begin the trigger-alert stream
+        # this is where we receive **back** messages
+        # about executions **from** the EMS actor
+        async for msg in stream:
+
+            # delete the line from view
+            oid = msg['oid']
+            resp = msg['msg']
+
+            if resp in ('active',):
+                print(f"order accepted: {msg}")
+
+                # show line label once order is live
+                order_mode.lines.commit_line(oid)
+
+                continue
+
+            elif resp in ('cancelled',):
+
+                # delete level from view
+                order_mode.lines.remove_line(uuid=oid)
+                print(f'deleting line with oid: {oid}')
+
+            elif resp in ('executed',):
+
+                order_mode.lines.remove_line(uuid=oid)
+                print(f'deleting line with oid: {oid}')
+
+                order_mode.arrows.add(
+                    oid,
+                    msg['index'],
+                    msg['price'],
+                    pointing='up' if msg['name'] == 'up' else 'down'
+                )
+
+                # DESKTOP NOTIFICATIONS
+                #
+                # TODO: this in another task?
+                # not sure if this will ever be a bottleneck,
+                # we probably could do graphics stuff first tho?
+
+                # XXX: linux only for now
+                result = await trio.run_process(
+                    [
+                        'notify-send',
+                        '-u', 'normal',
+                        '-t', '10000',
+                        'piker',
+                        f'alert: {msg}',
+                    ],
+                )
+                log.runtime(result)
diff --git a/piker/brokers/ib.py b/piker/brokers/ib.py
index 12f713ae..d0645dfe 100644
--- a/piker/brokers/ib.py
+++ b/piker/brokers/ib.py
@@ -49,9 +49,11 @@ from ..data import (
     attach_shm_array,
     # get_shm_token,
     subscribe_ohlc_for_increment,
+    _buffer,
 )
 from ..data._source import from_df
 from ._util import SymbolNotFound
+from .._async_utils import maybe_with_if
 
 
 log = get_logger(__name__)
@@ -355,11 +357,12 @@ class Client:
         symbol: str,
         to_trio,
         opts: Tuple[int] = ('375', '233',),
+        contract: Optional[Contract] = None,
         # opts: Tuple[int] = ('459',),
     ) -> None:
         """Stream a ticker using the std L1 api.
         """
-        contract = await self.find_contract(symbol)
+        contract = contract or (await self.find_contract(symbol))
         ticker: Ticker = self.ib.reqMktData(contract, ','.join(opts))
 
         # define a simple queue push routine that streams quote packets
@@ -386,6 +389,20 @@ class Client:
         # let the engine run and stream
         await self.ib.disconnectedEvent
 
+    async def get_quote(
+        self,
+        symbol: str,
+    ) -> Ticker:
+        """Return a single quote for symbol.
+
+        """
+        contract = await self.find_contract(symbol)
+        ticker: Ticker = self.ib.reqMktData(
+            contract,
+            snapshot=True,
+        )
+        return contract, (await ticker.updateEvent)
+
 
 # default config ports
 _tws_port: int = 7497
@@ -604,16 +621,21 @@ _local_buffer_writers = {}
 
 @asynccontextmanager
 async def activate_writer(key: str) -> (bool, trio.Nursery):
+    """Mark the current actor with module var determining
+    whether an existing shm writer task is already active.
+
+    This avoids more then one writer resulting in data
+    clobbering.
+    """
+    global _local_buffer_writers
+
     try:
-        writer_already_exists = _local_buffer_writers.get(key, False)
+        assert not _local_buffer_writers.get(key, False)
 
-        if not writer_already_exists:
-            _local_buffer_writers[key] = True
+        _local_buffer_writers[key] = True
 
-            async with trio.open_nursery() as n:
-                yield writer_already_exists, n
-        else:
-            yield writer_already_exists, None
+        async with trio.open_nursery() as n:
+            yield n
     finally:
         _local_buffer_writers.pop(key, None)
 
@@ -622,7 +644,7 @@ async def fill_bars(
     sym: str,
     first_bars: list,
     shm: 'ShmArray',  # type: ignore # noqa
-    # count: int = 20,  # NOTE: any more and we'll overrun the underlying buffer
+    # count: int = 20,  # NOTE: any more and we'll overrun underlying buffer
     count: int = 2,  # NOTE: any more and we'll overrun the underlying buffer
 ) -> None:
     """Fill historical bars into shared mem / storage afap.
@@ -692,8 +714,14 @@ async def stream_quotes(
     # TODO: support multiple subscriptions
     sym = symbols[0]
 
+    contract, first_ticker = await _trio_run_client_method(
+        method='get_quote',
+        symbol=sym,
+    )
+
     stream = await _trio_run_client_method(
         method='stream_ticker',
+        contract=contract,  # small speedup
         symbol=sym,
     )
 
@@ -701,14 +729,17 @@ async def stream_quotes(
 
         # check if a writer already is alive in a streaming task,
         # otherwise start one and mark it as now existing
-        async with activate_writer(
-            shm_token['shm_name']
-        ) as (writer_already_exists, ln):
 
-            # maybe load historical ohlcv in to shared mem
-            # check if shm has already been created by previous
-            # feed initialization
+        key = shm_token['shm_name']
+
+        writer_already_exists = _local_buffer_writers.get(key, False)
+
+        # maybe load historical ohlcv in to shared mem
+        # check if shm has already been created by previous
+        # feed initialization
+        async with trio.open_nursery() as ln:
             if not writer_already_exists:
+                _local_buffer_writers[key] = True
 
                 shm = attach_shm_array(
                     token=shm_token,
@@ -744,12 +775,33 @@ async def stream_quotes(
                 subscribe_ohlc_for_increment(shm, delay_s)
 
             # pass back token, and bool, signalling if we're the writer
+            # and that history has been written
             await ctx.send_yield((shm_token, not writer_already_exists))
 
-            # first quote can be ignored as a 2nd with newer data is sent?
-            first_ticker = await stream.__anext__()
+            # check for special contract types
+            if type(first_ticker.contract) not in (ibis.Commodity, ibis.Forex):
+                suffix = 'exchange'
+                # should be real volume for this contract
+                calc_price = False
+            else:
+                # commodities and forex don't have an exchange name and
+                # no real volume so we have to calculate the price
+                suffix = 'secType'
+                calc_price = True
+                ticker = first_ticker
 
-            quote = normalize(first_ticker)
+            # pass first quote asap
+            quote = normalize(first_ticker, calc_price=calc_price)
+            con = quote['contract']
+            topic = '.'.join((con['symbol'], con[suffix])).lower()
+            quote['symbol'] = topic
+
+            first_quote = {topic: quote}
+
+            # yield first quote asap
+            await ctx.send_yield(first_quote)
+
+            # ticker.ticks = []
 
             # ugh, clear ticks since we've consumed them
             # (ahem, ib_insync is stateful trash)
@@ -762,39 +814,31 @@ async def stream_quotes(
 
                 calc_price = False  # should be real volume for contract
 
-                async for ticker in stream:
+            # wait for real volume on feed (trading might be closed)
+            async for ticker in stream:
+
+                # for a real volume contract we rait for the first
+                # "real" trade to take place
+                if not calc_price and not ticker.rtTime:
                     # spin consuming tickers until we get a real market datum
-                    if not ticker.rtTime:
-                        log.debug(f"New unsent ticker: {ticker}")
-                        continue
-                    else:
-                        log.debug("Received first real volume tick")
-                        # ugh, clear ticks since we've consumed them
-                        # (ahem, ib_insync is truly stateful trash)
-                        ticker.ticks = []
+                    log.debug(f"New unsent ticker: {ticker}")
+                    continue
+                else:
+                    log.debug("Received first real volume tick")
+                    # ugh, clear ticks since we've consumed them
+                    # (ahem, ib_insync is truly stateful trash)
+                    ticker.ticks = []
 
-                        # XXX: this works because we don't use
-                        # ``aclosing()`` above?
-                        break
-            else:
-                # commodities don't have an exchange name for some reason?
-                suffix = 'secType'
-                calc_price = True
-                ticker = first_ticker
+                    # tell incrementer task it can start
+                    _buffer.shm_incrementing(key).set()
 
-            quote = normalize(ticker, calc_price=calc_price)
-            con = quote['contract']
-            topic = '.'.join((con['symbol'], con[suffix])).lower()
-            quote['symbol'] = topic
-
-            first_quote = {topic: quote}
-            ticker.ticks = []
-
-            # yield first quote asap
-            await ctx.send_yield(first_quote)
+                    # XXX: this works because we don't use
+                    # ``aclosing()`` above?
+                    break
 
             # real-time stream
             async for ticker in stream:
+
                 # print(ticker.vwap)
                 quote = normalize(
                     ticker,
diff --git a/piker/brokers/kraken.py b/piker/brokers/kraken.py
index 2289f743..dfbf3c0f 100644
--- a/piker/brokers/kraken.py
+++ b/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
@@ -34,6 +34,7 @@ import tractor
 from ._util import resproc, SymbolNotFound, BrokerError
 from ..log import get_logger, get_console_log
 from ..data import (
+    _buffer,
     # iterticks,
     attach_shm_array,
     get_shm_token,
@@ -266,6 +267,7 @@ def normalize(
     quote['broker_ts'] = quote['time']
     quote['brokerd_ts'] = time.time()
     quote['symbol'] = quote['pair'] = quote['pair'].replace('/', '')
+    quote['last'] = quote['close']
 
     # seriously eh? what's with this non-symmetry everywhere
     # in subscription systems...
@@ -381,6 +383,9 @@ async def stream_quotes(
                     # packetize as {topic: quote}
                     yield {topic: quote}
 
+                    # tell incrementer task it can start
+                    _buffer.shm_incrementing(shm_token['shm_name']).set()
+
                     # keep start of last interval for volume tracking
                     last_interval_start = ohlc_last.etime
 
diff --git a/piker/data/__init__.py b/piker/data/__init__.py
index fa26801c..579e596f 100644
--- a/piker/data/__init__.py
+++ b/piker/data/__init__.py
@@ -75,10 +75,12 @@ def get_ingestormod(name: str) -> ModuleType:
     return module
 
 
+# capable rpc modules
 _data_mods = [
     'piker.brokers.core',
     'piker.brokers.data',
     'piker.data',
+    'piker.data._buffer',
 ]
 
 
@@ -104,10 +106,13 @@ async def maybe_spawn_brokerd(
     brokermod = get_brokermod(brokername)
     dname = f'brokerd.{brokername}'
     async with tractor.find_actor(dname) as portal:
+
         # WTF: why doesn't this work?
         if portal is not None:
             yield portal
-        else:
+
+        else:  # no daemon has been spawned yet
+
             log.info(f"Spawning {brokername} broker daemon")
             tractor_kwargs = getattr(brokermod, '_spawn_kwargs', {})
             async with tractor.open_nursery() as nursery:
@@ -115,7 +120,7 @@ async def maybe_spawn_brokerd(
                     # spawn new daemon
                     portal = await nursery.start_actor(
                         dname,
-                        rpc_module_paths=_data_mods + [brokermod.__name__],
+                        enable_modules=_data_mods + [brokermod.__name__],
                         loglevel=loglevel,
                         **tractor_kwargs
                     )
@@ -140,7 +145,7 @@ class Feed:
     stream: AsyncIterator[Dict[str, Any]]
     shm: ShmArray
     # ticks: ShmArray
-    _broker_portal: tractor._portal.Portal
+    _brokerd_portal: tractor._portal.Portal
     _index_stream: Optional[AsyncIterator[Dict[str, Any]]] = None
 
     async def receive(self) -> dict:
@@ -151,9 +156,8 @@ class Feed:
             # XXX: this should be singleton on a host,
             # a lone broker-daemon per provider should be
             # created for all practical purposes
-            self._index_stream = await self._broker_portal.run(
-                'piker.data',
-                'increment_ohlc_buffer',
+            self._index_stream = await self._brokerd_portal.run(
+                increment_ohlc_buffer,
                 shm_token=self.shm.token,
                 topics=['index'],
             )
@@ -200,8 +204,7 @@ async def open_feed(
         loglevel=loglevel,
     ) as portal:
         stream = await portal.run(
-            mod.__name__,
-            'stream_quotes',
+            mod.stream_quotes,
             symbols=symbols,
             shm_token=shm.token,
 
@@ -225,5 +228,5 @@ async def open_feed(
             name=name,
             stream=stream,
             shm=shm,
-            _broker_portal=portal,
+            _brokerd_portal=portal,
         )
diff --git a/piker/data/_buffer.py b/piker/data/_buffer.py
index fed6b965..9a496e7a 100644
--- a/piker/data/_buffer.py
+++ b/piker/data/_buffer.py
@@ -27,6 +27,12 @@ from ._sharedmem import ShmArray
 
 
 _shms: Dict[int, ShmArray] = {}
+_start_increment: Dict[str, trio.Event] = {}
+
+
+def shm_incrementing(shm_token_name: str) -> trio.Event:
+    global _start_increment
+    return _start_increment.setdefault(shm_token_name, trio.Event())
 
 
 @tractor.msg.pub
@@ -47,6 +53,10 @@ async def increment_ohlc_buffer(
     Note that if **no** actor has initiated this task then **none** of
     the underlying buffers will actually be incremented.
     """
+
+    # wait for brokerd to signal we should start sampling
+    await shm_incrementing(shm_token['shm_name']).wait()
+
     # TODO: right now we'll spin printing bars if the last time stamp is
     # before a large period of no market activity.  Likely the best way
     # to solve this is to make this task aware of the instrument's
diff --git a/piker/data/_source.py b/piker/data/_source.py
index 26180443..a77839ff 100644
--- a/piker/data/_source.py
+++ b/piker/data/_source.py
@@ -17,6 +17,7 @@
 """
 numpy data source coversion helpers.
 """
+from typing import List
 import decimal
 from dataclasses import dataclass
 
@@ -81,6 +82,7 @@ class Symbol:
 
     """
     key: str = ''
+    brokers: List[str] = None
     min_tick: float = 0.01
     contract: str = ''
 
diff --git a/piker/ui/_axes.py b/piker/ui/_axes.py
index 08d2e1b5..f9893347 100644
--- a/piker/ui/_axes.py
+++ b/piker/ui/_axes.py
@@ -218,13 +218,13 @@ class AxisLabel(pg.GraphicsObject):
         p.drawRect(self.rect)
 
     def boundingRect(self):  # noqa
-        # if self.label_str:
-        #     self._size_br_from_str(self.label_str)
-        #     return self.rect
+        if self.label_str:
+            self._size_br_from_str(self.label_str)
+            return self.rect
 
-        # return QtCore.QRectF()
+        return QtCore.QRectF()
 
-        return self.rect or QtCore.QRectF()
+        # return self.rect or QtCore.QRectF()
 
     def _size_br_from_str(self, value: str) -> None:
         """Do our best to render the bounding rect to a set margin
diff --git a/piker/ui/_chart.py b/piker/ui/_chart.py
index 6cc8ecb1..39211e61 100644
--- a/piker/ui/_chart.py
+++ b/piker/ui/_chart.py
@@ -16,6 +16,7 @@
 
 """
 High level Qt chart widgets.
+
 """
 from typing import Tuple, Dict, Any, Optional, Callable
 from functools import partial
@@ -31,7 +32,7 @@ from ._axes import (
     PriceAxis,
 )
 from ._graphics._cursor import (
-    CrossHair,
+    Cursor,
     ContentsLabel,
 )
 from ._graphics._lines import (
@@ -56,8 +57,9 @@ from .. import data
 from ..data import maybe_open_shm_array
 from ..log import get_logger
 from ._exec import run_qtractor, current_screen
-from ._interaction import ChartView
+from ._interaction import ChartView, open_order_mode
 from .. import fsp
+from .._ems import spawn_router_stream_alerts
 
 
 log = get_logger(__name__)
@@ -123,10 +125,9 @@ class ChartSpace(QtGui.QWidget):
     # def init_strategy_ui(self):
     #     self.strategy_box = StrategyBoxWidget(self)
     #     self.toolbar_layout.addWidget(self.strategy_box)
-
     def load_symbol(
         self,
-        symbol: str,
+        symbol: Symbol,
         data: np.ndarray,
         ohlc: bool = True,
     ) -> None:
@@ -146,16 +147,15 @@ class ChartSpace(QtGui.QWidget):
         # self.symbol_label.setText(f'/`{symbol}`')
 
         linkedcharts = self._chart_cache.setdefault(
-            symbol,
-            LinkedSplitCharts()
+            symbol.key,
+            LinkedSplitCharts(symbol)
         )
-        s = Symbol(key=symbol)
 
         # remove any existing plots
         if not self.v_layout.isEmpty():
             self.v_layout.removeWidget(linkedcharts)
 
-        main_chart = linkedcharts.plot_ohlc_main(s, data)
+        main_chart = linkedcharts.plot_ohlc_main(symbol, data)
 
         self.v_layout.addWidget(linkedcharts)
 
@@ -181,10 +181,13 @@ class LinkedSplitCharts(QtGui.QWidget):
 
     zoomIsDisabled = QtCore.pyqtSignal(bool)
 
-    def __init__(self):
+    def __init__(
+        self,
+        symbol: Symbol,
+    ) -> None:
         super().__init__()
         self.signals_visible: bool = False
-        self._ch: CrossHair = None  # crosshair graphics
+        self._cursor: Cursor = None  # crosshair graphics
         self.chart: ChartPlotWidget = None  # main (ohlc) chart
         self.subplots: Dict[Tuple[str, ...], ChartPlotWidget] = {}
 
@@ -207,6 +210,13 @@ class LinkedSplitCharts(QtGui.QWidget):
         self.layout.setContentsMargins(0, 0, 0, 0)
         self.layout.addWidget(self.splitter)
 
+        # state tracker?
+        self._symbol: Symbol = symbol
+
+    @property
+    def symbol(self) -> Symbol:
+        return self._symbol
+
     def set_split_sizes(
         self,
         prop: float = 0.28  # proportion allocated to consumer subcharts
@@ -232,7 +242,7 @@ class LinkedSplitCharts(QtGui.QWidget):
         self.digits = symbol.digits()
 
         # add crosshairs
-        self._ch = CrossHair(
+        self._cursor = Cursor(
             linkedsplitcharts=self,
             digits=self.digits
         )
@@ -244,7 +254,7 @@ class LinkedSplitCharts(QtGui.QWidget):
             _is_main=True,
         )
         # add crosshair graphic
-        self.chart.addItem(self._ch)
+        self.chart.addItem(self._cursor)
 
         # axis placement
         if _xaxis_at == 'bottom':
@@ -291,18 +301,19 @@ class LinkedSplitCharts(QtGui.QWidget):
 
             array=array,
             parent=self.splitter,
+            linked_charts=self,
             axisItems={
                 'bottom': xaxis,
                 'right': PriceAxis(linked_charts=self)
             },
             viewBox=cv,
-            cursor=self._ch,
+            cursor=self._cursor,
             **cpw_kwargs,
         )
 
-        # give viewbox a reference to primary chart
-        # allowing for kb controls and interactions
-        # (see our custom view in `._interactions.py`)
+        # give viewbox as reference to chart
+        # allowing for kb controls and interactions on **this** widget
+        # (see our custom view mode in `._interactions.py`)
         cv.chart = cpw
 
         cpw.plotItem.vb.linked_charts = self
@@ -315,7 +326,7 @@ class LinkedSplitCharts(QtGui.QWidget):
         cpw.setXLink(self.chart)
 
         # add to cross-hair's known plots
-        self._ch.add_plot(cpw)
+        self._cursor.add_plot(cpw)
 
         # draw curve graphics
         if style == 'bar':
@@ -365,8 +376,9 @@ class ChartPlotWidget(pg.PlotWidget):
         # the data view we generate graphics from
         name: str,
         array: np.ndarray,
+        linked_charts: LinkedSplitCharts,
         static_yrange: Optional[Tuple[float, float]] = None,
-        cursor: Optional[CrossHair] = None,
+        cursor: Optional[Cursor] = None,
         **kwargs,
     ):
         """Configure chart display settings.
@@ -379,8 +391,8 @@ class ChartPlotWidget(pg.PlotWidget):
             useOpenGL=True,
             **kwargs
         )
-
         self.name = name
+        self._lc = linked_charts
 
         # self.setViewportMargins(0, 0, 0, 0)
         self._ohlc = array  # readonly view of ohlc data
@@ -407,10 +419,6 @@ class ChartPlotWidget(pg.PlotWidget):
 
         self.default_view()
 
-        # TODO: stick in config
-        # use cross-hair for cursor?
-        # self.setCursor(QtCore.Qt.CrossCursor)
-
         # Assign callback for rescaling y-axis automatically
         # based on data contents and ``ViewBox`` state.
         self.sigXRangeChanged.connect(self._set_yrange)
@@ -844,6 +852,8 @@ async def _async_main(
     # historical data fetch
     brokermod = brokers.get_brokermod(brokername)
 
+    symbol = Symbol(sym, [brokername])
+
     async with data.open_feed(
         brokername,
         [sym],
@@ -854,8 +864,7 @@ async def _async_main(
         bars = ohlcv.array
 
         # load in symbol's ohlc data
-        # await tractor.breakpoint()
-        linked_charts, chart = chart_app.load_symbol(sym, bars)
+        linked_charts, chart = chart_app.load_symbol(symbol, bars)
 
         # plot historical vwap if available
         wap_in_history = False
@@ -870,12 +879,13 @@ async def _async_main(
                     add_label=False,
                 )
 
+        # size view to data once at outset
         chart._set_yrange()
 
         # TODO: a data view api that makes this less shit
         chart._shm = ohlcv
 
-        # eventually we'll support some kind of n-compose syntax
+        # TODO: eventually we'll support some kind of n-compose syntax
         fsp_conf = {
             'rsi': {
                 'period': 14,
@@ -887,7 +897,8 @@ async def _async_main(
         }
 
         # make sure that the instrument supports volume history
-        # (sometimes this is not the case for some commodities and derivatives)
+        # (sometimes this is not the case for some commodities and
+        # derivatives)
         volm = ohlcv.array['volume']
         if (
             np.all(np.isin(volm, -1)) or
@@ -928,6 +939,7 @@ async def _async_main(
 
             # wait for a first quote before we start any update tasks
             quote = await feed.receive()
+
             log.info(f'Received first quote {quote}')
 
             n.start_soon(
@@ -938,8 +950,26 @@ async def _async_main(
                 linked_charts
             )
 
-            # probably where we'll eventually start the user input loop
-            await trio.sleep_forever()
+            async with open_order_mode(
+                chart,
+            ) as order_mode:
+
+                # TODO: this should probably be implicitly spawned
+                # inside the above mngr?
+
+                # spawn EMS actor-service
+                to_ems_chan = await n.start(
+                    spawn_router_stream_alerts,
+                    order_mode,
+                    symbol,
+                )
+
+                # wait for router to come up before setting
+                # enabling send channel on chart
+                linked_charts._to_ems = to_ems_chan
+
+                # probably where we'll eventually start the user input loop
+                await trio.sleep_forever()
 
 
 async def chart_from_quotes(
@@ -999,7 +1029,7 @@ async def chart_from_quotes(
         chart,
         # determine precision/decimal lengths
         digits=max(float_digits(last), 2),
-        size_digits=min(float_digits(volume), 3)
+        size_digits=min(float_digits(last), 3)
     )
 
     # TODO:
diff --git a/piker/ui/_graphics/_cursor.py b/piker/ui/_graphics/_cursor.py
index 83f0ee96..5d312f39 100644
--- a/piker/ui/_graphics/_cursor.py
+++ b/piker/ui/_graphics/_cursor.py
@@ -17,7 +17,7 @@
 Mouse interaction graphics
 
 """
-from typing import Optional, Tuple
+from typing import Optional, Tuple, Set, Dict
 
 import inspect
 import numpy as np
@@ -31,6 +31,10 @@ from .._style import (
     _font,
 )
 from .._axes import YAxisLabel, XAxisLabel
+from ...log import get_logger
+
+
+log = get_logger(__name__)
 
 # XXX: these settings seem to result in really decent mouse scroll
 # latency (in terms of perceived lag in cross hair) so really be sure
@@ -194,7 +198,7 @@ class ContentsLabel(pg.LabelItem):
             self.setText(f"{name}: {data:.2f}")
 
 
-class CrossHair(pg.GraphicsObject):
+class Cursor(pg.GraphicsObject):
 
     def __init__(
         self,
@@ -213,11 +217,21 @@ class CrossHair(pg.GraphicsObject):
             style=QtCore.Qt.DashLine,
         )
         self.lsc = linkedsplitcharts
-        self.graphics = {}
-        self.plots = []
+        self.graphics: Dict[str, pg.GraphicsObject] = {}
+        self.plots: List['PlotChartWidget'] = []  # type: ignore # noqa
         self.active_plot = None
-        self.digits = digits
-        self._lastx = None
+        self.digits: int = digits
+        self._datum_xy: Tuple[int, float] = (0, 0)
+
+        self._hovered: Set[pg.GraphicsObject] = set()
+        self._trackers: Set[pg.GraphicsObject] = set()
+
+    def add_hovered(
+        self,
+        item: pg.GraphicsObject,
+    ) -> None:
+        assert getattr(item, 'delete'), f"{item} must define a ``.delete()``"
+        self._hovered.add(item)
 
     def add_plot(
         self,
@@ -289,12 +303,17 @@ class CrossHair(pg.GraphicsObject):
     ) -> LineDot:
         # if this plot contains curves add line dot "cursors" to denote
         # the current sample under the mouse
-        cursor = LineDot(curve, index=plot._ohlc[-1]['index'], plot=plot)
+        cursor = LineDot(
+            curve,
+            index=plot._ohlc[-1]['index'],
+            plot=plot
+        )
         plot.addItem(cursor)
         self.graphics[plot].setdefault('cursors', []).append(cursor)
         return cursor
 
     def mouseAction(self, action, plot):  # noqa
+        log.debug(f"{(action, plot.name)}")
         if action == 'Enter':
             self.active_plot = plot
 
@@ -303,7 +322,6 @@ class CrossHair(pg.GraphicsObject):
             self.graphics[plot]['yl'].show()
 
         else:  # Leave
-            self.active_plot = None
 
             # hide horiz line and y-label
             self.graphics[plot]['hl'].hide()
@@ -332,15 +350,21 @@ class CrossHair(pg.GraphicsObject):
         # update y-range items
         self.graphics[plot]['hl'].setY(y)
 
+
         self.graphics[self.active_plot]['yl'].update_label(
             abs_pos=pos, value=y
         )
 
         # Update x if cursor changed after discretization calc
         # (this saves draw cycles on small mouse moves)
-        lastx = self._lastx
+        lastx, lasty = self._datum_xy
         ix = round(x)  # since bars are centered around index
 
+        # update all trackers
+        for item in self._trackers:
+            # print(f'setting {item} with {(ix, y)}')
+            item.on_tracked_source(ix, y)
+
         if ix != lastx:
             for plot, opts in self.graphics.items():
 
@@ -351,7 +375,6 @@ class CrossHair(pg.GraphicsObject):
                 plot.update_contents_labels(ix)
 
                 # update all subscribed curve dots
-                # first = plot._ohlc[0]['index']
                 for cursor in opts.get('cursors', ()):
                     cursor.setIndex(ix)
 
@@ -367,7 +390,7 @@ class CrossHair(pg.GraphicsObject):
                 value=x,
             )
 
-        self._lastx = ix
+        self._datum_xy = ix, y
 
     def boundingRect(self):
         try:
diff --git a/piker/ui/_graphics/_curve.py b/piker/ui/_graphics/_curve.py
index 7bf39cea..a9b24e7f 100644
--- a/piker/ui/_graphics/_curve.py
+++ b/piker/ui/_graphics/_curve.py
@@ -123,6 +123,18 @@ class FastAppendCurve(pg.PlotCurveItem):
             self.setCacheMode(QtGui.QGraphicsItem.DeviceCoordinateCache)
 
     def boundingRect(self):
+        if self.path is None:
+            return QtGui.QPainterPath().boundingRect()
+        else:
+            # dynamically override this method after initial
+            # path is created to avoid requiring the above None check
+            self.boundingRect = self._br
+            return self._br()
+
+    def _br(self):
+        """Post init ``.boundingRect()```.
+
+        """
         hb = self.path.controlPointRect()
         hb_size = hb.size()
         # print(f'hb_size: {hb_size}')
diff --git a/piker/ui/_graphics/_lines.py b/piker/ui/_graphics/_lines.py
index bd5b9de6..b697692b 100644
--- a/piker/ui/_graphics/_lines.py
+++ b/piker/ui/_graphics/_lines.py
@@ -18,6 +18,7 @@
 Lines for orders, alerts, L2.
 
 """
+from dataclasses import dataclass
 from typing import Tuple
 
 import pyqtgraph as pg
@@ -33,8 +34,6 @@ from .._axes import YSticky
 
 class LevelLabel(YSticky):
 
-    line_pen = pg.mkPen(hcolor('bracket'))
-
     _w_margin = 4
     _h_margin = 3
     level: float = 0
@@ -43,12 +42,16 @@ class LevelLabel(YSticky):
         self,
         chart,
         *args,
+        color: str = 'bracket',
         orient_v: str = 'bottom',
         orient_h: str = 'left',
         **kwargs
     ) -> None:
         super().__init__(chart, *args, **kwargs)
 
+        # TODO: this is kinda cludgy
+        self._pen = self.pen = pg.mkPen(hcolor(color))
+
         # orientation around axis options
         self._orient_v = orient_v
         self._orient_h = orient_h
@@ -75,7 +78,7 @@ class LevelLabel(YSticky):
         br = self.boundingRect()
         h, w = br.height(), br.width()
 
-        # this triggers ``.pain()`` implicitly?
+        # this triggers ``.paint()`` implicitly?
         self.setPos(QPointF(
             self._h_shift * w - offset,
             abs_pos.y() - (self._v_shift * h) - offset
@@ -85,10 +88,11 @@ class LevelLabel(YSticky):
         self.level = level
 
     def set_label_str(self, level: float):
-        # this is read inside ``.paint()``
         # self.label_str = '{size} x {level:.{digits}f}'.format(
-        self.label_str = '{level:.{digits}f}'.format(
             # size=self._size,
+
+        # this is read inside ``.paint()``
+        self.label_str = '{level:.{digits}f}'.format(
             digits=self.digits,
             level=level
         ).replace(',', ' ')
@@ -101,7 +105,7 @@ class LevelLabel(YSticky):
         p: QtGui.QPainter,
         rect: QtCore.QRectF
     ) -> None:
-        p.setPen(self.line_pen)
+        p.setPen(self._pen)
 
         if self._orient_v == 'bottom':
             lp, rp = rect.topLeft(), rect.topRight()
@@ -111,6 +115,14 @@ class LevelLabel(YSticky):
 
         p.drawLine(lp.x(), lp.y(), rp.x(), rp.y())
 
+    def highlight(self, pen) -> None:
+        self._pen = pen
+        self.update()
+
+    def unhighlight(self):
+        self._pen = self.pen
+        self.update()
+
 
 class L1Label(LevelLabel):
 
@@ -145,7 +157,7 @@ class L1Labels:
         self,
         chart: 'ChartPlotWidget',  # noqa
         digits: int = 2,
-        size_digits: int = 0,
+        size_digits: int = 3,
         font_size_inches: float = _down_2_font_inches_we_like,
     ) -> None:
 
@@ -181,29 +193,137 @@ class L1Labels:
 
 
 class LevelLine(pg.InfiniteLine):
+
+    # TODO: fill in these slots for orders
+    # .sigPositionChangeFinished.emit(self)
+
     def __init__(
         self,
+        chart: 'ChartPlotWidget',  # type: ignore # noqa
         label: LevelLabel,
+        highlight_color: str = 'default_light',
+        hl_on_hover: bool = True,
         **kwargs,
     ) -> None:
-        self.label = label
+
         super().__init__(**kwargs)
+        self.label = label
+
         self.sigPositionChanged.connect(self.set_level)
+        self._chart = chart
+        self._hoh = hl_on_hover
+
+        # use slightly thicker highlight
+        pen = pg.mkPen(hcolor(highlight_color))
+        pen.setWidth(2)
+        self.setHoverPen(pen)
+        self._track_cursor: bool = False
 
     def set_level(self, value: float) -> None:
         self.label.update_from_data(0, self.value())
 
+    def on_tracked_source(
+        self,
+        x: int,
+        y: float
+    ) -> None:
+        self.movable = True
+        self.setPos(y)  # implictly calls ``.set_level()``
+        self.update()
+
+    def setMouseHover(self, hover: bool) -> None:
+        """Mouse hover callback.
+
+        """
+        # XXX: currently we'll just return if _hoh is False
+        if self.mouseHovering == hover or not self._hoh:
+            return
+
+        self.mouseHovering = hover
+
+        chart = self._chart
+
+        if hover:
+
+            self.currentPen = self.hoverPen
+            self.label.highlight(self.hoverPen)
+
+            # add us to cursor state
+            chart._cursor.add_hovered(self)
+
+            # # hide y-crosshair
+            # chart._cursor.graphics[chart]['hl'].hide()
+
+        else:
+            self.currentPen = self.pen
+            self.label.unhighlight()
+
+            chart._cursor._hovered.remove(self)
+
+        # highlight any attached label
+
+        # self.setCursor(QtCore.Qt.OpenHandCursor)
+        # self.setCursor(QtCore.Qt.DragMoveCursor)
+        self.update()
+
+    def mouseDragEvent(self, ev):
+        chart = self._chart
+        # hide y-crosshair
+        chart._cursor.graphics[chart]['hl'].hide()
+
+        # highlight
+        self.currentPen = self.hoverPen
+        self.label.highlight(self.hoverPen)
+
+        # normal tracking behavior
+        super().mouseDragEvent(ev)
+
+        # This is the final position in the drag
+        if ev.isFinish():
+            # show y-crosshair again
+            chart = self._chart
+            chart._cursor.graphics[chart]['hl'].show()
+
+    def mouseDoubleClickEvent(
+        self,
+        ev: QtGui.QMouseEvent,
+    ) -> None:
+        print(f'double click {ev}')
+
+    # def mouseMoved(
+    #     self,
+    #     ev: Tuple[QtGui.QMouseEvent],
+    # ) -> None:
+    #     pos = evt[0]
+    #     print(pos)
+
+    def delete(self) -> None:
+        """Remove this line from containing chart/view/scene.
+
+        """
+        scene = self.scene()
+        if scene:
+            # self.label.parent.scene().removeItem(self.label)
+            scene.removeItem(self.label)
+
+        self._chart.plotItem.removeItem(self)
+
 
 def level_line(
     chart: 'ChartPlogWidget',  # noqa
     level: float,
     digits: int = 1,
+    color: str = 'default',
 
     # size 4 font on 4k screen scaled down, so small-ish.
     font_size_inches: float = _down_2_font_inches_we_like,
 
     show_label: bool = True,
 
+    # whether or not the line placed in view should highlight
+    # when moused over (aka "hovered")
+    hl_on_hover: bool = True,
+
     **linelabelkwargs
 ) -> LevelLine:
     """Convenience routine to add a styled horizontal line to a plot.
@@ -214,11 +334,13 @@ def level_line(
         parent=chart.getAxis('right'),
         # TODO: pass this from symbol data
         digits=digits,
-        opacity=1,
+        opacity=0.666,
         font_size_inches=font_size_inches,
+        color=color,
+
         # TODO: make this take the view's bg pen
         bg_color='papas_special',
-        fg_color='default',
+        fg_color=color,
         **linelabelkwargs
     )
     label.update_from_data(0, level)
@@ -227,12 +349,17 @@ def level_line(
     label._size_br_from_str(label.label_str)
 
     line = LevelLine(
+        chart,
         label,
+        # lookup "highlight" equivalent
+        highlight_color=color + '_light',
         movable=True,
         angle=0,
+        hl_on_hover=hl_on_hover,
     )
     line.setValue(level)
-    line.setPen(pg.mkPen(hcolor('default')))
+    line.setPen(pg.mkPen(hcolor(color)))
+
     # activate/draw label
     line.setValue(level)
 
diff --git a/piker/ui/_interaction.py b/piker/ui/_interaction.py
index f501238f..02c95230 100644
--- a/piker/ui/_interaction.py
+++ b/piker/ui/_interaction.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
@@ -17,7 +17,10 @@
 """
 UX interaction customs.
 """
-from typing import Optional
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from typing import Optional, Dict, Callable
+import uuid
 
 import pyqtgraph as pg
 from pyqtgraph import ViewBox, Point, QtCore, QtGui
@@ -26,6 +29,8 @@ import numpy as np
 
 from ..log import get_logger
 from ._style import _min_points_to_show, hcolor, _font
+from ._graphics._lines import level_line, LevelLine
+from .._ems import get_orders, OrderBook
 
 
 log = get_logger(__name__)
@@ -194,13 +199,260 @@ class SelectRect(QtGui.QGraphicsRectItem):
         self.hide()
 
 
+# global store of order-lines graphics
+# keyed by uuid4 strs - used to sync draw
+# order lines **after** the order is 100%
+# active in emsd
+_order_lines: Dict[str, LevelLine] = {}
+
+
+@dataclass
+class LineEditor:
+    view: 'ChartView'
+    _order_lines: field(default_factory=_order_lines)
+    chart: 'ChartPlotWidget' = None  # type: ignore # noqa
+    _active_staged_line: LevelLine = None
+    _stage_line: LevelLine = None
+
+    def stage_line(self, color: str = 'alert_yellow') -> LevelLine:
+        """Stage a line at the current chart's cursor position
+        and return it.
+
+        """
+        chart = self.chart._cursor.active_plot
+        chart.setCursor(QtCore.Qt.PointingHandCursor)
+        cursor = chart._cursor
+        y = chart._cursor._datum_xy[1]
+
+        line = self._stage_line
+        if not line:
+            # add a "staged" cursor-tracking line to view
+            # and cash it in a a var
+            line = level_line(
+                chart,
+                level=y,
+                digits=chart._lc.symbol.digits(),
+                color=color,
+
+                # don't highlight the "staging" line
+                hl_on_hover=False,
+            )
+            self._stage_line = line
+
+        else:
+            # use the existing staged line instead
+            # of allocating more mem / objects repeatedly
+            line.setValue(y)
+            line.show()
+            line.label.show()
+
+        self._active_staged_line = line
+
+        # hide crosshair y-line
+        cursor.graphics[chart]['hl'].hide()
+
+        # add line to cursor trackers
+        cursor._trackers.add(line)
+
+        return line
+
+    def unstage_line(self) -> LevelLine:
+        """Inverse of ``.stage_line()``.
+
+        """
+        chart = self.chart._cursor.active_plot
+        chart.setCursor(QtCore.Qt.ArrowCursor)
+        cursor = chart._cursor
+
+        # delete "staged" cursor tracking line from view
+        line = self._active_staged_line
+
+        cursor._trackers.remove(line)
+
+        if self._stage_line:
+            self._stage_line.hide()
+            self._stage_line.label.hide()
+
+        self._active_staged_line = None
+
+        # show the crosshair y line
+        hl = cursor.graphics[chart]['hl']
+        hl.show()
+
+    def create_line(self, uuid: str) -> LevelLine:
+
+        line = self._active_staged_line
+        if not line:
+            raise RuntimeError("No line commit is currently staged!?")
+
+        chart = self.chart._cursor.active_plot
+        y = chart._cursor._datum_xy[1]
+
+        line = level_line(
+            chart,
+            level=y,
+            color='alert_yellow',
+            digits=chart._lc.symbol.digits(),
+            show_label=False,
+        )
+
+        # register for later lookup/deletion
+        self._order_lines[uuid] = line
+        return line, y
+
+    def commit_line(self, uuid: str) -> LevelLine:
+        """Commit a "staged line" to view.
+
+        Submits the line graphic under the cursor as a (new) permanent
+        graphic in view.
+
+        """
+        line = self._order_lines[uuid]
+        line.oid = uuid
+        line.label.show()
+
+        # TODO: other flashy things to indicate the order is active
+
+        log.debug(f'Level active for level: {line.value()}')
+
+        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(
+        self,
+        line: LevelLine = None,
+        uuid: str = None,
+    ) -> None:
+        """Remove a line by refernce or uuid.
+
+        If no lines or ids are provided remove all lines under the
+        cursor position.
+
+        """
+        if line:
+            uuid = line.oid
+
+        # try to look up line from our registry
+        line = self._order_lines.pop(uuid)
+
+        # if hovered remove from cursor set
+        hovered = self.chart._cursor._hovered
+        if line in hovered:
+            hovered.remove(line)
+
+        line.delete()
+
+
+@dataclass
+class ArrowEditor:
+
+    chart: 'ChartPlotWidget'  # noqa
+    _arrows: field(default_factory=dict)
+
+    def add(
+        self,
+        uid: str,
+        x: float,
+        y: float,
+        color='default',
+        pointing: str = 'up',
+    ) -> pg.ArrowItem:
+        """Add an arrow graphic to view at given (x, y).
+
+        """
+        yb = pg.mkBrush(hcolor('alert_yellow'))
+
+        angle = 90 if pointing == 'up' else -90
+
+        arrow = pg.ArrowItem(
+            angle=angle,
+            baseAngle=0,
+            headLen=5,
+            headWidth=2,
+            tailLen=None,
+            brush=yb,
+        )
+        arrow.setPos(x, y)
+
+        self._arrows[uid] = arrow
+
+        # render to view
+        self.chart.plotItem.addItem(arrow)
+
+        return arrow
+
+    def remove(self, arrow) -> bool:
+        self.chart.plotItem.removeItem(arrow)
+
+
+@dataclass
+class OrderMode:
+    """Major mode for placing orders on a chart view.
+
+    """
+    chart: 'ChartPlotWidget'  #  type: ignore # noqa
+    book: OrderBook
+    lines: LineEditor
+    arrows: ArrowEditor
+    _arrow_colors = {
+        'alert': 'alert_yellow',
+        'buy': 'buy_green',
+        'sell': 'sell_red',
+    }
+
+    key_map: Dict[str, Callable] = field(default_factory=dict)
+
+    def uuid(self) -> str:
+        return str(uuid.uuid4())
+
+
+@asynccontextmanager
+async def open_order_mode(
+    chart,
+):
+    # global _order_lines
+
+    view = chart._vb
+    book = get_orders()
+    lines = LineEditor(view=view, _order_lines=_order_lines, chart=chart)
+    arrows = ArrowEditor(chart, {})
+
+    log.info("Opening order mode")
+
+    mode = OrderMode(chart, book, lines, arrows)
+    view.mode = mode
+
+    # # setup local ui event streaming channels for request/resp
+    # # streamging with EMS daemon
+    # global _to_ems, _from_order_book
+    # _to_ems, _from_order_book = trio.open_memory_channel(100)
+
+    try:
+        yield mode
+
+    finally:
+        # XXX special teardown handling like for ex.
+        # - cancelling orders if needed?
+        # - closing positions if desired?
+        # - switching special condition orders to safer/more reliable variants
+        log.info("Closing order mode")
+
+
 class ChartView(ViewBox):
     """Price chart view box with interaction behaviors you'd expect from
     any interactive platform:
 
         - zoom on mouse scroll that auto fits y-axis
-        - no vertical scrolling
-        - zoom to a "fixed point" on the y-axis
+        - vertical scrolling on y-axis
+        - zoom on x to most recent in view datum
+        - zoom on right-click-n-drag to cursor position
+
     """
     def __init__(
         self,
@@ -215,14 +467,21 @@ class ChartView(ViewBox):
         self.addItem(self.select_box, ignoreBounds=True)
         self._chart: 'ChartPlotWidget' = None  # noqa
 
+        # self._lines_editor = LineEditor(view=self, _lines=_lines)
+        self.mode = None
+
+        # kb ctrls processing
+        self._key_buffer = []
+
     @property
-    def chart(self) -> 'ChartPlotWidget':  # noqa
+    def chart(self) -> 'ChartPlotWidget':  # type: ignore # noqa
         return self._chart
 
     @chart.setter
-    def chart(self, chart: 'ChartPlotWidget') -> None:  # noqa
+    def chart(self, chart: 'ChartPlotWidget') -> None:  # type: ignore # noqa
         self._chart = chart
         self.select_box.chart = chart
+        # self._lines_editor.chart = chart
 
     def wheelEvent(self, ev, axis=None):
         """Override "center-point" location for scrolling.
@@ -286,6 +545,7 @@ class ChartView(ViewBox):
     ) -> None:
         #  if axis is specified, event will only affect that axis.
         ev.accept()  # we accept all buttons
+        button = ev.button()
 
         pos = ev.pos()
         lastPos = ev.lastPos()
@@ -299,13 +559,13 @@ class ChartView(ViewBox):
             mask[1-axis] = 0.0
 
         # Scale or translate based on mouse button
-        if ev.button() & (QtCore.Qt.LeftButton | QtCore.Qt.MidButton):
+        if button & (QtCore.Qt.LeftButton | QtCore.Qt.MidButton):
 
-            # zoom only y-axis when click-n-drag on it
+            # zoom y-axis ONLY when click-n-drag on it
             if axis == 1:
                 # set a static y range special value on chart widget to
                 # prevent sizing to data in view.
-                self._chart._static_yrange = 'axis'
+                self.chart._static_yrange = 'axis'
 
                 scale_y = 1.3 ** (dif.y() * -1 / 20)
                 self.setLimits(yMin=None, yMax=None)
@@ -338,6 +598,8 @@ class ChartView(ViewBox):
                     # update shape of scale box
                     # self.updateScaleBox(ev.buttonDownPos(), ev.pos())
             else:
+                # default bevavior: click to pan view
+
                 tr = self.childGroup.transform()
                 tr = fn.invertQTransform(tr)
                 tr = tr.map(dif*mask) - tr.map(Point(0, 0))
@@ -346,13 +608,16 @@ class ChartView(ViewBox):
                 y = tr.y() if mask[1] == 1 else None
 
                 self._resetTarget()
+
                 if x is not None or y is not None:
                     self.translateBy(x=x, y=y)
+
                 self.sigRangeChangedManually.emit(self.state['mouseEnabled'])
 
-        elif ev.button() & QtCore.Qt.RightButton:
+        elif button & QtCore.Qt.RightButton:
+
+            # right click zoom to center behaviour
 
-            # print "vb.rightDrag"
             if self.state['aspectLocked'] is not False:
                 mask[0] = 0
 
@@ -372,46 +637,119 @@ class ChartView(ViewBox):
             self.scaleBy(x=x, y=y, center=center)
             self.sigRangeChangedManually.emit(self.state['mouseEnabled'])
 
+    def mouseClickEvent(self, ev):
+        """Full-click callback.
+
+        """
+        button = ev.button()
+        # pos = ev.pos()
+
+        if button == QtCore.Qt.RightButton and self.menuEnabled():
+            ev.accept()
+            self.raiseContextMenu(ev)
+
+        elif button == QtCore.Qt.LeftButton:
+
+            ev.accept()
+
+            # self._lines_editor.commit_line()
+
+            # send order to EMS
+
+            # register the "staged" line under the cursor
+            # to be displayed when above order ack arrives
+            # (means the line graphic doesn't show on screen until the
+            # order is live in the emsd).
+            mode = self.mode
+            uuid = mode.uuid()
+
+            # make line graphic
+            line, y = mode.lines.create_line(uuid)
+
+            # send order cmd to ems
+            mode.book.alert(
+                uuid=uuid,
+                symbol=mode.chart._lc._symbol,
+                price=y
+            )
+
     def keyReleaseEvent(self, ev):
-        # print(f'release: {ev.text().encode()}')
+        """
+        Key release to normally to trigger release of input mode
+
+        """
+        # TODO: is there a global setting for this?
+        if ev.isAutoRepeat():
+            ev.ignore()
+            return
+
         ev.accept()
-        if ev.key() == QtCore.Qt.Key_Shift:
+        text = ev.text()
+        key = ev.key()
+        # mods = ev.modifiers()
+
+        if key == QtCore.Qt.Key_Shift:
             if self.state['mouseMode'] == ViewBox.RectMode:
                 self.setMouseMode(ViewBox.PanMode)
 
+        if text == 'a':
+            # draw "staged" line under cursor position
+            self.mode.lines.unstage_line()
+
     def keyPressEvent(self, ev):
         """
         This routine should capture key presses in the current view box.
-        """
-        # print(ev.text().encode())
-        ev.accept()
 
-        if ev.modifiers() == QtCore.Qt.ShiftModifier:
+        """
+        # TODO: is there a global setting for this?
+        if ev.isAutoRepeat():
+            ev.ignore()
+            return
+
+        ev.accept()
+        text = ev.text()
+        key = ev.key()
+        mods = ev.modifiers()
+
+        if mods == QtCore.Qt.ShiftModifier:
             if self.state['mouseMode'] == ViewBox.PanMode:
                 self.setMouseMode(ViewBox.RectMode)
 
         # ctl
-        if ev.modifiers() == QtCore.Qt.ControlModifier:
-            # print("CTRL")
+        if mods == QtCore.Qt.ControlModifier:
             # TODO: ctrl-c as cancel?
             # https://forum.qt.io/topic/532/how-to-catch-ctrl-c-on-a-widget/9
             # if ev.text() == 'c':
             #     self.rbScaleBox.hide()
-            pass
+            print(f"CTRL + key:{key} + text:{text}")
 
         # alt
-        if ev.modifiers() == QtCore.Qt.AltModifier:
+        if mods == QtCore.Qt.AltModifier:
             pass
-            # print("ALT")
 
         # esc
-        if ev.key() == QtCore.Qt.Key_Escape:
+        if key == QtCore.Qt.Key_Escape:
             self.select_box.clear()
 
-        if ev.text() == 'r':
+        self._key_buffer.append(text)
+
+        # order modes
+        if text == 'r':
             self.chart.default_view()
 
-        # Leaving this for light reference purposes
+        elif text == 'a':
+            # add a line at the current cursor
+            self.mode.lines.stage_line()
+
+        elif text == 'd':
+
+            # delete any lines under the cursor
+            mode = self.mode
+            for line in mode.lines.lines_under_cursor():
+                mode.book.cancel(uuid=line.oid)
+
+        # XXX: Leaving this for light reference purposes, there
+        # seems to be some work to at least gawk at for history mgmt.
 
         # Key presses are used only when mouse mode is RectMode
         # The following events are implemented:
diff --git a/piker/ui/_style.py b/piker/ui/_style.py
index 9208e13c..656877cc 100644
--- a/piker/ui/_style.py
+++ b/piker/ui/_style.py
@@ -166,4 +166,9 @@ def hcolor(name: str) -> str:
         'tina_green': '#00cc00',
         'tina_red': '#fa0000',
 
+
+        # orders and alerts
+        'alert_yellow': '#e2d083',
+        'alert_yellow_light': '#ffe366',
+
     }[name]
diff --git a/piker/ui/cli.py b/piker/ui/cli.py
index e14ef3f6..d2050bbc 100644
--- a/piker/ui/cli.py
+++ b/piker/ui/cli.py
@@ -150,5 +150,6 @@ def chart(config, symbol, date, rate, test, profile):
         tractor_kwargs={
             'debug_mode': True,
             'loglevel': tractorloglevel,
+            'rpc_module_paths': ['piker._ems'],
         },
     )