Skills › Scientific Agent Skills › exploratory-data-analysis › _tabular.py
_tabular.py
905 lines · skills/exploratory-data-analysis/scripts/_tabular.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded, redacted tabular profiling and EDA sensitivity calculations."""
4from __future__ import annotations
6import csv
7import hashlib
8import heapq
9import math
10import statistics
11from collections import Counter
12from collections.abc import Callable, Iterable, Sequence
13from dataclasses import dataclass, field
14from datetime import datetime, timezone
15from pathlib import Path
16from typing import Any
18from _common import (
19 DEFAULT_MAX_ROWS,
20 MAX_COLUMNS,
21 MAX_FIELD_CHARS,
22 MAX_ROWS,
23 CliError,
24 bounded_integer,
25 display_identifier,
26 finite_number,
27 stable_token,
28)
31DEFAULT_MISSING_TOKENS = ("",)
32PROFILE_SAMPLE_SIZE = 512
33DISTRIBUTION_SAMPLE_SIZE = 4096
34MAX_DISTRIBUTION_COLUMNS = 64
35MAX_DISTINCT_GROUPS = 1000
36MAX_TRACKED_LEAKAGE_KEYS = 200_000
39@dataclass
40class ScanSummary:
41 rows_scanned: int
42 truncated: bool
43 column_count: int
46def normalize_missing_tokens(tokens: Iterable[str] | None) -> frozenset[str]:
47 """Build an explicit, case-insensitive missing-code set."""
49 supplied = list(DEFAULT_MISSING_TOKENS if tokens is None else tokens)
50 normalized = {item.strip().casefold() for item in supplied}
51 normalized.add("")
52 return frozenset(normalized)
55def is_missing(value: str, missing_tokens: frozenset[str]) -> bool:
56 return value.strip().casefold() in missing_tokens
59def delimiter_for_path(path: Path) -> str:
60 name = path.name.casefold()
61 if name.endswith(".csv"):
62 return ","
63 if name.endswith(".tsv"):
64 return "\t"
65 raise CliError("tabular tools accept only .csv and .tsv files")
68def scan_table(
69 path: Path,
70 *,
71 max_rows: int = DEFAULT_MAX_ROWS,
72 max_columns: int = MAX_COLUMNS,
73 max_field_chars: int = MAX_FIELD_CHARS,
74 on_header: Callable[[list[str]], None],
75 on_row: Callable[[int, list[str]], None],
76) -> ScanSummary:
77 """Stream a rectangular UTF-8 table through bounded callbacks."""
79 max_rows = bounded_integer(
80 max_rows,
81 name="max rows",
82 minimum=1,
83 maximum=MAX_ROWS,
84 )
85 max_columns = bounded_integer(
86 max_columns,
87 name="max columns",
88 minimum=1,
89 maximum=MAX_COLUMNS,
90 )
91 max_field_chars = bounded_integer(
92 max_field_chars,
93 name="max field characters",
94 minimum=1,
95 maximum=MAX_FIELD_CHARS,
96 )
97 delimiter = delimiter_for_path(path)
98 previous_limit = csv.field_size_limit()
99 rows_scanned = 0
100 truncated = False
101 column_count = 0
102 try:
103 csv.field_size_limit(max_field_chars)
104 with path.open("r", encoding="utf-8-sig", newline="") as handle:
105 reader = csv.reader(
106 handle,
107 delimiter=delimiter,
108 strict=True,
109 )
110 try:
111 header = next(reader)
112 except StopIteration as exc:
113 raise CliError("the table is empty") from exc
114 if not header:
115 raise CliError("the table header is empty")
116 if len(header) > max_columns:
117 raise CliError(
118 f"the table has more than the {max_columns}-column limit"
119 )
120 if any(not name.strip() for name in header):
121 raise CliError("the table contains an empty column identifier")
122 column_count = len(header)
123 on_header(header)
124 for row_index, row in enumerate(reader):
125 if row_index >= max_rows:
126 truncated = True
127 break
128 if len(row) != column_count:
129 raise CliError("the table contains a non-rectangular row")
130 on_row(row_index, row)
131 rows_scanned += 1
132 except CliError:
133 raise
134 except (OSError, UnicodeError, csv.Error, OverflowError) as exc:
135 raise CliError("the table could not be parsed safely as UTF-8") from exc
136 finally:
137 csv.field_size_limit(previous_limit)
138 return ScanSummary(
139 rows_scanned=rows_scanned,
140 truncated=truncated,
141 column_count=column_count,
142 )
145def resolve_column(header: Sequence[str], requested: str | None) -> int | None:
146 """Resolve an exact identifier, rejecting absent or duplicate columns."""
148 if requested is None:
149 return None
150 matches = [index for index, name in enumerate(header) if name == requested]
151 if not matches:
152 raise CliError("a requested role column was not found")
153 if len(matches) > 1:
154 raise CliError("a requested role column is duplicated")
155 return matches[0]
158class PrioritySample:
159 """Keep a deterministic bounded sample selected by content and position hash."""
161 def __init__(self, limit: int) -> None:
162 self.limit = limit
163 self._heap: list[tuple[int, float]] = []
165 def add(self, value: float, *, row_index: int, column_index: int) -> None:
166 material = f"{row_index}\0{column_index}\0{value!r}".encode("ascii")
167 priority = int.from_bytes(
168 hashlib.blake2s(material, digest_size=8).digest(),
169 "big",
170 )
171 item = (-priority, value)
172 if len(self._heap) < self.limit:
173 heapq.heappush(self._heap, item)
174 elif priority < -self._heap[0][0]:
175 heapq.heapreplace(self._heap, item)
177 def values(self) -> list[float]:
178 return [value for _, value in self._heap]
181def _quantile(sorted_values: Sequence[float], probability: float) -> float | None:
182 if not sorted_values:
183 return None
184 if len(sorted_values) == 1:
185 return float(sorted_values[0])
186 position = probability * (len(sorted_values) - 1)
187 lower = math.floor(position)
188 upper = math.ceil(position)
189 if lower == upper:
190 return float(sorted_values[lower])
191 weight = position - lower
192 return float(
193 sorted_values[lower] * (1.0 - weight) + sorted_values[upper] * weight
194 )
197@dataclass
198class ColumnAccumulator:
199 column_index: int
200 sample_limit: int = PROFILE_SAMPLE_SIZE
201 total: int = 0
202 missing: int = 0
203 numeric: int = 0
204 integer_like: int = 0
205 boolean_like: int = 0
206 text: int = 0
207 mean: float = 0.0
208 m2: float = 0.0
209 minimum: float | None = None
210 maximum: float | None = None
211 text_length_total: int = 0
212 maximum_text_length: int = 0
213 unique_tokens: set[str] = field(default_factory=set)
214 unique_truncated: bool = False
215 frequent_tokens: Counter[str] = field(default_factory=Counter)
216 sample: PrioritySample = field(init=False)
218 def __post_init__(self) -> None:
219 self.sample = PrioritySample(self.sample_limit)
221 def add(
222 self,
223 value: str,
224 *,
225 row_index: int,
226 missing_tokens: frozenset[str],
227 ) -> None:
228 self.total += 1
229 if is_missing(value, missing_tokens):
230 self.missing += 1
231 return
232 token = stable_token(value, kind="value")
233 if len(self.unique_tokens) < 4096:
234 self.unique_tokens.add(token)
235 elif token not in self.unique_tokens:
236 self.unique_truncated = True
237 if len(self.frequent_tokens) < 4096 or token in self.frequent_tokens:
238 self.frequent_tokens[token] += 1
239 stripped = value.strip()
240 lowered = stripped.casefold()
241 if lowered in {"true", "false"}:
242 self.boolean_like += 1
243 number = finite_number(stripped)
244 if number is None:
245 self.text += 1
246 length = len(value)
247 self.text_length_total += length
248 self.maximum_text_length = max(self.maximum_text_length, length)
249 return
250 self.numeric += 1
251 if number.is_integer():
252 self.integer_like += 1
253 delta = number - self.mean
254 self.mean += delta / self.numeric
255 self.m2 += delta * (number - self.mean)
256 self.minimum = number if self.minimum is None else min(self.minimum, number)
257 self.maximum = number if self.maximum is None else max(self.maximum, number)
258 self.sample.add(
259 number,
260 row_index=row_index,
261 column_index=self.column_index,
262 )
264 def as_report(self, *, column_id: str) -> dict[str, Any]:
265 observed = self.total - self.missing
266 if observed == 0:
267 inferred = "all_missing_in_scanned_rows"
268 elif self.numeric == observed:
269 inferred = "integer" if self.integer_like == observed else "numeric"
270 elif self.boolean_like == observed:
271 inferred = "boolean"
272 elif self.text == observed:
273 inferred = "text"
274 else:
275 inferred = "mixed"
276 report: dict[str, Any] = {
277 "column_id": column_id,
278 "column_index": self.column_index,
279 "inferred_kind": inferred,
280 "missing_count": self.missing,
281 "missing_fraction": self.missing / self.total if self.total else None,
282 "non_missing_count": observed,
283 "numeric_parse_count": self.numeric,
284 "text_parse_count": self.text,
285 "distinct_value_count_or_lower_bound": len(self.unique_tokens),
286 "distinct_count_is_lower_bound": self.unique_truncated,
287 }
288 if self.numeric:
289 sample = sorted(self.sample.values())
290 report["numeric_aggregates"] = {
291 "count": self.numeric,
292 "mean": self.mean,
293 "sample_standard_deviation": (
294 math.sqrt(self.m2 / (self.numeric - 1))
295 if self.numeric > 1
296 else None
297 ),
298 "minimum": self.minimum,
299 "q1": _quantile(sample, 0.25),
300 "median": _quantile(sample, 0.5),
301 "q3": _quantile(sample, 0.75),
302 "maximum": self.maximum,
303 "quantiles_from_bounded_sample": len(sample) < self.numeric,
304 "quantile_sample_count": len(sample),
305 }
306 if self.text:
307 report["text_aggregates"] = {
308 "count": self.text,
309 "mean_character_count": self.text_length_total / self.text,
310 "maximum_character_count": self.maximum_text_length,
311 }
312 if self.frequent_tokens:
313 report["most_frequent_value_tokens"] = [
314 {"value_token": token, "count": count}
315 for token, count in sorted(
316 self.frequent_tokens.items(),
317 key=lambda item: (-item[1], item[0]),
318 )[:5]
319 ]
320 return report
323def profile_table(
324 path: Path,
325 *,
326 max_rows: int = DEFAULT_MAX_ROWS,
327 missing_tokens: Iterable[str] | None = None,
328 reveal_identifiers: bool = False,
329) -> dict[str, Any]:
330 """Build a bounded aggregate profile without emitting cell values."""
332 missing = normalize_missing_tokens(missing_tokens)
333 header: list[str] = []
334 accumulators: list[ColumnAccumulator] = []
335 duplicate_rows = 0
336 row_hashes: set[bytes] = set()
337 duplicate_tracking_truncated = False
339 def on_header(names: list[str]) -> None:
340 nonlocal header, accumulators
341 header = names
342 accumulators = [
343 ColumnAccumulator(column_index=index) for index in range(len(names))
344 ]
346 def on_row(row_index: int, row: list[str]) -> None:
347 nonlocal duplicate_rows, duplicate_tracking_truncated
348 for accumulator, value in zip(accumulators, row, strict=True):
349 accumulator.add(
350 value,
351 row_index=row_index,
352 missing_tokens=missing,
353 )
354 fingerprint = hashlib.blake2s(
355 "\0".join(row).encode("utf-8", errors="surrogatepass"),
356 digest_size=16,
357 ).digest()
358 if fingerprint in row_hashes:
359 duplicate_rows += 1
360 elif len(row_hashes) < MAX_TRACKED_LEAKAGE_KEYS:
361 row_hashes.add(fingerprint)
362 else:
363 duplicate_tracking_truncated = True
365 summary = scan_table(
366 path,
367 max_rows=max_rows,
368 on_header=on_header,
369 on_row=on_row,
370 )
371 columns = [
372 accumulator.as_report(
373 column_id=display_identifier(
374 header[index],
375 kind="column",
376 reveal_identifiers=reveal_identifiers,
377 )
378 )
379 for index, accumulator in enumerate(accumulators)
380 ]
381 return {
382 "profile_type": "tabular_schema_and_aggregate_profile",
383 "rows_scanned": summary.rows_scanned,
384 "row_limit_reached": summary.truncated,
385 "column_count": summary.column_count,
386 "columns": columns,
387 "duplicate_row_count_in_scanned_rows": duplicate_rows,
388 "duplicate_tracking_truncated": duplicate_tracking_truncated,
389 "missing_code_policy": {
390 "empty_or_whitespace_is_missing": True,
391 "additional_token_count": max(len(missing) - 1, 0),
392 "tokens_are_not_emitted": True,
393 },
394 "raw_values_emitted": False,
395 "identifier_redaction": "sanitized opt-in" if reveal_identifiers else "tokenized",
396 "limitations": [
397 "Dtypes are inferred from scanned text and are not a data dictionary.",
398 "Quantiles use a deterministic bounded sample when numeric counts exceed the sample limit.",
399 "Duplicate counts use exact hashes only until the documented tracking cap.",
400 ],
401 }
404def profile_json_structure(
405 document: Any,
406 *,
407 reveal_identifiers: bool = False,
408 max_nodes: int = 100_000,
409) -> dict[str, Any]:
410 """Summarize a parsed JSON value without emitting scalar values."""
412 stack: list[tuple[Any, int]] = [(document, 0)]
413 type_counts: Counter[str] = Counter()
414 top_level_fields: list[str] = []
415 maximum_depth = 0
416 array_lengths: list[int] = []
417 object_sizes: list[int] = []
418 nodes = 0
419 truncated = False
420 if isinstance(document, dict):
421 top_level_fields = [
422 display_identifier(
423 str(key),
424 kind="field",
425 reveal_identifiers=reveal_identifiers,
426 )
427 for key in list(document)[:MAX_COLUMNS]
428 ]
429 while stack:
430 value, depth = stack.pop()
431 if nodes >= max_nodes:
432 truncated = True
433 break
434 nodes += 1
435 maximum_depth = max(maximum_depth, depth)
436 if value is None:
437 type_counts["null"] += 1
438 elif isinstance(value, bool):
439 type_counts["boolean"] += 1
440 elif isinstance(value, (int, float)):
441 type_counts["number"] += 1
442 elif isinstance(value, str):
443 type_counts["string"] += 1
444 elif isinstance(value, list):
445 type_counts["array"] += 1
446 array_lengths.append(len(value))
447 stack.extend((item, depth + 1) for item in reversed(value))
448 elif isinstance(value, dict):
449 type_counts["object"] += 1
450 object_sizes.append(len(value))
451 stack.extend((item, depth + 1) for item in reversed(list(value.values())))
452 else:
453 raise CliError("the JSON parser produced an unsupported value type")
454 return {
455 "profile_type": "json_structural_profile",
456 "root_type": (
457 "object"
458 if isinstance(document, dict)
459 else "array"
460 if isinstance(document, list)
461 else "scalar"
462 ),
463 "nodes_visited": nodes,
464 "node_limit_reached": truncated,
465 "maximum_depth_visited": maximum_depth,
466 "type_counts": dict(sorted(type_counts.items())),
467 "top_level_field_ids": top_level_fields,
468 "top_level_fields_truncated": (
469 isinstance(document, dict) and len(document) > len(top_level_fields)
470 ),
471 "array_count": len(array_lengths),
472 "array_length_minimum": min(array_lengths) if array_lengths else None,
473 "array_length_maximum": max(array_lengths) if array_lengths else None,
474 "object_count": len(object_sizes),
475 "object_field_count_maximum": max(object_sizes) if object_sizes else None,
476 "raw_values_emitted": False,
477 "identifier_redaction": "sanitized opt-in" if reveal_identifiers else "tokenized",
478 }
481def _moment_skew(values: Sequence[float]) -> float | None:
482 if len(values) < 3:
483 return None
484 mean = statistics.fmean(values)
485 variance = statistics.fmean((value - mean) ** 2 for value in values)
486 if variance <= 0:
487 return 0.0
488 third = statistics.fmean((value - mean) ** 3 for value in values)
489 return third / (variance ** 1.5)
492def _distribution_report(
493 values: Sequence[float],
494 *,
495 numeric_count: int,
496 nonnumeric_count: int,
497 missing_count: int,
498) -> dict[str, Any]:
499 ordered = sorted(values)
500 if not ordered:
501 return {
502 "numeric_count": numeric_count,
503 "nonnumeric_count": nonnumeric_count,
504 "missing_count": missing_count,
505 "status": "no_finite_numeric_values_sampled",
506 }
507 q1 = _quantile(ordered, 0.25)
508 median = _quantile(ordered, 0.5)
509 q3 = _quantile(ordered, 0.75)
510 assert q1 is not None and median is not None and q3 is not None
511 iqr = q3 - q1
512 lower_fence = q1 - 1.5 * iqr
513 upper_fence = q3 + 1.5 * iqr
514 inliers = [value for value in ordered if lower_fence <= value <= upper_fence]
515 absolute_deviations = sorted(abs(value - median) for value in ordered)
516 mad = _quantile(absolute_deviations, 0.5)
517 trim = int(len(ordered) * 0.1)
518 trimmed = ordered[trim : len(ordered) - trim] if trim and len(ordered) > 2 * trim else ordered
519 winsorized = list(ordered)
520 if trim and len(ordered) > 2 * trim:
521 low = ordered[trim]
522 high = ordered[-trim - 1]
523 winsorized = [min(max(value, low), high) for value in ordered]
524 raw_mean = statistics.fmean(ordered)
525 transformed: dict[str, Any] = {
526 "raw_skewness_moment": _moment_skew(ordered),
527 }
528 if ordered[0] >= 0:
529 transformed["log1p_skewness_moment"] = _moment_skew(
530 [math.log1p(value) for value in ordered]
531 )
532 transformed["candidate"] = "log1p"
533 else:
534 transformed["signed_log1p_skewness_moment"] = _moment_skew(
535 [math.copysign(math.log1p(abs(value)), value) for value in ordered]
536 )
537 transformed["candidate"] = "signed_log1p"
538 return {
539 "numeric_count": numeric_count,
540 "nonnumeric_count": nonnumeric_count,
541 "missing_count": missing_count,
542 "bounded_numeric_sample_count": len(ordered),
543 "sample_is_bounded": len(ordered) < numeric_count,
544 "location_and_scale": {
545 "mean": raw_mean,
546 "sample_standard_deviation": (
547 statistics.stdev(ordered) if len(ordered) > 1 else None
548 ),
549 "median": median,
550 "median_absolute_deviation": mad,
551 "q1": q1,
552 "q3": q3,
553 "interquartile_range": iqr,
554 },
555 "outlier_sensitivity": {
556 "iqr_fence_low": lower_fence,
557 "iqr_fence_high": upper_fence,
558 "outside_iqr_fence_count_in_sample": len(ordered) - len(inliers),
559 "mean_without_iqr_fence_values": (
560 statistics.fmean(inliers) if inliers else None
561 ),
562 "ten_percent_trimmed_mean": statistics.fmean(trimmed),
563 "ten_percent_winsorized_mean": statistics.fmean(winsorized),
564 "raw_minus_trimmed_mean": raw_mean - statistics.fmean(trimmed),
565 "values_were_not_deleted_or_modified": True,
566 },
567 "transformation_sensitivity": {
568 **transformed,
569 "diagnostic_only": True,
570 "fit_on_training_data_only_if_later_adopted": True,
571 },
572 }
575def audit_distributions(
576 path: Path,
577 *,
578 columns: Sequence[str] | None = None,
579 max_rows: int = DEFAULT_MAX_ROWS,
580 missing_tokens: Iterable[str] | None = None,
581 reveal_identifiers: bool = False,
582) -> dict[str, Any]:
583 """Compare robust/classical summaries without deleting or transforming data."""
585 missing = normalize_missing_tokens(missing_tokens)
586 header: list[str] = []
587 selected_indices: list[int] = []
588 selected_ids: list[str] = []
589 accumulators: dict[int, ColumnAccumulator] = {}
590 skipped_columns = 0
592 def on_header(names: list[str]) -> None:
593 nonlocal header, selected_indices, selected_ids, accumulators, skipped_columns
594 header = names
595 if columns:
596 selected_indices = []
597 for requested in columns:
598 index = resolve_column(names, requested)
599 assert index is not None
600 if index in selected_indices:
601 raise CliError("distribution columns must not be duplicated")
602 selected_indices.append(index)
603 else:
604 selected_indices = list(range(min(len(names), MAX_DISTRIBUTION_COLUMNS)))
605 skipped_columns = max(len(names) - len(selected_indices), 0)
606 selected_ids = [
607 display_identifier(
608 names[index],
609 kind="column",
610 reveal_identifiers=reveal_identifiers,
611 )
612 for index in selected_indices
613 ]
614 accumulators = {
615 index: ColumnAccumulator(
616 column_index=index,
617 sample_limit=DISTRIBUTION_SAMPLE_SIZE,
618 )
619 for index in selected_indices
620 }
622 def on_row(row_index: int, row: list[str]) -> None:
623 for index in selected_indices:
624 accumulators[index].add(
625 row[index],
626 row_index=row_index,
627 missing_tokens=missing,
628 )
630 summary = scan_table(
631 path,
632 max_rows=max_rows,
633 on_header=on_header,
634 on_row=on_row,
635 )
636 reports: list[dict[str, Any]] = []
637 for index, column_id in zip(selected_indices, selected_ids, strict=True):
638 accumulator = accumulators[index]
639 report = _distribution_report(
640 accumulator.sample.values(),
641 numeric_count=accumulator.numeric,
642 nonnumeric_count=accumulator.text,
643 missing_count=accumulator.missing,
644 )
645 if accumulator.numeric:
646 reports.append({"column_id": column_id, **report})
647 return {
648 "audit_type": "distribution_and_outlier_sensitivity",
649 "rows_scanned": summary.rows_scanned,
650 "row_limit_reached": summary.truncated,
651 "numeric_columns_reported": len(reports),
652 "columns_skipped_by_default_limit": skipped_columns,
653 "columns": reports,
654 "interpretation": [
655 "IQR fences are descriptive flags, not deletion rules.",
656 "Transformation comparisons are exploratory diagnostics, not automatic recommendations.",
657 "No hypothesis tests, p-values, imputations, or causal claims are produced.",
658 "If later tests are run, define the hypothesis family and multiplicity procedure first.",
659 ],
660 "raw_values_emitted": False,
661 }
664def _parse_time(value: str) -> datetime | None:
665 text = value.strip()
666 if not text:
667 return None
668 if text.endswith("Z"):
669 text = text[:-1] + "+00:00"
670 try:
671 parsed = datetime.fromisoformat(text)
672 except ValueError:
673 return None
674 if parsed.tzinfo is not None:
675 return parsed.astimezone(timezone.utc).replace(tzinfo=None)
676 return parsed
679def audit_missingness_and_leakage(
680 path: Path,
681 *,
682 max_rows: int = DEFAULT_MAX_ROWS,
683 missing_tokens: Iterable[str] | None = None,
684 group_column: str | None = None,
685 entity_column: str | None = None,
686 split_column: str | None = None,
687 time_column: str | None = None,
688 reveal_identifiers: bool = False,
689) -> dict[str, Any]:
690 """Audit missingness and common split leakage without exposing identifiers."""
692 missing = normalize_missing_tokens(missing_tokens)
693 header: list[str] = []
694 group_index: int | None = None
695 entity_index: int | None = None
696 split_index: int | None = None
697 time_index: int | None = None
698 overall_missing: list[int] = []
699 by_group: dict[str, tuple[int, list[int]]] = {}
700 by_split: dict[str, tuple[int, list[int]]] = {}
701 entity_splits: dict[str, set[str]] = {}
702 group_splits: dict[str, set[str]] = {}
703 row_splits: dict[bytes, set[str]] = {}
704 time_intervals: dict[str, tuple[datetime, datetime]] = {}
705 time_parse_failures = 0
706 missing_split_rows = 0
707 tracking_truncated = False
709 def on_header(names: list[str]) -> None:
710 nonlocal header, group_index, entity_index, split_index, time_index
711 nonlocal overall_missing
712 header = names
713 group_index = resolve_column(names, group_column)
714 entity_index = resolve_column(names, entity_column)
715 split_index = resolve_column(names, split_column)
716 time_index = resolve_column(names, time_column)
717 overall_missing = [0] * len(names)
719 def update_partition(
720 mapping: dict[str, tuple[int, list[int]]],
721 key: str,
722 row: list[str],
723 ) -> None:
724 if key not in mapping:
725 if len(mapping) >= MAX_DISTINCT_GROUPS:
726 raise CliError("too many distinct group or split values")
727 mapping[key] = (0, [0] * len(row))
728 count, counts = mapping[key]
729 for index, value in enumerate(row):
730 if is_missing(value, missing):
731 counts[index] += 1
732 mapping[key] = (count + 1, counts)
734 def on_row(_row_index: int, row: list[str]) -> None:
735 nonlocal time_parse_failures, missing_split_rows, tracking_truncated
736 for index, value in enumerate(row):
737 if is_missing(value, missing):
738 overall_missing[index] += 1
739 group_token: str | None = None
740 split_token: str | None = None
741 if group_index is not None and not is_missing(row[group_index], missing):
742 group_token = stable_token(row[group_index], kind="group")
743 update_partition(by_group, group_token, row)
744 if split_index is not None:
745 if is_missing(row[split_index], missing):
746 missing_split_rows += 1
747 else:
748 split_token = stable_token(row[split_index], kind="split")
749 update_partition(by_split, split_token, row)
750 if split_token is None:
751 return
752 if entity_index is not None and not is_missing(row[entity_index], missing):
753 entity_token = stable_token(row[entity_index], kind="entity")
754 if entity_token in entity_splits or len(entity_splits) < MAX_TRACKED_LEAKAGE_KEYS:
755 entity_splits.setdefault(entity_token, set()).add(split_token)
756 elif entity_token not in entity_splits:
757 tracking_truncated = True
758 if group_token is not None:
759 if group_token in group_splits or len(group_splits) < MAX_TRACKED_LEAKAGE_KEYS:
760 group_splits.setdefault(group_token, set()).add(split_token)
761 elif group_token not in group_splits:
762 tracking_truncated = True
763 row_without_split = [
764 value for index, value in enumerate(row) if index != split_index
765 ]
766 fingerprint = hashlib.blake2s(
767 "\0".join(row_without_split).encode(
768 "utf-8",
769 errors="surrogatepass",
770 ),
771 digest_size=16,
772 ).digest()
773 if fingerprint in row_splits or len(row_splits) < MAX_TRACKED_LEAKAGE_KEYS:
774 row_splits.setdefault(fingerprint, set()).add(split_token)
775 elif fingerprint not in row_splits:
776 tracking_truncated = True
777 if time_index is not None:
778 parsed = _parse_time(row[time_index])
779 if parsed is None:
780 if not is_missing(row[time_index], missing):
781 time_parse_failures += 1
782 elif split_token not in time_intervals:
783 time_intervals[split_token] = (parsed, parsed)
784 else:
785 low, high = time_intervals[split_token]
786 time_intervals[split_token] = (min(low, parsed), max(high, parsed))
788 summary = scan_table(
789 path,
790 max_rows=max_rows,
791 on_header=on_header,
792 on_row=on_row,
793 )
794 column_ids = [
795 display_identifier(
796 name,
797 kind="column",
798 reveal_identifiers=reveal_identifiers,
799 )
800 for name in header
801 ]
802 overall = [
803 {
804 "column_id": column_ids[index],
805 "missing_count": count,
806 "missing_fraction": count / summary.rows_scanned
807 if summary.rows_scanned
808 else None,
809 }
810 for index, count in enumerate(overall_missing)
811 ]
813 def partition_report(
814 mapping: dict[str, tuple[int, list[int]]],
815 *,
816 key_name: str,
817 ) -> list[dict[str, Any]]:
818 reports: list[dict[str, Any]] = []
819 for key in sorted(mapping):
820 count, counts = mapping[key]
821 reports.append(
822 {
823 key_name: key,
824 "row_count": count,
825 "column_missingness": [
826 {
827 "column_id": column_ids[index],
828 "missing_count": value,
829 "missing_fraction": value / count if count else None,
830 }
831 for index, value in enumerate(counts)
832 ],
833 }
834 )
835 return reports
837 maximum_group_gaps: list[dict[str, Any]] = []
838 if len(by_group) >= 2:
839 for index, column_id in enumerate(column_ids):
840 rates = [
841 counts[index] / count
842 for count, counts in by_group.values()
843 if count
844 ]
845 if rates:
846 maximum_group_gaps.append(
847 {
848 "column_id": column_id,
849 "maximum_group_missingness_gap": max(rates) - min(rates),
850 }
851 )
852 temporal_overlap_pairs = 0
853 interval_items = sorted(time_intervals.items())
854 for left_index, (_, (left_low, left_high)) in enumerate(interval_items):
855 for _, (right_low, right_high) in interval_items[left_index + 1 :]:
856 if max(left_low, right_low) <= min(left_high, right_high):
857 temporal_overlap_pairs += 1
858 entity_overlap_count = sum(
859 len(splits) > 1 for splits in entity_splits.values()
860 )
861 group_overlap_count = sum(len(splits) > 1 for splits in group_splits.values())
862 duplicate_row_overlap_count = sum(
863 len(splits) > 1 for splits in row_splits.values()
864 )
865 leakage_flags = {
866 "entity_tokens_in_multiple_splits": entity_overlap_count,
867 "group_tokens_in_multiple_splits": group_overlap_count,
868 "identical_row_hashes_in_multiple_splits": duplicate_row_overlap_count,
869 "overlapping_split_time_interval_pairs": temporal_overlap_pairs,
870 }
871 any_flag = any(leakage_flags.values())
872 leakage_assessed = split_index is not None
873 return {
874 "audit_type": "missingness_group_and_split_leakage",
875 "rows_scanned": summary.rows_scanned,
876 "row_limit_reached": summary.truncated,
877 "overall_missingness": overall,
878 "missingness_by_group": partition_report(by_group, key_name="group_token"),
879 "missingness_by_split": partition_report(by_split, key_name="split_token"),
880 "maximum_group_missingness_gaps": maximum_group_gaps,
881 "leakage_audit": {
882 "status": (
883 "not_assessed_without_split_column"
884 if not leakage_assessed
885 else "potential_leakage_detected"
886 if any_flag
887 else "not_detected_in_scanned_rows"
888 ),
889 **leakage_flags,
890 "rows_with_missing_split": missing_split_rows,
891 "unparseable_nonmissing_time_values": time_parse_failures,
892 "tracking_truncated": tracking_truncated,
893 "scope_warning": (
894 "No finding is not proof of no leakage, especially for truncated scans "
895 "or unprovided entity/group/time roles."
896 ),
897 },
898 "missing_code_policy": {
899 "empty_or_whitespace_is_missing": True,
900 "additional_token_count": max(len(missing) - 1, 0),
901 "tokens_are_not_emitted": True,
902 },
903 "raw_values_emitted": False,
904 "automatic_imputation_or_deletion": False,
905 }