piker/_daemon.py

@@ -35,7 +35,7 @@ log = get_logger(__name__)
 
 _root_dname = 'pikerd'
 
-_registry_addr = ('127.0.0.1', 6116)
+_registry_addr = ('127.0.0.1', 1616)
 _tractor_kwargs: dict[str, Any] = {
     # use a different registry addr then tractor's default
     'arbiter_addr': _registry_addr

piker/brokers/ib/api.py

@@ -640,7 +640,6 @@ class Client:
         ready = ticker.updateEvent
 
         # ensure a last price gets filled in before we deliver quote
-        warnset: bool = False
         for _ in range(100):
             if isnan(ticker.last):
 
@@ -651,21 +650,17 @@ class Client:
                 if ready in done:
                     break
                 else:
-                    if not warnset:
+                    log.warning(
-                        log.warning(
+                        f'Quote for {symbol} timed out: market is closed?'
-                            f'Quote for {symbol} timed out: market is closed?'
+                    )
-                        )
-                        warnset = True
 
             else:
                 log.info(f'Got first quote for {symbol}')
                 break
         else:
-            if not warnset:
+            log.warning(
-                log.warning(
+                f'Symbol {symbol} is not returning a quote '
-                    f'Symbol {symbol} is not returning a quote '
+                'it may be outside trading hours?')
-                    'it may be outside trading hours?')
-                warnset = True
 
         return contract, ticker, details
 

piker/data/_ahab.py

@@ -19,7 +19,6 @@ Supervisor for docker with included specific-image service helpers.
 
 '''
 import os
-import time
 from typing import (
     Optional,
     Callable,
@@ -187,65 +186,45 @@ class Container:
 
     async def cancel(
         self,
-        stop_msg: str,
     ) -> None:
 
         cid = self.cntr.id
-        # first try a graceful cancel
-        log.cancel(
-            f'SIGINT cancelling container: {cid}\n'
-            f'waiting on stop msg: "{stop_msg}"'
-        )
         self.try_signal('SIGINT')
 
-        start = time.time()
+        with trio.move_on_after(0.5) as cs:
-        for _ in range(30):
+            cs.shield = True
+            await self.process_logs_until('initiating graceful shutdown')
+            await self.process_logs_until('exiting...',)
 
+        for _ in range(10):
             with trio.move_on_after(0.5) as cs:
                 cs.shield = True
-                await self.process_logs_until(stop_msg)
+                await self.process_logs_until('exiting...',)
-
-                # if we aren't cancelled on above checkpoint then we
-                # assume we read the expected stop msg and terminated.
                 break
 
-            try:
+            if cs.cancelled_caught:
-                log.info(f'Polling for container shutdown:\n{cid}')
+                # get out the big guns, bc apparently marketstore
+                # doesn't actually know how to terminate gracefully
+                # :eyeroll:...
+                self.try_signal('SIGKILL')
 
-                if self.cntr.status not in {'exited', 'not-running'}:
+                try:
+                    log.info('Waiting on container shutdown: {cid}')
                     self.cntr.wait(
                         timeout=0.1,
                         condition='not-running',
                     )
+                    break
 
-                break
+                except (
+                    ReadTimeout,
+                    ConnectionError,
+                ):
+                    log.error(f'failed to wait on container {cid}')
+                    raise
 
-            except (
-                ReadTimeout,
-            ):
-                log.info(f'Still waiting on container:\n{cid}')
-                continue
-
-            except (
-                docker.errors.APIError,
-                ConnectionError,
-            ):
-                log.exception('Docker connection failure')
-                break
         else:
-            delay = time.time() - start
+            raise RuntimeError('Failed to cancel container {cid}')
-            log.error(
-                f'Failed to kill container {cid} after {delay}s\n'
-                'sending SIGKILL..'
-            )
-            # get out the big guns, bc apparently marketstore
-            # doesn't actually know how to terminate gracefully
-            # :eyeroll:...
-            self.try_signal('SIGKILL')
-            self.cntr.wait(
-                timeout=3,
-                condition='not-running',
-            )
 
         log.cancel(f'Container stopped: {cid}')
 
@@ -266,16 +245,13 @@ async def open_ahabd(
         # params, etc. passing to ``Containter.run()``?
         # call into endpoint for container config/init
         ep_func = NamespacePath(endpoint).load_ref()
-        (
+        dcntr, cntr_config = ep_func(client)
-            dcntr,
-            cntr_config,
-            start_msg,
-            stop_msg,
-        ) = ep_func(client)
         cntr = Container(dcntr)
 
         with trio.move_on_after(1):
-            found = await cntr.process_logs_until(start_msg)
+            found = await cntr.process_logs_until(
+                "launching tcp listener for all services...",
+            )
 
             if not found and cntr not in client.containers.list():
                 raise RuntimeError(
@@ -295,9 +271,16 @@ async def open_ahabd(
             # callers to have root perms?
             await trio.sleep_forever()
 
-        finally:
+        except (
+            BaseException,
+            # trio.Cancelled,
+            # KeyboardInterrupt,
+        ):
+
             with trio.CancelScope(shield=True):
-                await cntr.cancel(stop_msg)
+                await cntr.cancel()
+
+            raise
 
 
 async def start_ahab(

piker/data/marketstore.py

@@ -127,15 +127,10 @@ def start_marketstore(
     import os
     import docker
     from .. import config
+
     get_console_log('info', name=__name__)
 
-    mktsdir = os.path.join(config._config_dir, 'marketstore')
+    yml_file = os.path.join(config._config_dir, 'mkts.yml')
-
-    # create when dne
-    if not os.path.isdir(mktsdir):
-        os.mkdir(mktsdir)
-
-    yml_file = os.path.join(mktsdir, 'mkts.yml')
     if not os.path.isfile(yml_file):
         log.warning(
             f'No `marketstore` config exists?: {yml_file}\n'
@@ -148,14 +143,14 @@ def start_marketstore(
     # create a mount from user's local piker config dir into container
     config_dir_mnt = docker.types.Mount(
         target='/etc',
-        source=mktsdir,
+        source=config._config_dir,
         type='bind',
     )
 
     # create a user config subdir where the marketstore
     # backing filesystem database can be persisted.
     persistent_data_dir = os.path.join(
-        mktsdir, 'data',
+        config._config_dir, 'data',
     )
     if not os.path.isdir(persistent_data_dir):
         os.mkdir(persistent_data_dir)
@@ -185,14 +180,7 @@ def start_marketstore(
         init=True,
         # remove=True,
     )
-    return (
+    return dcntr, _config
-        dcntr,
-        _config,
-
-        # expected startup and stop msgs
-        "launching tcp listener for all services...",
-        "exiting...",
-    )
 
 
 _tick_tbk_ids: tuple[str, str] = ('1Sec', 'TICK')
@@ -395,12 +383,7 @@ class Storage:
     ]:
 
         first_tsdb_dt, last_tsdb_dt = None, None
-        tsdb_arrays = await self.read_ohlcv(
+        tsdb_arrays = await self.read_ohlcv(fqsn)
-            fqsn,
-            # on first load we don't need to pull the max
-            # history per request size worth.
-            limit=3000,
-        )
         log.info(f'Loaded tsdb history {tsdb_arrays}')
 
         if tsdb_arrays:
@@ -418,7 +401,6 @@ class Storage:
         fqsn: str,
         timeframe: Optional[Union[int, str]] = None,
         end: Optional[int] = None,
-        limit: int = int(800e3),
 
     ) -> tuple[
         MarketstoreClient,
@@ -441,7 +423,7 @@ class Storage:
 
             # TODO: figure the max limit here given the
             # ``purepc`` msg size limit of purerpc: 33554432
-            limit=limit,
+            limit=int(800e3),
         )
 
         if timeframe is None:

piker/fsp/_engine.py

@@ -361,7 +361,7 @@ async def cascade(
                 ) -> tuple[TaskTracker, int]:
                     # TODO: adopt an incremental update engine/approach
                     # where possible here eventually!
-                    log.debug(f're-syncing fsp {func_name} to source')
+                    log.warning(f're-syncing fsp {func_name} to source')
                     tracker.cs.cancel()
                     await tracker.complete.wait()
                     tracker, index = await n.start(fsp_target)

piker/ui/_flows.py

@@ -750,15 +750,8 @@ class Flow(msgspec.Struct):  # , frozen=True):
             y = y[-uppx:]
             ymn, ymx = y.min(), y.max()
             # print(f'drawing uppx={uppx} mxmn line: {ymn}, {ymx}')
-            try:
-                iuppx = x[-uppx]
-            except IndexError:
-                # we're less then an x-px wide so just grab the start
-                # datum index.
-                iuppx = x[0]
-
             dsg._last_line = QLineF(
-                iuppx, ymn,
+                x[-uppx], ymn,
                 x[-1], ymx,
             )
             # print(f'updating DS curve {self.name}')

piker/ui/_fsp.py

@@ -440,7 +440,7 @@ class FspAdmin:
                         # if the chart isn't hidden try to update
                         # the data on screen.
                         if not self.linked.isHidden():
-                            log.debug(f'Re-syncing graphics for fsp: {ns_path}')
+                            log.info(f'Re-syncing graphics for fsp: {ns_path}')
                             self.linked.graphics_cycle(
                                 trigger_all=True,
                                 prepend_update_index=info['first'],

setup.py

@@ -57,7 +57,6 @@ setup(
         # from github currently (see requirements.txt)
         # 'trimeter',  # not released yet..
         # 'tractor',
-        # asyncvnc,
 
         # brokers
         'asks==2.4.8',
@@ -72,34 +71,32 @@ setup(
 
         # UI
         'PyQt5',
-        # 'pyqtgraph',  from our fork see reqs.txt
+        'pyqtgraph',
-        'qdarkstyle >= 3.0.2',  # themeing
+        'qdarkstyle >= 3.0.2',
-        'fuzzywuzzy[speedup]',  # fuzzy search
+        # fuzzy search
+        'fuzzywuzzy[speedup]',
 
         # tsdbs
-        # anyio-marketstore  # from gh see reqs.txt
+        'pymarketstore',
     ],
     extras_require={
+
+        # serialization
         'tsdb': [
             'docker',
         ],
 
     },
     tests_require=['pytest'],
-    python_requires=">=3.10",
+    python_requires=">=3.9",  # literally for ``datetime.datetime.fromisoformat``...
-    keywords=[
+    keywords=["async", "trading", "finance", "quant", "charting"],
-        "async",
-        "trading",
-        "finance",
-        "quant",
-        "charting",
-    ],
     classifiers=[
         'Development Status :: 3 - Alpha',
         'License :: OSI Approved :: ',
         'Operating System :: POSIX :: Linux',
         "Programming Language :: Python :: Implementation :: CPython",
         "Programming Language :: Python :: 3 :: Only",
+        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         'Intended Audience :: Financial and Insurance Industry',
         'Intended Audience :: Science/Research',