#!/usr/bin/env python """Deterministic dataset profiler — the honest core of CSV -> Insights (#12). Everything numeric a user ever sees is computed HERE, in plain pandas, not by an LLM. The narrator (narrator.py) only turns this structured profile into prose; it never invents a statistic. This is the same "grounded, no hallucinated numbers" discipline as the #10 RAG app. from profiler import profile_csv prof = profile_csv("samples/seattle-weather.csv") # -> JSON-able dict """ import warnings import numpy as np import pandas as pd MAX_TOP = 8 # top categorical values reported per column OUTLIER_K = 1.5 # IQR multiplier for outlier flagging def _round(x, n=3): """JSON-safe rounding: NaN/inf -> None, numpy scalars -> python floats/ints.""" try: if x is None or (isinstance(x, float) and (np.isnan(x) or np.isinf(x))): return None if isinstance(x, (np.integer,)): return int(x) if isinstance(x, (np.floating, float)): v = float(x) return None if (np.isnan(v) or np.isinf(v)) else round(v, n) return x except Exception: return None def _detect_datetime(s: pd.Series) -> pd.Series | None: """Return a parsed datetime Series if the column looks like dates, else None. Object columns are parsed leniently; a column qualifies only if most values parse (avoids mangling free text).""" if pd.api.types.is_datetime64_any_dtype(s): return s if s.dtype == object: name = str(s.name).lower() looks_datey = any(k in name for k in ("date", "time", "day", "month", "year", "timestamp")) sample = s.dropna().astype(str).head(50) if sample.empty: return None with warnings.catch_warnings(): warnings.simplefilter("ignore") # dateutil "could not infer format" is expected here # format="mixed" parses each value on its own terms — robust across pandas versions, # where bare inference could coerce valid ISO dates to NaT and mis-type the column. parsed = pd.to_datetime(sample, errors="coerce", format="mixed") if parsed.notna().mean() >= (0.6 if looks_datey else 0.95): return pd.to_datetime(s, errors="coerce", format="mixed") return None def _numeric_summary(s: pd.Series) -> dict: d = s.dropna() if d.empty: return {"count": 0} q1, q3 = d.quantile(0.25), d.quantile(0.75) iqr = q3 - q1 lo, hi = q1 - OUTLIER_K * iqr, q3 + OUTLIER_K * iqr outliers = int(((d < lo) | (d > hi)).sum()) if iqr > 0 else 0 return { "count": int(d.count()), "mean": _round(d.mean()), "std": _round(d.std()), "min": _round(d.min()), "q1": _round(q1), "median": _round(d.median()), "q3": _round(q3), "max": _round(d.max()), "skew": _round(d.skew()) if d.count() > 2 else None, "outliers": outliers, "outlier_pct": _round(100 * outliers / len(d), 1), } def _categorical_summary(s: pd.Series) -> dict: d = s.dropna().astype(str) vc = d.value_counts() top = [{"value": k[:60], "count": int(v), "pct": _round(100 * v / len(d), 1)} for k, v in vc.head(MAX_TOP).items()] return {"unique": int(s.nunique(dropna=True)), "top": top} def profile_df(df: pd.DataFrame, name: str = "dataset") -> dict: n_rows, n_cols = df.shape columns, numeric_cols, cat_cols, dt_info = [], [], [], [] for col in df.columns: s = df[col] missing = int(s.isna().sum()) info = { "name": str(col), "dtype": str(s.dtype), "missing": missing, "missing_pct": _round(100 * missing / n_rows, 1) if n_rows else None, "role": None, } dt = _detect_datetime(s) if dt is not None and dt.notna().any(): info["role"] = "datetime" valid = dt.dropna() span_days = int((valid.max() - valid.min()).days) if len(valid) > 1 else 0 dt_info.append({"name": str(col), "start": str(valid.min().date()), "end": str(valid.max().date()), "span_days": span_days}) elif pd.api.types.is_numeric_dtype(s): info["role"] = "numeric" info["stats"] = _numeric_summary(s) numeric_cols.append(str(col)) else: info["role"] = "categorical" info["stats"] = _categorical_summary(s) cat_cols.append(str(col)) columns.append(info) # correlations (strongest numeric pairs) corr_pairs = [] if len(numeric_cols) >= 2: corr = df[numeric_cols].corr(numeric_only=True) seen = set() for a in numeric_cols: for b in numeric_cols: if a == b or (b, a) in seen: continue seen.add((a, b)) r = corr.loc[a, b] if pd.notna(r): corr_pairs.append({"a": a, "b": b, "r": _round(r)}) corr_pairs.sort(key=lambda p: abs(p["r"] or 0), reverse=True) # data-quality flags (deterministic, plain rules) flags = [] for c in columns: if c["missing_pct"] and c["missing_pct"] >= 20: flags.append(f"'{c['name']}' is {c['missing_pct']}% missing") if c.get("role") == "categorical" and c["stats"]["unique"] <= 1 and n_rows: flags.append(f"'{c['name']}' has a single constant value") if c.get("role") == "numeric" and c["stats"].get("outlier_pct"): if c["stats"]["outlier_pct"] >= 5: flags.append(f"'{c['name']}' has {c['stats']['outlier_pct']}% outliers (IQR)") dup = int(df.duplicated().sum()) if dup: flags.append(f"{dup} duplicate row(s)") return { "name": name, "shape": {"rows": int(n_rows), "cols": int(n_cols)}, "columns": columns, "numeric_cols": numeric_cols, "categorical_cols": cat_cols, "datetime_cols": dt_info, "top_correlations": corr_pairs[:5], "duplicate_rows": dup, "flags": flags, } def profile_csv(path_or_buffer, name: str = None, max_rows: int = 200_000) -> dict: df = pd.read_csv(path_or_buffer, nrows=max_rows) if isinstance(path_or_buffer, str) and name is None: name = path_or_buffer.replace("\\", "/").split("/")[-1] return profile_df(df, name=name or "dataset") if __name__ == "__main__": import json, sys p = sys.argv[1] if len(sys.argv) > 1 else "samples/seattle-weather.csv" print(json.dumps(profile_csv(p), indent=2))