fix(streaming): real recent-window dedup, dtype-stable row hashes, drift for constant columns

docs/streaming.md

@@ -154,7 +154,8 @@ Each batch is compared against the locked schema baseline and the running state,
 - a column's dtype changes;
 - its missing ratio jumps sharply (`drift_missing_jump`);
 - its cardinality explodes (`drift_cardinality_factor`);
-- its numeric mean shifts past `drift_zscore` σ from the running mean.
+- its numeric mean shifts past `drift_zscore` σ from the running mean, or moves at all
+  for a column that has been constant so far.
 
 ## Input formats
 
@@ -231,7 +232,8 @@ fewer rows are processed than requested.
     - Streaming mode is **micro-batch**, not true row-by-row real time.
     - **Global** cross-batch duplicate detection is limited: by default duplicates are
       scoped within a batch; enabling `global_duplicates` uses a *bounded recent-window*
-      that can miss duplicates older than the window.
+      (the `window_size` most recently seen distinct rows) that can miss duplicates
+      older than the window.
     - Medians/quantiles are **approximate** (reservoir-sampled), and the top-k category
       summary is approximate when a column saturates `max_categories`.
     - Kafka and Arrow Flight are **optional** integrations.

src/freshdata/streaming/_cleaner.py

@@ -12,15 +12,17 @@
 from __future__ import annotations
 
 import dataclasses
+from collections import OrderedDict
 from collections.abc import Iterable, Iterator
 from typing import TYPE_CHECKING, Any
 
+import numpy as np
 import pandas as pd
 
 if TYPE_CHECKING:
     from ..context import ContextPolicy
 
-from .._util import exceeds_float64_exact, fill_na_exact
+from .._util import FLOAT64_EXACT_INT, exceeds_float64_exact, fill_na_exact
 from ..cleaner import run_pipeline
 from ..config import CleanConfig, merge_options
 from ..engine.context import infer_role
@@ -75,7 +77,9 @@ def __init__(self, *, config: CleanConfig | None = None,
         )
         self._roles: dict[str, str] = {}
         self.report_: CleanReport | None = None
-        self._seen_hashes: set[int] = set()  # recent-window cross-batch dedup
+        # Recent-window cross-batch dedup: row hashes in least- to most-recently-seen
+        # order, capped at ``window_size`` (see ``_dedup_window``).
+        self._seen_hashes: OrderedDict[int, None] = OrderedDict()
         self._n_imputed = 0
         self._n_deferred = 0
         self._gate_failures = 0
@@ -373,12 +377,19 @@ def _lock_roles(self, df: pd.DataFrame) -> None:
                 self._roles[name] = infer_role(name, df[col], self.config)
 
     def _dedup_window(self, df: pd.DataFrame, report: CleanReport) -> pd.DataFrame:
-        # ponytail: bounded recent-window dedup (cap = window_size), not true global.
-        hashes = pd.util.hash_pandas_object(df, index=False).to_numpy()
-        keep = [h not in self._seen_hashes for h in hashes]
-        for h, k in zip(hashes, keep):
-            if k and len(self._seen_hashes) < self.scfg.window_size:
-                self._seen_hashes.add(int(h))
+        # Bounded recent-window dedup, not true global: a row is dropped when it
+        # repeats one of the ``window_size`` most recently seen distinct rows of
+        # earlier batches. Within-batch duplicates are left to the pipeline.
+        hashes = _row_hashes(df)
+        seen = self._seen_hashes
+        keep = [h not in seen for h in hashes]
+        for h in hashes:
+            if h in seen:
+                seen.move_to_end(h)  # a repeat counts as recent again
+            else:
+                seen[h] = None
+        while len(seen) > self.scfg.window_size:
+            seen.popitem(last=False)  # evict the least recently seen row
         removed = len(df) - sum(keep)
         if removed:
             report.add("duplicates", f"removed {removed} cross-batch duplicate row(s) "
@@ -507,3 +518,33 @@ def _preserve(self, df: pd.DataFrame, col: str, miss: int, report: CleanReport,
 
 def _round(value: float | None) -> float | None:
     return round(value, 2) if value is not None else None
+
+
+def _row_hashes(df: pd.DataFrame) -> list[int]:
+    """Per-row hashes that are stable when a numeric column's dtype flips.
+
+    ``hash_pandas_object`` hashes the dtype-specific bytes, so ``1`` (int64) and
+    ``1.0`` (float64, e.g. an int column promoted by a missing value) differ.
+    Numeric non-bool columns, nullable ones included, are hashed as float64
+    (missing -> NaN, ``-0.0`` -> ``0.0``). An integer column holding a value
+    beyond ±2**53 keeps its own dtype, because float64 would merge distinct values.
+    """
+    columns: dict[int, pd.Series] = {}
+    changed = False
+    for i in range(df.shape[1]):
+        s = df.iloc[:, i]
+        kind = getattr(s.dtype, "kind", None)
+        if kind in ("i", "u", "f") and (kind == "f" or _ints_fit_float64(s)):
+            values = s.to_numpy(dtype="float64", na_value=np.nan) + 0.0
+            s = pd.Series(values, index=df.index)
+            changed = True
+        columns[i] = s
+    frame = pd.DataFrame(columns, index=df.index) if changed else df
+    return pd.util.hash_pandas_object(frame, index=False).tolist()
+
+
+def _ints_fit_float64(s: pd.Series) -> bool:
+    present = s.dropna()
+    if present.empty:
+        return True
+    return int(present.max()) <= FLOAT64_EXACT_INT and int(present.min()) >= -FLOAT64_EXACT_INT

src/freshdata/streaming/_config.py

@@ -40,6 +40,9 @@ class StreamingCleanConfig:
     global_duplicates:
         When False (default), duplicate removal is scoped **within each batch**
         only — cross-batch duplicate detection would require unbounded state.
+        When True, a row is also dropped if it repeats one of the ``window_size``
+        most recently seen distinct rows from earlier batches (least recently
+        seen rows are evicted first).
     drift_missing_jump:
         Absolute jump in a column's missing ratio (batch vs. running) that flags
         missing-rate drift.
@@ -48,7 +51,8 @@ class StreamingCleanConfig:
         count flags a cardinality explosion.
     drift_zscore:
         |batch mean − running mean| / running std above this flags a numeric
-        distribution shift.
+        distribution shift. For a column that has been constant so far (running
+        std 0), any change of the batch mean flags it.
     seed:
         Seed for reservoir sampling, for reproducible approximate quantiles.
     """

src/freshdata/streaming/_drift.py

@@ -112,12 +112,24 @@ def _distribution_drift(s, cstate, config, findings) -> None:
     if not (is_numeric_dtype(s) and not is_bool_dtype(s)):
         return
     snap = cstate.numeric_snapshot()
-    if snap.count < 2 or snap.std <= 0:
+    if snap.count < 2:
         return
     nonnull = pd.to_numeric(s, errors="coerce").dropna()
     if nonnull.empty:
         return
-    z = abs(float(nonnull.mean()) - snap.mean) / snap.std
+    batch_mean = float(nonnull.mean())
+    if snap.std <= 0:
+        # Constant so far: the z-score is undefined (any move is infinite σ), so
+        # flag a mean that leaves the constant by more than a float tolerance.
+        if abs(batch_mean - snap.mean) > 1e-9 * max(1.0, abs(snap.mean)):
+            findings.append(DriftFinding(
+                "distribution", cstate.name,
+                f"mean of '{cstate.name}' moved to {batch_mean:.4g} from a column "
+                f"that was constant at {snap.mean:.4g}",
+                risk="high",
+            ))
+        return
+    z = abs(batch_mean - snap.mean) / snap.std
     if z > config.drift_zscore:
         findings.append(DriftFinding(
             "distribution", cstate.name,

tests/test_streaming_dedup_drift.py

@@ -0,0 +1,148 @@
+"""Regression tests for streaming cross-batch dedup (#292, #293) and drift (#294)."""
+
+from __future__ import annotations
+
+import pandas as pd
+
+import freshdata as fd
+
+
+def _keys_batch(keys):
+    return pd.DataFrame({"k": keys, "s": [f"x{i}" for i in keys]})
+
+
+def _run(cleaner, batches):
+    return [cleaner.clean_batch(b)[0] for b in batches]
+
+
+# -- #292: the window keeps the most recent rows, not the first ones -------------------
+
+
+def test_duplicates_of_recent_rows_are_removed_after_window_fills():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False)
+    outs = _run(cleaner, [_keys_batch([1, 2]), _keys_batch([3, 4]), _keys_batch([3, 4])])
+    assert outs[2]["k"].tolist() == []
+
+
+def test_duplicates_after_more_than_window_size_distinct_rows():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=3, verbose=False)
+    batches = [_keys_batch([1, 2, 3]), _keys_batch([4, 5, 6]), _keys_batch([7, 8]),
+               _keys_batch([6, 7, 8])]
+    outs = _run(cleaner, batches)
+    assert outs[3]["k"].tolist() == []
+
+
+def test_oldest_rows_are_evicted_and_window_stays_bounded():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False)
+    outs = _run(cleaner, [_keys_batch([1, 2]), _keys_batch([3, 4]), _keys_batch([1])])
+    assert outs[2]["k"].tolist() == [1]  # 1 fell out of the 2-row window
+    assert len(cleaner._seen_hashes) <= 2
+
+
+def test_repeat_refreshes_recency():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False)
+    outs = _run(cleaner, [
+        _keys_batch([1, 2]),
+        _keys_batch([1]),  # duplicate: removed, and 1 becomes most recent
+        _keys_batch([3]),  # evicts 2, the least recently seen
+        _keys_batch([1]),
+        _keys_batch([2]),
+    ])
+    assert [len(o) for o in outs] == [2, 0, 1, 0, 1]
+
+
+def test_batch_larger_than_window_keeps_only_latest_rows():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False)
+    outs = _run(cleaner, [_keys_batch([1, 2, 3, 4]), _keys_batch([1, 3, 4])])
+    assert outs[1]["k"].tolist() == [1]
+    assert len(cleaner._seen_hashes) == 2
+
+
+# -- #293: hashing survives int64 <-> float64 flips ------------------------------------
+
+
+def test_int_row_repeats_in_nan_promoted_float_batch():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"k": [1, 2], "v": [10, 20]}))
+    out, report = cleaner.clean_batch(pd.DataFrame({"k": [1, 3], "v": [10, None]}))
+    assert out["k"].tolist() == [3]
+    assert report.duplicates_removed == 1
+
+
+def test_float_row_repeats_in_later_int_batch():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"k": [1, 2], "v": [10.0, None]}))
+    out, _ = cleaner.clean_batch(pd.DataFrame({"k": [2, 5], "v": [20, 30]}))
+    assert out["k"].tolist() == [2, 5]  # (2, NaN) != (2, 20)
+    out, _ = cleaner.clean_batch(pd.DataFrame({"k": [1, 6], "v": [10, 60]}))
+    assert out["k"].tolist() == [6]
+
+
+def test_nullable_integer_matches_plain_float():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"v": pd.array([1, None, 3], dtype="Int64")}))
+    out, _ = cleaner.clean_batch(pd.DataFrame({"v": [3.0, 4.0]}))
+    assert out["v"].tolist() == [4.0]
+
+
+def test_negative_zero_matches_integer_zero():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"k": ["a", "b"], "v": [0, 1]}))
+    out, _ = cleaner.clean_batch(pd.DataFrame({"k": ["a", "c"], "v": [-0.0, 2.5]}))
+    assert out["k"].tolist() == ["c"]
+
+
+def test_distinct_numeric_values_are_not_merged():
+    cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"k": ["a"], "v": [1]}))
+    out, _ = cleaner.clean_batch(pd.DataFrame({"k": ["a"], "v": [1.5]}))
+    assert len(out) == 1
+
+
+def test_integers_beyond_float64_precision_stay_distinct():
+    big = 2**53
+    cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"id": [big + 1]}))
+    out, _ = cleaner.clean_batch(pd.DataFrame({"id": [big]}))
+    assert out["id"].tolist() == [big]
+
+
+# -- #294: drift on a column that was constant ------------------------------------------
+
+
+def _drift_actions(report):
+    return [a for a in report.actions if a.step == "drift"]
+
+
+def test_constant_column_jump_reports_distribution_drift():
+    cleaner = fd.StreamingCleaner(verbose=False)
+    for val in (5, 5, 5, 1_000_000):
+        _, report = cleaner.clean_batch(pd.DataFrame({"x": [val] * 50, "y": list(range(50))}))
+    assert report.streaming["schema_drift_detected"] is True
+    actions = _drift_actions(report)
+    assert [a.column for a in actions] == ["x"]
+    assert actions[0].risk == "high"
+    assert cleaner.state.drift_log[-1]["kind"] == "distribution"
+
+
+def test_constant_column_unchanged_reports_no_drift():
+    cleaner = fd.StreamingCleaner(verbose=False)
+    for _ in range(4):
+        _, report = cleaner.clean_batch(pd.DataFrame({"x": [5.0] * 50, "y": list(range(50))}))
+        assert _drift_actions(report) == []
+    assert cleaner.state.drift_log == []
+
+
+def test_constant_column_within_float_tolerance_reports_no_drift():
+    cleaner = fd.StreamingCleaner(verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"x": [0.1] * 10, "y": list(range(10))}))
+    _, report = cleaner.clean_batch(
+        pd.DataFrame({"x": [0.1 + 1e-15] * 10, "y": list(range(10))}))
+    assert _drift_actions(report) == []
+
+
+def test_single_prior_value_does_not_report_drift():
+    cleaner = fd.StreamingCleaner(verbose=False)
+    cleaner.clean_batch(pd.DataFrame({"x": [5], "y": ["a"]}))
+    _, report = cleaner.clean_batch(pd.DataFrame({"x": [9], "y": ["a"]}))
+    assert _drift_actions(report) == []