Skip stale trade ticks in 1s sampler

Queued IB startup trades can arrive after samplerd has advanced
the current 1s bar. Writing them into that bar creates a false
price and volume spike. Compare each stamped trade with the
current bar before updating either SHM buffer. Warn whenever a
next-slot trade must wait for samplerd; skip other mismatches
and leave untimestamped feeds on the receive-time path.

Add focused tests for epoch units, queued trades, and warning
before the next-slot wait. Note that incomplete provider bars
still need a correction path for late trades.

Prompt-IO: ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md

(this patch was generated in some part by `codex` using `gpt-6` (`openai`))

ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md

@@ -0,0 +1,45 @@
+---
+model: gpt-6
+service: codex
+session: sampler_tick_time_guard
+timestamp: 2026-09-25T19:05:23Z
+git_ref: c0dae89a
+scope: code
+substantive: true
+raw_file: 20260925T190523Z_c0dae89a_prompt_io.raw.md
+---
+
+## Prompt
+
+The user observed an oversized MNQ 1s bar just after boot and asked
+to keep queued startup trades out of a different timestamped bar.
+After direct chart and separate pikerd restarts showed no new spike,
+the user requested the fix on a separate branch based on
+`ib_connectivity_harden`, followed by `$commit-plan`.
+
+## Response summary
+
+Normalize provider trade timestamps and compare them with the
+sampler's current 1s bar. Wait briefly for a one-step-ahead tick,
+warn that its price update is delayed, skip other timestamp
+mismatches, and preserve behavior for feeds without tick
+timestamps. Note the remaining late-bar correction race and add
+focused sampler tests.
+
+## Files changed
+
+- `piker/data/_sampling.py` — guard writes by tick and bar time.
+- `tests/test_sampling_tick_times.py` — exercise startup backlog and
+  next-step behavior.
+
+## Human edits
+
+The human identified the oversized MNQ boot bar, proposed a sampler
+startup race, directed the timestamp check, and live-tested the fix
+in both direct-chart and separate-pikerd setups. The human chose to
+isolate this fix from the ongoing segment-repair review on a branch
+based on `ib_connectivity_harden`. During local tuicr review, the
+human required a warning before each next-slot wait because chart
+price latency matters, and called out the race between incomplete
+history bars and late trade ticks. The agent applied the edits and
+added a TODO for IB and provider-neutral bar reconciliation.

ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.raw.md

@@ -0,0 +1,34 @@
+---
+model: gpt-6
+service: codex
+timestamp: 2026-09-25T19:05:23Z
+git_ref: c0dae89a
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+## Prompt and generated output
+
+The user reported a large MNQ 1s startup bar and asked us to ensure
+queued trades are not aggregated into the current timestep when their
+timestamps indicate otherwise. After a direct chart restart and a
+separate pikerd run showed no spike, the user requested the tested fix
+on a separate worktree branch based on `ib_connectivity_harden`, then
+requested `$commit-plan`.
+
+> `git diff HEAD~1..HEAD -- piker/data/_sampling.py`
+
+Generated `trade_time_s()` to normalize common epoch units. In
+`sample_and_broadcast()`, timestamped trades reach the current 1s bar
+only when their rounded time matches the bar. A trade for the next bar
+waits briefly for samplerd's step. Older or otherwise mismatched
+trades are skipped, with one warning per symbol. Untimestamped trades
+retain the existing receive-time behavior.
+
+> `git diff HEAD~1..HEAD -- tests/test_sampling_tick_times.py`
+
+Generated focused tests for epoch units, startup backlog trades,
+untimestamped trades, and the next-step wait.
+
+The isolated worktree's focused test run passed: 8 tests. Ruff and
+`git diff --check` passed. The user also live-tested the same guard
+on direct chart and separate pikerd starts without seeing the spike.

piker/data/_sampling.py

@@ -25,6 +25,7 @@ from collections import (
     defaultdict,
 )
 from contextlib import asynccontextmanager as acm
+from math import isfinite
 import time
 from typing import (
     Any,
@@ -67,6 +68,42 @@ if TYPE_CHECKING:
 # tick-event stream.
 _default_delay_s: float = 1.0
 
+# Give the independent samplerd incrementer one step to catch up
+# before deciding a tick stamped for the next 1s bar is unusable.
+_next_tick_step_wait_s: float = 1.5
+
+
+def trade_time_s(tick: dict) -> float|None:
+    '''
+    Read a trade's epoch time in seconds, if one was supplied.
+
+    Broker feeds may omit timestamps. IB's `TickData.time` uses
+    nanoseconds; other feeds may use seconds, milliseconds, or
+    microseconds. An absent or invalid value retains the existing
+    receive-time sampling behavior.
+
+    '''
+    raw: Any = tick.get('time')
+    if (
+        isinstance(raw, bool)
+        or
+        not isinstance(raw, (int, float))
+    ):
+        return None
+
+    seconds: float = float(raw)
+    if not isfinite(seconds) or seconds <= 0:
+        return None
+
+    if seconds >= 1e17:
+        seconds /= 1e9
+    elif seconds >= 1e14:
+        seconds /= 1e6
+    elif seconds >= 1e11:
+        seconds /= 1e3
+
+    return seconds
+
 
 class Sampler:
     '''
@@ -610,6 +647,7 @@ async def sample_and_broadcast(
     log.info("Started shared mem bar writer")
 
     overruns = Counter()
+    mismatched_ticks: Counter[str] = Counter()
 
     # NOTE, only used for debugging live-data-feed issues, though
     # this should be resolved more correctly in the future using the
@@ -658,6 +696,60 @@ async def sample_and_broadcast(
                 # write trade events to shm last OHLC sample
                 if ticktype in ('trade', 'utrade'):
 
+                    tick_s: float|None = trade_time_s(tick)
+                    bar_s: float = float(rt_shm.array[-1]['time'])
+                    if (
+                        tick_s is not None
+                        and
+                        round(tick_s) == round(bar_s) + 1
+                    ):
+                        # The chart cannot see this trade price until
+                        # `Sampler.increment_ohlc_buffer()` advances.
+                        # TODO: emit a `FeedDelayed` event to feed
+                        # subscribers if this occurs beyond startup.
+                        log.warning(
+                            'DELAYING LIVE TRADE SHM UPDATE until '
+                            f'next 1s OHLC slot for '
+                            f'{broker_symbol}!\n'
+                            f'tick_time_s: {tick_s}\n'
+                            f'current_bar_s: {bar_s}\n'
+                            f'max_wait_s: {_next_tick_step_wait_s}'
+                        )
+                        with trio.move_on_after(
+                            _next_tick_step_wait_s
+                        ):
+                            while round(tick_s) > round(bar_s):
+                                await trio.sleep(0.01)
+                                bar_s = float(
+                                    rt_shm.array[-1]['time']
+                                )
+                    if (
+                        tick_s is not None
+                        and
+                        round(tick_s) != round(bar_s)
+                    ):
+                        # The newest provider frame may already hold
+                        # queued startup trades. Replaying them into
+                        # the sampler's current bar misdates them and
+                        # can double-count provider history. Leave
+                        # past rows to the history/backfill path.
+                        # Sampler.increment_ohlc_buffer() rounds its
+                        # 1s step timestamp, so use the same bucket.
+                        # TODO: `Client.bars()` fetches a one-shot
+                        # latest frame; a late trade may be absent
+                        # from its unfinished last bar. Reconcile IB
+                        # `barUpdateEvent` from `keepUpToDate=True`,
+                        # or add a provider-neutral back-bar
+                        # correction path.
+                        mismatched_ticks[broker_symbol] += 1
+                        if mismatched_ticks[broker_symbol] == 1:
+                            log.warning(
+                                'Skipping trade outside current 1s '
+                                f'bar for {broker_symbol}: '
+                                f'tick={tick_s}, bar={bar_s}'
+                            )
+                        continue
+
                     last = tick['price']
 
                     # more compact inline-way to do this assignment

tests/test_sampling_tick_times.py

@@ -0,0 +1,242 @@
+# piker: trading gear for hackers Copyright (C) 2018-present  Tyler
+# Goodlet (in stewardship of pikers)
+
+# 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/>.
+
+'''
+Live sampler checks for timestamped startup trade backlogs.
+
+'''
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+import trio
+
+from piker.data._sampling import (
+    sample_and_broadcast,
+    trade_time_s,
+)
+from piker.data import _sampling
+
+
+@pytest.mark.parametrize(
+    ('raw', 'expected'),
+    [
+        (1_790_360_484, 1_790_360_484.),  # seconds
+        (1_790_360_484_000, 1_790_360_484.),  # milliseconds
+        (1_790_360_484_000_000, 1_790_360_484.),
+        (1_790_360_484_000_000_000, 1_790_360_484.),
+        (None, None),  # provider without a tick timestamp
+    ],
+)
+def test_trade_time_s_accepts_common_epoch_units(
+    raw: int|None,
+    expected: float|None,
+) -> None:
+    assert trade_time_s({'time': raw}) == expected
+
+
+@pytest.mark.parametrize(
+    ('bar_time', 'current_time_ns'),
+    [
+        (
+            1_790_360_484.,
+            1_790_360_484_250_000_000,
+        ),
+        (
+            1_790_360_485.,
+            1_790_360_484_750_000_000,
+        ),
+    ],
+)
+def test_queued_trade_is_not_written_to_a_later_bar(
+    bar_time: float,
+    current_time_ns: int,
+) -> None:
+    '''
+    A 13:21:19 IB trade queued until the 13:21:24 1s bar must not
+    add its price or volume to either live SHM tail. A current tick
+    in the same input channel still updates both bars.
+
+    '''
+    dtype = [
+        ('time', 'f8'),
+        ('open', 'f8'),
+        ('high', 'f8'),
+        ('low', 'f8'),
+        ('close', 'f8'),
+        ('volume', 'f8'),
+    ]
+    rt = np.array(
+        [(bar_time, 30_918., 30_918.,
+          30_918., 30_918., 19.)],
+        dtype=dtype,
+    )
+    hist = np.array(
+        [(1_790_360_460., 30_918., 30_918.,
+          30_918., 30_918., 248.)],
+        dtype=dtype,
+    )
+    rt_shm = SimpleNamespace(array=rt)
+    hist_shm = SimpleNamespace(array=hist)
+    bus = SimpleNamespace(get_subs=lambda key: set())
+
+    async def replay() -> None:
+        send, receive = trio.open_memory_channel(4)
+        async with send:
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': 1_790_360_479_217_767_927,
+                        'price': 30_923.,
+                        'size': 12_013.,
+                    }],
+                    'volume': 2_222_721.,
+                },
+            })
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': current_time_ns,
+                        'price': 30_918.25,
+                        'size': 2.,
+                    }],
+                    'volume': 2_222_723.,
+                },
+            })
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': int(
+                            (bar_time + 2.) * 1_000_000_000
+                        ),
+                        'price': 30_925.,
+                        'size': 100.,
+                    }],
+                    'volume': 2_222_823.,
+                },
+            })
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'price': 30_918.5,
+                        'size': 1.,
+                    }],
+                    'volume': 2_222_824.,
+                },
+            })
+        async with receive:
+            await sample_and_broadcast(
+                bus,
+                rt_shm,
+                hist_shm,
+                receive,
+                'ib',
+            )
+
+    trio.run(replay)
+    assert float(rt['high'][-1]) == 30_918.5
+    assert float(rt['volume'][-1]) == 22.
+    assert float(hist['high'][-1]) == 30_918.5
+    assert float(hist['volume'][-1]) == 251.
+
+
+def test_next_second_trade_waits_for_sampler_step(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    '''
+    A trade for the next 1s slot must not update the older SHM bar.
+
+    The independent `Sampler.increment_ohlc_buffer()` can lag the
+    tick stream. Waiting avoids a misdated price but delays chart
+    updates, so a warning must fire before the slot advances. Hold
+    the old slot until that warning, then advance it and verify the
+    trade reaches both SHM tails without being dropped.
+
+    '''
+    dtype = [
+        ('time', 'f8'),
+        ('open', 'f8'),
+        ('high', 'f8'),
+        ('low', 'f8'),
+        ('close', 'f8'),
+        ('volume', 'f8'),
+    ]
+    rt = np.array(
+        [(1_790_360_484., 30_918., 30_918.,
+          30_918., 30_918., 0.)],
+        dtype=dtype,
+    )
+    hist = rt.copy()
+    rt_shm = SimpleNamespace(array=rt)
+    hist_shm = SimpleNamespace(array=hist)
+    bus = SimpleNamespace(get_subs=lambda key: set())
+    warnings: list[str] = []
+    wait_started = trio.Event()
+
+    def record_warning(message: str) -> None:
+        warnings.append(message)
+        wait_started.set()
+
+    monkeypatch.setattr(
+        _sampling.log,
+        'warning',
+        record_warning,
+    )
+
+    async def replay() -> None:
+        send, receive = trio.open_memory_channel(1)
+        async with send:
+            await send.send({
+                'mnq.cme.20261218': {
+                    'ticks': [{
+                        'type': 'trade',
+                        'time': 1_790_360_485_250_000_000,
+                        'price': 30_919.,
+                        'size': 2.,
+                    }],
+                    'volume': 2_222_723.,
+                },
+            })
+
+        async def step() -> None:
+            with trio.fail_after(2):
+                await wait_started.wait()
+            rt['time'][-1] = 1_790_360_485.
+
+        async with (
+            receive,
+            trio.open_nursery() as nursery,
+        ):
+            nursery.start_soon(step)
+            await sample_and_broadcast(
+                bus,
+                rt_shm,
+                hist_shm,
+                receive,
+                'ib',
+            )
+
+    trio.run(replay)
+    assert float(rt['high'][-1]) == 30_919.
+    assert float(rt['volume'][-1]) == 2.
+    assert float(hist['volume'][-1]) == 2.
+    assert len(warnings) == 1
+    assert 'DELAYING LIVE TRADE SHM UPDATE' in warnings[0]