"""EDAgent - a lightweight EDA agent for Hugging Face Spaces (ZeroGPU). Give it a Hugging Face dataset id and a prompt. It builds a compact schema card, writes its own matplotlib/seaborn analysis code, runs that code in a guarded sandbox, repairs it once if it crashes, and returns an EDA report. Architecture: PYTHON COMPUTES, THE LLM PHRASES. Every number, table, association, recommendation and hypothesis is produced by pandas and scipy. The model writes plotting code and rephrases sentences that were already assembled from computed values, so no statistic in the report can be hallucinated - and audit_numbers() flags any numeral in the prose that is absent from the computed evidence. """ import os import warnings import logging os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1") os.environ.setdefault("TRANSFORMERS_VERBOSITY", "error") os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") warnings.filterwarnings("ignore") logging.getLogger("huggingface_hub").setLevel(logging.ERROR) import io import re import glob import time import shutil import base64 import traceback import contextlib import subprocess import concurrent.futures import numpy as np import pandas as pd import torch import spaces import gradio as gr import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import seaborn as sns from matplotlib.axes import Axes from matplotlib.ticker import (FuncFormatter, LogLocator, NullFormatter, ScalarFormatter) from scipy import stats from huggingface_hub import HfApi, hf_hub_download from transformers import (AutoTokenizer, AutoModelForCausalLM, StoppingCriteria, StoppingCriteriaList) # ZeroGPU is an RTX Pro 6000 Blackwell (compute 12.x), so bf16 is native. The # CPU fallback exists only so the Space still boots if the hardware setting is # wrong - a report on CPU takes 15-40 minutes and is not usable. DEVICE = "cuda" if torch.cuda.is_available() else "cpu" DTYPE = torch.bfloat16 if DEVICE == "cuda" else torch.float32 PLOTS_DIR = "plots" os.makedirs(PLOTS_DIR, exist_ok=True) # =========================================================================== # # Data layer # =========================================================================== # TABULAR_EXT = (".csv", ".tsv", ".parquet", ".json", ".jsonl") def _read_any(path: str) -> pd.DataFrame: """Dispatch to the right pandas reader based on extension.""" if path.endswith(".parquet"): return pd.read_parquet(path) if path.endswith(".tsv"): return pd.read_csv(path, sep="\t") if path.endswith((".json", ".jsonl")): return pd.read_json(path, lines=path.endswith(".jsonl")) return pd.read_csv(path) def load_hf_dataframe(hf_dataset: str) -> pd.DataFrame: """Hub dataset id -> DataFrame. Tries load_dataset, falls back to raw files. The fallback matters: datasets>=3.0 dropped script-based repos, so datasets such as mstz/titanic can no longer be loaded the standard way. """ try: from datasets import load_dataset ds = load_dataset(hf_dataset) split = "train" if "train" in ds else list(ds.keys())[0] print(f"[loader] load_dataset OK split='{split}'", flush=True) return ds[split].to_pandas() except Exception as e: print(f"[loader] load_dataset failed -> {type(e).__name__}: {str(e)[:120]}", flush=True) files = HfApi().list_repo_files(hf_dataset, repo_type="dataset") cands = [f for f in files if f.lower().endswith(TABULAR_EXT)] if not cands: raise RuntimeError(f"No tabular file found in {hf_dataset}. Files: {files}") cands.sort(key=lambda f: (("train" not in f.lower()), len(f))) pick = cands[0] print(f"[loader] fallback -> downloading '{pick}'", flush=True) df = _read_any(hf_hub_download(hf_dataset, pick, repo_type="dataset")) print("[loader] fallback OK", flush=True) return df def drop_index_cols(df: pd.DataFrame): """Remove monotonic unique integer columns - row ids, not variables.""" drop = [c for c in df.columns if pd.api.types.is_integer_dtype(df[c]) and df[c].is_monotonic_increasing and df[c].nunique() == len(df)] return df.drop(columns=drop), drop # =========================================================================== # # Normalize and context card # =========================================================================== # MAX_CONTEXT_COLS = 40 # columns described to the model MAX_HEAD_COLS = 10 # columns in the sample rows: head(3) over 470 columns # was 99,000 chars on its own, 62% of the whole card def normalize(df: pd.DataFrame) -> pd.DataFrame: """Strip literal quote chars, turn empty strings into real NaN, re-infer numerics.""" df = df.copy() for c in df.select_dtypes(include="object").columns: s = df[c].astype(str).str.strip() s = s.str.replace(r"^'(.*)'$", r"\1", regex=True) df[c] = s.str.strip().replace( {"": np.nan, "nan": np.nan, "None": np.nan, "?": np.nan}) for c in df.select_dtypes(include="object").columns: conv = pd.to_numeric(df[c], errors="coerce") if conv.isna().sum() == df[c].isna().sum(): df[c] = conv return df def pick_context_cols(df: pd.DataFrame, max_cols: int = MAX_CONTEXT_COLS): """Which columns the model gets to see. Returns (columns, n_omitted). Constant columns are dropped first - they cost tokens and say nothing. If the table is still too wide, sample EVENLY across it rather than taking the first N: on the restaurants dataset the first 40 columns are all address fields, which would have told the model nothing about the 400 metric columns that follow. """ cols = [c for c in df.columns if df[c].nunique(dropna=False) > 1] or list(df.columns) if len(cols) <= max_cols: return cols, len(df.columns) - len(cols) idx = np.unique(np.linspace(0, len(cols) - 1, max_cols).round().astype(int)) keep = [cols[i] for i in idx] return keep, len(df.columns) - len(keep) def build_context(df: pd.DataFrame, name: str) -> str: """Compact, model-facing description of the table. One line per column.""" shown, omitted = pick_context_cols(df) lines = [f"Dataset: {name}", f"Shape: {df.shape[0]} rows x {df.shape[1]} columns"] if omitted: lines.append(f"(describing {len(shown)} representative columns; " f"{omitted} constant or similar columns omitted)") lines += ["", "Columns:"] for c in shown: s, nulls = df[c], df[c].isna().sum() if pd.api.types.is_numeric_dtype(s): detail = f"min={s.min():.4g} max={s.max():.4g} mean={s.mean():.4g}" else: detail = "examples=" + ", ".join(map(str, s.dropna().unique()[:4]))[:70] lines.append(f"- {c} ({s.dtype}) nulls={nulls} ({100*nulls/len(df):.1f}%) " f"unique={s.nunique(dropna=True)} {detail}") head_cols = shown[:MAX_HEAD_COLS] lines += ["", f"First 3 rows (columns {', '.join(map(str, head_cols))}):", df[head_cols].head(3).to_string(max_colwidth=22)] return "\n".join(lines) # =========================================================================== # # Model layer # =========================================================================== # MODEL_ID = "Qwen/Qwen2.5-Coder-1.5B-Instruct" MODEL_WINDOW = 32768 # Qwen2.5-Coder-1.5B-Instruct MAX_PROMPT_TOKENS = 6000 # our own budget, far below the window: a T4 also # has to hold the KV cache, and a 1.5B model reasons # worse the more irrelevant schema you hand it tok = AutoTokenizer.from_pretrained(MODEL_ID) model = AutoModelForCausalLM.from_pretrained( MODEL_ID, dtype=DTYPE, device_map="cuda", ) model.eval() if tok.pad_token_id is None: tok.pad_token = tok.eos_token # decoder-only models must be LEFT padded for batched generation, otherwise the # padding sits between the prompt and the first generated token and the output # is garbage. Harmless for a batch of one. tok.padding_side = "left" print(f"loaded {MODEL_ID}") print(f"params : {model.num_parameters()/1e9:.2f}B") print(f"vram : {torch.cuda.memory_allocated()/1e9:.2f} GB") print(f"window : {MODEL_WINDOW:,} tokens (prompt budget {MAX_PROMPT_TOKENS:,})") class StopOnClosingFence(StoppingCriteria): """Stop once a fenced block has been opened and closed. Decoding the tail is more reliable than matching token ids: ``` tokenizes differently depending on what precedes it. Decoding on every step would be wasteful, so we only check every `check_every` tokens. """ def __init__(self, tokenizer, prompt_len, fences=2, check_every=8): self.tok, self.prompt_len = tokenizer, prompt_len self.fences, self.check_every = fences, check_every self._step = 0 def __call__(self, input_ids, scores, **kwargs): self._step += 1 done = False if self._step % self.check_every == 0: text = self.tok.decode(input_ids[0][self.prompt_len:], skip_special_tokens=True) done = text.count("```") >= self.fences return torch.full((input_ids.shape[0],), done, dtype=torch.bool, device=input_ids.device) def _check_budget(n_tokens: int, where: str): """Fail with a diagnosis instead of an OOM or a silent garbage completion. A 470-column dataset built a 40,000-token context card - larger than the model's whole window. The old code passed it straight to model.generate and the request died with an unexplained error. Now the message names the cause. """ if n_tokens > MAX_PROMPT_TOKENS: raise ValueError( f"{where}: prompt is {n_tokens:,} tokens, over the " f"{MAX_PROMPT_TOKENS:,}-token budget ({MODEL_ID} has a " f"{MODEL_WINDOW:,}-token window and a T4 must also hold the KV " f"cache). This usually means a very wide table - lower " f"MAX_CONTEXT_COLS in the context-card cell.") @torch.inference_mode() def generate(system: str, user: str, max_new_tokens: int = 900, stop_fences: int = 0) -> str: """Single entry point to the LLM. Returns only the newly generated text. stop_fences=2 halts as soon as the closing ``` of a code block is emitted, instead of running on to max_new_tokens with trailing prose. """ msgs = [{"role": "system", "content": system}, {"role": "user", "content": user}] text = tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True) inputs = tok(text, return_tensors="pt") prompt_len = inputs["input_ids"].shape[1] _check_budget(prompt_len, "generate") inputs = inputs.to(model.device) criteria = None if stop_fences: criteria = StoppingCriteriaList( [StopOnClosingFence(tok, prompt_len, fences=stop_fences)]) out = model.generate( **inputs, max_new_tokens=max_new_tokens, do_sample=False, pad_token_id=tok.pad_token_id, stopping_criteria=criteria, ) return tok.decode(out[0][prompt_len:], skip_special_tokens=True) @torch.inference_mode() def generate_batch(pairs: list, max_new_tokens: int = 160) -> list: """N independent (system, user) prompts in ONE forward pass. At 1.5B, decoding is memory-bandwidth-bound: a batch of six costs about the same wall time as a batch of one. The narrative calls are independent, so running them sequentially was paying six times for nothing. Cost is set by the LONGEST row, so group prompts with similar output length. """ if not pairs: return [] texts = [tok.apply_chat_template( [{"role": "system", "content": s}, {"role": "user", "content": u}], tokenize=False, add_generation_prompt=True) for s, u in pairs] enc = tok(texts, return_tensors="pt", padding=True) # the batch is padded to the longest row, so that row sets the cost _check_budget(enc["input_ids"].shape[1], "generate_batch") enc = enc.to(model.device) out = model.generate( **enc, max_new_tokens=max_new_tokens, do_sample=False, pad_token_id=tok.pad_token_id, ) gen = out[:, enc["input_ids"].shape[1]:] return [tok.decode(g, skip_special_tokens=True) for g in gen] # =========================================================================== # # Code generation # =========================================================================== # CODE_SYSTEM = """You are a data analyst. You write Python code for exploratory data analysis. Rules: - A pandas DataFrame named `df` is already loaded. Never load, download or create data. - Use pandas, matplotlib and seaborn. - print() every result you want reported. - Save each figure with plt.savefig("plots/.png", bbox_inches="tight") then plt.close(). Never call plt.show(). - Reply with exactly one ```python code block and no other text.""" PLOT_MAX_PIE_LEVELS = 5 # more wedges than this is unreadable PLOT_MAX_GROUP_LEVELS = 10 # bars / boxes per chart def _pl_num(d): """Numeric columns that actually vary.""" return [c for c in d.select_dtypes(include="number").columns if d[c].nunique(dropna=True) > 1] def _pl_cat(d, max_levels): return [c for c in d.columns if 2 <= d[c].nunique(dropna=True) <= max_levels] def _pl_datetime(d): return list(d.select_dtypes(include=["datetime64[ns]", "datetimetz"]).columns) # Every plot type carries the condition under which it means anything. Asking a # 1.5B model for a chart the data cannot support wastes generation budget and # produces something misleading - an "area plot across the sorted row order" is # a time series with no time axis, which is a picture of the sort order. PLOT_MENU = { "Histogram": { "desc": "histograms of the main numeric columns", "needs": lambda d: bool(_pl_num(d)), "why": "no numeric column varies", }, "Bar Graph": { "desc": ("bar charts of value counts for the low-cardinality " "categorical columns"), "needs": lambda d: bool(_pl_cat(d, PLOT_MAX_GROUP_LEVELS)), "why": f"no categorical column with 2-{PLOT_MAX_GROUP_LEVELS} levels", }, "Scatter Plot": { "desc": ("scatter plots of a few pairs of numeric columns that look " "related"), "needs": lambda d: len(_pl_num(d)) >= 2, "why": "fewer than two varying numeric columns", }, "Box plot by group": { "desc": ("box plots of one numeric column split by a low-cardinality " "categorical column"), "needs": lambda d: bool(_pl_num(d)) and bool(_pl_cat(d, PLOT_MAX_GROUP_LEVELS)), "why": "needs both a numeric column and a low-cardinality categorical one", }, "Missingness matrix": { "desc": ("a missingness matrix drawn with " "plt.imshow(df.isna().values, aspect='auto', cmap='gray_r'), " "with the column names as x tick labels"), "needs": lambda d: bool(d.isna().any().any()), "why": "no missing values anywhere", }, "Correlation Heatmap": { "desc": "one correlation heatmap of the numeric columns", "needs": lambda d: len(_pl_num(d)) >= 2, "why": "fewer than two varying numeric columns", }, "Pie Plot": { "desc": ("a pie chart of the share of each level in one categorical " "column"), "needs": lambda d: bool(_pl_cat(d, PLOT_MAX_PIE_LEVELS)), "why": f"no categorical column with at most {PLOT_MAX_PIE_LEVELS} levels", }, "Area Plot": { "desc": ("an area plot of one numeric column against the datetime " "column, sorted by that datetime"), "needs": lambda d: bool(_pl_datetime(d)) and bool(_pl_num(d)), "why": ("no datetime column - an area plot over row order shows the " "sort order, not a trend"), }, } # Single source of truth for the UI and for every default call site. ALL_PLOTS = list(PLOT_MENU) # Four by default. CODE_MAX_TOKENS is 500 and each plot block costs roughly # 60-90 tokens, so asking for more risks the script being cut off mid-figure. # The rest stay available in ALL_PLOTS. DEFAULT_PLOTS = ["Histogram", "Bar Graph", "Correlation Heatmap", "Missingness matrix"] def build_code_prompt(context: str, instruction: str, selected_plots: list | None = None, df: pd.DataFrame | None = None) -> str: """Ask only for plots this dataset can actually support. Passing df is what makes the gating real: without it every plot type is requested and the model invents a datetime axis or a 40-wedge pie chart. """ plot_req = "" if selected_plots: wants, skipped = [], [] for p in selected_plots: spec = PLOT_MENU.get(p) if spec is None: continue if df is not None and not spec["needs"](df): skipped.append(f"{p} ({spec['why']})") continue wants.append(spec["desc"]) if skipped: print(f"[agent] plot types skipped: {'; '.join(skipped)}", flush=True) if wants: plot_req = (f"\nCreate only these plot types: {'; '.join(wants)}. " "Do not create any other plots.") return f"{context}\n\nTask: {instruction}{plot_req}" def extract_code(text: str) -> str: """Pull python out of markdown fences; concatenate if the model emitted several.""" blocks = re.findall(r"```(?:python|py)?\s*\n(.*?)```", text, flags=re.DOTALL) return "\n\n".join(b.strip() for b in blocks) if blocks else text.strip() # =========================================================================== # # Execution sandbox # =========================================================================== # BANNED = re.compile(r"\b(pd\.read_\w+|load_dataset|sns\.load_dataset)\s*\(") CORR_FIX = re.compile(r"\.corr\(\s*\)") MAX_ANNOT_COLS = 12 MAX_TICKS = 20 MAX_CAT_LEVELS = 20 MAX_PLOT_LEVELS = 30 # bars in any single chart. A 5,000-category bar # chart costs ~3.5s to build and is unreadable; # a loop over eight such columns exhausts the # whole execution budget before anything is saved. MAX_PAIRPLOT_COLS = 5 # a pair grid is quadratic in columns MAX_PAIRPLOT_ROWS = 2000 # ... and linear in rows, per panel WIDE_SPAN_RATIO = 50 # if one series' range is >50x another's, they do # not belong on a shared axis - Price (to 37M) # overlaid on Number_Beds (1-7) renders as a blank # canvas with two invisible spikes at zero MAX_FIG_INCHES = 40 # matplotlib hard-fails above 2^16 px per side; # 40in x 110dpi = 4,400px, comfortably under PLOT_SAMPLE_ROWS = 5000 # plotting sees a sample; the report's stats do not CODE_TIMEOUT_S = 30 # a working EDA script on a 5,000-row sample finishes # in under 10s. Each second here costs two seconds # of worst case, because of the repair attempt. TIMEOUT_MARK = "TimeoutError:" SKIP_ATTR = "_edagent_skip" RANGE_ATTR = "_edagent_ranges" TITLE_ATTR = "_edagent_title" def sanitize(code: str) -> str: """Neutralize data reloads; patch the one error the model can't reliably fix.""" out = [] for line in code.splitlines(): if BANNED.search(line): out.append("# [removed by agent] " + line.strip()) else: out.append(CORR_FIX.sub(".corr(numeric_only=True)", line)) return "\n".join(out) def _short(s, n=18): s = str(s) return s if len(s) <= n else f"{s[:n//2 - 1]}…{s[-(n//2 - 1):]}" def _tick(v, _=None): """Plain numbers with thousands separators - never 1e7.""" return f"{v:,.0f}" if abs(v) >= 1000 else f"{v:g}" def _tidy(fig): """Make any figure legible: thin dense ticks, rotate, kill scientific notation.""" for ax in fig.get_axes(): if len(ax.get_xticklabels()) > MAX_TICKS: step = max(1, len(ax.get_xticks()) // MAX_TICKS) ax.set_xticks(ax.get_xticks()[::step]) for axis, scale in ((ax.xaxis, ax.get_xscale()), (ax.yaxis, ax.get_yscale())): if scale == "linear" and isinstance(axis.get_major_formatter(), ScalarFormatter): axis.set_major_formatter(FuncFormatter(_tick)) plt.setp(ax.get_xticklabels(), rotation=45, ha="right", fontsize=8) plt.setp(ax.get_yticklabels(), fontsize=8) def _timed(name, fn): """Announce every plotting call with its cost. These prints go through the redirected stdout, so they survive a timeout and land in the Execution output tab - which is the only way to see WHICH call was running when the budget ran out. """ def wrapper(*a, **kw): t0 = time.time() try: return fn(*a, **kw) finally: dt = time.time() - t0 if dt > 0.25: print(f"[plot] {name} took {dt:.1f}s") return wrapper def _install_plot_guards(): """Patch the real modules, so `import matplotlib.pyplot as plt` can't bypass us.""" real = {"savefig": plt.savefig, "heatmap": sns.heatmap, "histplot": sns.histplot, "boxplot": sns.boxplot, "violinplot": sns.violinplot, "countplot": sns.countplot, "pairplot": sns.pairplot, "pie": plt.pie, "hist": plt.hist} def _vars(a, kw): """Recover the (name, Series) pairs a seaborn call is plotting.""" d = kw.get("data", a[0] if a else None) out = [] for key in ("x", "y"): v = kw.get(key) if isinstance(v, str) and hasattr(d, "columns"): out.append((v, d[v])) elif isinstance(v, pd.Series): out.append((v.name, v)) if not out and isinstance(d, pd.Series): out.append((d.name, d)) return out def _counts_instead(name, s): """A box plot of labels is meaningless; show how often each level occurs.""" ax = plt.gca() s.value_counts().head(MAX_CAT_LEVELS).sort_values().plot(kind="barh", ax=ax) ax.set_xlabel("count") ax.set_ylabel(str(name or "")) setattr(plt.gcf(), TITLE_ATTR, f"Counts of {name}" if name else "Counts") return ax def _categorical_guard(a, kw): """Return an axes if we handled it, else None to fall through.""" pairs = _vars(a, kw) if not pairs or any(pd.api.types.is_numeric_dtype(s) for _, s in pairs): return None name, s = pairs[0] if s.nunique(dropna=True) > MAX_CAT_LEVELS: setattr(plt.gcf(), SKIP_ATTR, True) return plt.gca() return _counts_instead(name, s) def savefig(fname, *a, **kw): fig = plt.gcf() if getattr(fig, SKIP_ATTR, False): return None title = getattr(fig, TITLE_ATTR, None) if title and fig.get_axes(): fig.get_axes()[0].set_title(title) w, h = fig.get_size_inches() # Clamp BOTH ways. df.hist() over hundreds of columns asks for a canvas # matplotlib refuses to render, and that ValueError used to escape. fig.set_size_inches(min(max(w, 8), MAX_FIG_INCHES), min(max(h, 5), MAX_FIG_INCHES)) _tidy(fig) kw.setdefault("bbox_inches", "tight") kw.setdefault("dpi", 110) return real["savefig"](fname, *a, **kw) def heatmap(data, *a, **kw): n = getattr(data, "shape", (0, 0))[1] if n > MAX_ANNOT_COLS: kw["annot"] = False kw.setdefault("cmap", "coolwarm") side = min(max(7, 0.5 * n + 4), 20) plt.gcf().set_size_inches(side, side * 0.85) ax = real["heatmap"](data, *a, **kw) if hasattr(data, "columns"): fs = 8 if n <= 15 else 6 ax.set_xticklabels([_short(c, 24) for c in data.columns], rotation=45, ha="right", fontsize=fs) ax.set_yticklabels([_short(c, 24) for c in data.index], rotation=0, fontsize=fs) return ax def _numeric_series(a, kw): for _, s in _vars(a, kw): s = pd.to_numeric(s, errors="coerce").dropna() if len(s): return s return None def histplot(*a, **kw): # A histogram of a text column is a categorical bar chart in disguise: # sns.histplot(df["ticker"]) on 5,000 unique values costs ~2.9s and # says nothing. boxplot/violinplot/countplot already guarded this; # histplot did not. handled = _categorical_guard(a, kw) if handled is not None: return handled s = _numeric_series(a, kw) logged = False if (s is not None and len(s) > 20 and s.min() > 0 and s.max() / max(s.median(), 1e-9) > 50 and "log_scale" not in kw): kw["log_scale"] = (True, False) logged = True ax = real["histplot"](*a, **kw) if logged: ax.xaxis.set_major_locator(LogLocator(base=10, subs=(1.0, 2.0, 5.0), numticks=12)) ax.xaxis.set_major_formatter(FuncFormatter(_tick)) ax.xaxis.set_minor_formatter(NullFormatter()) ax.set_xlabel(f"{ax.get_xlabel()} (log scale)") return ax def boxplot(*a, **kw): handled = _categorical_guard(a, kw) return handled if handled is not None else real["boxplot"](*a, **kw) def violinplot(*a, **kw): handled = _categorical_guard(a, kw) return handled if handled is not None else real["violinplot"](*a, **kw) def countplot(*a, **kw): pairs = _vars(a, kw) if pairs and pairs[0][1].nunique(dropna=True) > MAX_CAT_LEVELS: setattr(plt.gcf(), SKIP_ATTR, True) return plt.gca() return real["countplot"](*a, **kw) def pairplot(*a, **kw): """A pair grid is O(cols^2) panels x rows points. Unbounded, it hangs.""" d = kw.get("data", a[0] if a else None) if isinstance(d, pd.DataFrame): num = d.select_dtypes(include="number") keep = list(num.columns)[:MAX_PAIRPLOT_COLS] d2 = num[keep] if keep else num if len(d2) > MAX_PAIRPLOT_ROWS: d2 = d2.sample(MAX_PAIRPLOT_ROWS, random_state=0) if "data" in kw: kw["data"] = d2 return real["pairplot"](*a, **kw) return real["pairplot"](d2, *a[1:], **kw) return real["pairplot"](*a, **kw) def _series_ranges(x): """(min, max) per series in a hist argument, whatever shape it arrives in. matplotlib accepts three shapes and they mean different things: a list of arrays is one dataset per element, a 2-D array is one dataset per COLUMN, and anything else is a single dataset. Treating a list as an array transposes it and reports one "series" per row. """ try: if isinstance(x, (list, tuple)) and x and hasattr(x[0], "__len__"): parts = [np.asarray(p, dtype=float).ravel() for p in x] elif hasattr(x, "columns"): parts = [x[c].to_numpy(dtype=float) for c in x.columns] else: arr = np.asarray(x, dtype=float) parts = ([arr] if arr.ndim == 1 else [arr[:, i] for i in range(arr.shape[1])]) except Exception: return [] out = [] for c in parts: c = c[np.isfinite(c)] if len(c): out.append((float(c.min()), float(c.max()))) return out real["axes_hist"] = Axes.hist def axes_hist(self, x, *a, **kw): """Catch the overlay wherever it comes from. Guarding plt.hist only covers ONE call site. The common pattern is a loop - `for c in num_cols: plt.hist(df[c], label=c)` - where every call passes a single series and looks harmless on its own, yet they all land on the same axes. Accumulating the ranges seen by THIS axes is the only way to notice that Price (to 37,000,000) and Number_Beds (1-7) are now sharing one linear x-axis. """ out = real["axes_hist"](self, x, *a, **kw) seen = list(getattr(self, RANGE_ATTR, [])) + _series_ranges(x) setattr(self, RANGE_ATTR, seen) spans = [hi - lo for lo, hi in seen if hi > lo] if (len(spans) > 1 and min(spans) > 0 and max(spans) / min(spans) > WIDE_SPAN_RATIO and all(lo > 0 for lo, _ in seen) and self.get_xscale() == "linear"): self.set_xscale("log") self.xaxis.set_major_locator(LogLocator(base=10, subs=(1.0, 2.0, 5.0), numticks=12)) self.xaxis.set_major_formatter(FuncFormatter(_tick)) self.xaxis.set_minor_formatter(NullFormatter()) if "log scale" not in self.get_xlabel(): self.set_xlabel(f"{self.get_xlabel() or 'Value'} (log scale)") print(f"[plot] {len(seen)} series on one axis spanning " f"{max(spans)/min(spans):,.0f}x - switched to a log x-scale " f"so the small ones are visible") return out def hist(x, *a, **kw): """plt.hist([price, beds, baths]) shares ONE set of bin edges across every series. With Price running to 37,000,000 and Number_Beds to 7, all of beds and baths fall inside the first bin and the chart is blank. This cannot be detected after drawing - the bars all sit on the same grid - so it has to be caught here, while the raw series are still visible. Log-spaced bins plus a log axis give each series its own region of the width. """ try: cols = ([x[c] for c in x.columns] if hasattr(x, "columns") else (list(x) if isinstance(x, (list, tuple)) and len(x) > 1 and hasattr(x[0], "__len__") else None)) except Exception: cols = None if cols and len(cols) > 1: series = [pd.to_numeric(pd.Series(np.asarray(c).ravel()), errors="coerce").dropna() for c in cols] series = [c for c in series if len(c)] spans = [float(c.max() - c.min()) for c in series] good = [v for v in spans if v > 0] if (len(good) > 1 and max(good) / min(good) > WIDE_SPAN_RATIO and all(float(c.min()) > 0 for c in series)): lo = min(float(c.min()) for c in series) hi = max(float(c.max()) for c in series) nbins = kw.get("bins") nbins = nbins if isinstance(nbins, int) else 40 kw["bins"] = np.logspace(np.log10(lo), np.log10(hi), nbins + 1) print(f"[plot] plt.hist: series ranges differ by " f"{max(good)/min(good):,.0f}x - using log-spaced bins so " f"the small ones are not swallowed by the first bin") out = real["hist"](x, *a, **kw) ax = plt.gca() ax.set_xscale("log") ax.xaxis.set_major_locator(LogLocator(base=10, subs=(1.0, 2.0, 5.0), numticks=12)) ax.xaxis.set_major_formatter(FuncFormatter(_tick)) ax.xaxis.set_minor_formatter(NullFormatter()) ax.set_xlabel(f"{ax.get_xlabel() or 'Value'} (log scale)") return out return real["hist"](x, *a, **kw) def pie(x, *a, **kw): """Hundreds of labelled wedges is neither fast nor readable.""" try: n = len(x) except Exception: n = 0 if n > MAX_CAT_LEVELS: setattr(plt.gcf(), SKIP_ATTR, True) return ([], [], []) if kw.get("autopct") else ([], []) return real["pie"](x, *a, **kw) real["plot_accessor"] = pd.plotting.PlotAccessor.__call__ def plot_accessor(self, *a, **kw): """The one bar path the seaborn guards never saw. df["ticker"].value_counts().plot(kind="bar") goes nowhere near seaborn or plt.pie - it builds 5,000 Rectangle patches directly. Capping at the accessor rather than at Axes.bar keeps histograms and line/scatter plots exactly as written; pandas passes positional indices to Axes.bar, so a guard down there cannot tell a bar chart from a histogram. """ kind = kw.get("kind") or (a[0] if a and isinstance(a[0], str) else "line") obj = getattr(self, "_parent", None) # df[["Price", "Number_Beds"]].plot(kind="hist") puts every column on # ONE shared axis. When the ranges differ by orders of magnitude the # small ones vanish and the chart shows nothing. Give each column its # own panel instead - same information, actually readable. if (kind == "hist" and isinstance(obj, pd.DataFrame) and not kw.get("subplots")): num = obj.select_dtypes(include="number") spans = (num.max() - num.min()).replace(0, np.nan).dropna() if len(spans) > 1 and spans.max() / spans.min() > WIDE_SPAN_RATIO: print(f"[plot] .plot(kind='hist'): {len(spans)} columns whose " f"ranges differ by {spans.max()/spans.min():,.0f}x - " f"drawing one panel per column instead of one shared axis") kw["subplots"] = True kw.setdefault("layout", (len(spans), 1)) kw.setdefault("sharex", False) kw.setdefault("figsize", (8, 2.8 * len(spans))) kw.setdefault("legend", False) return real["plot_accessor"](num[list(spans.index)].plot, *a, **kw) if (kind in ("bar", "barh", "pie") and obj is not None and len(obj) > MAX_PLOT_LEVELS): print(f"[plot] .plot(kind='{kind}'): {len(obj)} categories capped " f"to the {MAX_PLOT_LEVELS} largest") obj = (obj.astype(float).nlargest(MAX_PLOT_LEVELS) if isinstance(obj, pd.Series) else obj.head(MAX_PLOT_LEVELS)) return real["plot_accessor"](obj.plot, *a, **kw) return real["plot_accessor"](self, *a, **kw) pd.plotting.PlotAccessor.__call__ = plot_accessor Axes.hist = axes_hist plt.savefig, plt.pie, plt.hist = _timed("savefig", savefig), pie, hist sns.heatmap = _timed("heatmap", heatmap) sns.histplot = _timed("histplot", histplot) sns.boxplot = _timed("boxplot", boxplot) sns.violinplot = _timed("violinplot", violinplot) sns.countplot = _timed("countplot", countplot) sns.pairplot = _timed("pairplot", pairplot) return real def _restore_plot_guards(real): pd.plotting.PlotAccessor.__call__ = real["plot_accessor"] Axes.hist = real["axes_hist"] plt.savefig, plt.pie, plt.hist = real["savefig"], real["pie"], real["hist"] sns.heatmap, sns.histplot = real["heatmap"], real["histplot"] sns.boxplot, sns.violinplot = real["boxplot"], real["violinplot"] sns.countplot, sns.pairplot = real["countplot"], real["pairplot"] def _exec_with_timeout(code, ns, seconds): """Run generated code with a wall clock. Python cannot kill a running thread, so on timeout the worker is abandoned rather than stopped - this unblocks the UI, it does not free the CPU. That is exactly why the repair loop must NOT retry a timeout: a second hanging thread stacks on top of the first and both keep contending for the GIL. """ pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) try: fut = pool.submit(exec, code, ns) try: fut.result(timeout=seconds) return None except concurrent.futures.TimeoutError: return (f"{TIMEOUT_MARK} the generated code was still running after " f"{seconds}s and was abandoned. The worker thread cannot be " f"killed and may keep consuming CPU until the runtime is " f"restarted.\n") except Exception: return traceback.format_exc(limit=3) finally: pool.shutdown(wait=False) def run_code(code: str, df: pd.DataFrame) -> dict: """Execute code with df in scope. Return stdout, error traceback, new plot paths. The sandbox gets a SAMPLE of large frames. Plots of a 5,000-row sample are visually identical and finish in seconds; every number in the report is computed separately, on the full frame, by the synthesis layer. """ before = set(glob.glob(f"{PLOTS_DIR}/*.png")) plot_df, note = df, "" if len(df) > PLOT_SAMPLE_ROWS: plot_df = df.sample(PLOT_SAMPLE_ROWS, random_state=0) note = (f"[agent] plotting on a random sample of {PLOT_SAMPLE_ROWS:,} of " f"{len(df):,} rows; report statistics use every row.\n") ns = {"df": plot_df.copy(), "pd": pd, "np": np, "plt": plt, "sns": sns} real = _install_plot_guards() buf = io.StringIO() err, timed_out, skipped = None, False, [] try: with contextlib.redirect_stdout(buf): err = _exec_with_timeout(code, ns, CODE_TIMEOUT_S) timed_out = bool(err) and err.startswith(TIMEOUT_MARK) finally: # After a timeout the abandoned thread is STILL drawing. Harvesting its # figures from this thread is unsafe (matplotlib is not thread-safe) and # can block for minutes - a second, UNBOUNDED stall hiding behind the # bounded one. Take whatever already reached disk and leave the rest. # Consequence: those figures leak for the rest of the session. Leaking # memory is recoverable; deadlocking the UI is not. if not timed_out: for i, num in enumerate(plt.get_fignums(), start=1): # One unrenderable figure must not take the whole request down. # This loop is in a finally block, so anything it raises escapes # run_code, escapes generate_and_run, and reaches the user as a # bare "Error" with no message. try: fig = plt.figure(num) if fig.get_axes() and not getattr(fig, SKIP_ATTR, False): w, h = fig.get_size_inches() fig.set_size_inches(min(w, MAX_FIG_INCHES), min(h, MAX_FIG_INCHES)) fig.savefig(f"{PLOTS_DIR}/plot_{i}.png") except Exception as e: skipped.append(f"{type(e).__name__}: {str(e)[:120]}") plt.close("all") _restore_plot_guards(real) if skipped: note += "[agent] %d figure(s) could not be saved: %s\n" % ( len(skipped), "; ".join(skipped)) return {"ok": err is None, "stdout": note + buf.getvalue(), "error": err, "plots": sorted(set(glob.glob(f"{PLOTS_DIR}/*.png")) - before)} # =========================================================================== # # Repair loop # =========================================================================== # FIX_SYSTEM = """You fix broken Python data-analysis code. Rules: - A pandas DataFrame named `df` is already loaded. Never load or create data. - Return the complete corrected script, not just the changed lines. - Reply with exactly one ```python code block and no other text.""" CODE_MAX_TOKENS = 500 # observed scripts are ~500 tokens; 1200 only bought rambling def format_error(err: str, code: str) -> str: """Turn a harness traceback into a signal about the model's own code. exec'd code has no source file, so Python prints the failing line number with no source text. We reconstruct it from the code we sent. """ last = err.strip().splitlines()[-1] hits = re.findall(r'File "", line (\d+)', err) if not hits: return last ln, src = int(hits[-1]), code.splitlines() line = src[ln - 1].strip() if 0 < ln <= len(src) else "" return f"Line {ln}: {line}\n{last}" def _timed_out(res) -> bool: return bool(res["error"]) and res["error"].lstrip().startswith(TIMEOUT_MARK) def generate_and_run_stream(context: str, instruction: str, df: pd.DataFrame, retries: int = 1, selected_plots: list | None = None): """Generator. Yields ('status', text) as each phase starts, then ('result', dict) once. This stage used to sit between two Gradio yields, so the browser showed one frozen message for its whole duration and the only timings were printed in the Colab cell. Streaming the phases means the elapsed seconds are visible where the person actually is. Worst case is bounded by construction: generate (<=CODE_MAX_TOKENS) + execute (<=CODE_TIMEOUT_S) + repair (<=CODE_MAX_TOKENS) + execute (<=CODE_TIMEOUT_S) and a timeout skips the repair entirely. """ shutil.rmtree(PLOTS_DIR, ignore_errors=True) os.makedirs(PLOTS_DIR, exist_ok=True) t_start = time.time() def say(msg): print(f"[agent] {msg}", flush=True) return ("status", f"{msg} · {time.time() - t_start:.0f}s elapsed") yield say(f"generating plot code (up to {CODE_MAX_TOKENS} tokens, " f"~20s on a T4) …") t0 = time.time() raw = generate(CODE_SYSTEM, build_code_prompt(context, instruction, selected_plots, df), max_new_tokens=CODE_MAX_TOKENS, stop_fences=2) t_gen = time.time() - t0 code = sanitize(extract_code(raw)) yield say(f"generated in {t_gen:.0f}s. Running it (hard limit " f"{CODE_TIMEOUT_S}s) …") t0 = time.time() res = run_code(code, df) t_exec = time.time() - t0 print(f"[attempt 1] ok={res['ok']} generate={t_gen:.1f}s execute={t_exec:.1f}s", flush=True) for attempt in range(retries): if res["ok"]: break if _timed_out(res): yield say(f"the code was still running after {CODE_TIMEOUT_S}s and was " f"abandoned. NOT repairing - the corrected version would " f"hang on the same construct.") break yield say(f"attempt 1 failed after {t_exec:.0f}s " f"({format_error(res['error'], code).splitlines()[-1][:90]}). " f"Repairing …") fix_prompt = (f"{context}\n\nThis code failed:\n```python\n{code}\n```\n\n" f"Error:\n{format_error(res['error'], code)}\n\n" f"Return the corrected script.") t0 = time.time() code = sanitize(extract_code( generate(FIX_SYSTEM, fix_prompt, max_new_tokens=CODE_MAX_TOKENS, stop_fences=2))) t_gen = time.time() - t0 yield say(f"repair generated in {t_gen:.0f}s. Running it …") t0 = time.time() res = run_code(code, df) t_exec = time.time() - t0 print(f"[attempt {attempt + 2}] ok={res['ok']} generate={t_gen:.1f}s " f"execute={t_exec:.1f}s", flush=True) res["code"] = code res["plots"] = sorted(glob.glob(f"{PLOTS_DIR}/*.png")) res["seconds"] = time.time() - t_start print(f"[agent] plot stage finished in {res['seconds']:.1f}s", flush=True) yield ("result", res) def generate_and_run(context: str, instruction: str, df: pd.DataFrame, retries: int = 1, selected_plots: list | None = None) -> dict: """Blocking wrapper for the notebook path. Drains the stream, returns the result.""" res = None for kind, payload in generate_and_run_stream(context, instruction, df, retries, selected_plots): if kind == "result": res = payload return res # =========================================================================== # # Synthesis layer - EDA structure per Lecture 02 # =========================================================================== # REDUNDANT_R = 0.99 # at/above this an association is a duplicate encoding MAX_GROUP_BLOCKS = 10 IQR_K = 1.5 # course-standard outlier fence OUTLIER_FLAG_PCT = 5.0 # flag a column when >5% of rows are IQR outliers DROP_MISSING_PCT = 70.0 # drop a column when >70% missing ... unless the # missingness itself carries signal (see below) ONEHOT_LEVELS = 2 # one-hot ONLY binary columns MAX_ASSOC_COLS = 25 # bound the O(n^2) association sweep MAX_ASSOC_LEVELS = 20 # above this a column is not a usable category FREETEXT_UNIQ_RATIO = 0.9 # nearly unique per row ... FREETEXT_MIN_LEN = 15 # ... and long ... FREETEXT_MIN_TOKENS = 1.0 # ... or multi-word => free text, not an ID MAX_TABLE_ROWS = 25 # a 470-column dataset would otherwise print a # 400-row distribution table nobody reads MAX_CHECKLIST_ROWS = 15 # ... per action group CO_MISSING_R = 0.5 # report column pairs whose missingness co-occurs MISSING_SIGNAL_STD = 0.2 # |gap| in target, in target standard deviations, # above which missingness is itself a predictor TARGET_MAX_LEVELS = 10 # a target with more levels than this is not a # classification label we can summarise simply MAX_TARGET_BREAKDOWNS = 3 # columns to break the target down by MAX_HYPOTHESES = 4 TARGET_NAME_HINTS = ("survived", "target", "label", "class", "outcome") ACTION_ICON = {"DROP": "🗑️", "IMPUTE": "🔄", "ENCODE": "🔢", "OUTLIERS": "⚠️", "FEATURE": "✍️"} def fmt_num(v): """Plain, comma-separated numbers. Never 1.02011e+06.""" if pd.isna(v): return "" v = float(v) if v.is_integer(): return f"{int(v):,}" if abs(v) >= 1000: return f"{v:,.0f}" return f"{v:,.2f}" def md_table(rows, n: int = MAX_TABLE_ROWS, drop=(), **kw) -> str: """Render at most n rows and say how many were left out.""" if not len(rows): return "" keep = [{k: v for k, v in r.items() if k not in drop} for r in rows[:n]] tbl = pd.DataFrame(keep).to_markdown(index=False, disable_numparse=True, **kw) if len(rows) > n: tbl += f"\n\n_+{len(rows) - n} more rows omitted._" return tbl def schema_only(df: pd.DataFrame, name: str) -> str: """Context minus sample rows, so stray values can't be copied into prose.""" return (f"Dataset: {name}\nShape: {df.shape[0]} rows x {df.shape[1]} columns\n" "Columns: " + ", ".join(f"{c} ({df[c].dtype})" for c in df.columns)) # --------------------------------------------------------------------------- # # Target variable # --------------------------------------------------------------------------- # def infer_target(df: pd.DataFrame): """Best guess at the column being predicted, or None. Name first - a column called `has_survived` is the target however the rest of the table looks. Otherwise the LAST column with a small number of levels, which is where labels conventionally sit. None if nothing qualifies, and every downstream caller must tolerate that. """ def usable(c): return 2 <= df[c].nunique(dropna=True) <= TARGET_MAX_LEVELS for c in df.columns: if any(h in str(c).lower() for h in TARGET_NAME_HINTS) and usable(c): return c for c in reversed(list(df.columns)): if usable(c): return c return None def target_series(df: pd.DataFrame, target): """(numeric series, label) for the target, or (None, None). Everything downstream averages the target, so a non-numeric one has to become a number first. A two-level label becomes the share of its minority level - which is the interesting rate, and is named in the label so no reader has to guess which class the number refers to. """ if target is None or target not in df.columns: return None, None y = df[target] if pd.api.types.is_numeric_dtype(y) and y.nunique(dropna=True) > 1: return pd.to_numeric(y, errors="coerce"), f"mean {target}" levels = y.value_counts(dropna=True) if len(levels) == 2: pos = levels.index[-1] # the minority level num = (y == pos).astype(float) return num.where(y.notna()), f"share of {target} = {pos}" return None, None def target_balance(df: pd.DataFrame, target): """Class balance as counts and percentages.""" if target is None or target not in df.columns: return [] vc = df[target].value_counts(dropna=False) n = int(vc.sum()) return [{"level": ("(missing)" if pd.isna(k) else str(k)), "n": fmt_num(v), "share %": round(100 * v / max(n, 1), 1)} for k, v in vc.items()] def target_predictors(checks: dict, target): """Pull the target's rows out of the association sweep already computed. Deliberately a filter, not a second sweep: association_pairs is O(cols^2) with a chi-square per pair, and running it twice for the same numbers was a bug fixed earlier in this project. """ if target is None: return [] real, _ = checks.get("assoc") or ([], []) out = [] for r in real: if target in (r.get("a"), r.get("b")): other = r["b"] if r.get("a") == target else r["a"] out.append({"predictor": other, "type": r["type"], "strength": r["strength"], "direction": r["direction"]}) return out def target_rate_by_level(df: pd.DataFrame, target, checks: dict, max_cols: int = MAX_TARGET_BREAKDOWNS): """Target rate per level of the strongest categorical predictors, with n. The n matters: a level with three rows and a 100% rate is noise, and the table has to make that visible rather than letting the rate stand alone. """ y, label = target_series(df, target) if y is None: return [], None cats = [r["predictor"] for r in target_predictors(checks, target) if r["type"] in ("cat-cat", "cat-num") and r["predictor"] in df.columns and 2 <= df[r["predictor"]].nunique(dropna=True) <= MAX_ASSOC_LEVELS] seen, rows = set(), [] for c in cats: if c in seen or len(seen) >= max_cols: continue seen.add(c) g = pd.DataFrame({"lvl": df[c], "y": y}).dropna(subset=["lvl"]) agg = g.groupby("lvl")["y"].agg(["mean", "count"]) for lvl, r in agg.sort_values("mean", ascending=False).iterrows(): rows.append({"column": c, "level": str(lvl), label: round(float(r["mean"]), 3), "n": fmt_num(r["count"])}) return rows, label # --------------------------------------------------------------------------- # # Missingness structure # --------------------------------------------------------------------------- # def co_missingness(df: pd.DataFrame, r_min: float = CO_MISSING_R): """Column pairs that go missing together. Two columns missing on the same rows is one mechanism, not two problems - it usually means a whole block of the record was never captured, and imputing them independently pretends they are unrelated. """ m = df.isna() cols = [c for c in m.columns if 0 < m[c].sum() < len(m)][:MAX_ASSOC_COLS] if len(cols) < 2: return [] c = m[cols].astype(float).corr() rows = [] for i, a in enumerate(cols): for b in cols[i + 1:]: r = c.loc[a, b] if pd.notna(r) and abs(float(r)) > r_min: rows.append({"pair": f"{a} ~ {b}", "correlation of missingness": round(float(r), 3), "both missing": fmt_num(int((m[a] & m[b]).sum()))}) rows.sort(key=lambda x: abs(x["correlation of missingness"]), reverse=True) return rows def missing_vs_target(df: pd.DataFrame, target): """Is a column's MISSINGNESS predictive of the target? This is the check that stops the agent recommending you delete Titanic's `cabin`. 77% of it is missing, so the blind >70% rule says drop - but whether a cabin was recorded at all separates survivors from non-survivors almost as well as any recorded feature. The absence is the signal. """ y, label = target_series(df, target) if y is None: return [], None sd = float(y.std()) rows = [] for c in df.columns: if c == target: continue miss = df[c].isna() n_miss, n_have = int(miss.sum()), int((~miss).sum()) if n_miss < 5 or n_have < 5: continue a = float(y[miss].mean()) b = float(y[~miss].mean()) if pd.isna(a) or pd.isna(b): continue gap = a - b std_gap = abs(gap) / sd if sd and sd > 0 else 0.0 rows.append({"column": c, f"{label} when missing": round(a, 3), f"{label} when present": round(b, 3), "gap": round(gap, 3), "n missing": fmt_num(n_miss), "n present": fmt_num(n_have), "_std_gap": round(std_gap, 3), "_predictive": bool(std_gap >= MISSING_SIGNAL_STD)}) rows.sort(key=lambda r: abs(r["gap"]), reverse=True) return rows, label # --------------------------------------------------------------------------- # # Column roles: a nearly-unique text column is usually free text (names, # descriptions), not an identifier. Dropping it throws away real signal. # --------------------------------------------------------------------------- # def detect_free_text(df: pd.DataFrame) -> list: out = [] for c in df.select_dtypes(include=["object", "category"]).columns: s = df[c].dropna().astype(str) if len(s) < 5: continue uniq_ratio = s.nunique() / len(s) mean_len = s.str.len().mean() mean_tok = s.str.split().str.len().mean() if uniq_ratio > FREETEXT_UNIQ_RATIO and (mean_len > FREETEXT_MIN_LEN or mean_tok > FREETEXT_MIN_TOKENS): out.append(c) return out # --------------------------------------------------------------------------- # # Mixed-type associations. Pearson r, Cramer's V and the correlation ratio all # live on 0..1 in absolute value, so a single ranking is meaningful. # --------------------------------------------------------------------------- # def cramers_v(a, b): """Categorical <-> categorical. Uncorrected; fine for ranking, not inference.""" ct = pd.crosstab(a, b) if ct.shape[0] < 2 or ct.shape[1] < 2: return np.nan chi2 = stats.chi2_contingency(ct, correction=False)[0] n = ct.to_numpy().sum() if n <= 0: return np.nan phi2 = chi2 / n r, k = ct.shape return float(np.sqrt(phi2 / max(min(k - 1, r - 1), 1))) def correlation_ratio(cats, vals): """Categorical <-> numeric: eta = sqrt(between-group var / total var).""" d = pd.DataFrame({"c": cats, "v": pd.to_numeric(vals, errors="coerce")}).dropna() if d["c"].nunique() < 2 or len(d) < 3: return np.nan grand = d["v"].mean() num = sum(len(g) * (g["v"].mean() - grand) ** 2 for _, g in d.groupby("c")) den = float(((d["v"] - grand) ** 2).sum()) return float(np.sqrt(num / den)) if den > 0 else np.nan def association_pairs(df: pd.DataFrame, free_text=(), id_like=()): """Every pair of usable columns, ranked by absolute effect size. Each row keeps the two column names separately as "a" and "b" as well as the display string, so the target section can filter by identity instead of substring-matching "pair" - which would match `age` inside `percentage`. This sweep is O(cols^2) with a chi-square or a groupby per pair. It runs once, in run_advanced_eda_checks; every other caller reads checks["assoc"]. """ skip = set(free_text) | set(id_like) num = [c for c in df.select_dtypes(include="number").columns if c not in skip and df[c].nunique(dropna=True) > 1][:MAX_ASSOC_COLS] cat = [c for c in df.columns if c not in skip and c not in num and 2 <= df[c].nunique(dropna=True) <= MAX_ASSOC_LEVELS][:MAX_ASSOC_COLS] rows = [] for i, a in enumerate(num): # numeric <-> numeric for b in num[i + 1:]: d = df[[a, b]].dropna() if len(d) < 3: continue r = d[a].corr(d[b]) if pd.isna(r): continue rows.append({"a": a, "b": b, "pair": f"{a} ~ {b}", "type": "num-num", "measure": "Pearson r", "strength": round(abs(float(r)), 3), "direction": "positive" if r > 0 else "negative"}) for i, a in enumerate(cat): # categorical <-> categorical for b in cat[i + 1:]: v = cramers_v(df[a], df[b]) if pd.isna(v): continue rows.append({"a": a, "b": b, "pair": f"{a} ~ {b}", "type": "cat-cat", "measure": "Cramer's V", "strength": round(float(v), 3), "direction": "n/a"}) for a in cat: # categorical <-> numeric for b in num: e = correlation_ratio(df[a], df[b]) if pd.isna(e): continue rows.append({"a": a, "b": b, "pair": f"{a} ~ {b}", "type": "cat-num", "measure": "corr. ratio", "strength": round(float(e), 3), "direction": "n/a"}) rows.sort(key=lambda r: r["strength"], reverse=True) return ([r for r in rows if r["strength"] < REDUNDANT_R], [r for r in rows if r["strength"] >= REDUNDANT_R]) # --------------------------------------------------------------------------- # # Distribution shape, not just describe() # --------------------------------------------------------------------------- # def shape_label(skew, exkurt): if pd.isna(skew): return "undetermined" if abs(skew) < 0.5: lab = "symmetric" elif abs(skew) < 1: lab = "moderately right-skewed" if skew > 0 else "moderately left-skewed" else: lab = "strongly right-skewed" if skew > 0 else "strongly left-skewed" if not pd.isna(exkurt) and exkurt > 3: lab += ", heavy-tailed" return lab def distribution_profile(df: pd.DataFrame) -> list: """Per numeric column: location, spread, shape. pandas.kurtosis is excess.""" rows = [] for c in df.select_dtypes(include="number").columns: s = df[c].dropna() if len(s) < 3 or s.nunique() <= 1: continue sk = float(s.skew()) ku = float(s.kurtosis()) if len(s) > 3 else np.nan med = float(s.median()) ratio = float(s.mean() / med) if med != 0 else np.nan rows.append({"column": c, "count": int(s.count()), "mean": fmt_num(s.mean()), "median": fmt_num(med), "std": fmt_num(s.std()), "min": fmt_num(s.min()), "max": fmt_num(s.max()), "skew": round(sk, 2), "kurtosis": ("" if pd.isna(ku) else round(ku, 2)), "mean/median": ("" if pd.isna(ratio) else round(ratio, 2)), "shape": shape_label(sk, ku)}) return rows def run_advanced_eda_checks(df: pd.DataFrame, target=None) -> dict: """Column roles, integrity findings, missingness structure, shapes, IQR outliers, target relationships and the checklist. All pandas/scipy. The model never computes any of this. target may be None. """ n_rows, n_cols = df.shape checks = {"target": target if (target in df.columns) else None} target = checks["target"] # ---- column roles and integrity findings ---- free_text = detect_free_text(df) dup_rows = int(df.duplicated().sum()) const_cols = [c for c in df.columns if df[c].nunique(dropna=False) <= 1] # unique per row, but neither a float measurement nor free text id_like = [c for c in df.columns if n_rows > 0 and df[c].nunique(dropna=True) == n_rows and not pd.api.types.is_float_dtype(df[c]) and c not in free_text] miss_pct = (df.isna().sum() / max(n_rows, 1) * 100) overall_missing = float(df.isna().sum().sum()) / max(n_rows * n_cols, 1) * 100 cols_with_missing = int((miss_pct > 0).sum()) checks["free_text"] = free_text checks["dup_rows"] = dup_rows checks["dup_pct"] = 100 * dup_rows / max(n_rows, 1) checks["const_cols"] = const_cols checks["id_like"] = id_like checks["overall_missing_pct"] = overall_missing checks["cols_with_missing"] = cols_with_missing # ---- missingness structure ---- checks["co_missing"] = co_missingness(df) mvt, mvt_label = missing_vs_target(df, target) checks["missing_vs_target"] = mvt checks["target_metric"] = mvt_label checks["predictive_missing"] = {r["column"] for r in mvt if r["_predictive"]} # ---- distribution shapes ---- checks["distribution"] = distribution_profile(df) # ---- outliers by the 1.5x IQR method, reported actionably ---- outliers = [] for c in df.select_dtypes(include="number").columns: s = df[c].dropna() if len(s) < 10 or s.nunique() <= 2: # too small / binary: fences meaningless continue q1, q3 = s.quantile(0.25), s.quantile(0.75) iqr = q3 - q1 if iqr == 0: continue lo, hi = q1 - IQR_K * iqr, q3 + IQR_K * iqr ex = s[(s < lo) | (s > hi)] cnt = int(len(ex)) if not cnt: continue pct = round(100 * cnt / len(s), 1) far = ex.reindex((ex - s.median()).abs().sort_values(ascending=False).index) examples = ", ".join(fmt_num(v) for v in far.head(3)) if pct > 10: rec = "large share - check whether these are a distinct sub-population" elif pct > OUTLIER_FLAG_PCT: rec = "inspect domain validity; consider winsorizing or filtering" else: rec = "few extremes - usually safe to keep" outliers.append({"column": c, "pct": pct, "count": cnt, "examples": examples, "recommendation": rec, "lower_fence": round(float(lo), 2), "upper_fence": round(float(hi), 2)}) outliers.sort(key=lambda r: r["pct"], reverse=True) checks["outliers"] = outliers # ---- categorical handling. One-hot ONLY for binary columns. ---- cats = [] for c in df.select_dtypes(include=["object", "category"]).columns: k = df[c].nunique(dropna=True) if c in free_text: rec = "free text - extract features (length, tokens, title) rather than dropping" elif k == ONEHOT_LEVELS: rec = "binary - one-hot encode with pd.get_dummies" elif k < ONEHOT_LEVELS: rec = "single level - carries no information, drop" elif k <= 10: rec = "more than 2 levels - keep as is or group rare levels" else: rec = "high cardinality - group rare levels or drop" cats.append({"column": c, "levels": k, "recommendation": rec}) checks["categoricals"] = cats # ---- associations: computed ONCE, read by every other caller ---- checks["assoc"] = association_pairs(df, free_text, id_like) # ---- target relationships (all no-ops when target is None) ---- checks["target_balance"] = target_balance(df, target) checks["target_predictors"] = target_predictors(checks, target) rates, rate_label = target_rate_by_level(df, target, checks) checks["target_rates"] = rates checks["target_rate_label"] = rate_label # ---- ML preprocessing checklist, (action, column, reason) ---- cl = [] for c in df.columns: if c in free_text or c == target: continue # free text is engineered, never dropped if miss_pct.get(c, 0) > DROP_MISSING_PCT: # The blind rule says drop. It is wrong whenever the missingness is # itself informative, which is exactly the Titanic `cabin` case, so # the recommendation is downgraded to an encoding instead. if c in checks["predictive_missing"]: row = next(r for r in mvt if r["column"] == c) cl.append(("ENCODE", c, f"{miss_pct[c]:.1f}% missing, but the missingness itself " f"predicts {target} (gap {row['gap']:+.3f}, " f"{row['_std_gap']} SD) - add `{c}_was_missing` and use " f"THAT as the feature; the column is too sparse to " f"impute meaningfully")) else: cl.append(("DROP", c, f"{miss_pct[c]:.1f}% missing (over {DROP_MISSING_PCT:.0f}%), " f"and the missingness does not predict the target")) for c in const_cols: cl.append(("DROP", c, "constant column - zero variance")) for c in id_like: cl.append(("DROP", c, "unique per row - an identifier, not a feature")) dropped_set = {c for a, c, _ in cl if a == "DROP"} encoded_missing = {c for a, c, _ in cl if a == "ENCODE"} for c in free_text: if c not in dropped_set: cl.append(("FEATURE", c, "free text - extract features (length, token count, title) " "rather than dropping")) for c in df.columns: # encoded_missing columns are deliberately excluded: a column kept for # its missingness indicator must not also be told to impute 78% of # itself with one mode value. The two recommendations contradict. if (c in dropped_set or c in encoded_missing or c in free_text or c == target or miss_pct.get(c, 0) == 0): continue if pd.api.types.is_numeric_dtype(df[c]): cl.append(("IMPUTE", c, f"{miss_pct[c]:.1f}% missing - fill with the median " f"({fmt_num(df[c].median())})")) else: mode = df[c].mode(dropna=True) mv = str(mode.iloc[0]) if len(mode) else "?" cl.append(("IMPUTE", c, f"{miss_pct[c]:.1f}% missing - fill with the mode ('{mv}')")) for row in cats: if (row["column"] not in dropped_set and row["column"] not in free_text and row["column"] != target and row["column"] not in encoded_missing and row["levels"] == ONEHOT_LEVELS): cl.append(("ENCODE", row["column"], "2 levels - one-hot encode with pd.get_dummies")) for row in outliers: if (row["pct"] > OUTLIER_FLAG_PCT and row["column"] not in dropped_set and row["column"] != target): cl.append(("OUTLIERS", row["column"], f"{row['pct']}% extreme values (e.g. {row['examples']}) - " f"{row['recommendation']}")) checks["checklist"] = cl return checks def integrity_findings(checks: dict) -> list: """Data quality as findings, not a score. A single number invited the wrong reading - a synthetic dataset with uniform columns and zero correlations scored 100/100. These lines say what was actually checked and what was actually found. """ out = [] if checks["dup_rows"]: out.append(f"⚠️ **{fmt_num(checks['dup_rows'])} duplicate rows** " f"({checks['dup_pct']:.1f}% of the table) — consider `df.drop_duplicates()`.") else: out.append("✅ No duplicate rows.") if checks["const_cols"]: out.append(f"⚠️ **{len(checks['const_cols'])} constant column(s)**: " f"`{'`, `'.join(checks['const_cols'])}` — zero variance, drop them.") else: out.append("✅ No constant columns.") if checks["id_like"]: out.append(f"⚠️ **{len(checks['id_like'])} identifier column(s)**: " f"`{'`, `'.join(checks['id_like'])}` — unique per row, not features.") else: out.append("✅ No identifier columns.") if checks["free_text"]: out.append(f"✍️ **{len(checks['free_text'])} free-text column(s)**: " f"`{'`, `'.join(checks['free_text'])}` — extract features " f"(length, tokens, title) rather than dropping.") if checks["overall_missing_pct"] > 0: out.append(f"⚠️ **{checks['overall_missing_pct']:.1f}% of all cells are missing**, " f"across {checks['cols_with_missing']} column(s).") else: out.append("✅ No missing values anywhere.") if checks.get("predictive_missing"): cols = "`, `".join(sorted(checks["predictive_missing"])) out.append(f"🔎 **Missingness carries signal** in `{cols}` — whether the " f"value was recorded predicts the target. Encode it, do not " f"drop it.") return out def format_checklist(cl: list) -> str: """A grouped, scannable bullet list. A dry table nobody reads is a wasted section.""" if not cl: return "_Nothing to do - the dataset is ML-ready as loaded._" out = [] for act in ("DROP", "IMPUTE", "ENCODE", "OUTLIERS", "FEATURE"): rows = [r for r in cl if r[0] == act] if not rows: continue # The action is already in the heading; repeating the icon and the # [TAG] on every bullet was noise that pushed the actual column name # and reason to the right. out += [f"**{ACTION_ICON[act]} {act} — {len(rows)} column(s)**", ""] out += [f"- `{col}`: {why}" for _, col, why in rows[:MAX_CHECKLIST_ROWS]] if len(rows) > MAX_CHECKLIST_ROWS: out += [f"- _...and {len(rows) - MAX_CHECKLIST_ROWS} more columns " f"needing {act}._"] out += [""] return "\n".join(out).strip() def hypothesis_candidates(df, checks: dict, facts: dict, target) -> list: """Hypotheses, computed - not invented. Tukey defined EDA as the hypothesis-generating phase, so the section belongs in the report. What does not belong is a language model inventing the hypotheses: every line here is assembled in Python from numbers already in `checks`, and the model is later asked only to phrase them. """ h = [] preds = checks.get("target_predictors") or [] if target and preds: p = preds[0] h.append(f"{p['predictor']} is the strongest available predictor of " f"{target} (effect size {p['strength']}); test whether it " f"survives controlling for the other top predictors") if checks.get("predictive_missing"): c = sorted(checks["predictive_missing"])[0] h.append(f"whether {c} was recorded at all predicts {target}; add a " f"{c}_was_missing indicator and test it against the imputed column") if checks.get("co_missing"): cm = checks["co_missing"][0] h.append(f"{cm['pair']} go missing together " f"(missingness correlation {cm['correlation of missingness']}); " f"they likely share one collection failure and should be " f"imputed jointly, not independently") dist = checks.get("distribution") or [] if dist: w = max(dist, key=lambda r: abs(r["skew"])) if abs(w["skew"]) >= 1: h.append(f"{w['column']} is {w['shape']} (skew {w['skew']}); a log " f"or rank transform should help any model that assumes " f"symmetry") outs = checks.get("outliers") or [] if outs and outs[0]["pct"] > OUTLIER_FLAG_PCT: o = outs[0] h.append(f"the {o['pct']}% of rows outside {o['column']}'s IQR fences may " f"be a distinct sub-population rather than errors; compare them " f"against the rest before removing anything") bal = checks.get("target_balance") or [] if len(bal) >= 2: top = max(bal, key=lambda r: r["share %"]) if top["share %"] >= 65: h.append(f"{target} is imbalanced ({top['level']} is {top['share %']}% " f"of rows); use stratified splits and judge models on " f"precision-recall rather than accuracy") _, dup = checks.get("assoc") or ([], []) if dup: h.append(f"{dup[0]['pair']} associate at {dup[0]['strength']}, which is " f"duplicate encoding rather than a finding; drop one before " f"modelling") return h[:MAX_HYPOTHESES] def make_tables(df: pd.DataFrame, checks: dict, max_card: int = 6) -> dict: """Every table rendered by pandas. The model never transcribes one.""" t, n_rows = {}, len(df) target = checks.get("target") # --- Data quality --- t["health"] = "\n".join(f"- {line}" for line in integrity_findings(checks)) if checks["dup_rows"]: t["duplicates"] = (f"**{fmt_num(checks['dup_rows'])} duplicate rows** " f"({checks['dup_pct']:.1f}% of the table) — consider " f"`df.drop_duplicates()`.") # --- Target analysis --- if target: t["target_balance"] = md_table(checks["target_balance"]) or "_No levels._" preds = checks["target_predictors"] t["target_predictors"] = ( md_table(preds, n=10) if preds else "_No column shows a measurable association with the target._") if checks["target_rates"]: t["target_rates"] = md_table(checks["target_rates"], n=MAX_TABLE_ROWS) t["target_rate_caption"] = ( f"{checks.get('target_rate_label') or 'Target rate'} by level, " f"with group size") # --- Missing values, duplicates and missingness structure --- n = df.isna().sum() n = n[n > 0].sort_values(ascending=False) if len(n): head = n.head(15) strategy = ["median" if pd.api.types.is_numeric_dtype(df[c]) else "mode" for c in head.index] tbl = pd.DataFrame({"missing": head.map(fmt_num), "pct": (100 * head / n_rows).round(1), "imputation": strategy}).to_markdown(disable_numparse=True) if len(n) > 15: tbl += f"\n\n_+{len(n) - 15} more columns with missing values._" t["missing"] = tbl else: t["missing"] = "_No missing values._" if checks["co_missing"]: t["co_missing"] = md_table(checks["co_missing"], n=10) if checks["missing_vs_target"]: rows = [r for r in checks["missing_vs_target"] if r["_predictive"]] if rows: t["missing_vs_target"] = md_table(rows, n=10, drop=("_std_gap", "_predictive")) # --- distribution shape, not just describe() --- if checks["distribution"]: # widest first: the columns with the most extreme shape are the ones # worth the reader's attention, not whichever happened to come first rows = sorted(checks["distribution"], key=lambda r: abs(r["skew"]), reverse=True) t["describe"] = md_table(rows) else: t["describe"] = "_No numeric columns with enough variation to profile._" # --- categorical handling --- if checks["categoricals"]: rows = [{"column": r["column"], "unique levels": r["levels"], "recommendation": r["recommendation"]} for r in checks["categoricals"]] t["encoding"] = md_table(rows) else: t["encoding"] = "_No categorical columns._" # --- group differences + mixed-type associations --- num = df.select_dtypes(include="number") bins = [c for c in num.columns if df[c].nunique() == 2] lowcard = [c for c in df.columns if df[c].nunique(dropna=True) <= max_card] # Count first, build second. The old comprehension materialised EVERY # groupby + to_markdown and then threw all but ten away - thousands of # wasted operations on a wide table. n_possible = sum(1 for c in lowcard for b in bins if b != c) blocks = [] for c in lowcard: for b in bins: if b == c: continue blocks.append(f"**mean `{b}` by `{c}`**\n\n" + df.groupby(c)[b].mean().round(3).to_frame().to_markdown()) if len(blocks) >= MAX_GROUP_BLOCKS: break if len(blocks) >= MAX_GROUP_BLOCKS: break if blocks: t["groups"] = "\n\n".join(blocks) if n_possible > len(blocks): t["groups"] += (f"\n\n_+{n_possible - len(blocks)} more group " "comparisons omitted._") else: t["groups"] = "_No low-cardinality groupings._" real, dup = checks.get("assoc") or association_pairs( df, checks["free_text"], checks["id_like"]) note = ("_Ranked by absolute effect size across mixed types: Pearson r for " "numeric pairs, Cramer's V for categorical pairs, and the correlation " "ratio for categorical-numeric pairs._\n\n") if real and real[0]["strength"] < 0.1: note += (f"_All associations are near zero (largest = " f"{real[0]['strength']:.3f}) - the columns appear mutually " f"independent, often a sign of synthetic data._\n\n") # "measure" is redundant with "type": num-num is always Pearson r, cat-cat # always Cramer's V, cat-num always the correlation ratio - and the note # above the table already says so. "a"/"b" exist only for filtering. t["corr"] = note + (md_table(real, n=10, drop=("measure", "a", "b")) if real else "_Too few usable columns._") if dup: t["redundant"] = ( f"_{len(dup)} column pairs associate at >= {REDUNDANT_R} - likely " "duplicate encodings rather than findings._\n\n" + md_table(dup, n=8, drop=("measure", "a", "b", "direction"))) # --- outliers, actionable rather than raw quartiles --- if checks["outliers"]: t["outliers"] = md_table( [{"column": r["column"], "outliers %": r["pct"], "example extreme values": r["examples"], "recommendation": r["recommendation"]} for r in checks["outliers"]]) else: t["outliers"] = f"_No values fall outside the {IQR_K}x IQR fences._" # --- the checklist --- t["checklist"] = format_checklist(checks["checklist"]) return t def key_facts(df: pd.DataFrame, checks: dict, max_card: int = 6) -> dict: """Every superlative computed in pandas, never looked up by the model.""" target = checks.get("target") issues = [] if checks["dup_rows"]: issues.append(f"{fmt_num(checks['dup_rows'])} duplicate rows") if checks["const_cols"]: issues.append(f"{len(checks['const_cols'])} constant columns") if checks["id_like"]: issues.append(f"{len(checks['id_like'])} identifier columns") if checks["overall_missing_pct"] > 0: issues.append(f"{checks['overall_missing_pct']:.1f}% of cells missing") f = {"health": ("the integrity checks found " + ", ".join(issues) + f", and {len(checks['checklist'])} preprocessing actions " "are recommended") if issues else ("the integrity checks found no duplicate rows, no constant " "columns, no identifier columns and no missing values")} n = df.isna().sum() n = n[n > 0] if len(n): c = n.idxmax() f["missing"] = (f"{c} has the most missing values: " f"{fmt_num(n.max())} ({100 * n.max() / len(df):.1f}%)") if checks["distribution"]: worst = max(checks["distribution"], key=lambda r: abs(r["skew"])) f["shape"] = (f"{worst['column']} is the most skewed numeric column " f"(skew {worst['skew']}), described as {worst['shape']}") if checks["outliers"]: r = checks["outliers"][0] f["outliers"] = (f"{r['column']} has the highest share of extreme values: " f"{r['pct']}% of its rows") real, _ = checks.get("assoc") or association_pairs( df, checks["free_text"], checks["id_like"]) if real: r = real[0] if r["strength"] < 0.1: f["corr"] = (f"no meaningful association exists between the columns: " f"the largest is {r['pair']} at {r['strength']}") else: direction = "" if r["direction"] == "n/a" else f", {r['direction']}" f["corr"] = (f"the strongest association is {r['pair']} " f"({r['measure']} = {r['strength']}{direction})") if target: bal = checks.get("target_balance") or [] preds = checks.get("target_predictors") or [] bits = [f"the target is {target}"] if bal: top = max(bal, key=lambda r: r["share %"]) bits.append(f"its most common level is {top['level']} at " f"{top['share %']}% of rows") if preds: bits.append(f"its strongest predictor is {preds[0]['predictor']} " f"(effect size {preds[0]['strength']})") f["target"] = ", and ".join(bits) return f def first_sentence(text: str) -> str: """First real sentence: skip heading-like fragments, stop at the first period.""" lines = [l.strip().lstrip("-*# ").strip() for l in text.splitlines() if l.strip()] lines = [l for l in lines if len(l.split()) >= 4] or lines s = lines[0] if lines else "" m = re.match(r"(.+?[.!?])(\s|$)", s) return m.group(1) if m else s[:220] def first_sentences(text: str, n: int = 3) -> str: """Keep the first n sentences; drop any list or heading scaffolding.""" lines = [l.strip().lstrip("-*# ").strip() for l in text.splitlines() if l.strip()] lines = [l for l in lines if len(l.split()) >= 4] or lines blob = " ".join(lines) parts = re.findall(r"[^.!?]+[.!?]", blob) return " ".join(p.strip() for p in parts[:n]) or blob[:400] SEMANTIC_SYSTEM = ( "You explain what a dataset is about, using only its name and column names.\n" "Rules:\n" "- Two or three short sentences of plain prose.\n" "- Never state a count, percentage or any other statistic.\n" "- No lists, no headings, no code." ) ONELINER_SYSTEM = ("Rewrite the given fact as one short sentence. Add nothing. " "No lists, no headings.") def audit_numbers(md: str, allowed_text: str) -> list: """Flag numerals in the prose absent from the evidence. Table rows and headings are skipped: heading numbers ("## 7.") are structure, not claims. Detects fabrication, not misinterpretation. """ allowed = set(re.findall(r"\d+\.?\d*", allowed_text.replace(",", ""))) prose = "\n".join(l for l in md.splitlines() if not l.strip().startswith(("|", "#"))) return sorted({x for x in re.findall(r"\d+\.?\d*", prose.replace(",", "")) if x not in allowed}) # --------------------------------------------------------------------------- # # Caches keyed by dataset id (and target, which changes every statistic below # it). Changing nothing but the instruction must not re-download the data, # recompute the statistics, or re-run the narrative calls. # --------------------------------------------------------------------------- # _DATA_CACHE: dict = {} # name -> (df, dropped, context) _STATS_CACHE: dict = {} # (name, target) -> (checks, tables, facts, hypotheses) _NARR_CACHE: dict = {} # (name, target) -> {section: sentence} def get_dataframe(name: str): """Load + normalize once per dataset per session.""" if name not in _DATA_CACHE: df = normalize(load_hf_dataframe(name)) df, dropped = drop_index_cols(df) if dropped: print(f"[agent] dropped index-like columns: {dropped}") _DATA_CACHE[name] = (df, dropped, build_context(df, name)) return _DATA_CACHE[name] def get_stats(name: str, df: pd.DataFrame, target=None): """All deterministic analysis, computed once per (dataset, target).""" k = (name, target) if k not in _STATS_CACHE: t0 = time.time() checks = run_advanced_eda_checks(df, target) tables = make_tables(df, checks) facts = key_facts(df, checks) hyps = hypothesis_candidates(df, checks, facts, checks.get("target")) # Put the raw candidates in the tables dict so their numbers reach # audit_numbers' allow-list. Nothing renders this key. tables["hypotheses_src"] = "\n".join(hyps) _STATS_CACHE[k] = (checks, tables, facts, hyps) print(f"[stats] computed in {time.time() - t0:.1f}s " f"(target={checks.get('target')})", flush=True) return _STATS_CACHE[k] def get_narrative(name: str, df: pd.DataFrame, facts: dict, hypotheses=(), target=None): """All narrative in TWO batched calls instead of one call per sentence. Short section one-liners and the hypothesis phrasings go in one batch (they are all one sentence); the longer dataset overview goes in the other, because a batch costs whatever its longest row costs. The model PHRASES the hypotheses. It never composes them - every one arrives already written by hypothesis_candidates() from computed numbers, and if a phrasing comes back empty the computed line is used verbatim. """ k = (name, target) if k in _NARR_CACHE: return _NARR_CACHE[k] t0 = time.time() keys = [x for x in ("missing", "shape", "corr", "outliers", "target") if x in facts] hypotheses = list(hypotheses) # batch 1: every one-sentence job at once shorts = generate_batch( [(ONELINER_SYSTEM, facts[x]) for x in keys] + [(ONELINER_SYSTEM, h) for h in hypotheses], max_new_tokens=70) nar = {x: first_sentence(txt) for x, txt in zip(keys, shorts[:len(keys)])} phrased = [] for raw, out in zip(hypotheses, shorts[len(keys):]): s = first_sentence(out) phrased.append(s if len(s.split()) >= 5 else raw) nar["hypotheses"] = ("\n".join(f"- {h}" for h in phrased) if phrased else "_No hypotheses could be derived from this dataset._") # batch 2: the dataset overview. Only the columns the model was SHOWN - # pasting all 470 column names of a wide dataset is a ~6,300-token prompt on # its own, over MAX_PROMPT_TOKENS, and _check_budget would raise. shown, _ = pick_context_cols(df) cols = ", ".join(map(str, shown)) longs = generate_batch( [(SEMANTIC_SYSTEM, f"Dataset name: {name}\nColumns: {cols}\n\n" "What kind of data is this, and what do the main columns appear to measure?")], max_new_tokens=240) nar["overview"] = first_sentences(longs[0], 3) print(f"[narrative] {len(keys)} one-liners + {len(phrased)} hypotheses " f"+ overview in 2 batched calls, {time.time() - t0:.1f}s", flush=True) _NARR_CACHE[k] = nar return nar def clear_caches(name: str | None = None): """Drop one dataset (or everything) from the caches.""" _DATA_CACHE.pop(name, None) if name else _DATA_CACHE.clear() for c in (_STATS_CACHE, _NARR_CACHE): if name: for k in [k for k in c if k[0] == name]: c.pop(k, None) else: c.clear() def assemble_report(df, name, instruction, dropped, tables, facts, narrative, plots, target=None) -> tuple: """Pure string assembly - no computation, no model calls. target may be None.""" t, nar = tables, narrative P = [f"# EDA Report — {name}", "", f"*{instruction}*", ""] P += ["## Dataset Overview & Schema", f"**{df.shape[0]:,} rows x {df.shape[1]} columns.** " + nar["overview"], ""] if dropped: P += [f"_Index-like columns excluded from analysis: {', '.join(dropped)}._", ""] # Target analysis sits directly after the overview: once a target exists, # every other section is read in relation to it. if target and "target_balance" in t: P += [f"## Target Analysis — `{target}`", "", "**Class balance**", "", t["target_balance"], ""] P += ["**Predictors, ranked by effect size**", "", t["target_predictors"], ""] if "target_rates" in t: P += [f"**{t.get('target_rate_caption', 'Target rate by level')}**", "", t["target_rates"], ""] if "target" in nar: P += [nar["target"], ""] P += ["## Data Quality", t["health"], ""] P += ["## Missing Values & Duplicate Rows", t["missing"], ""] if t.get("duplicates"): P += [t["duplicates"], ""] if "co_missing" in t: P += ["**Columns that go missing together**", "", t["co_missing"], ""] if "missing_vs_target" in t: P += ["**Missingness that predicts the target**", "", t["missing_vs_target"], ""] if "missing" in nar: P += [nar["missing"], ""] P += ["## Distribution & Summary Characteristics", t["describe"], ""] if "shape" in nar: P += [nar["shape"], ""] P += ["## Required Categorical Transformations", t["encoding"], ""] P += ["## Correlations & Group Differences", t["corr"], ""] if "corr" in nar: P += [nar["corr"], ""] P += [t["groups"], ""] if "redundant" in t: P += ["**Redundant columns**", "", t["redundant"], ""] P += [f"## Outliers Detection ({IQR_K}x IQR Method)", t["outliers"], ""] if "outliers" in nar: P += [nar["outliers"], ""] P += ["## Machine Learning Readiness & Preprocessing Checklist", t["checklist"], ""] P += ["## Visualizations", ""] for p in plots: P += [f"**{os.path.basename(p)[:-4].replace('_', ' ')}**", "", f"![]({p})", ""] P += ["## Hypotheses & Next Steps", nar.get("hypotheses", "_No hypotheses could be derived._"), ""] report = "\n".join(P) allowed = "\n".join([*t.values(), *facts.values(), schema_only(df, name)]) return report, audit_numbers(report, allowed) def build_report(df, res, name, instruction, dropped=None, target=None) -> tuple: """Cache-aware wrapper: only the plot list is new on a re-run.""" checks, tables, facts, hyps = get_stats(name, df, target) nar = get_narrative(name, df, facts, hyps, checks.get("target")) return assemble_report(df, name, instruction, dropped, tables, facts, nar, res["plots"], checks.get("target")) # =========================================================================== # # Report rendering # =========================================================================== # def embed_images(md: str) -> str: """Inline PNGs as data URIs so they render inside Colab's output iframe.""" def repl(m): path = m.group(1) if not os.path.exists(path): return m.group(0) return ("](data:image/png;base64," + base64.b64encode(open(path, "rb").read()).decode() + ")") return re.sub(r"\]\(([^)]+\.png)\)", repl, md) def report_to_pdf(report_md: str, plots: list, path: str = "eda_report.pdf") -> str: """Render the markdown report plus every plot to a single PDF.""" from fpdf import FPDF from fpdf.enums import XPos, YPos def clean(s): for a, b in [("—", "-"), ("…", "..."), ("≥", ">="), ("×", "x"), ("−", "-"), ("’", "'"), ("“", '"'), ("”", '"')]: s = s.replace(a, b) # fpdf2's core fonts are latin-1: drop emoji rather than render them as "?" s = re.sub(r"[^\x00-\xff]", "", s) return s.encode("latin-1", "replace").decode("latin-1") pdf = FPDF(format="A4") pdf.set_auto_page_break(True, margin=15) pdf.add_page() usable = pdf.w - pdf.l_margin - pdf.r_margin max_table_chars = int(usable / (6.5 * 0.6 * 0.3528)) # Courier 0.6 em, 1pt=0.3528mm def write(text, style="", size=10, h=5): """Always start at the left margin — otherwise multi_cell width goes to zero.""" pdf.set_font("Courier" if style == "mono" else "Helvetica", "B" if style == "B" else "", size) pdf.set_x(pdf.l_margin) pdf.multi_cell(0, h, text, new_x=XPos.LMARGIN, new_y=YPos.NEXT) for raw in report_md.splitlines(): line = clean(raw.rstrip()) if line.startswith("!["): continue is_table = line.startswith("|") if not is_table: line = re.sub(r"[*_`]", "", line) line = " ".join(w if len(w) <= 50 else " ".join(w[i:i + 50] for i in range(0, len(w), 50)) for w in line.split()) elif len(line) > max_table_chars: line = line[:max_table_chars - 3] + "..." if line.startswith("# "): write(line[2:], "B", 15, 8) pdf.ln(1) elif line.startswith("## "): pdf.ln(2) write(line[3:], "B", 12, 7) elif is_table: write(line, "mono", 6.5, 3.4) elif line.strip(): write(line, "", 10, 5) else: pdf.ln(2) for p in plots: pdf.add_page() write(clean(os.path.basename(p)[:-4].replace("_", " ")), "B", 11, 7) pdf.image(p, w=usable) pdf.output(path) return path # =========================================================================== # # Pipeline # =========================================================================== # DEFAULT_INSTRUCTION = ("Run a comprehensive exploratory data analysis, highlighting " "missing values, distributions, and key feature correlations. " "Include a correlation heatmap of the numeric columns.") DEFAULT_DATASET = "mstz/titanic" AUTO_TARGET, NO_TARGET = "(auto-detect)", "(none)" GPU_DURATION = 90 # seconds requested per run. Budget on Blackwell: # narrative (2 batched calls) ~5s # plot code generation ~6s # execution (hard limit) <=30s # repair generation + execution <=36s # ZeroGPU checks the REQUESTED duration against the visitor's # remaining quota before starting, and a shorter request also # buys higher queue priority. _CACHE = {} # full-run cache: (instruction, dataset, target choice) # pdf, code, stdout, gallery, report, target-dropdown. gr.update() with no # arguments leaves the dropdown alone, which is what intermediate yields want. BLANK = (None, "", "", [], "", gr.update()) def resolve_target(df, choice): """UI choice -> a column name or None. Tolerates anything typed.""" if choice in (None, "", NO_TARGET): return None if choice == AUTO_TARGET: return infer_target(df) return choice if choice in df.columns else infer_target(df) @spaces.GPU(duration=GPU_DURATION) def _gpu_stages(instruction, dataset, df, dropped, context, tables, facts, hyps, target, dd): """The GPU half: narrative + code generation + execution + assembly. A generator, so the browser sees each phase as it starts. Download and statistics deliberately run OUTSIDE this function: they are pure pandas, and holding a GPU slot through a dataset download burns quota for nothing. """ yield ("Writing the narrative ...", *BLANK) narrative = get_narrative(dataset, df, facts, hyps, target) res = None for kind, payload in generate_and_run_stream(context, instruction, df, selected_plots=DEFAULT_PLOTS): if kind == "status": yield (payload, *BLANK) else: res = payload if res["error"] and res["error"].startswith(TIMEOUT_MARK): status = (f"Generated code exceeded the {CODE_TIMEOUT_S}s limit and was " f"abandoned - report built from the data alone") elif res["ok"]: status = "Code ran successfully" else: status = "Code still failing after one repair - report built from data only" yield (f"{status}. Assembling the report ...", None, res["code"], res["stdout"], res["plots"], "", gr.update()) report, flags = assemble_report(df, dataset, instruction, dropped, tables, facts, narrative, res["plots"], target) # The fabrication check runs on every report. It appears in the report # itself, and only when it has something to say. if flags: report += ("\n\n---\n\n_Fabrication check: " f"{len(flags)} number(s) appear in the prose but not in the " f"computed evidence - {', '.join(flags[:10])}._") safe = re.sub(r"[^\w.-]", "_", dataset) try: pdf = report_to_pdf(report, res["plots"], path=f"eda_report_{safe}.pdf") pdf_note = "" except Exception as e: pdf, pdf_note = None, f" (PDF unavailable: {type(e).__name__})" yield (f"Done in {res['seconds']:.0f}s - {status.lower()}, " f"{len(res['plots'])} plots.{pdf_note}", pdf, res["code"], res["stdout"], res["plots"], embed_images(report), dd) def run_agent(instruction, dataset, target_choice): """Streams: status, pdf, code, stdout, gallery, report, target dropdown.""" instruction = (instruction or "").strip() or DEFAULT_INSTRUCTION dataset = (dataset or "").strip() if not dataset: yield ("Enter a Hugging Face dataset id, for example `mstz/titanic`.", *BLANK) return key = (instruction, dataset, target_choice) if key in _CACHE: # Served without touching the GPU, so a repeat view costs the visitor # nothing from their ZeroGPU quota. status, *rest = _CACHE[key] yield (status + " _(cached)_", *rest) return # ---- CPU: download ---- yield (("Reading cached data ..." if dataset in _DATA_CACHE else f"Downloading `{dataset}` ..."), *BLANK) try: df, dropped, context = get_dataframe(dataset) except Exception as e: yield (f"**Could not load `{dataset}`** - {type(e).__name__}: {e}", *BLANK) return # The dropdown cannot be pre-populated: the columns are not known until the # dataset is downloaded. So it offers (auto-detect)/(none) up front and is # refilled with the real columns - and the resolved target - once loaded. target = resolve_target(df, target_choice) dd = gr.update(choices=[AUTO_TARGET, NO_TARGET] + [str(c) for c in df.columns], value=(str(target) if target else NO_TARGET)) # ---- CPU: statistics ---- warm = (dataset, target) in _STATS_CACHE tgt_note = f"target `{target}`" if target else "no target" yield ((f"**{df.shape[0]:,} x {df.shape[1]}**, {tgt_note} - " + ("statistics cached ..." if warm else "computing statistics ...")), *BLANK) try: checks, tables, facts, hyps = get_stats(dataset, df, target) target = checks.get("target") except Exception as e: yield (f"**{type(e).__name__}** - {e}\n\n" f"
traceback\n\n```\n" f"{traceback.format_exc(limit=6)}\n```\n
", *BLANK) return # ---- GPU ---- final = None try: for out in _gpu_stages(instruction, dataset, df, dropped, context, tables, facts, hyps, target, dd): final = out yield out except Exception as e: yield (f"**{type(e).__name__}** - {e}\n\n" f"
traceback\n\n```\n" f"{traceback.format_exc(limit=6)}\n```\n
", *BLANK) return if final is not None: _CACHE[key] = final def reset_form(): return DEFAULT_INSTRUCTION, DEFAULT_DATASET, AUTO_TARGET # =========================================================================== # # UI # =========================================================================== # CSS = ".hero {text-align:center;}" with gr.Blocks(title="EDAgent", theme=gr.themes.Soft(), css=CSS) as demo: gr.Markdown("# EDAgent\nGive it a Hugging Face dataset. Get a full EDA report.", elem_classes="hero") with gr.Row(): instruction = gr.Textbox(label="Prompt instruction", value=DEFAULT_INSTRUCTION, lines=3, scale=3) with gr.Column(scale=1): dataset = gr.Textbox(label="Hugging Face dataset id", value=DEFAULT_DATASET, placeholder="owner/name", lines=1) target_dd = gr.Dropdown(label="Target column", choices=[AUTO_TARGET, NO_TARGET], value=AUTO_TARGET, allow_custom_value=True) with gr.Row(): run_btn = gr.Button("Run", variant="primary", size="lg", scale=4) reset_btn = gr.Button("Reset", variant="secondary", size="lg", scale=1) status = gr.Markdown() pdf_file = gr.File(label="Report as PDF") with gr.Tabs(): with gr.Tab("Report"): report_md = gr.Markdown() with gr.Tab("Plots"): gallery = gr.Gallery(label=None, columns=2, height=560) with gr.Tab("Code"): code_box = gr.Code(language="python", label=None) with gr.Tab("Output"): stdout_box = gr.Textbox(label=None, lines=18, interactive=False) live_outputs = [status, pdf_file, code_box, stdout_box, gallery, report_md, target_dd] run_btn.click(run_agent, [instruction, dataset, target_dd], live_outputs) reset_btn.click(reset_form, None, [instruction, dataset, target_dd]) if __name__ == "__main__": demo.queue(max_size=12).launch()