From 976db6c33c4b8ed07eee3047226417d3928d9cca Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 21 Jul 2026 22:07:13 -0400 Subject: [PATCH] .tsp: normalize gaps after sorted dedupe - recompute datetime and delta cols after final timestamp ordering - detect only positive sample-step gaps - document inclusive null grouping with an updated visual map - check all small null layouts against an independent test oracle - honor selected null cols and reject discontinuous frame indexes - share exact null grouping across NumPy and Polars Prompt-IO: ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260722T001357Z_16f3cd6f_prompt_io.md | 46 +++ ...20260722T001357Z_16f3cd6f_prompt_io.raw.md | 40 +++ piker/tsp/_anal.py | 278 +++++---------- piker/tsp/_dedupe_smart.py | 16 +- tests/test_tsp_analysis.py | 318 ++++++++++++++++++ 5 files changed, 494 insertions(+), 204 deletions(-) create mode 100644 ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md create mode 100644 tests/test_tsp_analysis.py diff --git a/ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md b/ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md new file mode 100644 index 00000000..e168f3de --- /dev/null +++ b/ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md @@ -0,0 +1,46 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-07-22T00:13:57Z +git_ref: 16f3cd6f +scope: code +substantive: true +raw_file: 20260722T001357Z_16f3cd6f_prompt_io.raw.md +--- + +## Prompt + +Continue the gap-checker and history-backfiller repair plan. Start +with deterministic analysis characterization, then fix canonical +sorting, dedupe-derived columns, positive gap detection, and null +grouping. + +During review, the user challenged the replacement null-grouping +algorithm because its initial tests did not assert exact endpoints. +The user explicitly required retaining detailed, human-oriented +algorithm documentation such as the existing ASCII segment map. + +## Response summary + +Added deterministic analysis regressions and corrected normalization +so sorting and duplicate resolution precede derived-column +calculation. Unified null grouping for NumPy and Polars inputs, +documented its exact inclusive-boundary contract with an adapted +visual map, and verified all small null layouts against an independent +state-machine oracle. Retained sorted smart-dedupe output. + +## Files changed + +- `piker/tsp/_anal.py` - canonical gap and documented null grouping +- `piker/tsp/_dedupe_smart.py` - sorted duplicate resolution +- `tests/test_tsp_analysis.py` - exact and exhaustive regressions +- `ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md` + - unedited response record +- `ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.md` + - provenance metadata and response summary + +## Human edits + +None - generated changes have not yet been edited by the human. diff --git a/ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md b/ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md new file mode 100644 index 00000000..29723c19 --- /dev/null +++ b/ai/prompt-io/opencode/20260722T001357Z_16f3cd6f_prompt_io.raw.md @@ -0,0 +1,40 @@ +--- +model: gpt-5.6-sol +provider: openai +service: opencode +timestamp: 2026-07-22T00:13:57Z +git_ref: 16f3cd6f +diff_cmd: git diff HEAD~1..HEAD +--- + +> `git diff HEAD~1..HEAD -- piker/tsp/_anal.py` + +Generated canonical gap and null-segment normalization. Derived +datetime and delta columns are recomputed after final sorting and +duplicate resolution, and negative deltas no longer become ordinary +gaps. + +NumPy and Polars null grouping share one implementation while +retaining the human-oriented visual map of contiguous zero runs. The +documented contract defines absolute inclusive endpoints, margin +expansion, frame boundary clamping, contiguous-index validation, and +requested-column selection. + +> `git diff HEAD~1..HEAD -- piker/tsp/_dedupe_smart.py` + +Adjusted smart dedupe to honor sorted output and resolve duplicate +timestamps without retaining stale order. + +> `git diff HEAD~1..HEAD -- tests/test_tsp_analysis.py` + +Added deterministic synthetic OHLCV coverage for sorted dedupe, fresh +derived columns, and positive gap detection. Null grouping is checked +against readable endpoint examples and an independent state-machine +oracle over every layout through six rows, three margin widths, and +both NumPy and Polars inputs. + +Verification generated with the patch: + +`timeout -k 5 30 python -m pytest -q tests/test_tsp_analysis.py` + +Result: 11 passed. diff --git a/piker/tsp/_anal.py b/piker/tsp/_anal.py index 14d34238..53284c84 100644 --- a/piker/tsp/_anal.py +++ b/piker/tsp/_anal.py @@ -233,7 +233,7 @@ def get_null_segs( ) -> tuple[ # Seq, # TODO: can we make it an array-type instead? list[ - list[int, int], + list[int], ], Seq, Frame @@ -246,218 +246,92 @@ def get_null_segs( Filter to all such zero (time) segments and return the corresponding frame zeroed segment's, - - gap absolute (in buffer terms) indices-endpoints as - `absi_zsegs` - - abs indices of all rows with zeroed `col` values as `absi_zeros` + - gap absolute (in buffer terms), inclusive boundary-row + index endpoints as `absi_zsegs`; each null run is expanded + by `imargin` and clamped to the input frame + - abs indices of all rows with zeroed `col` values as + `absi_zeros` - the corresponding frame's row-entries (view) which are zeroed for the `col` as `zero_t` - ''' - times: Seq = frame['time'] - zero_pred: Seq = (times == 0) + For an interior run, the default one-row margin makes each + endpoint a valid datum immediately outside the nulls. At a frame + edge the missing side is clamped and may remain null; callers that + require query boundaries must reject or defer that segment. - if isinstance(frame, np.ndarray): - tis_zeros: int = zero_pred.any() - else: - tis_zeros: int = zero_pred.any() + Consumers converting a pair to a Python slice must add one to its + inclusive end index. + + ''' + values: Seq = frame[col] + zero_pred: Seq = (values == 0) + tis_zeros: bool = bool(zero_pred.any()) if not tis_zeros: return None - # TODO: use ndarray for this?! - absi_zsegs: list[list[int, int]] = [] + if imargin < 0: + raise ValueError('`imargin` must be >= 0') if isinstance(frame, np.ndarray): - # view of ONLY the zero segments as one continuous chunk zero_t: np.ndarray = frame[zero_pred] - # abs indices of said zeroed rows - absi_zeros = zero_t['index'] - # diff of abs index steps between each zeroed row - absi_zdiff: np.ndarray = np.diff(absi_zeros) - - if zero_t.size < 2: - idx: int = zero_t['index'][0] - idx_before: int = idx - 1 - idx_after: int = idx + 1 - index = frame['index'] - before_cond = idx_before <= index - after_cond = index <= idx_after - bars: np.ndarray = frame[ - before_cond - & - after_cond - ] - time: np.ndarray = bars['time'] - from pendulum import ( - from_timestamp, - Interval, - ) - gap: Interval = ( - from_timestamp(time[-1]) - - - from_timestamp(time[0]) - ) - log.warning( - f'Single OHLCV-bar null-segment detected??\n' - f'gap -> {gap}\n' - ) - - # ^^XXX, if you want to debug the above bar-gap^^ - # try: - # breakpoint() - # except RuntimeError: - # # XXX, if greenback not active from - # # piker store ldshm cmd.. - # log.exception( - # "Can't debug single-sample null!\n" - # ) - - return None - - # scan for all frame-indices where the - # zeroed-row-abs-index-step-diff is greater then the - # expected increment of 1. - # data 1st zero seg data zeros - # ---- ------------ ---- ----- ------ ---- - # ||||..000000000000..||||..00000..||||||..0000 - # ---- ------------ ---- ----- ------ ---- - # ^zero_t[0] ^zero_t[-1] - # ^fi_zgaps[0] ^fi_zgaps[1] - # ^absi_zsegs[0][0] ^---^ => absi_zsegs[1]: tuple - # absi_zsegs[0][1]^ - # - # NOTE: the first entry in `fi_zgaps` is where - # the first (absolute) index step diff is > 1. - # and it is a frame-relative index into `zero_t`. - fi_zgaps = np.argwhere( - absi_zdiff > 1 - # NOTE: +1 here is ensure we index to the "start" of each - # segment (if we didn't the below loop needs to be - # re-written to expect `fi_end_rows`! - ) + 1 - # the rows from the contiguous zeroed segments which have - # abs-index steps >1 compared to the previous zero row - # (indicating an end of zeroed segment). - fi_zseg_start_rows = zero_t[fi_zgaps] - - # TODO: equiv for pl.DataFrame case! + absi_zeros: np.ndarray = zero_t['index'] + frame_indexes: np.ndarray = frame['index'] + zero_indexes: np.ndarray = absi_zeros else: - izeros: pl.Series = zero_pred.arg_true() - zero_t: pl.DataFrame = frame[izeros] + zero_t: pl.DataFrame = frame.filter(zero_pred) + absi_zeros: pl.Series = zero_t['index'] + frame_indexes = frame['index'].to_numpy() + zero_indexes = absi_zeros.to_numpy() - absi_zeros = zero_t['index'] - absi_zdiff: pl.Series = absi_zeros.diff() - fi_zgaps = (absi_zdiff > 1).arg_true() + # Null-run detection depends on absolute indexes stepping by + # exactly one per frame row. Without this invariant an + # index jump between adjacent null rows is indistinguishable from + # valid data rows separating two null segments. + if np.any(np.diff(frame_indexes) != 1): + raise ValueError( + 'OHLCV frame indexes must be contiguous' + ) - # XXX: our goal (in this func) is to select out slice index - # pairs (zseg0_start, zseg_end) in abs index units for each - # null-segment portion detected throughout entire input frame. + # Scan zero-row absolute indexes for steps larger than one. Each + # such step begins another contiguous null segment: + # + # data 1st zero seg data zeros data zeros + # ---- ------------ ---- ----- ------ ----- + # ||||..000000000000..||||..00000...||||||..0000 + # ---- ------------ ---- ----- ------ ----- + # ^zero_indexes[0] ^zero_indexes[-1] + # ^split_at[0] ^split_at[1] + # ^zero_groups[0] ^zero_groups[1] ^zero_groups[2] + # + # `split_at` entries are frame-relative positions in + # `zero_indexes`, not absolute buffer indexes. `np.split()` keeps + # each run together so its first and last absolute indexes can be + # expanded into the surrounding query boundary rows. + # + # Plain arrays also keep the NumPy and Polars paths on + # exactly the same segment-grouping implementation. + split_at: np.ndarray = ( + np.flatnonzero(np.diff(zero_indexes) != 1) + + + 1 + ) + zero_groups: list[np.ndarray] = np.split( + zero_indexes, + split_at, + ) - # only up to one null-segment in entire frame? - num_gaps: int = fi_zgaps.size + 1 - if num_gaps < 1: - if absi_zeros.size > 1: - absi_zsegs = [[ - # TODO: maybe mk these max()/min() limits func - # consts instead of called more then once? - max( - absi_zeros[0] - 1, - 0, - ), - # NOTE: need the + 1 to guarantee we index "up to" - # the next non-null row-datum. - min( - absi_zeros[-1] + 1, - frame['index'][-1], - ), - ]] - else: - # XXX EDGE CASE: only one null-datum found so - # mark the start abs index as None to trigger - # a full frame-len query to the respective backend? - absi_zsegs = [[ - # see `get_hist()` in backend, should ALWAYS be - # able to handle a `start_dt=None`! - # None, - None, - absi_zeros[0] + 1, - ]] - - # XXX NOTE XXX: if >= 2 zeroed segments are found, there should - # ALWAYS be more then one zero-segment-abs-index-step-diff row - # in `absi_zdiff`, so loop through all such - # abs-index-step-diffs >1 (i.e. the entries of `absi_zdiff`) - # and add them as the "end index" entries for each segment. - # Then, iif NOT iterating the first such segment end, look back - # for the prior segments zero-segment start indext by relative - # indexing the `zero_t` frame by -1 and grabbing the abs index - # of what should be the prior zero-segment abs start index. - else: - # NOTE: since `absi_zdiff` will never have a row - # corresponding to the first zero-segment's row, we add it - # manually here. + frame_start: int = int(frame_indexes[0]) + frame_end: int = int(frame_indexes[-1]) + absi_zsegs: list[list[int]] = [] + for group in zero_groups: + zero_start: int = int(group[0]) + zero_end: int = int(group[-1]) absi_zsegs.append([ - max( - absi_zeros[0] - 1, - 0, - ), - None, + max(frame_start, zero_start - imargin), + min(frame_end, zero_end + imargin), ]) - # TODO: can we do it with vec ops? - for i, ( - fi, # frame index of zero-seg start - zseg_start_row, # full row for ^ - ) in enumerate(zip( - fi_zgaps, - fi_zseg_start_rows, - )): - assert (zseg_start_row == zero_t[fi]).all() - iabs: int = zseg_start_row['index'][0] - absi_zsegs.append([ - iabs - 1, - None, # backfilled on next iter - ]) - - # final iter case, backfill FINAL end iabs! - if (i + 1) == fi_zgaps.size: - absi_zsegs[-1][1] = absi_zeros[-1] + 1 - - # NOTE: only after the first segment (due to `.diff()` - # usage above) can we do a lookback to the prior - # segment's end row and determine it's abs index to - # retroactively insert to the prior - # `absi_zsegs[i-1][1]` entry Bo - last_end: int = absi_zsegs[i][1] - if last_end is None: - prev_zseg_row = zero_t[fi - 1] - absi_post_zseg = prev_zseg_row['index'][0] + 1 - # XXX: MUST BACKFILL previous end iabs! - absi_zsegs[i][1] = absi_post_zseg - - else: - if 0 < num_gaps < 2: - absi_zsegs[-1][1] = min( - absi_zeros[-1] + 1, - frame['index'][-1], - ) - - iabs_first: int = frame['index'][0] - for start, end in absi_zsegs: - - ts_start: float = times[start - iabs_first] - ts_end: float = times[end - iabs_first] - if ( - (ts_start == 0 and not start == 0) - or - ts_end == 0 - ): - import pdbp - pdbp.set_trace() - - assert end - assert start < end - log.warning( f'Frame has {len(absi_zsegs)} NULL GAPS!?\n' f'period: {period}\n' @@ -492,12 +366,12 @@ def iter_null_segs( ], None, ]: - if not ( - null_segs := get_null_segs( + if null_segs is None: + null_segs = get_null_segs( frame, period=timeframe, ) - ): + if not null_segs: return absi_pairs_zsegs: list[list[float, float]] @@ -609,7 +483,7 @@ def detect_time_gaps( # first select by any sample-period (in seconds unit) step size # greater then expected. step_gaps: pl.DataFrame = w_dts.filter( - pl.col('s_diff').abs() > expect_period + pl.col('s_diff') > expect_period ) if gap_dt_unit is None: @@ -701,10 +575,8 @@ def dedupe( ''' wdts: pl.DataFrame = with_dts(src_df) - deduped = wdts - # remove duplicated datetime samples/sections - deduped: pl.DataFrame = wdts.unique( + deduped: pl.DataFrame = src_df.unique( # subset=['dt'], subset=['time'], maintain_order=True, @@ -717,6 +589,8 @@ def dedupe( # -[ ] report in log msg # -[ ] possibly return segment sections which were moved? + deduped = with_dts(deduped) + diff: int = ( wdts.height - diff --git a/piker/tsp/_dedupe_smart.py b/piker/tsp/_dedupe_smart.py index 8c0ac55a..2470ddb0 100644 --- a/piker/tsp/_dedupe_smart.py +++ b/piker/tsp/_dedupe_smart.py @@ -50,8 +50,18 @@ def dedupe_ohlcv_smart( ) if dupes.is_empty(): - # No duplicates, return as-is - return (wdts, wdts, 0, None, None) + deduped: pl.DataFrame = ( + src_df.sort(time_col) + if sort + else src_df + ) + return ( + wdts, + with_dts(deduped), + 0, + None, + None, + ) # Analyze duplicate groups for validation dupe_analysis: pl.DataFrame = ( @@ -195,6 +205,8 @@ def dedupe_ohlcv_smart( if sort: deduped = deduped.sort(by=time_col) + deduped = with_dts(deduped.select(src_df.columns)) + diff: int = wdts.height - deduped.height return ( diff --git a/tests/test_tsp_analysis.py b/tests/test_tsp_analysis.py new file mode 100644 index 00000000..a3d7c2a9 --- /dev/null +++ b/tests/test_tsp_analysis.py @@ -0,0 +1,318 @@ +''' +Deterministic time-series analysis regressions. + +''' +import numpy as np +import polars as pl +import pytest + +from piker.data._source import def_iohlcv_fields +from piker.tsp import _anal +from piker.tsp._anal import ( + dedupe, + detect_time_gaps, + get_null_segs, + iter_null_segs, + np2pl, + with_dts, +) +from piker.tsp._dedupe_smart import dedupe_ohlcv_smart + + +def mk_ohlcv( + times: list[int], + indexes: list[int]|None = None, +) -> np.ndarray: + ''' + Build a synthetic OHLCV frame from timestamps. + + ''' + size: int = len(times) + frame = np.zeros( + size, + dtype=np.dtype(def_iohlcv_fields), + ) + frame['index'] = ( + np.arange(size) + if indexes is None + else indexes + ) + frame['time'] = times + frame['open'] = np.arange(size) + 10 + frame['high'] = frame['open'] + 2 + frame['low'] = frame['open'] - 2 + frame['close'] = frame['open'] + 1 + frame['volume'] = np.arange(size) + 100 + return frame + + +def expected_null_segments( + null_positions: list[int], + size: int, + offset: int, + margin: int, + +) -> list[list[int]]: + ''' + Group relative null positions with a simple state machine. + + ''' + runs: list[list[int]] = [] + for pos in null_positions: + if ( + not runs + or + pos != runs[-1][1] + 1 + ): + runs.append([pos, pos]) + else: + runs[-1][1] = pos + + frame_end: int = offset + size - 1 + return [ + [ + max(offset, offset + start - margin), + min(frame_end, offset + end + margin), + ] + for start, end in runs + ] + + +def test_dedupe_recomputes_dts_after_sort() -> None: + ''' + Derived columns describe final sorted, unique rows. + + ''' + src: pl.DataFrame = np2pl(mk_ohlcv([ + 60, + 180, + 120, + 180, + ])) + + _, deduped, diff = dedupe(src) + + assert diff == 1 + assert deduped['time'].to_list() == [60, 120, 180] + assert deduped['time_prev'].to_list() == [None, 60, 120] + assert deduped['s_diff'].to_list() == [None, 60, 60] + assert detect_time_gaps( + deduped, + expect_period=60, + ).is_empty() + + +def test_negative_delta_is_not_missing_history() -> None: + ''' + Reversed timestamps are ordering faults, not gaps. + + ''' + wdts: pl.DataFrame = with_dts( + np2pl(mk_ohlcv([120, 60])) + ) + + assert detect_time_gaps( + wdts, + expect_period=30, + ).is_empty() + + +def test_single_null_row_is_a_segment() -> None: + ''' + A one-row null segment remains repairable. + + ''' + frame: np.ndarray = mk_ohlcv([ + 60, + 120, + 0, + 240, + 300, + ]) + + nulls = get_null_segs(frame, period=60) + + assert nulls is not None + segments, zero_indexes, zero_rows = nulls + assert segments == [[1, 3]] + assert zero_indexes.tolist() == [2] + assert zero_rows['index'].tolist() == [2] + + +@pytest.mark.parametrize( + ('times', 'expected'), + [ + ([0, 0, 180, 240, 300, 360], [[10, 12]]), + ([60, 0, 0, 240, 0, 360], [[10, 13], [13, 15]]), + ([60, 120, 180, 240, 0, 0], [[13, 15]]), + ], +) +def test_null_segment_boundaries_match_numpy_and_polars( + times: list[int], + expected: list[list[int]], +) -> None: + ''' + Segments have exact, inclusive, margin-expanded endpoints. + + ''' + frame: np.ndarray = mk_ohlcv( + times, + indexes=list(range(10, 16)), + ) + + np_nulls = get_null_segs(frame, period=60) + pl_nulls = get_null_segs(np2pl(frame), period=60) + + assert np_nulls is not None + assert pl_nulls is not None + assert np_nulls[0] == expected + assert pl_nulls[0] == expected + assert np_nulls[1].tolist() == pl_nulls[1].to_list() + + +def test_null_segment_grouping_exhaustive() -> None: + ''' + Every null layout through six rows matches an independent oracle. + + ''' + offset: int = 10 + for size in range(1, 7): + for mask in range(1, 1 << size): + null_positions: list[int] = [ + pos + for pos in range(size) + if mask & (1 << pos) + ] + frame = mk_ohlcv( + [ + 0 + if pos in null_positions + else (pos + 1) * 60 + for pos in range(size) + ], + indexes=list(range(offset, offset + size)), + ) + + for margin in range(3): + expected = expected_null_segments( + null_positions, + size, + offset, + margin, + ) + for candidate in (frame, np2pl(frame)): + nulls = get_null_segs( + candidate, + period=60, + imargin=margin, + ) + assert nulls is not None + assert nulls[0] == expected + + +def test_null_segment_margin_and_column() -> None: + ''' + The requested column and index margin define each segment. + + ''' + frame: np.ndarray = mk_ohlcv( + [60, 120, 180, 240, 300], + indexes=[10, 11, 12, 13, 14], + ) + frame['volume'][1:3] = 0 + + for candidate in (frame, np2pl(frame)): + no_margin = get_null_segs( + candidate, + period=60, + imargin=0, + col='volume', + ) + wide_margin = get_null_segs( + candidate, + period=60, + imargin=2, + col='volume', + ) + + assert no_margin is not None + assert wide_margin is not None + assert no_margin[0] == [[11, 12]] + assert wide_margin[0] == [[10, 14]] + assert list(no_margin[1]) == [11, 12] + + +def test_null_segment_input_invariants() -> None: + ''' + Empty results and invalid index/margin inputs are explicit. + + ''' + frame = mk_ohlcv([60, 120, 180]) + assert get_null_segs(frame, period=60) is None + + with pytest.raises(ValueError, match='imargin'): + get_null_segs( + mk_ohlcv([60, 0, 180]), + period=60, + imargin=-1, + ) + + with pytest.raises(ValueError, match='contiguous'): + get_null_segs( + mk_ohlcv( + [60, 0, 180], + indexes=[10, 12, 13], + ), + period=60, + ) + + +def test_iter_null_segs_uses_precomputed_segments( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Iteration preserves inclusive boundary rows without rescanning. + + ''' + frame = mk_ohlcv([60, 120, 0, 240, 300]) + nulls = get_null_segs(frame, period=60) + assert nulls is not None + + def fail_rescan(*args, **kwargs): + raise AssertionError('precomputed segments were ignored') + + monkeypatch.setattr(_anal, 'get_null_segs', fail_rescan) + segments = list(iter_null_segs( + timeframe=60, + frame=frame, + null_segs=nulls, + )) + + assert len(segments) == 1 + assert segments[0][:6] == ( + 1, + 3, + 1, + 3, + 120, + 240, + ) + + +def test_smart_dedupe_honors_sort_without_dupes() -> None: + ''' + The sort option applies even when no duplicates exist. + + ''' + src: pl.DataFrame = np2pl(mk_ohlcv([ + 180, + 60, + 120, + ])) + + _, deduped, diff, _, issues = dedupe_ohlcv_smart(src) + + assert diff == 0 + assert issues is None + assert deduped['time'].to_list() == [60, 120, 180] + assert deduped['s_diff'].to_list() == [None, 60, 60]