Data Profiling Tool (python, written by Claude Code)

Claude Code 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: Data Profiling Tool

Write a program that profiles tabular datasets by analyzing column types, distributions, missing values, correlations, and data quality issues, producing a comprehensive data quality report.

FUNCTIONAL REQUIREMENTS:
- Accept a CSV or JSON data file path as a command-line argument
- Auto-detect column data types: numeric (integer vs float), string, boolean, date/timestamp, and categorical (low cardinality strings)
- For numeric columns: compute min, max, mean, median, standard deviation, skewness, and percentiles (25th, 50th, 75th, 95th, 99th)
- For string columns: compute min/max/average length, most common values (top 10), and unique count
- For all columns: count total values, missing/null values, missing percentage, and unique value count
- Detect potential data quality issues: columns that are entirely null, columns with a single unique value, numeric columns with extreme outliers (beyond 4 standard deviations), and string columns that look like misclassified numbers or dates
- Compute a pairwise correlation matrix for all numeric columns
- Print a formatted summary report to console showing key statistics per column
- Save the full profiling report as a JSON file with --output flag (default: data_profile.json)
- If no input file is given, generate a sample dataset with 1000 rows across at least 8 columns of mixed types including intentional quality issues, then profile it
- Handle files with inconsistent delimiters or encoding issues gracefully

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
"""
Data Profiling Tool - Trial 1
Dependencies: pandas, scipy, matplotlib
Profiles tabular datasets: column types, distributions, missing values,
correlations, data quality report.
"""

import sys
import os
import json
import math
import io
import tempfile
from datetime import datetime

import pandas as pd
import scipy.stats as stats
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt


def generate_sample_dataset(path="sample_data.csv"):
    """Generate a messy sample dataset for demonstration."""
    import random
    random.seed(42)

    rows = 200
    data = {
        "id": list(range(1, rows + 1)),
        "name": [random.choice(["Alice", "Bob", "Charlie", "Diana", "Eve",
                                 None, "Frank", "Grace", "", "Heidi"])
                 for _ in range(rows)],
        "age": [random.choice([random.randint(18, 80), None, -1, 999])
                for _ in range(rows)],
        "salary": [random.choice([round(random.uniform(25000, 150000), 2),
                                   None, 0, -500])
                   for _ in range(rows)],
        "department": [random.choice(["Engineering", "Sales", "HR",
                                       "Marketing", None, "Engineering",
                                       "Sales", ""])
                       for _ in range(rows)],
        "score": [random.choice([round(random.uniform(0, 100), 1), None,
                                  float("nan")])
                  for _ in range(rows)],
        "join_date": [random.choice(["2020-01-15", "2021-06-30",
                                      "2019-12-01", None, "invalid-date",
                                      "2022-03-10"])
                      for _ in range(rows)],
        "is_active": [random.choice([True, False, None, "yes", "no", 1, 0])
                      for _ in range(rows)],
    }

    df = pd.DataFrame(data)
    df.to_csv(path, index=False)
    print(f"[INFO] Generated sample dataset: {path} ({rows} rows)")
    return path


def load_data(file_path):
    """Load CSV or JSON file into a DataFrame. Handles messy data."""
    ext = os.path.splitext(file_path)[1].lower()
    if ext == ".json":
        df = pd.read_json(file_path)
    elif ext == ".csv":
        df = pd.read_csv(file_path, na_values=["", "NA", "N/A", "null",
                                                 "None", "nan", "NaN"])
    else:
        raise ValueError(f"Unsupported file format: {ext}. Use .csv or .json")
    return df


def detect_column_types(df):
    """Auto-detect semantic column types for each column."""
    type_map = {}
    for col in df.columns:
        series = df[col].dropna()
        if series.empty:
            type_map[col] = "empty"
            continue

        # Check if numeric
        if pd.api.types.is_numeric_dtype(series):
            if set(series.unique()).issubset({0, 1, True, False}):
                type_map[col] = "boolean"
            elif pd.api.types.is_integer_dtype(series):
                type_map[col] = "integer"
            else:
                type_map[col] = "float"
            continue

        # Try to parse as numeric
        coerced = pd.to_numeric(series, errors="coerce")
        if coerced.notna().sum() / len(series) > 0.8:
            type_map[col] = "numeric_mixed"
            continue

        # Try to parse as datetime
        try:
            parsed = pd.to_datetime(series, errors="coerce", infer_datetime_format=True)
            if parsed.notna().sum() / len(series) > 0.7:
                type_map[col] = "datetime"
                continue
        except Exception:
            pass

        # Check boolean-like
        lower_vals = series.astype(str).str.lower().str.strip()
        bool_vals = {"true", "false", "yes", "no", "1", "0", "y", "n"}
        if set(lower_vals.unique()).issubset(bool_vals):
            type_map[col] = "boolean"
            continue

        # Categorical vs text
        nunique = series.nunique()
        if nunique / len(series) < 0.5 or nunique <= 20:
            type_map[col] = "categorical"
        else:
            type_map[col] = "text"

    return type_map


def compute_numeric_stats(series):
    """Compute statistics for numeric columns."""
    clean = pd.to_numeric(series, errors="coerce").dropna()
    if clean.empty:
        return {}

    result = {
        "count": int(clean.count()),
        "mean": round(float(clean.mean()), 4),
        "median": round(float(clean.median()), 4),
        "std": round(float(clean.std()), 4) if len(clean) > 1 else 0.0,
        "min": round(float(clean.min()), 4),
        "max": round(float(clean.max()), 4),
        "q1": round(float(clean.quantile(0.25)), 4),
        "q3": round(float(clean.quantile(0.75)), 4),
        "skewness": round(float(stats.skew(clean)), 4),
        "kurtosis": round(float(stats.kurtosis(clean)), 4),
        "zeros": int((clean == 0).sum()),
        "negatives": int((clean < 0).sum()),
    }
    return result


def compute_categorical_stats(series):
    """Compute statistics for categorical columns."""
    clean = series.dropna().astype(str).str.strip()
    clean = clean[clean != ""]
    if clean.empty:
        return {}

    value_counts = clean.value_counts()
    top_n = 10

    result = {
        "unique_count": int(clean.nunique()),
        "top_values": {str(k): int(v) for k, v in
                       value_counts.head(top_n).items()},
        "mode": str(value_counts.index[0]) if len(value_counts) > 0 else None,
        "mode_frequency": int(value_counts.iloc[0]) if len(value_counts) > 0 else 0,
    }
    return result


def compute_distributions(df, col, col_type):
    """Compute value distribution for a column."""
    if col_type in ("integer", "float", "numeric_mixed"):
        clean = pd.to_numeric(df[col], errors="coerce").dropna()
        if clean.empty:
            return {}
        hist, bin_edges = pd.cut(clean, bins=10, retbins=True)
        counts = hist.value_counts().sort_index()
        return {
            "type": "histogram",
            "bins": [round(float(e), 4) for e in bin_edges],
            "counts": [int(c) for c in counts.values],
        }
    elif col_type in ("categorical", "boolean", "text"):
        vc = df[col].dropna().astype(str).value_counts().head(20)
        return {
            "type": "frequency",
            "values": {str(k): int(v) for k, v in vc.items()},
        }
    return {}


def compute_correlation_matrix(df):
    """Compute correlation matrix for numeric columns."""
    numeric_df = df.select_dtypes(include=["number"])
    if numeric_df.shape[1] < 2:
        return {}

    corr = numeric_df.corr(method="pearson")
    result = {}
    for col in corr.columns:
        result[col] = {}
        for row in corr.index:
            val = corr.loc[row, col]
            result[col][row] = round(float(val), 4) if not math.isnan(val) else None
    return result


def compute_data_quality_score(df, column_profiles):
    """Compute overall data quality score (0-100)."""
    total_cells = df.shape[0] * df.shape[1]
    if total_cells == 0:
        return 0.0

    # Completeness: % of non-missing values
    missing_total = df.isnull().sum().sum()
    completeness = 1.0 - (missing_total / total_cells)

    # Uniqueness: average uniqueness across columns
    uniqueness_scores = []
    for col in df.columns:
        non_null = df[col].dropna()
        if len(non_null) > 0:
            nunique = non_null.nunique()
            uniqueness_scores.append(nunique / len(non_null))
    avg_uniqueness = sum(uniqueness_scores) / len(uniqueness_scores) if uniqueness_scores else 0

    # Consistency: columns with uniform types
    consistency_scores = []
    for col in df.columns:
        non_null = df[col].dropna()
        if len(non_null) > 0:
            coerced = pd.to_numeric(non_null, errors="coerce")
            numeric_ratio = coerced.notna().sum() / len(non_null)
            consistency_scores.append(max(numeric_ratio, 1 - numeric_ratio))
    avg_consistency = (sum(consistency_scores) / len(consistency_scores)
                       if consistency_scores else 0)

    # Weighted score
    score = (completeness * 0.5 + avg_consistency * 0.3 + min(avg_uniqueness, 1.0) * 0.2) * 100
    return round(score, 2)


def generate_correlation_plot(df, output_dir):
    """Generate correlation matrix heatmap using matplotlib."""
    numeric_df = df.select_dtypes(include=["number"])
    if numeric_df.shape[1] < 2:
        return None

    corr = numeric_df.corr()
    fig, ax = plt.subplots(figsize=(10, 8))
    cax = ax.matshow(corr, cmap="coolwarm", vmin=-1, vmax=1)
    fig.colorbar(cax)
    ax.set_xticks(range(len(corr.columns)))
    ax.set_yticks(range(len(corr.columns)))
    ax.set_xticklabels(corr.columns, rotation=45, ha="left")
    ax.set_yticklabels(corr.columns)
    ax.set_title("Correlation Matrix", pad=20)
    plt.tight_layout()
    plot_path = os.path.join(output_dir, "correlation_matrix.png")
    fig.savefig(plot_path, dpi=100)
    plt.close(fig)
    print(f"[INFO] Correlation plot saved: {plot_path}")
    return plot_path


def profile_dataset(file_path):
    """Main profiling function."""
    print(f"\n{'='*60}")
    print(f"  DATA PROFILING TOOL - Trial 1 (pandas + scipy + matplotlib)")
    print(f"{'='*60}\n")

    df = load_data(file_path)
    output_dir = os.path.dirname(os.path.abspath(file_path))

    print(f"[INFO] Loaded file: {file_path}")
    print(f"[INFO] Shape: {df.shape[0]} rows x {df.shape[1]} columns\n")

    # Detect column types
    col_types = detect_column_types(df)

    # Profile each column
    column_profiles = {}
    for col in df.columns:
        col_type = col_types.get(col, "unknown")
        total = len(df[col])
        missing = int(df[col].isnull().sum())
        missing_pct = round(missing / total * 100, 2) if total > 0 else 0.0

        profile = {
            "name": col,
            "detected_type": col_type,
            "total_count": total,
            "missing_count": missing,
            "missing_percentage": missing_pct,
            "unique_count": int(df[col].nunique()),
        }

        if col_type in ("integer", "float", "numeric_mixed"):
            profile["numeric_stats"] = compute_numeric_stats(df[col])
        elif col_type in ("categorical", "boolean", "text"):
            profile["categorical_stats"] = compute_categorical_stats(df[col])

        profile["distribution"] = compute_distributions(df, col, col_type)
        column_profiles[col] = profile

    # Correlation matrix
    corr_matrix = compute_correlation_matrix(df)

    # Data quality score
    quality_score = compute_data_quality_score(df, column_profiles)

    # Generate plots
    generate_correlation_plot(df, output_dir)

    # Build report
    report = {
        "file": os.path.basename(file_path),
        "generated_at": datetime.now().isoformat(),
        "dataset_overview": {
            "rows": df.shape[0],
            "columns": df.shape[1],
            "total_cells": df.shape[0] * df.shape[1],
            "total_missing": int(df.isnull().sum().sum()),
            "total_missing_pct": round(
                df.isnull().sum().sum() / (df.shape[0] * df.shape[1]) * 100, 2
            ),
            "memory_usage_bytes": int(df.memory_usage(deep=True).sum()),
        },
        "data_quality_score": quality_score,
        "column_profiles": column_profiles,
        "correlation_matrix": corr_matrix,
    }

    # Console output
    print_console_report(report)

    # Save JSON report
    report_path = os.path.join(output_dir, "profile_report.json")
    with open(report_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"\n[INFO] Profile report saved: {report_path}")

    return report


def print_console_report(report):
    """Print human-readable report to console."""
    overview = report["dataset_overview"]
    print(f"--- Dataset Overview ---")
    print(f"  Rows:            {overview['rows']}")
    print(f"  Columns:         {overview['columns']}")
    print(f"  Total cells:     {overview['total_cells']}")
    print(f"  Total missing:   {overview['total_missing']} ({overview['total_missing_pct']}%)")
    print(f"  Memory usage:    {overview['memory_usage_bytes']:,} bytes")
    print(f"\n  Data Quality Score: {report['data_quality_score']} / 100\n")

    print(f"--- Column Profiles ---")
    for col_name, profile in report["column_profiles"].items():
        print(f"\n  [{col_name}]")
        print(f"    Type:          {profile['detected_type']}")
        print(f"    Missing:       {profile['missing_count']} ({profile['missing_percentage']}%)")
        print(f"    Unique:        {profile['unique_count']}")

        if "numeric_stats" in profile and profile["numeric_stats"]:
            ns = profile["numeric_stats"]
            print(f"    Min:           {ns.get('min', 'N/A')}")
            print(f"    Max:           {ns.get('max', 'N/A')}")
            print(f"    Mean:          {ns.get('mean', 'N/A')}")
            print(f"    Median:        {ns.get('median', 'N/A')}")
            print(f"    Std:           {ns.get('std', 'N/A')}")
            print(f"    Skewness:      {ns.get('skewness', 'N/A')}")
            print(f"    Kurtosis:      {ns.get('kurtosis', 'N/A')}")
            print(f"    Zeros:         {ns.get('zeros', 0)}")
            print(f"    Negatives:     {ns.get('negatives', 0)}")

        if "categorical_stats" in profile and profile["categorical_stats"]:
            cs = profile["categorical_stats"]
            print(f"    Unique values: {cs.get('unique_count', 'N/A')}")
            print(f"    Mode:          {cs.get('mode', 'N/A')} (freq: {cs.get('mode_frequency', 0)})")
            print(f"    Top values:")
            for val, cnt in list(cs.get("top_values", {}).items())[:5]:
                print(f"      {val}: {cnt}")

    if report["correlation_matrix"]:
        print(f"\n--- Correlation Matrix ---")
        cols = list(report["correlation_matrix"].keys())
        header = "".ljust(15) + "".join(c[:12].ljust(14) for c in cols)
        print(f"  {header}")
        for row in cols:
            vals = "".join(
                str(report["correlation_matrix"][col].get(row, "N/A"))[:12].ljust(14)
                for col in cols
            )
            print(f"  {row[:15].ljust(15)}{vals}")


def main():
    if len(sys.argv) > 1:
        file_path = sys.argv[1]
    else:
        print("[INFO] No input file provided. Generating sample dataset...")
        file_path = generate_sample_dataset()

    if not os.path.exists(file_path):
        print(f"[ERROR] File not found: {file_path}")
        sys.exit(1)

    profile_dataset(file_path)


if __name__ == "__main__":
    main()

# Data Profiling Tool - Python Trial 1

## Dependencies
- pandas 2.2.2
- scipy 1.13.1
- matplotlib 3.9.0

## Setup
```bash
pip install -r requirements.txt
```

## Usage
```bash
# Profile a CSV file
python profiler.py data.csv

# Profile a JSON file
python profiler.py data.json

# Generate and profile sample dataset
python profiler.py
```

## Output
- Console report with column profiles, distributions, and correlations
- `profile_report.json` with full profiling results
- `correlation_matrix.png` heatmap visualization

pandas==2.2.2
scipy==1.13.1
matplotlib==3.9.0