audit-labs/audit-tools
A collection of scripts, queries, and other goodies you can use in an audit.
clone: git clone https://gitbay.org/audit-labs/audit-tools.git
d04c321a220805e9dc6ed73bd52b193ec0e4e4bc
verified · cmc
author: Christian Cleberg <hello@cleberg.net> · 2026-07-07T22:12:23Z
.gitignore | 2 + README.org | 2 +- requirements.txt | 4 + sampling/README.md | 125 +++++++++-- sampling/audit_sample.py | 15 ++ sampling/examples/changes_population.csv | 15 ++ sampling/examples/stratified_config.yml | 7 + sampling/examples/users_population.csv | 13 ++ sampling/sampling_tool/__init__.py | 3 + sampling/sampling_tool/cli.py | 367 +++++++++++++++++++++++++++++++ sampling/sampling_tool/filters.py | 44 ++++ sampling/sampling_tool/io.py | 60 +++++ sampling/sampling_tool/manifest.py | 58 +++++ sampling/sampling_tool/methods.py | 125 +++++++++++ sampling/sampling_tool/reconciliation.py | 41 ++++ sampling/sampling_tool/reporting.py | 87 ++++++++ sampling/sampling_tool/validation.py | 157 +++++++++++++ sampling/tests/test_random_sample.py | 70 ++++++ sampling/tests/test_reconciliation.py | 44 ++++ sampling/tests/test_stratified_sample.py | 66 ++++++ sampling/tests/test_validation.py | 81 +++++++ 21 files changed, 1362 insertions(+), 24 deletions(-) @@ -4,8 +4,10 @@ readme.html # Python __pycache__/ +**/__pycache__/ *.py[cod] # Audit output applications/github/output/ output/ +**/output/ @@ -28,7 +28,7 @@ specific audit environments. *Clone the Repository* #+begin_src bash -git clone https://github.com/audit-lab/audit-toolss +git clone https://github.com/audit-labs/audit-tools cd audit-tools #+end_src @@ -1,4 +1,8 @@ pandas +openpyxl +xlrd +PyYAML +pytest requests dash plotly @@ -1,31 +1,110 @@ -# `sample.py` +# Audit Sampling Tools -``` bash -python ./sample.py +This directory contains simple audit sampling utilities. The production-ready +CLI is `audit_sample.py`; the older `sample.py`, `sample.html`, and +`stratified_sample.py` examples remain for compatibility and learning. + +## Purpose + +`audit_sample.py` generates reproducible, documented samples from CSV and Excel +populations. Each run writes the selected sample, validated population, +reconciliation files, methodology notes, a manifest, and a log suitable for an +audit workpaper package. + +## Installation + +Install the repository requirements: + +```bash +pip install -r requirements.txt +``` + +Supported input formats are `.csv`, `.xlsx`, `.xls`, and `.xlsm`. + +## Random Sampling Example + +```bash +python sampling/audit_sample.py \ + --input sampling/examples/users_population.csv \ + --id-column "User ID" \ + --method random \ + --sample-size 5 \ + --seed 20260707 \ + --out ./output ``` -``` text -Dataframe size (rows, columns): (100, 9) -Sample size: 5 -Sample: - Index Organization Id ... Industry Number of employees -79 80 cBa7EFe5D05Adaf ... Online Publishing 7805 -97 98 E7df80C60Abd7f9 ... Broadcast Media 236 -3 4 2bFC1Be8a4ce42f ... Automotive 921 -42 43 A2D89Ab9bCcAd4e ... Capital Markets / Hedge Fund / Private Equity 3816 -70 71 32BB9Ff4d939788 ... Wireless 6146 - -[5 rows x 9 columns] +## Stratified Counts Example + +```bash +python sampling/audit_sample.py \ + --input sampling/examples/changes_population.csv \ + --id-column "Change ID" \ + --method stratified \ + --stratify-column "Change Type" \ + --strata-counts "Normal=3,Emergency=2,Standard=2" \ + --seed 20260707 \ + --out ./output ``` -# `sample.html` +## Stratified Proportions Example + +```bash +python sampling/audit_sample.py \ + --input sampling/examples/changes_population.csv \ + --id-column "Change ID" \ + --method stratified \ + --stratify-column "Change Type" \ + --strata-proportions "Normal=0.50,Emergency=0.25,Standard=0.25" \ + --sample-size 8 \ + --seed 20260707 \ + --out ./output +``` + +## Validate-Only Example + +```bash +python sampling/audit_sample.py \ + --input sampling/examples/changes_population.csv \ + --id-column "Change ID" \ + --method validate-only \ + --filter "Status=Closed" \ + --out ./output +``` + +YAML config files are also supported. CLI arguments override config values: + +```bash +python sampling/audit_sample.py --config sampling/examples/stratified_config.yml +``` + +## Output Files + +Each run creates `sample_<YYYY-MM-DD>_<HHMMSS>` under the selected output +directory. + +- `sample.csv`: selected sample rows for random and stratified runs. +- `population_validated.csv`: population after filters, blank-ID handling, and + dedupe handling. +- `population_reconciliation.csv`: row-count tie-out metrics. +- `excluded_rows.csv`: rows removed by filters, blank-ID exclusion, or dedupe. +- `duplicate_ids.csv`: duplicate ID rows when duplicates are identified. +- `strata_summary.csv`: requested and actual counts for stratified runs. +- `methodology.txt`: audit workpaper narrative. +- `manifest.json`: machine-readable run metadata, input hash, options, and + output list. +- `run.log`: start time, warnings, errors, output folder, and status. + +## Reproducibility + +Provide `--seed` to make row selection reproducible for the same input and +options. If no seed is provided for sampling, the tool generates one, prints a +warning, and records the generated seed in `manifest.json` and +`methodology.txt`. -This is an interactive web page that allows users to submit their -population size, sample size(s), and generate a psuedo-random sample -list of numbers to use when sampling against their population. +The tool samples without replacement. Stratified sampling derives each stratum +seed from the base seed by adding the stratum index. -Samples can be re-generated and validated using the seed numbers -provided during the original generation. +## Limitations -<span class="spurious-link" -target="sample-html.png">*sample-html.png*</span> +Filters are exact matches in `Column=Value` form only. The tool does not yet +perform monetary-unit sampling or statistical sample-size calculation. new file mode 100644 @@ -0,0 +1,15 @@ +#!/usr/bin/env python3 +"""Command-line entrypoint for the audit sampling tool.""" + +from pathlib import Path +import sys + + +if __package__ is None or __package__ == "": + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from sampling.sampling_tool.cli import main + + +if __name__ == "__main__": + raise SystemExit(main()) new file mode 100644 @@ -0,0 +1,15 @@ +Change ID,Change Type,Risk Rating,Status,In Scope +CHG001,Normal,High,Closed,Yes +CHG002,Normal,Medium,Closed,Yes +CHG003,Normal,Low,Open,No +CHG004,Normal,Medium,Closed,Yes +CHG005,Normal,High,Closed,Yes +CHG006,Emergency,High,Closed,Yes +CHG007,Emergency,High,Closed,Yes +CHG008,Emergency,Medium,Open,Yes +CHG009,Emergency,Low,Closed,No +CHG010,Standard,Low,Closed,Yes +CHG011,Standard,Medium,Closed,Yes +CHG012,Standard,Low,Open,No +CHG013,Standard,High,Closed,Yes +CHG014,Standard,Medium,Closed,Yes new file mode 100644 @@ -0,0 +1,7 @@ +input: sampling/examples/changes_population.csv +id_column: Change ID +method: stratified +stratify_column: Change Type +strata_counts: Normal=3,Emergency=2,Standard=2 +seed: 20260707 +out: ./output new file mode 100644 @@ -0,0 +1,13 @@ +User ID,Name,Role,Status,In Scope +U001,Avery Chen,Administrator,Active,Yes +U002,Blake Rivera,Developer,Active,Yes +U003,Casey Morgan,Analyst,Inactive,No +U004,Drew Patel,Reviewer,Active,Yes +U005,Emerson Lee,Developer,Active,Yes +U006,Finley Brooks,Analyst,Inactive,No +U007,Gray Taylor,Administrator,Active,Yes +U008,Harper Smith,Developer,Active,Yes +U009,Indigo Clark,Reviewer,Active,No +U010,Jordan Kim,Analyst,Active,Yes +U011,Kai Johnson,Developer,Inactive,No +U012,Logan Davis,Reviewer,Active,Yes new file mode 100644 @@ -0,0 +1,3 @@ +"""Reusable audit sampling helpers.""" + +__version__ = "1.0.0" new file mode 100644 @@ -0,0 +1,367 @@ +"""CLI orchestration for the audit sampling tool.""" + +from __future__ import annotations + +from argparse import ArgumentParser, Namespace +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace +import sys + +import pandas as pd + +from .filters import apply_filters, parse_filters +from .io import AuditSamplingError, load_population, sha256_file, write_csv +from .manifest import build_manifest, write_manifest +from .methods import ( + ensure_seed, + largest_remainder_allocation, + random_sample, + stratified_sample, +) +from .reconciliation import build_reconciliation, build_strata_summary +from .reporting import RunLogger, build_methodology +from .validation import validate_and_prepare + + +def build_parser() -> ArgumentParser: + parser = ArgumentParser(description="Generate documented audit samples.") + parser.add_argument("--input") + parser.add_argument("--sheet") + parser.add_argument("--id-column") + parser.add_argument("--method", choices=["random", "stratified", "validate-only"]) + parser.add_argument("--sample-size", type=int) + parser.add_argument("--stratify-column") + parser.add_argument("--strata-counts") + parser.add_argument("--strata-proportions") + parser.add_argument("--seed", type=int) + parser.add_argument("--out") + parser.add_argument("--exclude-blank-id", action="store_true", default=None) + parser.add_argument("--dedupe-id", choices=["fail", "first", "last"]) + parser.add_argument("--filter", action="append", dest="filter_values") + parser.add_argument("--config") + parser.add_argument("--allow-shortfall", action="store_true", default=None) + return parser + + +def load_config(path: str | None) -> dict[str, object]: + if not path: + return {} + try: + import yaml + except ImportError as exc: + raise AuditSamplingError( + "YAML config support requires PyYAML. Install requirements.txt." + ) from exc + + config_path = Path(path) + with config_path.open("r", encoding="utf-8") as handle: + data = yaml.safe_load(handle) or {} + if not isinstance(data, dict): + raise AuditSamplingError("Config file must contain a YAML mapping.") + return data + + +def merge_options(args: Namespace, config: dict[str, object]) -> SimpleNamespace: + mapping = { + "input": "input", + "sheet": "sheet", + "id_column": "id_column", + "method": "method", + "sample_size": "sample_size", + "stratify_column": "stratify_column", + "strata_counts": "strata_counts", + "strata_proportions": "strata_proportions", + "seed": "seed", + "out": "out", + "exclude_blank_id": "exclude_blank_id", + "dedupe_id": "dedupe_id", + "allow_shortfall": "allow_shortfall", + } + merged: dict[str, object] = {} + for attr, key in mapping.items(): + cli_value = getattr(args, attr) + merged[attr] = cli_value if cli_value is not None else config.get(key) + + config_filters = config.get("filters", {}) + if isinstance(config_filters, list): + config_filters = parse_filters(config_filters) + if not isinstance(config_filters, dict): + raise AuditSamplingError("Config filters must be a mapping or list.") + cli_filters = parse_filters(args.filter_values) + merged["filters"] = {**config_filters, **cli_filters} + + merged["out"] = merged["out"] or "./output" + merged["dedupe_id"] = merged["dedupe_id"] or "fail" + merged["exclude_blank_id"] = bool(merged["exclude_blank_id"]) + merged["allow_shortfall"] = bool(merged["allow_shortfall"]) + if merged["input"] is None: + raise AuditSamplingError("--input is required unless provided by --config.") + if merged["method"] is None: + raise AuditSamplingError("--method is required unless provided by --config.") + if merged["method"] not in {"random", "stratified", "validate-only"}: + raise AuditSamplingError("--method must be random, stratified, or validate-only.") + if merged["sample_size"] is not None: + merged["sample_size"] = int(merged["sample_size"]) + if merged["seed"] is not None: + merged["seed"] = int(merged["seed"]) + return SimpleNamespace(**merged) + + +def create_run_dir(out_dir: Path, now: datetime) -> Path: + run_dir = out_dir / f"sample_{now.strftime('%Y-%m-%d_%H%M%S')}" + suffix = 1 + while True: + candidate = run_dir if suffix == 1 else out_dir / f"{run_dir.name}_{suffix}" + try: + candidate.mkdir(parents=True, exist_ok=False) + return candidate + except FileExistsError: + suffix += 1 + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + options = merge_options(args, load_config(args.config)) + run(options) + return 0 + except AuditSamplingError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + + +def run(options) -> Path: + now = datetime.now(timezone.utc) + timestamp = now.replace(microsecond=0).isoformat().replace("+00:00", "Z") + run_dir = create_run_dir(Path(options.out), now) + logger = RunLogger() + logger.log(f"start timestamp: {timestamp}") + logger.log(f"input path: {options.input}") + logger.log(f"method: {options.method}") + logger.log(f"output folder: {run_dir}") + print(f"Writing audit sample package to {run_dir}") + + output_files: list[str] = [] + input_path = Path(options.input) + input_hash = "" + source = pd.DataFrame() + filtered = pd.DataFrame() + validated = pd.DataFrame() + excluded_rows = pd.DataFrame() + duplicate_rows = pd.DataFrame() + sample = pd.DataFrame() + strata_rows: list[dict[str, object]] = [] + random_seed: int | None = options.seed + effective_id_column = None + id_column_omitted = False + blank_id_count = 0 + duplicate_id_count = 0 + + try: + input_hash = sha256_file(input_path) + source = load_population(input_path, options.sheet) + filtered, filter_excluded = apply_filters(source, options.filters) + validated = filtered.copy() + excluded_rows = filter_excluded.copy() + validation = validate_and_prepare(filtered, options) + validated = validation.population + excluded_rows = _concat_nonempty([filter_excluded, validation.excluded_rows]) + duplicate_rows = validation.duplicate_rows + effective_id_column = validation.effective_id_column + id_column_omitted = validation.id_column_omitted + blank_id_count = validation.blank_id_count + duplicate_id_count = validation.duplicate_id_count + for warning in validation.warnings: + print(f"WARNING: {warning}") + logger.warning(warning) + + if options.method in {"random", "stratified"}: + random_seed, generated = ensure_seed(options.seed) + if generated: + warning = f"No seed provided; generated seed {random_seed}." + print(f"WARNING: {warning}") + logger.warning(warning) + + if options.method == "random": + sample = random_sample(validated, options.sample_size, random_seed) + sample = add_sample_metadata(sample, options, random_seed, timestamp, None) + elif options.method == "stratified": + counts = validation.strata_counts + if validation.strata_proportions: + counts = largest_remainder_allocation( + options.sample_size, validation.strata_proportions + ) + sampled, strata_rows = stratified_sample( + validated, + options.stratify_column, + counts, + random_seed, + options.allow_shortfall, + ) + sample = add_sample_metadata( + sampled, options, random_seed, timestamp, options.stratify_column + ) + + _write_outputs( + run_dir, + options, + source, + validated, + excluded_rows, + duplicate_rows, + sample, + strata_rows, + output_files, + ) + logger.log("final status: success") + print("Audit sampling run completed.") + except AuditSamplingError as exc: + duplicate_rows = exc.artifacts.get("duplicate_rows", duplicate_rows) + logger.error(str(exc)) + print(f"ERROR: {exc}", file=sys.stderr) + _write_failure_outputs( + run_dir, + options, + source, + filtered, + excluded_rows, + duplicate_rows, + output_files, + ) + logger.log("final status: failed") + raise + finally: + reconciliation = build_reconciliation( + source_rows=len(source), + blank_id_count=blank_id_count, + duplicate_id_count=duplicate_id_count or len(duplicate_rows), + excluded_rows=len(excluded_rows), + validated_rows=len(validated), + requested_sample_size=_requested_sample_size(options, strata_rows), + final_sample_size=len(sample), + ) + write_csv(reconciliation, run_dir / "population_reconciliation.csv") + _track(output_files, "population_reconciliation.csv") + methodology = build_methodology( + input_file=input_path, + input_sheet=options.sheet, + input_sha256=input_hash, + source_row_count=len(source), + options=options, + effective_id_column=effective_id_column, + id_column_omitted=id_column_omitted, + duplicate_id_count=duplicate_id_count or len(duplicate_rows), + blank_id_count=blank_id_count, + sample_size_actual=len(sample), + random_seed=random_seed, + strata_summary=strata_rows, + ) + (run_dir / "methodology.txt").write_text(methodology) + _track(output_files, "methodology.txt") + manifest = build_manifest( + run_timestamp_utc=timestamp, + input_file=input_path, + input_sha256=input_hash, + input_sheet=options.sheet, + options=options, + effective_id_column=effective_id_column, + id_column_omitted=id_column_omitted, + sample_size_actual=len(sample), + random_seed=random_seed, + source_row_count=len(source), + validated_population_count=len(validated), + excluded_row_count=len(excluded_rows), + duplicate_id_count=duplicate_id_count or len(duplicate_rows), + blank_id_count=blank_id_count, + output_files=sorted(output_files + ["run.log", "manifest.json"]), + ) + write_manifest(manifest, run_dir / "manifest.json") + logger.write(run_dir / "run.log") + return run_dir + + +def add_sample_metadata( + sample: pd.DataFrame, + options, + seed: int, + timestamp: str, + stratum_column: str | None, +) -> pd.DataFrame: + source_columns = [column for column in sample.columns if column != "_audit_row_id"] + output = sample[source_columns].copy() + output.insert(0, "_selected_at_utc", timestamp) + output.insert(0, "_random_seed", seed) + output.insert(0, "_stratum", output[stratum_column] if stratum_column else "") + output.insert(0, "_selection_method", options.method) + output.insert(0, "_source_row_number", output.pop("_source_row_number")) + output.insert(0, "_sample_id", range(1, len(output) + 1)) + return output + + +def _write_outputs( + run_dir: Path, + options, + source: pd.DataFrame, + validated: pd.DataFrame, + excluded_rows: pd.DataFrame, + duplicate_rows: pd.DataFrame, + sample: pd.DataFrame, + strata_rows: list[dict[str, object]], + output_files: list[str], +) -> None: + write_csv(validated, run_dir / "population_validated.csv") + _track(output_files, "population_validated.csv") + if options.method in {"random", "stratified"}: + write_csv(sample, run_dir / "sample.csv") + _track(output_files, "sample.csv") + if not excluded_rows.empty: + write_csv(excluded_rows, run_dir / "excluded_rows.csv") + _track(output_files, "excluded_rows.csv") + if not duplicate_rows.empty: + write_csv(duplicate_rows, run_dir / "duplicate_ids.csv") + _track(output_files, "duplicate_ids.csv") + if options.method == "stratified": + write_csv(build_strata_summary(strata_rows), run_dir / "strata_summary.csv") + _track(output_files, "strata_summary.csv") + + +def _write_failure_outputs( + run_dir: Path, + options, + source: pd.DataFrame, + filtered: pd.DataFrame, + excluded_rows: pd.DataFrame, + duplicate_rows: pd.DataFrame, + output_files: list[str], +) -> None: + if not filtered.empty: + write_csv(filtered, run_dir / "population_validated.csv") + _track(output_files, "population_validated.csv") + if not excluded_rows.empty: + write_csv(excluded_rows, run_dir / "excluded_rows.csv") + _track(output_files, "excluded_rows.csv") + if not duplicate_rows.empty: + write_csv(duplicate_rows, run_dir / "duplicate_ids.csv") + _track(output_files, "duplicate_ids.csv") + + +def _concat_nonempty(frames: list[pd.DataFrame]) -> pd.DataFrame: + nonempty = [frame for frame in frames if frame is not None and not frame.empty] + if not nonempty: + return pd.DataFrame() + return pd.concat(nonempty, ignore_index=True) + + +def _track(output_files: list[str], filename: str) -> None: + if filename not in output_files: + output_files.append(filename) + + +def _requested_sample_size(options, strata_rows: list[dict[str, object]]) -> int | None: + if options.sample_size is not None: + return options.sample_size + if strata_rows: + return int(sum(row["Requested Sample Count"] for row in strata_rows)) + return None new file mode 100644 @@ -0,0 +1,44 @@ +"""Exact-match filter parsing and application.""" + +from __future__ import annotations + +import pandas as pd + +from .io import AuditSamplingError + + +def parse_filters(filter_values: list[str] | None) -> dict[str, str]: + parsed: dict[str, str] = {} + for value in filter_values or []: + if "=" not in value: + raise AuditSamplingError( + f"Invalid filter '{value}'. Expected format: Column=Value" + ) + column, expected = value.split("=", 1) + column = column.strip() + if not column: + raise AuditSamplingError( + f"Invalid filter '{value}'. Filter column cannot be blank." + ) + parsed[column] = expected.strip() + return parsed + + +def apply_filters( + population: pd.DataFrame, filters: dict[str, str] +) -> tuple[pd.DataFrame, pd.DataFrame]: + if not filters: + return population.copy(), population.iloc[0:0].copy() + + missing = [column for column in filters if column not in population.columns] + if missing: + raise AuditSamplingError(f"Filter column(s) not found: {', '.join(missing)}") + + keep_mask = pd.Series(True, index=population.index) + for column, expected in filters.items(): + keep_mask &= population[column].astype("string").fillna("") == expected + + excluded = population.loc[~keep_mask].copy() + if not excluded.empty: + excluded["_exclusion_reason"] = "Filtered out" + return population.loc[keep_mask].copy(), excluded new file mode 100644 @@ -0,0 +1,60 @@ +"""Input and output helpers for audit sampling.""" + +from __future__ import annotations + +from pathlib import Path +import hashlib + +import pandas as pd + + +SUPPORTED_EXCEL_SUFFIXES = {".xlsx", ".xls", ".xlsm"} + + +class AuditSamplingError(Exception): + """Raised when the sampling request cannot be completed.""" + + def __init__(self, message: str, **artifacts) -> None: + super().__init__(message) + self.artifacts = artifacts + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def load_population(input_path: Path, sheet: str | None = None) -> pd.DataFrame: + if not input_path.exists(): + raise AuditSamplingError(f"Input file does not exist: {input_path}") + + suffix = input_path.suffix.lower() + if suffix == ".csv": + frame = pd.read_csv(input_path) + elif suffix in SUPPORTED_EXCEL_SUFFIXES: + excel = pd.ExcelFile(input_path) + if sheet is None: + if len(excel.sheet_names) != 1: + names = ", ".join(excel.sheet_names) + raise AuditSamplingError( + "Excel workbook has multiple sheets. Provide --sheet. " + f"Available sheets: {names}" + ) + sheet = excel.sheet_names[0] + frame = pd.read_excel(input_path, sheet_name=sheet) + else: + supported = ".csv, .xlsx, .xls, .xlsm" + raise AuditSamplingError( + f"Unsupported input extension '{suffix}'. Supported: {supported}" + ) + + frame = frame.copy() + frame.insert(0, "_source_row_number", range(2, len(frame) + 2)) + return frame + + +def write_csv(frame: pd.DataFrame, path: Path) -> None: + frame.to_csv(path, index=False) new file mode 100644 @@ -0,0 +1,58 @@ +"""Manifest generation.""" + +from __future__ import annotations + +from pathlib import Path +import json + +from . import __version__ + + +def build_manifest( + *, + run_timestamp_utc: str, + input_file: Path, + input_sha256: str, + input_sheet: str | None, + options, + effective_id_column: str | None, + id_column_omitted: bool, + sample_size_actual: int, + random_seed: int | None, + source_row_count: int, + validated_population_count: int, + excluded_row_count: int, + duplicate_id_count: int, + blank_id_count: int, + output_files: list[str], +) -> dict[str, object]: + return { + "tool": "audit_sample", + "version": __version__, + "run_timestamp_utc": run_timestamp_utc, + "input_file": str(input_file), + "input_sha256": input_sha256, + "input_sheet": input_sheet, + "method": options.method, + "id_column": options.id_column, + "effective_id_column": effective_id_column, + "id_column_omitted": id_column_omitted, + "stratify_column": options.stratify_column, + "sample_size_requested": options.sample_size, + "sample_size_actual": sample_size_actual, + "random_seed": random_seed, + "filters": options.filters, + "dedupe_id": options.dedupe_id, + "exclude_blank_id": options.exclude_blank_id, + "allow_shortfall": options.allow_shortfall, + "source_row_count": source_row_count, + "validated_population_count": validated_population_count, + "excluded_row_count": excluded_row_count, + "duplicate_id_count": duplicate_id_count, + "blank_id_count": blank_id_count, + "output_files": output_files, + } + + +def write_manifest(manifest: dict[str, object], path: Path) -> None: + path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") new file mode 100644 @@ -0,0 +1,125 @@ +"""Sampling methods.""" + +from __future__ import annotations + +import math +import random + +import pandas as pd + +from .io import AuditSamplingError + + +def ensure_seed(seed: int | None) -> tuple[int, bool]: + if seed is not None: + return int(seed), False + return random.SystemRandom().randint(1, 2_147_483_647), True + + +def parse_key_ints(value: str | None, label: str) -> dict[str, int]: + if not value: + return {} + parsed: dict[str, int] = {} + for part in value.split(","): + if "=" not in part: + raise AuditSamplingError(f"Invalid {label} entry '{part}'. Use Name=Count.") + key, raw_count = part.split("=", 1) + key = key.strip() + try: + count = int(raw_count.strip()) + except ValueError as exc: + raise AuditSamplingError( + f"Invalid {label} count for '{key}': {raw_count}" + ) from exc + if count <= 0: + raise AuditSamplingError(f"{label} count for '{key}' must be positive.") + parsed[key] = count + return parsed + + +def parse_key_floats(value: str | None, label: str) -> dict[str, float]: + if not value: + return {} + parsed: dict[str, float] = {} + for part in value.split(","): + if "=" not in part: + raise AuditSamplingError( + f"Invalid {label} entry '{part}'. Use Name=Proportion." + ) + key, raw_proportion = part.split("=", 1) + key = key.strip() + try: + proportion = float(raw_proportion.strip()) + except ValueError as exc: + raise AuditSamplingError( + f"Invalid {label} proportion for '{key}': {raw_proportion}" + ) from exc + if proportion <= 0: + raise AuditSamplingError( + f"{label} proportion for '{key}' must be positive." + ) + parsed[key] = proportion + if not math.isclose(sum(parsed.values()), 1.0, rel_tol=1e-9, abs_tol=1e-9): + raise AuditSamplingError(f"{label} proportions must sum to 1.0.") + return parsed + + +def largest_remainder_allocation( + sample_size: int, proportions: dict[str, float] +) -> dict[str, int]: + raw = { + stratum: { + "floor": math.floor(sample_size * proportion), + "remainder": sample_size * proportion + - math.floor(sample_size * proportion), + } + for stratum, proportion in proportions.items() + } + allocation = {stratum: values["floor"] for stratum, values in raw.items()} + remaining = sample_size - sum(allocation.values()) + ranked = sorted( + raw, + key=lambda stratum: (-raw[stratum]["remainder"], stratum), + ) + for stratum in ranked[:remaining]: + allocation[stratum] += 1 + return allocation + + +def random_sample(population: pd.DataFrame, sample_size: int, seed: int) -> pd.DataFrame: + return population.sample(n=sample_size, random_state=seed) + + +def stratified_sample( + population: pd.DataFrame, + stratify_column: str, + counts: dict[str, int], + seed: int, + allow_shortfall: bool, +) -> tuple[pd.DataFrame, list[dict[str, object]]]: + samples: list[pd.DataFrame] = [] + summary: list[dict[str, object]] = [] + for index, (stratum, requested) in enumerate(counts.items()): + stratum_population = population[population[stratify_column] == stratum] + actual_count = min(requested, len(stratum_population)) + if actual_count < requested and not allow_shortfall: + raise AuditSamplingError( + f"Stratum '{stratum}' has {len(stratum_population)} rows; " + f"requested {requested}. Use --allow-shortfall to continue." + ) + if actual_count: + sampled = stratum_population.sample(n=actual_count, random_state=seed + index) + samples.append(sampled) + summary.append( + { + "Stratum": stratum, + "Population Count": len(stratum_population), + "Requested Sample Count": requested, + "Actual Sample Count": actual_count, + "Shortfall": requested - actual_count, + } + ) + + if samples: + return pd.concat(samples), summary + return population.iloc[0:0].copy(), summary new file mode 100644 @@ -0,0 +1,41 @@ +"""Reconciliation output helpers.""" + +from __future__ import annotations + +import pandas as pd + + +def build_reconciliation( + source_rows: int, + blank_id_count: int, + duplicate_id_count: int, + excluded_rows: int, + validated_rows: int, + requested_sample_size: int | None, + final_sample_size: int, +) -> pd.DataFrame: + unsampled = validated_rows - final_sample_size + rows = [ + ("Source rows", source_rows), + ("Rows with blank ID", blank_id_count), + ("Duplicate IDs", duplicate_id_count), + ("Excluded rows", excluded_rows), + ("Validated population rows", validated_rows), + ("Requested sample size", "" if requested_sample_size is None else requested_sample_size), + ("Final sample size", final_sample_size), + ("Unsampled population rows", unsampled), + ] + return pd.DataFrame(rows, columns=["Metric", "Value"]) + + +def build_strata_summary(summary_rows: list[dict[str, object]]) -> pd.DataFrame: + return pd.DataFrame( + summary_rows, + columns=[ + "Stratum", + "Population Count", + "Requested Sample Count", + "Actual Sample Count", + "Shortfall", + ], + ) new file mode 100644 @@ -0,0 +1,87 @@ +"""Human-readable reporting outputs.""" + +from __future__ import annotations + +from pathlib import Path + + +class RunLogger: + def __init__(self) -> None: + self.lines: list[str] = [] + + def log(self, message: str) -> None: + self.lines.append(message) + + def warning(self, message: str) -> None: + self.log(f"WARNING: {message}") + + def error(self, message: str) -> None: + self.log(f"ERROR: {message}") + + def write(self, path: Path) -> None: + path.write_text("\n".join(self.lines) + "\n") + + +def build_methodology( + *, + input_file: Path, + input_sheet: str | None, + input_sha256: str, + source_row_count: int, + options, + effective_id_column: str | None, + id_column_omitted: bool, + duplicate_id_count: int, + blank_id_count: int, + sample_size_actual: int, + random_seed: int | None, + strata_summary: list[dict[str, object]] | None, +) -> str: + lines = [ + "Audit Sampling Methodology", + "", + f"Source file: {input_file}", + f"Source sheet: {input_sheet or ''}", + f"Input SHA-256: {input_sha256}", + f"Source row count: {source_row_count}", + ] + if id_column_omitted: + lines.append( + "ID column: omitted; _audit_row_id was generated from _source_row_number." + ) + else: + lines.append(f"ID column: {options.id_column}") + lines.extend( + [ + f"Effective ID column: {effective_id_column or ''}", + f"Duplicate ID rows identified: {duplicate_id_count}", + f"Blank ID rows identified: {blank_id_count}", + f"Blank ID handling: {'excluded' if options.exclude_blank_id else 'retained'}", + f"Filters applied: {options.filters or {}}", + f"Sampling method: {options.method}", + f"Sample size requested: {options.sample_size or ''}", + f"Sample size selected: {sample_size_actual}", + f"Random seed used: {random_seed or ''}", + ] + ) + if options.method == "stratified": + lines.append(f"Stratification column: {options.stratify_column}") + lines.append(f"Strata counts: {options.strata_counts or ''}") + lines.append(f"Strata proportions: {options.strata_proportions or ''}") + if strata_summary: + lines.append("Strata summary:") + for row in strata_summary: + lines.append( + " " + f"{row['Stratum']}: population={row['Population Count']}, " + f"requested={row['Requested Sample Count']}, " + f"actual={row['Actual Sample Count']}, " + f"shortfall={row['Shortfall']}" + ) + lines.extend( + [ + "Sampling was performed without replacement.", + "Selected rows retain source row numbers and original source fields.", + ] + ) + return "\n".join(lines) + "\n" new file mode 100644 @@ -0,0 +1,157 @@ +"""Validation and population preparation.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pandas as pd + +from .io import AuditSamplingError +from .methods import parse_key_floats, parse_key_ints + + +@dataclass +class ValidationResult: + population: pd.DataFrame + excluded_rows: pd.DataFrame + duplicate_rows: pd.DataFrame + effective_id_column: str + id_column_omitted: bool + blank_id_count: int + duplicate_id_count: int + warnings: list[str] + strata_counts: dict[str, int] + strata_proportions: dict[str, float] + + +def blank_id_mask(series: pd.Series) -> pd.Series: + return series.isna() | (series.astype("string").fillna("").str.strip() == "") + + +def validate_and_prepare(population: pd.DataFrame, options) -> ValidationResult: + warnings: list[str] = [] + excluded_parts: list[pd.DataFrame] = [] + working = population.copy() + + if options.id_column: + if options.id_column not in working.columns: + raise AuditSamplingError(f"ID column not found: {options.id_column}") + effective_id_column = options.id_column + id_column_omitted = False + else: + effective_id_column = "_audit_row_id" + id_column_omitted = True + working[effective_id_column] = working["_source_row_number"] + warnings.append( + "--id-column omitted; using _source_row_number as generated _audit_row_id." + ) + + if options.stratify_column and options.stratify_column not in working.columns: + raise AuditSamplingError( + f"Stratification column not found: {options.stratify_column}" + ) + + id_blank_mask = blank_id_mask(working[effective_id_column]) + blank_id_count = int(id_blank_mask.sum()) + if blank_id_count: + if options.exclude_blank_id: + excluded = working.loc[id_blank_mask].copy() + excluded["_exclusion_reason"] = "Blank ID" + excluded_parts.append(excluded) + working = working.loc[~id_blank_mask].copy() + else: + warnings.append( + f"{blank_id_count} row(s) have blank IDs and were retained." + ) + + nonblank_ids = ~blank_id_mask(working[effective_id_column]) + duplicate_mask = working.loc[nonblank_ids, effective_id_column].duplicated( + keep=False + ) + duplicate_rows = working.loc[nonblank_ids].loc[duplicate_mask].copy() + duplicate_id_count = int(len(duplicate_rows)) + if duplicate_id_count: + if options.dedupe_id == "fail": + raise AuditSamplingError( + f"Duplicate IDs found in '{effective_id_column}'. " + "See duplicate_ids.csv.", + duplicate_rows=duplicate_rows, + ) + keep = "first" if options.dedupe_id == "first" else "last" + drop_mask = working[effective_id_column].duplicated(keep=keep) & ~blank_id_mask( + working[effective_id_column] + ) + excluded = working.loc[drop_mask].copy() + excluded["_exclusion_reason"] = f"Duplicate ID removed by dedupe={keep}" + excluded_parts.append(excluded) + working = working.loc[~drop_mask].copy() + warnings.append( + f"{len(excluded)} duplicate ID row(s) removed using dedupe={keep}." + ) + + if options.sample_size is not None and options.sample_size <= 0: + raise AuditSamplingError("--sample-size must be a positive integer.") + needs_sample_size = options.method == "random" or ( + options.method == "stratified" and bool(options.strata_proportions) + ) + if needs_sample_size: + if options.sample_size is None: + raise AuditSamplingError(f"--sample-size is required for {options.method}.") + if options.sample_size > len(working) and not ( + options.method == "stratified" and options.allow_shortfall + ): + raise AuditSamplingError( + "--sample-size cannot exceed the validated population size." + ) + + strata_counts: dict[str, int] = {} + strata_proportions: dict[str, float] = {} + if options.method == "stratified": + if not options.stratify_column: + raise AuditSamplingError("--stratify-column is required for stratified.") + has_counts = bool(options.strata_counts) + has_proportions = bool(options.strata_proportions) + if has_counts == has_proportions: + raise AuditSamplingError( + "Use exactly one of --strata-counts or --strata-proportions." + ) + strata_counts = parse_key_ints(options.strata_counts, "strata") + strata_proportions = parse_key_floats(options.strata_proportions, "strata") + requested_strata = set(strata_counts or strata_proportions) + actual_strata = set(working[options.stratify_column].dropna().astype(str)) + missing = sorted(requested_strata - actual_strata) + if missing: + raise AuditSamplingError( + "Requested strata not found in population: " + ", ".join(missing) + ) + + if strata_counts: + _validate_stratum_counts_fit(working, options, strata_counts) + + if excluded_parts: + excluded_rows = pd.concat(excluded_parts, ignore_index=True) + else: + excluded_rows = working.iloc[0:0].copy() + + return ValidationResult( + population=working, + excluded_rows=excluded_rows, + duplicate_rows=duplicate_rows, + effective_id_column=effective_id_column, + id_column_omitted=id_column_omitted, + blank_id_count=blank_id_count, + duplicate_id_count=duplicate_id_count, + warnings=warnings, + strata_counts=strata_counts, + strata_proportions=strata_proportions, + ) + + +def _validate_stratum_counts_fit(population, options, counts: dict[str, int]) -> None: + for stratum, requested in counts.items(): + available = int((population[options.stratify_column] == stratum).sum()) + if requested > available and not options.allow_shortfall: + raise AuditSamplingError( + f"Stratum '{stratum}' has {available} rows; requested {requested}. " + "Use --allow-shortfall to continue." + ) new file mode 100644 @@ -0,0 +1,70 @@ +from types import SimpleNamespace + +import pandas as pd + +from sampling.sampling_tool.cli import run + + +def _write_population(path, rows=20): + frame = pd.DataFrame( + { + "ID": [f"ID{i:03d}" for i in range(rows)], + "Status": ["Closed"] * rows, + } + ) + frame.to_csv(path, index=False) + + +def _options(input_path, out_path, seed=123, sample_size=5, method="random", **kwargs): + values = { + "input": str(input_path), + "sheet": None, + "id_column": "ID", + "method": method, + "sample_size": sample_size, + "stratify_column": None, + "strata_counts": None, + "strata_proportions": None, + "seed": seed, + "out": str(out_path), + "exclude_blank_id": False, + "dedupe_id": "fail", + "filters": {}, + "allow_shortfall": False, + } + values.update(kwargs) + return SimpleNamespace(**values) + + +def test_random_sample_returns_correct_size(tmp_path): + source = tmp_path / "population.csv" + _write_population(source) + + run_dir = run(_options(source, tmp_path / "out")) + + sample = pd.read_csv(run_dir / "sample.csv") + assert len(sample) == 5 + + +def test_same_seed_returns_same_selected_ids(tmp_path): + source = tmp_path / "population.csv" + _write_population(source) + + first = run(_options(source, tmp_path / "out1", seed=20260707)) + second = run(_options(source, tmp_path / "out2", seed=20260707)) + + first_ids = pd.read_csv(first / "sample.csv")["ID"].tolist() + second_ids = pd.read_csv(second / "sample.csv")["ID"].tolist() + assert first_ids == second_ids + + +def test_different_seed_can_return_different_selected_ids(tmp_path): + source = tmp_path / "population.csv" + _write_population(source) + + first = run(_options(source, tmp_path / "out1", seed=1)) + second = run(_options(source, tmp_path / "out2", seed=2)) + + first_ids = pd.read_csv(first / "sample.csv")["ID"].tolist() + second_ids = pd.read_csv(second / "sample.csv")["ID"].tolist() + assert first_ids != second_ids new file mode 100644 @@ -0,0 +1,44 @@ +from types import SimpleNamespace + +import pandas as pd + +from sampling.sampling_tool.cli import run + + +def test_reconciliation_math_ties_out(tmp_path): + source = tmp_path / "population.csv" + pd.DataFrame( + {"ID": ["A", "B", "C", "D"], "Status": ["Closed", "Open", "Closed", "Open"]} + ).to_csv(source, index=False) + options = SimpleNamespace( + input=str(source), + sheet=None, + id_column="ID", + method="random", + sample_size=1, + stratify_column=None, + strata_counts=None, + strata_proportions=None, + seed=7, + out=str(tmp_path / "out"), + exclude_blank_id=False, + dedupe_id="fail", + filters={"Status": "Closed"}, + allow_shortfall=False, + ) + + run_dir = run(options) + recon = pd.read_csv(run_dir / "population_reconciliation.csv") + metrics = dict(zip(recon["Metric"], recon["Value"])) + + assert int(metrics["Source rows"]) == 4 + assert int(metrics["Excluded rows"]) == 2 + assert int(metrics["Validated population rows"]) == 2 + assert int(metrics["Final sample size"]) == 1 + assert int(metrics["Unsampled population rows"]) == 1 + assert int(metrics["Source rows"]) == ( + int(metrics["Validated population rows"]) + int(metrics["Excluded rows"]) + ) + assert int(metrics["Validated population rows"]) == ( + int(metrics["Final sample size"]) + int(metrics["Unsampled population rows"]) + ) new file mode 100644 @@ -0,0 +1,66 @@ +from types import SimpleNamespace + +import pandas as pd + +from sampling.sampling_tool.cli import run + + +def _write_population(path): + frame = pd.DataFrame( + { + "ID": [f"ID{i:03d}" for i in range(12)], + "Type": ["A"] * 5 + ["B"] * 4 + ["C"] * 3, + } + ) + frame.to_csv(path, index=False) + + +def _options(input_path, out_path, **kwargs): + values = { + "input": str(input_path), + "sheet": None, + "id_column": "ID", + "method": "stratified", + "sample_size": None, + "stratify_column": "Type", + "strata_counts": "A=2,B=2,C=1", + "strata_proportions": None, + "seed": 50, + "out": str(out_path), + "exclude_blank_id": False, + "dedupe_id": "fail", + "filters": {}, + "allow_shortfall": False, + } + values.update(kwargs) + return SimpleNamespace(**values) + + +def test_stratified_counts_select_exact_requested_counts(tmp_path): + source = tmp_path / "population.csv" + _write_population(source) + + run_dir = run(_options(source, tmp_path / "out")) + + counts = pd.read_csv(run_dir / "sample.csv")["Type"].value_counts().to_dict() + assert counts == {"A": 2, "B": 2, "C": 1} + + +def test_stratified_proportions_use_largest_remainder(tmp_path): + source = tmp_path / "population.csv" + _write_population(source) + + run_dir = run( + _options( + source, + tmp_path / "out", + sample_size=7, + strata_counts=None, + strata_proportions="A=0.50,B=0.30,C=0.20", + ) + ) + + sample = pd.read_csv(run_dir / "sample.csv") + counts = sample["Type"].value_counts().to_dict() + assert len(sample) == 7 + assert counts == {"A": 4, "B": 2, "C": 1} new file mode 100644 @@ -0,0 +1,81 @@ +from types import SimpleNamespace + +import pandas as pd +import pytest + +from sampling.sampling_tool.cli import run +from sampling.sampling_tool.io import AuditSamplingError + + +def _options(input_path, out_path, **kwargs): + values = { + "input": str(input_path), + "sheet": None, + "id_column": "ID", + "method": "validate-only", + "sample_size": None, + "stratify_column": None, + "strata_counts": None, + "strata_proportions": None, + "seed": None, + "out": str(out_path), + "exclude_blank_id": False, + "dedupe_id": "fail", + "filters": {}, + "allow_shortfall": False, + } + values.update(kwargs) + return SimpleNamespace(**values) + + +def test_duplicate_ids_fail_by_default_and_write_duplicate_file(tmp_path): + source = tmp_path / "population.csv" + pd.DataFrame({"ID": ["A", "A", "B"], "Status": ["Closed"] * 3}).to_csv( + source, index=False + ) + + with pytest.raises(AuditSamplingError): + run(_options(source, tmp_path / "out")) + + run_dir = next((tmp_path / "out").glob("sample_*")) + duplicates = pd.read_csv(run_dir / "duplicate_ids.csv") + assert duplicates["ID"].tolist() == ["A", "A"] + + +def test_blank_ids_are_excluded_when_requested(tmp_path): + source = tmp_path / "population.csv" + pd.DataFrame({"ID": ["A", "", "B"], "Status": ["Closed"] * 3}).to_csv( + source, index=False + ) + + run_dir = run(_options(source, tmp_path / "out", exclude_blank_id=True)) + + validated = pd.read_csv(run_dir / "population_validated.csv") + excluded = pd.read_csv(run_dir / "excluded_rows.csv") + assert len(validated) == 2 + assert excluded["_exclusion_reason"].tolist() == ["Blank ID"] + + +def test_filters_reduce_population_and_write_excluded_rows(tmp_path): + source = tmp_path / "population.csv" + pd.DataFrame( + {"ID": ["A", "B", "C"], "Status": ["Closed", "Open", "Closed"]} + ).to_csv(source, index=False) + + run_dir = run(_options(source, tmp_path / "out", filters={"Status": "Closed"})) + + validated = pd.read_csv(run_dir / "population_validated.csv") + excluded = pd.read_csv(run_dir / "excluded_rows.csv") + assert validated["ID"].tolist() == ["A", "C"] + assert excluded["ID"].tolist() == ["B"] + + +def test_validate_only_writes_no_sample_csv(tmp_path): + source = tmp_path / "population.csv" + pd.DataFrame({"ID": ["A", "B", "C"], "Status": ["Closed"] * 3}).to_csv( + source, index=False + ) + + run_dir = run(_options(source, tmp_path / "out")) + + assert not (run_dir / "sample.csv").exists()