#!/usr/bin/env python3 """Bounded, redacted tabular profiling and EDA sensitivity calculations.""" from __future__ import annotations import csv import hashlib import heapq import math import statistics from collections import Counter from collections.abc import Callable, Iterable, Sequence from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any from _common import ( DEFAULT_MAX_ROWS, MAX_COLUMNS, MAX_FIELD_CHARS, MAX_ROWS, CliError, bounded_integer, display_identifier, finite_number, stable_token, ) DEFAULT_MISSING_TOKENS = ("",) PROFILE_SAMPLE_SIZE = 512 DISTRIBUTION_SAMPLE_SIZE = 4096 MAX_DISTRIBUTION_COLUMNS = 64 MAX_DISTINCT_GROUPS = 1000 MAX_TRACKED_LEAKAGE_KEYS = 200_000 @dataclass class ScanSummary: rows_scanned: int truncated: bool column_count: int def normalize_missing_tokens(tokens: Iterable[str] | None) -> frozenset[str]: """Build an explicit, case-insensitive missing-code set.""" supplied = list(DEFAULT_MISSING_TOKENS if tokens is None else tokens) normalized = {item.strip().casefold() for item in supplied} normalized.add("") return frozenset(normalized) def is_missing(value: str, missing_tokens: frozenset[str]) -> bool: return value.strip().casefold() in missing_tokens def delimiter_for_path(path: Path) -> str: name = path.name.casefold() if name.endswith(".csv"): return "," if name.endswith(".tsv"): return "\t" raise CliError("tabular tools accept only .csv and .tsv files") def scan_table( path: Path, *, max_rows: int = DEFAULT_MAX_ROWS, max_columns: int = MAX_COLUMNS, max_field_chars: int = MAX_FIELD_CHARS, on_header: Callable[[list[str]], None], on_row: Callable[[int, list[str]], None], ) -> ScanSummary: """Stream a rectangular UTF-8 table through bounded callbacks.""" max_rows = bounded_integer( max_rows, name="max rows", minimum=1, maximum=MAX_ROWS, ) max_columns = bounded_integer( max_columns, name="max columns", minimum=1, maximum=MAX_COLUMNS, ) max_field_chars = bounded_integer( max_field_chars, name="max field characters", minimum=1, maximum=MAX_FIELD_CHARS, ) delimiter = delimiter_for_path(path) previous_limit = csv.field_size_limit() rows_scanned = 0 truncated = False column_count = 0 try: csv.field_size_limit(max_field_chars) with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.reader( handle, delimiter=delimiter, strict=True, ) try: header = next(reader) except StopIteration as exc: raise CliError("the table is empty") from exc if not header: raise CliError("the table header is empty") if len(header) > max_columns: raise CliError( f"the table has more than the {max_columns}-column limit" ) if any(not name.strip() for name in header): raise CliError("the table contains an empty column identifier") column_count = len(header) on_header(header) for row_index, row in enumerate(reader): if row_index >= max_rows: truncated = True break if len(row) != column_count: raise CliError("the table contains a non-rectangular row") on_row(row_index, row) rows_scanned += 1 except CliError: raise except (OSError, UnicodeError, csv.Error, OverflowError) as exc: raise CliError("the table could not be parsed safely as UTF-8") from exc finally: csv.field_size_limit(previous_limit) return ScanSummary( rows_scanned=rows_scanned, truncated=truncated, column_count=column_count, ) def resolve_column(header: Sequence[str], requested: str | None) -> int | None: """Resolve an exact identifier, rejecting absent or duplicate columns.""" if requested is None: return None matches = [index for index, name in enumerate(header) if name == requested] if not matches: raise CliError("a requested role column was not found") if len(matches) > 1: raise CliError("a requested role column is duplicated") return matches[0] class PrioritySample: """Keep a deterministic bounded sample selected by content and position hash.""" def __init__(self, limit: int) -> None: self.limit = limit self._heap: list[tuple[int, float]] = [] def add(self, value: float, *, row_index: int, column_index: int) -> None: material = f"{row_index}\0{column_index}\0{value!r}".encode("ascii") priority = int.from_bytes( hashlib.blake2s(material, digest_size=8).digest(), "big", ) item = (-priority, value) if len(self._heap) < self.limit: heapq.heappush(self._heap, item) elif priority < -self._heap[0][0]: heapq.heapreplace(self._heap, item) def values(self) -> list[float]: return [value for _, value in self._heap] def _quantile(sorted_values: Sequence[float], probability: float) -> float | None: if not sorted_values: return None if len(sorted_values) == 1: return float(sorted_values[0]) position = probability * (len(sorted_values) - 1) lower = math.floor(position) upper = math.ceil(position) if lower == upper: return float(sorted_values[lower]) weight = position - lower return float( sorted_values[lower] * (1.0 - weight) + sorted_values[upper] * weight ) @dataclass class ColumnAccumulator: column_index: int sample_limit: int = PROFILE_SAMPLE_SIZE total: int = 0 missing: int = 0 numeric: int = 0 integer_like: int = 0 boolean_like: int = 0 text: int = 0 mean: float = 0.0 m2: float = 0.0 minimum: float | None = None maximum: float | None = None text_length_total: int = 0 maximum_text_length: int = 0 unique_tokens: set[str] = field(default_factory=set) unique_truncated: bool = False frequent_tokens: Counter[str] = field(default_factory=Counter) sample: PrioritySample = field(init=False) def __post_init__(self) -> None: self.sample = PrioritySample(self.sample_limit) def add( self, value: str, *, row_index: int, missing_tokens: frozenset[str], ) -> None: self.total += 1 if is_missing(value, missing_tokens): self.missing += 1 return token = stable_token(value, kind="value") if len(self.unique_tokens) < 4096: self.unique_tokens.add(token) elif token not in self.unique_tokens: self.unique_truncated = True if len(self.frequent_tokens) < 4096 or token in self.frequent_tokens: self.frequent_tokens[token] += 1 stripped = value.strip() lowered = stripped.casefold() if lowered in {"true", "false"}: self.boolean_like += 1 number = finite_number(stripped) if number is None: self.text += 1 length = len(value) self.text_length_total += length self.maximum_text_length = max(self.maximum_text_length, length) return self.numeric += 1 if number.is_integer(): self.integer_like += 1 delta = number - self.mean self.mean += delta / self.numeric self.m2 += delta * (number - self.mean) self.minimum = number if self.minimum is None else min(self.minimum, number) self.maximum = number if self.maximum is None else max(self.maximum, number) self.sample.add( number, row_index=row_index, column_index=self.column_index, ) def as_report(self, *, column_id: str) -> dict[str, Any]: observed = self.total - self.missing if observed == 0: inferred = "all_missing_in_scanned_rows" elif self.numeric == observed: inferred = "integer" if self.integer_like == observed else "numeric" elif self.boolean_like == observed: inferred = "boolean" elif self.text == observed: inferred = "text" else: inferred = "mixed" report: dict[str, Any] = { "column_id": column_id, "column_index": self.column_index, "inferred_kind": inferred, "missing_count": self.missing, "missing_fraction": self.missing / self.total if self.total else None, "non_missing_count": observed, "numeric_parse_count": self.numeric, "text_parse_count": self.text, "distinct_value_count_or_lower_bound": len(self.unique_tokens), "distinct_count_is_lower_bound": self.unique_truncated, } if self.numeric: sample = sorted(self.sample.values()) report["numeric_aggregates"] = { "count": self.numeric, "mean": self.mean, "sample_standard_deviation": ( math.sqrt(self.m2 / (self.numeric - 1)) if self.numeric > 1 else None ), "minimum": self.minimum, "q1": _quantile(sample, 0.25), "median": _quantile(sample, 0.5), "q3": _quantile(sample, 0.75), "maximum": self.maximum, "quantiles_from_bounded_sample": len(sample) < self.numeric, "quantile_sample_count": len(sample), } if self.text: report["text_aggregates"] = { "count": self.text, "mean_character_count": self.text_length_total / self.text, "maximum_character_count": self.maximum_text_length, } if self.frequent_tokens: report["most_frequent_value_tokens"] = [ {"value_token": token, "count": count} for token, count in sorted( self.frequent_tokens.items(), key=lambda item: (-item[1], item[0]), )[:5] ] return report def profile_table( path: Path, *, max_rows: int = DEFAULT_MAX_ROWS, missing_tokens: Iterable[str] | None = None, reveal_identifiers: bool = False, ) -> dict[str, Any]: """Build a bounded aggregate profile without emitting cell values.""" missing = normalize_missing_tokens(missing_tokens) header: list[str] = [] accumulators: list[ColumnAccumulator] = [] duplicate_rows = 0 row_hashes: set[bytes] = set() duplicate_tracking_truncated = False def on_header(names: list[str]) -> None: nonlocal header, accumulators header = names accumulators = [ ColumnAccumulator(column_index=index) for index in range(len(names)) ] def on_row(row_index: int, row: list[str]) -> None: nonlocal duplicate_rows, duplicate_tracking_truncated for accumulator, value in zip(accumulators, row, strict=True): accumulator.add( value, row_index=row_index, missing_tokens=missing, ) fingerprint = hashlib.blake2s( "\0".join(row).encode("utf-8", errors="surrogatepass"), digest_size=16, ).digest() if len(row_hashes) < MAX_TRACKED_LEAKAGE_KEYS: if fingerprint in row_hashes: duplicate_rows += 1 row_hashes.add(fingerprint) else: duplicate_tracking_truncated = True summary = scan_table( path, max_rows=max_rows, on_header=on_header, on_row=on_row, ) columns = [ accumulator.as_report( column_id=display_identifier( header[index], kind="column", reveal_identifiers=reveal_identifiers, ) ) for index, accumulator in enumerate(accumulators) ] return { "profile_type": "tabular_schema_and_aggregate_profile", "rows_scanned": summary.rows_scanned, "row_limit_reached": summary.truncated, "column_count": summary.column_count, "columns": columns, "duplicate_row_count_in_scanned_rows": duplicate_rows, "duplicate_tracking_truncated": duplicate_tracking_truncated, "missing_code_policy": { "empty_or_whitespace_is_missing": True, "additional_token_count": max(len(missing) - 1, 0), "tokens_are_not_emitted": True, }, "raw_values_emitted": False, "identifier_redaction": "sanitized opt-in" if reveal_identifiers else "tokenized", "limitations": [ "Dtypes are inferred from scanned text and are not a data dictionary.", "Quantiles use a deterministic bounded sample when numeric counts exceed the sample limit.", "Duplicate counts use exact hashes only until the documented tracking cap.", ], } def profile_json_structure( document: Any, *, reveal_identifiers: bool = False, max_nodes: int = 100_000, ) -> dict[str, Any]: """Summarize a parsed JSON value without emitting scalar values.""" stack: list[tuple[Any, int]] = [(document, 0)] type_counts: Counter[str] = Counter() top_level_fields: list[str] = [] maximum_depth = 0 array_lengths: list[int] = [] object_sizes: list[int] = [] nodes = 0 truncated = False if isinstance(document, dict): top_level_fields = [ display_identifier( str(key), kind="field", reveal_identifiers=reveal_identifiers, ) for key in list(document)[:MAX_COLUMNS] ] while stack: value, depth = stack.pop() if nodes >= max_nodes: truncated = True break nodes += 1 maximum_depth = max(maximum_depth, depth) if value is None: type_counts["null"] += 1 elif isinstance(value, bool): type_counts["boolean"] += 1 elif isinstance(value, (int, float)): type_counts["number"] += 1 elif isinstance(value, str): type_counts["string"] += 1 elif isinstance(value, list): type_counts["array"] += 1 array_lengths.append(len(value)) stack.extend((item, depth + 1) for item in reversed(value)) elif isinstance(value, dict): type_counts["object"] += 1 object_sizes.append(len(value)) stack.extend((item, depth + 1) for item in reversed(list(value.values()))) else: raise CliError("the JSON parser produced an unsupported value type") return { "profile_type": "json_structural_profile", "root_type": ( "object" if isinstance(document, dict) else "array" if isinstance(document, list) else "scalar" ), "nodes_visited": nodes, "node_limit_reached": truncated, "maximum_depth_visited": maximum_depth, "type_counts": dict(sorted(type_counts.items())), "top_level_field_ids": top_level_fields, "top_level_fields_truncated": ( isinstance(document, dict) and len(document) > len(top_level_fields) ), "array_count": len(array_lengths), "array_length_minimum": min(array_lengths) if array_lengths else None, "array_length_maximum": max(array_lengths) if array_lengths else None, "object_count": len(object_sizes), "object_field_count_maximum": max(object_sizes) if object_sizes else None, "raw_values_emitted": False, "identifier_redaction": "sanitized opt-in" if reveal_identifiers else "tokenized", } def _moment_skew(values: Sequence[float]) -> float | None: if len(values) < 3: return None mean = statistics.fmean(values) variance = statistics.fmean((value - mean) ** 2 for value in values) if variance <= 0: return 0.0 third = statistics.fmean((value - mean) ** 3 for value in values) return third / (variance ** 1.5) def _distribution_report( values: Sequence[float], *, numeric_count: int, nonnumeric_count: int, missing_count: int, ) -> dict[str, Any]: ordered = sorted(values) if not ordered: return { "numeric_count": numeric_count, "nonnumeric_count": nonnumeric_count, "missing_count": missing_count, "status": "no_finite_numeric_values_sampled", } q1 = _quantile(ordered, 0.25) median = _quantile(ordered, 0.5) q3 = _quantile(ordered, 0.75) assert q1 is not None and median is not None and q3 is not None iqr = q3 - q1 lower_fence = q1 - 1.5 * iqr upper_fence = q3 + 1.5 * iqr inliers = [value for value in ordered if lower_fence <= value <= upper_fence] absolute_deviations = sorted(abs(value - median) for value in ordered) mad = _quantile(absolute_deviations, 0.5) trim = int(len(ordered) * 0.1) trimmed = ordered[trim : len(ordered) - trim] if trim and len(ordered) > 2 * trim else ordered winsorized = list(ordered) if trim and len(ordered) > 2 * trim: low = ordered[trim] high = ordered[-trim - 1] winsorized = [min(max(value, low), high) for value in ordered] raw_mean = statistics.fmean(ordered) transformed: dict[str, Any] = { "raw_skewness_moment": _moment_skew(ordered), } if ordered[0] >= 0: transformed["log1p_skewness_moment"] = _moment_skew( [math.log1p(value) for value in ordered] ) transformed["candidate"] = "log1p" else: transformed["signed_log1p_skewness_moment"] = _moment_skew( [math.copysign(math.log1p(abs(value)), value) for value in ordered] ) transformed["candidate"] = "signed_log1p" return { "numeric_count": numeric_count, "nonnumeric_count": nonnumeric_count, "missing_count": missing_count, "bounded_numeric_sample_count": len(ordered), "sample_is_bounded": len(ordered) < numeric_count, "location_and_scale": { "mean": raw_mean, "sample_standard_deviation": ( statistics.stdev(ordered) if len(ordered) > 1 else None ), "median": median, "median_absolute_deviation": mad, "q1": q1, "q3": q3, "interquartile_range": iqr, }, "outlier_sensitivity": { "iqr_fence_low": lower_fence, "iqr_fence_high": upper_fence, "outside_iqr_fence_count_in_sample": len(ordered) - len(inliers), "mean_without_iqr_fence_values": ( statistics.fmean(inliers) if inliers else None ), "ten_percent_trimmed_mean": statistics.fmean(trimmed), "ten_percent_winsorized_mean": statistics.fmean(winsorized), "raw_minus_trimmed_mean": raw_mean - statistics.fmean(trimmed), "values_were_not_deleted_or_modified": True, }, "transformation_sensitivity": { **transformed, "diagnostic_only": True, "fit_on_training_data_only_if_later_adopted": True, }, } def audit_distributions( path: Path, *, columns: Sequence[str] | None = None, max_rows: int = DEFAULT_MAX_ROWS, missing_tokens: Iterable[str] | None = None, reveal_identifiers: bool = False, ) -> dict[str, Any]: """Compare robust/classical summaries without deleting or transforming data.""" missing = normalize_missing_tokens(missing_tokens) header: list[str] = [] selected_indices: list[int] = [] selected_ids: list[str] = [] accumulators: dict[int, ColumnAccumulator] = {} skipped_columns = 0 def on_header(names: list[str]) -> None: nonlocal header, selected_indices, selected_ids, accumulators, skipped_columns header = names if columns: selected_indices = [] for requested in columns: index = resolve_column(names, requested) assert index is not None if index in selected_indices: raise CliError("distribution columns must not be duplicated") selected_indices.append(index) else: selected_indices = list(range(min(len(names), MAX_DISTRIBUTION_COLUMNS))) skipped_columns = max(len(names) - len(selected_indices), 0) selected_ids = [ display_identifier( names[index], kind="column", reveal_identifiers=reveal_identifiers, ) for index in selected_indices ] accumulators = { index: ColumnAccumulator( column_index=index, sample_limit=DISTRIBUTION_SAMPLE_SIZE, ) for index in selected_indices } def on_row(row_index: int, row: list[str]) -> None: for index in selected_indices: accumulators[index].add( row[index], row_index=row_index, missing_tokens=missing, ) summary = scan_table( path, max_rows=max_rows, on_header=on_header, on_row=on_row, ) reports: list[dict[str, Any]] = [] for index, column_id in zip(selected_indices, selected_ids, strict=True): accumulator = accumulators[index] report = _distribution_report( accumulator.sample.values(), numeric_count=accumulator.numeric, nonnumeric_count=accumulator.text, missing_count=accumulator.missing, ) if accumulator.numeric: reports.append({"column_id": column_id, **report}) return { "audit_type": "distribution_and_outlier_sensitivity", "rows_scanned": summary.rows_scanned, "row_limit_reached": summary.truncated, "numeric_columns_reported": len(reports), "columns_skipped_by_default_limit": skipped_columns, "columns": reports, "interpretation": [ "IQR fences are descriptive flags, not deletion rules.", "Transformation comparisons are exploratory diagnostics, not automatic recommendations.", "No hypothesis tests, p-values, imputations, or causal claims are produced.", "If later tests are run, define the hypothesis family and multiplicity procedure first.", ], "raw_values_emitted": False, } def _parse_time(value: str) -> datetime | None: text = value.strip() if not text: return None if text.endswith("Z"): text = text[:-1] + "+00:00" try: parsed = datetime.fromisoformat(text) except ValueError: return None if parsed.tzinfo is not None: return parsed.astimezone(timezone.utc).replace(tzinfo=None) return parsed def audit_missingness_and_leakage( path: Path, *, max_rows: int = DEFAULT_MAX_ROWS, missing_tokens: Iterable[str] | None = None, group_column: str | None = None, entity_column: str | None = None, split_column: str | None = None, time_column: str | None = None, reveal_identifiers: bool = False, ) -> dict[str, Any]: """Audit missingness and common split leakage without exposing identifiers.""" missing = normalize_missing_tokens(missing_tokens) header: list[str] = [] group_index: int | None = None entity_index: int | None = None split_index: int | None = None time_index: int | None = None overall_missing: list[int] = [] by_group: dict[str, tuple[int, list[int]]] = {} by_split: dict[str, tuple[int, list[int]]] = {} entity_splits: dict[str, set[str]] = {} group_splits: dict[str, set[str]] = {} row_splits: dict[bytes, set[str]] = {} time_intervals: dict[str, tuple[datetime, datetime]] = {} time_parse_failures = 0 missing_split_rows = 0 tracking_truncated = False def on_header(names: list[str]) -> None: nonlocal header, group_index, entity_index, split_index, time_index nonlocal overall_missing header = names group_index = resolve_column(names, group_column) entity_index = resolve_column(names, entity_column) split_index = resolve_column(names, split_column) time_index = resolve_column(names, time_column) overall_missing = [0] * len(names) def update_partition( mapping: dict[str, tuple[int, list[int]]], key: str, row: list[str], ) -> None: if key not in mapping: if len(mapping) >= MAX_DISTINCT_GROUPS: raise CliError("too many distinct group or split values") mapping[key] = (0, [0] * len(row)) count, counts = mapping[key] for index, value in enumerate(row): if is_missing(value, missing): counts[index] += 1 mapping[key] = (count + 1, counts) def on_row(_row_index: int, row: list[str]) -> None: nonlocal time_parse_failures, missing_split_rows, tracking_truncated for index, value in enumerate(row): if is_missing(value, missing): overall_missing[index] += 1 group_token: str | None = None split_token: str | None = None if group_index is not None and not is_missing(row[group_index], missing): group_token = stable_token(row[group_index], kind="group") update_partition(by_group, group_token, row) if split_index is not None: if is_missing(row[split_index], missing): missing_split_rows += 1 else: split_token = stable_token(row[split_index], kind="split") update_partition(by_split, split_token, row) if split_token is None: return if entity_index is not None and not is_missing(row[entity_index], missing): entity_token = stable_token(row[entity_index], kind="entity") if len(entity_splits) < MAX_TRACKED_LEAKAGE_KEYS: entity_splits.setdefault(entity_token, set()).add(split_token) elif entity_token not in entity_splits: tracking_truncated = True if group_token is not None: if len(group_splits) < MAX_TRACKED_LEAKAGE_KEYS: group_splits.setdefault(group_token, set()).add(split_token) elif group_token not in group_splits: tracking_truncated = True row_without_split = [ value for index, value in enumerate(row) if index != split_index ] fingerprint = hashlib.blake2s( "\0".join(row_without_split).encode( "utf-8", errors="surrogatepass", ), digest_size=16, ).digest() if len(row_splits) < MAX_TRACKED_LEAKAGE_KEYS: row_splits.setdefault(fingerprint, set()).add(split_token) elif fingerprint not in row_splits: tracking_truncated = True if time_index is not None: parsed = _parse_time(row[time_index]) if parsed is None: if not is_missing(row[time_index], missing): time_parse_failures += 1 elif split_token not in time_intervals: time_intervals[split_token] = (parsed, parsed) else: low, high = time_intervals[split_token] time_intervals[split_token] = (min(low, parsed), max(high, parsed)) summary = scan_table( path, max_rows=max_rows, on_header=on_header, on_row=on_row, ) column_ids = [ display_identifier( name, kind="column", reveal_identifiers=reveal_identifiers, ) for name in header ] overall = [ { "column_id": column_ids[index], "missing_count": count, "missing_fraction": count / summary.rows_scanned if summary.rows_scanned else None, } for index, count in enumerate(overall_missing) ] def partition_report( mapping: dict[str, tuple[int, list[int]]], *, key_name: str, ) -> list[dict[str, Any]]: reports: list[dict[str, Any]] = [] for key in sorted(mapping): count, counts = mapping[key] reports.append( { key_name: key, "row_count": count, "column_missingness": [ { "column_id": column_ids[index], "missing_count": value, "missing_fraction": value / count if count else None, } for index, value in enumerate(counts) ], } ) return reports maximum_group_gaps: list[dict[str, Any]] = [] if len(by_group) >= 2: for index, column_id in enumerate(column_ids): rates = [ counts[index] / count for count, counts in by_group.values() if count ] if rates: maximum_group_gaps.append( { "column_id": column_id, "maximum_group_missingness_gap": max(rates) - min(rates), } ) temporal_overlap_pairs = 0 interval_items = sorted(time_intervals.items()) for left_index, (_, (left_low, left_high)) in enumerate(interval_items): for _, (right_low, right_high) in interval_items[left_index + 1 :]: if max(left_low, right_low) <= min(left_high, right_high): temporal_overlap_pairs += 1 entity_overlap_count = sum( len(splits) > 1 for splits in entity_splits.values() ) group_overlap_count = sum(len(splits) > 1 for splits in group_splits.values()) duplicate_row_overlap_count = sum( len(splits) > 1 for splits in row_splits.values() ) leakage_flags = { "entity_tokens_in_multiple_splits": entity_overlap_count, "group_tokens_in_multiple_splits": group_overlap_count, "identical_row_hashes_in_multiple_splits": duplicate_row_overlap_count, "overlapping_split_time_interval_pairs": temporal_overlap_pairs, } any_flag = any(leakage_flags.values()) leakage_assessed = split_index is not None return { "audit_type": "missingness_group_and_split_leakage", "rows_scanned": summary.rows_scanned, "row_limit_reached": summary.truncated, "overall_missingness": overall, "missingness_by_group": partition_report(by_group, key_name="group_token"), "missingness_by_split": partition_report(by_split, key_name="split_token"), "maximum_group_missingness_gaps": maximum_group_gaps, "leakage_audit": { "status": ( "not_assessed_without_split_column" if not leakage_assessed else "potential_leakage_detected" if any_flag else "not_detected_in_scanned_rows" ), **leakage_flags, "rows_with_missing_split": missing_split_rows, "unparseable_nonmissing_time_values": time_parse_failures, "tracking_truncated": tracking_truncated, "scope_warning": ( "No finding is not proof of no leakage, especially for truncated scans " "or unprovided entity/group/time roles." ), }, "missing_code_policy": { "empty_or_whitespace_is_missing": True, "additional_token_count": max(len(missing) - 1, 0), "tokens_are_not_emitted": True, }, "raw_values_emitted": False, "automatic_imputation_or_deletion": False, }