Cache EMS trade relay tasks on feed fqsn

Except for paper accounts (in which case we need a trades dialog and
paper engine per symbol to enable simulated clearing) we can rely on the
instrument feed (symbol name) to be the caching key. Utilize
`tractor.trionics.maybe_open_context()` and the new key-as-callable
support in the paper case to ensure we have separate paper clearing
loops per symbol.

Requires https://github.com/goodboy/tractor/pull/329

piker/clearing/_ems.py

@@ -31,6 +31,7 @@ from typing import (
     AsyncIterator,
     Any,
     Callable,
+    Hashable,
     Optional,
 )
 
@@ -1265,6 +1266,24 @@ async def process_client_order_cmds(
                 )
 
 
+@acm
+async def maybe_open_trade_relays(
+    router: Router,
+    fqsn: str,
+    exec_mode: str,  # ('paper', 'live')
+    loglevel: str = 'info',
+
+) -> tuple:
+
+    relay, feed, client_ready = await _router.nursery.start(
+        _router.open_trade_relays,
+        fqsn,
+        exec_mode,
+        loglevel,
+    )
+    yield relay, feed, client_ready
+
+
 @tractor.context
 async def _emsd_main(
     ctx: tractor.Context,
@@ -1329,17 +1348,34 @@ async def _emsd_main(
     feed: Feed
     client_ready: trio.Event
 
-    # open a stream with the brokerd backend for order flow dialogue
-    # only open if one isn't already up: we try to keep as few duplicate
-    # streams as necessary.
-    # TODO: should we try using `tractor.trionics.maybe_open_context()`
-    # here?
-    relay, feed, client_ready = await _router.nursery.start(
-        _router.open_trade_relays,
-        fqsn,
-        exec_mode,
-        loglevel,
-    )
+    # NOTE: open a stream with the brokerd backend for order flow
+    # dialogue and dark clearing but only open one: we try to keep as
+    # few duplicate streams as necessary per ems actor.
+    def cache_on_fqsn_unless_paper(
+        router: Router,
+        fqsn: str,
+        exec_mode: str,  # ('paper', 'live')
+        loglevel: str = 'info',
+    ) -> Hashable:
+        if exec_mode == 'paper':
+            return f'paper_{fqsn}'
+        else:
+            return fqsn
+
+    async with tractor.trionics.maybe_open_context(
+        acm_func=maybe_open_trade_relays,
+        kwargs={
+            'router': _router,
+            'fqsn': fqsn,
+            'exec_mode': exec_mode,
+            'loglevel': loglevel,
+        },
+        key=cache_on_fqsn_unless_paper,
+
+    ) as (
+        cache_hit,
+        (relay, feed, client_ready)
+    ):
         brokerd_stream = relay.brokerd_stream
         dark_book = _router.get_dark_book(broker)