diff --git a/CHANGELOG.md b/CHANGELOG.md index 8cc50e0..3f46902 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -96,6 +96,11 @@ adheres to [Semantic Versioning](https://semver.org/). - The missing-pyarrow error names the feature that needs it (for example Arrow output), and Parquet metadata reads no longer fail with `AttributeError` in a fresh process (#215). +- Docs: the streaming per-batch field table described `rolling_trust_score` as + row-weighted. It is an unweighted mean of the per-batch scores in the window, + so a 1-row batch counts as much as a 999-row one; only + `cumulative_trust_score` is row-weighted. Documentation fix only — the score + is unchanged (#349). ## [2.0.0] - 2026-07-20 diff --git a/docs/streaming.md b/docs/streaming.md index 8db5cc2..56b74a9 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -97,7 +97,7 @@ in `report.to_dict()`) carries: | `batch_id` | 1-based index of this batch | | `rows_in_batch` / `rows_seen_total` | rows in this batch / cumulative across the stream | | `batch_trust_score` | trust score of this cleaned batch (0-100) | -| `rolling_trust_score` | row-weighted trust over the recent window | +| `rolling_trust_score` | unweighted mean of the last `rolling_trust_window` batch scores | | `cumulative_trust_score` | row-weighted trust over the whole stream | | `schema_drift_detected` | whether drift was flagged for this batch | | `warmup_phase` | whether this batch was still in warmup | diff --git a/tests/test_streaming_state.py b/tests/test_streaming_state.py index 8279792..602d099 100644 --- a/tests/test_streaming_state.py +++ b/tests/test_streaming_state.py @@ -2,7 +2,9 @@ import numpy as np import pandas as pd +import pytest +from freshdata.streaming import StreamingCleaner from freshdata.streaming._state import ColumnState, StreamingState from freshdata.streaming._stats import BoundedCounter, ReservoirSampler, Welford @@ -90,6 +92,53 @@ def test_streaming_state_tracks_rows_and_trust(): assert abs(cumulative - (80 * 3 + 90 * 1) / 4) < 1e-9 # rows-weighted +# --- documented behaviour: the rolling score is NOT row-weighted ------------- +# docs/streaming.md called rolling_trust_score "row-weighted trust over the +# recent window", but it is an unweighted mean of the per-batch scores in the +# window — a 1-row batch counts exactly as much as a 999-row one. Only +# cumulative_trust_score is row-weighted. The docs were corrected to match; +# these tests pin the arithmetic so the two cannot drift apart again. + +def test_rolling_trust_is_unweighted_while_cumulative_is_row_weighted(): + state = StreamingState(rolling_trust_window=8) + state.record_trust(0.0, rows=1) # tiny, terrible batch + rolling, cumulative = state.record_trust(100.0, rows=999) # huge, clean batch + + assert rolling == 50.0 # (0 + 100) / 2 — rows ignored + assert abs(cumulative - (0 * 1 + 100 * 999) / 1000) < 1e-9 # 99.9, rows honoured + assert rolling != pytest.approx(cumulative) # the two must not be conflated + + +def test_rolling_trust_ignores_batch_size_end_to_end(): + """The issue's reproduction: a 1-row all-None batch, then 999 clean rows. + + ``drop_empty_rows=False`` matters here: the streaming representation pass + clears ``drop_empty_columns`` but keeps the default row dropping, so the + all-None row would otherwise be removed before the batch is scored. That + leaves a 0-row batch, and a 0-vs-999 split cannot demonstrate anything + about weighting — the tiny batch would carry no weight under either rule. + """ + cleaner = StreamingCleaner(verbose=False, drop_empty_rows=False) + + _, tiny_rep = cleaner.clean_batch(pd.DataFrame({"a": [None], "b": [None]})) + _, big_rep = cleaner.clean_batch(pd.DataFrame({ + "a": list(range(999)), "b": [float(i) for i in range(999)], + })) + + tiny, big = tiny_rep.streaming, big_rep.streaming + scores = [tiny["batch_trust_score"], big["batch_trust_score"]] + rows = [tiny["rows_in_batch"], big["rows_in_batch"]] + assert rows == [1, 999] # the 1-vs-999 split the issue describes + assert scores[0] < scores[1] # and the batches really do score differently + + # Derived from the observed scores, so an unrelated change to how a batch is + # scored will not break this — only a change to the *weighting* will. + assert big["rolling_trust_score"] == pytest.approx(sum(scores) / 2) + assert big["rolling_trust_score"] != pytest.approx( + sum(s * r for s, r in zip(scores, rows)) / sum(rows) + ) + + def test_state_to_dict_is_json_friendly(): state = StreamingState() state.observe_batch(pd.DataFrame({"n": [1.0, 2.0], "c": ["x", "y"]}),