From a26e060c3a25452f62e24bc0010037437f420c4d Mon Sep 17 00:00:00 2001 From: goodboy Date: Fri, 25 Sep 2026 17:54:42 -0400 Subject: [PATCH] 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`)) --- .../20260925T190523Z_c0dae89a_prompt_io.md | 45 ++++ ...20260925T190523Z_c0dae89a_prompt_io.raw.md | 34 +++ piker/data/_sampling.py | 92 +++++++ tests/test_sampling_tick_times.py | 242 ++++++++++++++++++ 4 files changed, 413 insertions(+) create mode 100644 ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md create mode 100644 ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.raw.md create mode 100644 tests/test_sampling_tick_times.py diff --git a/ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md b/ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.md new file mode 100644 index 00000000..2a5a669f --- /dev/null +++ b/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. diff --git a/ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.raw.md b/ai/prompt-io/codex/20260925T190523Z_c0dae89a_prompt_io.raw.md new file mode 100644 index 00000000..a8633a16 --- /dev/null +++ b/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. diff --git a/piker/data/_sampling.py b/piker/data/_sampling.py index 4f9b3d9d..b837c565 100644 --- a/piker/data/_sampling.py +++ b/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 diff --git a/tests/test_sampling_tick_times.py b/tests/test_sampling_tick_times.py new file mode 100644 index 00000000..9f9cda6f --- /dev/null +++ b/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 +# . + +''' +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]