fix(timeseries): epoch and mixed-offset timestamps, MAD zero-scale, stable anomaly columns, cdc tz (#360)

Time-series streaming parsed the timestamp column with a bare
pd.to_datetime(errors="coerce"). Integer epoch seconds were read as
nanoseconds (1970), unparseable strings silently became NaT, and a batch
with mixed UTC offsets (a DST change) raised ValueError.

- #227: numeric timestamp and event-time columns are read as epochs. The
  unit comes from the new TimeSeriesCleanConfig.timestamp_unit
  ("s"/"ms"/"us"/"ns"), or is inferred from the median magnitude
  (<1e11 s, <1e14 ms, <1e17 us, else ns); a "timeseries_timestamp_parse"
  action records the unit. Values that still fail to parse keep their row
  and are recorded in report.coerced_cells / coerced_rows (masked for
  sensitive_columns) with a medium-risk action and a report warning.
- #250: string parsing goes through coerce_datetimes(). When the plain
  parse raises, returns an object column, or loses values, it retries with
  utc=True and, on pandas >= 2 (which infers one format from the first
  value), with format="ISO8601", keeping the attempt that loses the fewest
  values. Clean batches still pay for a single parse, and a single-offset
  or naive batch keeps its zone.
- #290: anomaly_window_size=1 is rejected in the config ("must be 0 or
  >= 2") instead of failing inside pandas on the first batch.
- #291: MAD treated a zero MAD as a flat window, so the minority value of
  any sparse or two-valued series was flagged and, with cap, overwritten
  by the median. A window is now flat only when every other point in it
  is identical. A zero MAD in a non-flat window gives no scale, so the
  point is not flagged and its fences are NaN. A spike in a truly flat
  window is still flagged and capped.
- #248 part b: anomaly target columns are chosen from the (locked)
  numeric role, not from the batch dtype, so a numeric column holding a
  stray string still emits <col>_anomaly. It is scored with
  pd.to_numeric(errors="coerce"): non-numeric cells are neither flagged
  nor capped. Interpolation and seasonal imputation still require a
  numeric dtype.
- #233 part 4: cdc_profile parses event times with the same
  coerce_datetimes() helper, so mixed offsets no longer crash cummax. The
  default now is the current UTC time instead of local wall-clock time
  for naive events. now= and watermark= are converted to the event
  times' zone, reading naive values (and naive events) as UTC.

Closes #227
Closes #250
Closes #290
Closes #291
Refs #248 (part b: time-series anomaly columns)
Refs #233 (part 4: cdc_profile)

Author: kevincostner17

src/freshdata/cdc.py

@@ -27,7 +27,7 @@
 
 import pandas as pd
 
-from .streaming._timeseries import to_timedelta
+from .streaming._timeseries import coerce_datetimes, to_timedelta
 
 __all__ = ["CDCDefect", "CDCReport", "cdc_profile"]
 
@@ -156,6 +156,31 @@ def _sample_keys(
     return tuple(str(v) for v in vals)
 
 
+def _parse_event_time(raw: pd.Series) -> pd.Series:
+    """Parse the event-time column; unparseable values become ``NaT``.
+
+    Datetime columns pass through. Mixed UTC offsets (e.g. across a DST change)
+    or a mix of naive and offset-aware values are normalised to UTC, reading
+    naive values as UTC (see :func:`~freshdata.streaming._timeseries.coerce_datetimes`).
+    """
+    return coerce_datetimes(raw)
+
+
+def _to_event_tz(value: object, tz: Any) -> pd.Timestamp:
+    """Express a ``now`` / ``watermark`` reference in the event times' zone.
+
+    Naive references and naive event times are both read as UTC, so a naive
+    value compared with tz-aware events is localised to UTC first, and an
+    aware value compared with naive events becomes naive UTC.
+    """
+    stamp = pd.Timestamp(value)
+    if tz is None:
+        return stamp if stamp.tz is None else stamp.tz_convert("UTC").tz_localize(None)
+    if stamp.tz is None:
+        stamp = stamp.tz_localize("UTC")
+    return stamp.tz_convert(tz)
+
+
 def _running_max(ts: pd.Series, key_series: pd.Series | None) -> pd.Series:
     """Per-key (or global) running max of event time, in arrival order."""
     if key_series is not None:
@@ -214,7 +239,8 @@ def _freshness(
     max_ts = ts.max()
     if pd.isna(max_ts):
         return None
-    reference = now if now is not None else pd.Timestamp.now(tz=max_ts.tz)
+    # The default reference is the current UTC time (naive event times are UTC).
+    reference = now if now is not None else _to_event_tz(pd.Timestamp.now(tz="UTC"), max_ts.tz)
     freshness_seconds = float((reference - max_ts).total_seconds())
     if stale_after_td is not None and freshness_seconds > stale_after_td.total_seconds():
         defects.append(
@@ -274,6 +300,8 @@ def cdc_profile(
     now, stale_after:
         Reference 'now' and the max age before the batch is ``stale``. ``now``
         defaults to the current UTC time; pass it explicitly for determinism.
+        Naive ``now`` / ``watermark`` values and naive event times are read as
+        UTC; references are converted to the event times' zone before comparing.
     replay_threshold:
         Fraction of (duplicate + late) rows above which a ``replay_risk`` batch
         warning is raised.
@@ -292,10 +320,10 @@ def cdc_profile(
     n_rows = len(df)
     lateness_td = to_timedelta(lateness) or pd.Timedelta(0)
     stale_after_td = to_timedelta(stale_after)
-    now_ts = pd.Timestamp(now) if now is not None else None
 
-    raw = df[event_time]
-    ts = raw if pd.api.types.is_datetime64_any_dtype(raw) else pd.to_datetime(raw, errors="coerce")
+    ts = _parse_event_time(df[event_time])
+    event_tz = ts.dt.tz
+    now_ts = _to_event_tz(now, event_tz) if now is not None else None
 
     defects: list[CDCDefect] = []
 
@@ -315,7 +343,7 @@ def cdc_profile(
 
     # 2) ordering: out-of-order + late (relative to watermark / running watermark).
     if watermark is not None:
-        wm = pd.Timestamp(watermark)
+        wm = _to_event_tz(watermark, event_tz)
         late_mask = ts < wm
         n_late = int(late_mask.sum())
         counts = {"out_of_order": 0, "late": n_late}