Tolerate various "bad data" cases in `markup_gaps()`

Namely such that when the previous-df-row by our shm-abs-'index' doesn't
exist we ignore certain cases which are likely due to borked-but-benign
samples written to the tsdb or rt shm buffers prior.

Particularly we now ignore,
- any `dt`/`prev_dt` values which are UNIX-epoch timestamped (val of 0).
- any row-is-first-row in the df; there is no previous.
- any missing previous datum by 'index', in which case we lookup the
  `wdts` prior row and use that instead.
  * this would indicate a missing sample for the time-step but we can
    still detect a "gap" by looking at the prior row, by df-abs-index
    `i`, and use its timestamp to determine the period/size of missing
    samples (which need to likely still be retrieved).
  * in this case i'm leaving in a pause-point for introspecting these
    rarer cases when `--pdb` is passed via CLI.

Relatedly in the `piker store` CLI ep,
- add `--pdb` flag to `piker store`, pass it verbatim as `debug_mode`.
- when `times` has only a single row, don't calc a `period_s` median.
- only trace `null_segs` when in debug mode.
- always markup/dedupe gaps for `period_s==60`

piker/brokers/binance/feed.py

@@ -108,7 +108,6 @@ class AggTrade(Struct, frozen=True):
     m: bool  # Is the buyer the market maker?
     M: bool|None = None  # Ignore
     nq: float|None = None # Normal quantity without the trades involving RPI orders
-    # ^XXX https://developers.binance.com/docs/derivatives/change-log#2025-12-29
 
 
 async def stream_messages(

piker/tsp/_history.py

@@ -75,6 +75,7 @@ from piker.brokers._util import (
 )
 from piker.storage import TimeseriesNotFound
 from ._anal import (
+
     dedupe,
     get_null_segs,
     iter_null_segs,
@@ -119,16 +120,15 @@ _rt_buffer_start = int((_days_worth - 1) * _secs_in_day)
 
 def diff_history(
     array: np.ndarray,
-    append_until_dt: datetime|None = None,
-    prepend_until_dt: datetime|None = None,
+    append_until_dt: datetime | None = None,
+    prepend_until_dt: datetime | None = None,
 
 ) -> np.ndarray:
 
     # no diffing with tsdb dt index possible..
     if (
         prepend_until_dt is None
-        and
-        append_until_dt is None
+        and append_until_dt is None
     ):
         return array
 
@@ -140,26 +140,15 @@ def diff_history(
         return array[times >= prepend_until_dt.timestamp()]
 
 
+# TODO: can't we just make this a sync func now?
 async def shm_push_in_between(
     shm: ShmArray,
     to_push: np.ndarray,
     prepend_index: int,
-    backfill_until_dt: datetime,
 
     update_start_on_prepend: bool = False,
 
 ) -> int:
-
-    # XXX, try to catch bad inserts by peeking at the first/last
-    # times and ensure we don't violate order.
-    f_times: np.ndarray = to_push['time']
-    f_start: float = f_times[0]
-    f_start_dt = from_timestamp(f_start)
-    if (
-       f_start_dt < backfill_until_dt
-    ):
-        await tractor.pause()
-
     # XXX: extremely important, there can be no checkpoints
     # in the body of this func to avoid entering new ``frames``
     # values while we're pipelining the current ones to
@@ -192,7 +181,6 @@ async def maybe_fill_null_segments(
     get_hist: Callable,
     sampler_stream: tractor.MsgStream,
     mkt: MktPair,
-    backfill_until_dt: datetime,
 
     task_status: TaskStatus[trio.Event] = trio.TASK_STATUS_IGNORED,
 
@@ -203,11 +191,7 @@ async def maybe_fill_null_segments(
 
     frame: Frame = shm.array
 
-    # TODO, put in parent task/daemon root!
-    import greenback
-    await greenback.ensure_portal()
-
-    null_segs: tuple|None = get_null_segs(
+    null_segs: tuple | None = get_null_segs(
         frame,
         period=timeframe,
     )
@@ -253,7 +237,6 @@ async def maybe_fill_null_segments(
             shm,
             to_push,
             prepend_index=absi_end,
-            backfill_until_dt=backfill_until_dt,
             update_start_on_prepend=False,
         )
         # TODO: UI side needs IPC event to update..
@@ -369,12 +352,15 @@ async def start_backfill(
     mkt: MktPair,
     shm: ShmArray,
     timeframe: float,
+
     backfill_from_shm_index: int,
     backfill_from_dt: datetime,
+
     sampler_stream: tractor.MsgStream,
 
-    backfill_until_dt: datetime|None = None,
-    storage: StorageClient|None = None,
+    backfill_until_dt: datetime | None = None,
+    storage: StorageClient | None = None,
+
     write_tsdb: bool = True,
 
     task_status: TaskStatus[tuple] = trio.TASK_STATUS_IGNORED,
@@ -509,14 +495,7 @@ async def start_backfill(
 
             assert time[-1] == next_end_dt.timestamp()
 
-            expected_dur: Interval = (
-                last_start_dt.subtract(
-                    seconds=timeframe
-                    # ^XXX, always "up to" the bar *before*
-                )
-                -
-                next_start_dt
-            )
+            expected_dur: Interval = last_start_dt - next_start_dt
 
             # frame's worth of sample-period-steps, in seconds
             frame_size_s: float = len(array) * timeframe
@@ -577,7 +556,6 @@ async def start_backfill(
                     shm,
                     to_push,
                     prepend_index=next_prepend_index,
-                    backfill_until_dt=backfill_until_dt,
                     update_start_on_prepend=update_start_on_prepend,
                 )
                 await sampler_stream.send({
@@ -607,7 +585,6 @@ async def start_backfill(
                     shm,
                     to_push,
                     prepend_index=next_prepend_index,
-                    backfill_until_dt=backfill_until_dt,
                     update_start_on_prepend=update_start_on_prepend,
                 )
                 await sampler_stream.send({
@@ -1079,7 +1056,7 @@ async def tsdb_backfill(
                 trio.open_nursery() as tn,
             ):
 
-                bf_done: trio.Event = await tn.start(
+                bf_done = await tn.start(
                     partial(
                         start_backfill,
                         get_hist=get_hist,
@@ -1099,10 +1076,8 @@ async def tsdb_backfill(
                         write_tsdb=True,
                     )
                 )
-                nulls_detected: trio.Event|None = None
-
+                nulls_detected: trio.Event | None = None
                 if last_tsdb_dt is not None:
-
                     # calc the index from which the tsdb data should be
                     # prepended, presuming there is a gap between the
                     # latest frame (loaded/read above) and the latest
@@ -1173,7 +1148,7 @@ async def tsdb_backfill(
                     # TODO: ideally these can never exist!
                     # -[ ] somehow it seems sometimes we're writing zero-ed
                     #    segments to tsdbs during teardown?
-                    # -[ ] can we ensure that the backfiller tasks do this
+                    # -[ ] can we ensure that the backcfiller tasks do this
                     #    work PREVENTAVELY instead?
                     # -[ ] fill in non-zero epoch time values ALWAYS!
                     # await maybe_fill_null_segments(
@@ -1185,7 +1160,6 @@ async def tsdb_backfill(
                         get_hist=get_hist,
                         sampler_stream=sampler_stream,
                         mkt=mkt,
-                        backfill_until_dt=last_tsdb_dt,
                     ))
 
             # 2nd nursery END