CSV Statistical Analyzer (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: CSV Statistical Analyzer

Write a program that reads a CSV file and performs comprehensive statistical analysis on every numeric column. It should handle real-world messy data — missing values, mixed types, malformed rows — and produce both a human-readable console report and a machine-readable JSON output.

FUNCTIONAL REQUIREMENTS:
- Accept a CSV file path as a command-line argument
- Auto-detect which columns are numeric vs categorical
- For each numeric column compute: mean, median, standard deviation, variance, min, max, 25th/50th/75th percentiles, and non-missing value count
- Detect outliers using the IQR method (values below Q1 - 1.5*IQR or above Q3 + 1.5*IQR) and list them per column
- For each categorical column compute: unique count, most frequent value, and top 10 value frequencies
- Print a formatted summary table to the console with aligned columns
- Save the complete analysis to report.json including all stats, outlier details, and column type classifications
- If no input file is given, generate a sample CSV with at least 200 rows across 5 numeric and 2 categorical columns, then analyze it
- Handle gracefully: empty files, header-only files, columns with all missing values, single-row files, quoted fields containing commas

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

#!/usr/bin/env python3
"""CSV statistical analyzer using pandas and SciPy.

The tool ingests a CSV file, computes descriptive statistics, and reports them as JSON.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Dict, Iterable, List, Sequence

import numpy as np
import pandas as pd
from scipy import stats

DEFAULT_PERCENTILES = [0.05, 0.25, 0.5, 0.75, 0.95]


def log(message: str) -> None:
    """Write a diagnostic message to stderr."""
    print(message, file=sys.stderr)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Analyze CSV numeric columns using pandas and SciPy statistics."
    )
    parser.add_argument(
        "csv_path",
        help="Path to the CSV file that will be analyzed.",
    )
    parser.add_argument(
        "--columns",
        nargs="+",
        help="Optional subset of columns to analyze. Defaults to every numeric column.",
    )
    parser.add_argument(
        "--percentiles",
        nargs="+",
        type=float,
        help="Percentile values expressed between 0 and 1 (e.g. 0.9 for the 90th percentile).",
    )
    parser.add_argument(
        "--output",
        help="Optional path to write the JSON report. Stdout is used when not provided.",
    )
    parser.add_argument(
        "--json-indent",
        type=int,
        default=2,
        help="Indentation level for the JSON report (default: 2).",
    )
    parser.add_argument(
        "--z-threshold",
        type=float,
        default=3.0,
        help="Z-score threshold for outlier flagging (default: 3.0).",
    )
    parser.add_argument(
        "--iqr-factor",
        type=float,
        default=1.5,
        help="IQR multiplier used for Tukey-style outlier detection (default: 1.5).",
    )
    return parser.parse_args()


def sanitize_percentiles(values: Sequence[float] | None) -> List[float]:
    """Validate percentile values and return a sorted unique list."""
    if not values:
        return DEFAULT_PERCENTILES
    sanitized: List[float] = []
    for value in values:
        if not 0 <= value <= 1:
            raise ValueError(f"Percentile {value} is outside the inclusive 0-1 range.")
        sanitized.append(float(value))
    ordered = sorted(set(sanitized))
    if not ordered:
        raise ValueError("At least one percentile must be supplied.")
    return ordered


def load_dataframe(csv_path: str) -> pd.DataFrame:
    """Read the CSV file into a pandas DataFrame."""
    path = Path(csv_path).expanduser()
    if not path.exists():
        raise FileNotFoundError(f"CSV file '{path}' was not found.")
    return pd.read_csv(path)


def determine_columns(df: pd.DataFrame, requested: Sequence[str] | None) -> List[str]:
    """Return the columns that will be analyzed."""
    if requested:
        missing = [column for column in requested if column not in df.columns]
        if missing:
            raise ValueError(f"Columns not present in CSV: {', '.join(missing)}")
        ordered_unique = list(dict.fromkeys(requested))
        return ordered_unique
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    if not numeric_cols:
        raise ValueError("No numeric columns were detected in the CSV file.")
    return numeric_cols


def ensure_numeric(series: pd.Series) -> pd.Series:
    """Ensure the series is numeric, coercing values when necessary."""
    if pd.api.types.is_numeric_dtype(series):
        return series
    log(f"Column '{series.name}' is not numeric; attempting to coerce values.")
    return pd.to_numeric(series, errors="coerce")


def format_float(value: float | np.floating | np.integer | None) -> float | None:
    """Convert numpy scalars to builtin floats for JSON serialization."""
    if value is None:
        return None
    if isinstance(value, (np.floating, np.integer)):
        value = float(value)
    if isinstance(value, float) and np.isnan(value):
        return None
    return value


def compute_percentiles(series: pd.Series, percentiles: Sequence[float]) -> Dict[str, float]:
    """Compute percentile values for a pandas Series."""
    if series.empty:
        return {}
    values = np.quantile(series.to_numpy(dtype=float), percentiles)
    result: Dict[str, float] = {}
    for pct, val in zip(percentiles, values):
        label = f"{pct * 100:.2f}"
        label = label.rstrip("0").rstrip(".") + "%"
        result[label] = float(val)
    return result


def normalize_index(values: Iterable[object]) -> List[object]:
    """Convert pandas index values into JSON-serializable primitives."""
    normalized: List[object] = []
    for value in values:
        if isinstance(value, (np.integer, int)):
            normalized.append(int(value))
        else:
            normalized.append(str(value))
    return normalized


def detect_outliers(series: pd.Series, z_thresh: float, iqr_factor: float) -> Dict[str, object]:
    """Identify outliers using both z-score and IQR rules."""
    clean = series.dropna()
    if clean.empty:
        return {
            "z_score": {"threshold": z_thresh, "count": 0, "indices": []},
            "iqr": {
                "factor": iqr_factor,
                "count": 0,
                "indices": [],
                "lower_bound": None,
                "upper_bound": None,
            },
        }

    numeric = clean.to_numpy(dtype=float)
    z_scores = stats.zscore(numeric, nan_policy="omit")
    if np.isnan(z_scores).all():
        z_mask = np.zeros_like(numeric, dtype=bool)
    else:
        z_mask = np.abs(z_scores) > z_thresh
    z_indices = normalize_index(clean.index[z_mask].tolist())

    q1, q3 = np.quantile(numeric, [0.25, 0.75])
    iqr = q3 - q1
    lower = q1 - iqr_factor * iqr
    upper = q3 + iqr_factor * iqr
    if iqr == 0:
        iqr_mask = (numeric < q1) | (numeric > q3)
    else:
        iqr_mask = (numeric < lower) | (numeric > upper)
    iqr_indices = normalize_index(clean.index[iqr_mask].tolist())

    return {
        "z_score": {"threshold": z_thresh, "count": len(z_indices), "indices": z_indices},
        "iqr": {
            "factor": iqr_factor,
            "count": len(iqr_indices),
            "indices": iqr_indices,
            "lower_bound": format_float(lower),
            "upper_bound": format_float(upper),
        },
    }


def summarize_column(
    series: pd.Series,
    percentiles: Sequence[float],
    z_thresh: float,
    iqr_factor: float,
) -> Dict[str, object]:
    """Summarize a numeric column."""
    numeric = ensure_numeric(series)
    clean = numeric.dropna()
    values = clean.to_numpy(dtype=float)

    summary: Dict[str, object] = {
        "count": int(clean.count()),
        "missing": int(series.isna().sum()),
        "mean": format_float(clean.mean()),
        "median": format_float(clean.median()),
        "std_dev": format_float(clean.std(ddof=1)),
        "variance": format_float(clean.var(ddof=1)),
        "min": format_float(clean.min()),
        "max": format_float(clean.max()),
        "percentiles": compute_percentiles(clean, percentiles),
    }

    if values.size >= 3:
        summary["skewness"] = format_float(stats.skew(values, bias=False))
    else:
        summary["skewness"] = None

    if values.size >= 4:
        summary["kurtosis"] = format_float(stats.kurtosis(values, fisher=True, bias=False))
    else:
        summary["kurtosis"] = None

    summary["outliers"] = detect_outliers(clean, z_thresh, iqr_factor)
    return summary


def analyze_dataframe(
    df: pd.DataFrame,
    csv_path: str,
    columns: Sequence[str],
    percentiles: Sequence[float],
    z_thresh: float,
    iqr_factor: float,
) -> Dict[str, object]:
    """Generate statistics for each requested column."""
    stats_map: Dict[str, Dict[str, object]] = {}
    for column in columns:
        stats_map[column] = summarize_column(df[column], percentiles, z_thresh, iqr_factor)

    return {
        "file": str(Path(csv_path).expanduser().resolve()),
        "row_count": int(len(df)),
        "columns_analyzed": list(columns),
        "statistics": stats_map,
    }


def write_report(report: Dict[str, object], destination: str | None, indent: int) -> None:
    """Emit the JSON report either to disk or stdout."""
    payload = json.dumps(report, indent=indent)
    if destination:
        Path(destination).write_text(payload + "\n", encoding="utf-8")
    else:
        print(payload)


def main() -> None:
    args = parse_args()
    try:
        percentiles = sanitize_percentiles(args.percentiles)
        df = load_dataframe(args.csv_path)
        columns = determine_columns(df, args.columns)
        report = analyze_dataframe(
            df=df,
            csv_path=args.csv_path,
            columns=columns,
            percentiles=percentiles,
            z_thresh=args.z_threshold,
            iqr_factor=args.iqr_factor,
        )
        write_report(report, args.output, args.json_indent)
    except Exception as exc:  # pragma: no cover - CLI safety net
        log(f"Error: {exc}")
        sys.exit(1)


if __name__ == "__main__":
    main()

# CSV Statistical Analyzer

This tool loads a CSV file with pandas, computes descriptive statistics, derives configurable percentiles, and flags potential outliers using SciPy. The results are emitted as a JSON document that can be consumed by downstream automation or reporting scripts.

## Dependencies

- Python 3.10+
- pandas
- SciPy
- NumPy

Install the dependencies with:

```
pip install -r requirements.txt
```

## Usage

Run the analyzer by pointing it to a CSV file. By default every numeric column is analyzed, but you can constrain the run to specific columns and tweak the percentile and outlier configuration.

```
python analyzer.py data.csv --output report.json --percentiles 0.1 0.5 0.9 --columns price quantity
```

Key arguments:

- `--columns`: names of the columns to analyze. Omit to auto-detect numeric columns.
- `--percentiles`: floating point values between 0 and 1 (e.g., 0.95 for the 95th percentile).
- `--output`: path to write the JSON report. Defaults to stdout.
- `--z-threshold`: absolute z-score used when flagging outliers.
- `--iqr-factor`: Tukey IQR multiplier for the second outlier detector.

The generated JSON bundles per-column stats, percentile tables, SciPy skewness/kurtosis metrics, and indices of rows that violate the outlier thresholds. You can feed the JSON into dashboards, alerting pipelines, or further Python notebooks for downstream exploration.

pandas==2.2.2
scipy==1.13.1
numpy==1.26.4