Imported from zstary/DDC_Skills_for_AI_Agents_in_Construction (2_DDC_Book/2.1-Data-Types-Classification/data-profiler/SKILL.md). Install upstream with npx skills add zstary/DDC_Skills_for_AI_Agents_in_Construction --skill data-profiler. Copyright stays with the author.

Data Profiler for Construction

Overview

Analyze construction data to understand its characteristics, distributions, quality, and patterns. Essential for data quality assessment, ETL planning, and identifying data issues before they impact projects.

Business Case

Before using any construction data, you need to understand:

• What data types are present
• Distribution of values
• Missing data patterns
• Anomalies and outliers
• Referential integrity issues

This skill profiles data to answer these questions and provides actionable insights.

Technical Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Tuple
import pandas as pd
import numpy as np
from datetime import datetime
import json

@dataclass
class ColumnProfile:
    name: str
    data_type: str
    inferred_type: str  # More specific: project_id, cost, date, csi_code, etc.
    total_count: int
    null_count: int
    null_percentage: float
    unique_count: int
    uniqueness_ratio: float
    # For numeric columns
    min_value: Optional[float] = None
    max_value: Optional[float] = None
    mean_value: Optional[float] = None
    median_value: Optional[float] = None
    std_dev: Optional[float] = None
    # For string columns
    min_length: Optional[int] = None
    max_length: Optional[int] = None
    avg_length: Optional[float] = None
    # Top values
    top_values: List[Tuple[Any, int]] = field(default_factory=list)
    # Patterns
    common_patterns: List[str] = field(default_factory=list)
    # Quality flags
    quality_issues: List[str] = field(default_factory=list)

@dataclass
class DataProfile:
    source_name: str
    row_count: int
    column_count: int
    columns: List[ColumnProfile]
    duplicate_rows: int
    memory_usage: str
    profiled_at: datetime
    quality_score: float
    recommendations: List[str]

class ConstructionDataProfiler:
    """Profile construction data for quality and characteristics."""

    # Known construction data patterns
    CONSTRUCTION_PATTERNS = {
        'csi_code': r'^\d{2}\s?\d{2}\s?\d{2}$',
        'project_id': r'^[A-Z]{2,4}[-_]?\d{3,6}$',
        'cost_code': r'^\d{2}[-.]?\d{2,4}$',
        'wbs': r'^[\d.]+$',
        'phone': r'^\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}$',
        'email': r'^[\w.-]+@[\w.-]+\.\w+$',
        'date_iso': r'^\d{4}-\d{2}-\d{2}',
        'date_us': r'^\d{1,2}/\d{1,2}/\d{2,4}$',
        'currency': r'^\$?[\d,]+\.?\d{0,2}$',
        'percentage': r'^\d+\.?\d*%?$',
    }

    # Construction-specific column name patterns
    COLUMN_TYPE_HINTS = {
        'project': ['project_id', 'project_name', 'proj', 'job'],
        'cost': ['cost', 'amount', 'price', 'total', 'budget', 'actual'],
        'date': ['date', 'start', 'finish', 'end', 'created', 'modified'],
        'quantity': ['qty', 'quantity', 'count', 'units'],
        'csi': ['csi', 'division', 'masterformat', 'spec'],
        'location': ['location', 'area', 'zone', 'floor', 'level'],
        'person': ['owner', 'manager', 'superintendent', 'foreman', 'contact'],
    }

    def __init__(self):
        self.profiles: Dict[str, DataProfile] = {}

    def profile_dataframe(self, df: pd.DataFrame, source_name: str) -> DataProfile:
        """Profile a pandas DataFrame."""
        columns = []

        for col in df.columns:
            col_profile = self._profile_column(df[col], col)
            columns.append(col_profile)

        # Calculate duplicates
        duplicate_rows = len(df) - len(df.drop_duplicates())

        # Calculate memory usage
        memory_bytes = df.memory_usage(deep=True).sum()
        if memory_bytes < 1024:
            memory_usage = f"{memory_bytes} B"
        elif memory_bytes < 1024**2:
            memory_usage = f"{memory_bytes/1024:.1f} KB"
        else:
            memory_usage = f"{memory_bytes/1024**2:.1f} MB"

        # Calculate overall quality score
        quality_score = self._calculate_quality_score(columns)

        # Generate recommendations
        recommendations = self._generate_recommendations(columns, df)

        profile = DataProfile(
            source_name=source_name,
            row_count=len(df),
            column_count=len(df.columns),
            columns=columns,
            duplicate_rows=duplicate_rows,
            memory_usage=memory_usage,
            profiled_at=datetime.now(),
            quality_score=quality_score,
            recommendations=recommendations
        )

        self.profiles[source_name] = profile
        return profile

    def _profile_column(self, series: pd.Series, name: str) -> ColumnProfile:
        """Profile a single column."""
        total_count = len(series)
        null_count = series.isnull().sum()
        null_percentage = (null_count / total_count * 100) if total_count > 0 else 0

        # Get non-null values for analysis
        non_null = series.dropna()
        unique_count = non_null.nunique()
        uniqueness_ratio = unique_count / len(non_null) if len(non_null) > 0 else 0

        profile = ColumnProfile(
            name=name,
            data_type=str(series.dtype),
            inferred_type=self._infer_construction_type(series, name),
            total_count=total_count,
            null_count=null_count,
            null_percentage=round(null_percentage, 2),
            unique_count=unique_count,
            uniqueness_ratio=round(uniqueness_ratio, 4)
        )

        # Numeric analysis
        if pd.api.types.is_numeric_dtype(series):
            profile.min_value = float(non_null.min()) if len(non_null) > 0 else None
            profile.max_value = float(non_null.max()) if len(non_null) > 0 else None
            profile.mean_value = float(non_null.mean()) if len(non_null) > 0 else None
            profile.median_value = float(non_null.median()) if len(non_null) > 0 else None
            profile.std_dev = float(non_null.std()) if len(non_null) > 1 else None

            # Check for outliers
            if len(non_null) > 10 and profile.std_dev:
                outliers = non_null[abs(non_null - profile.mean_value) > 3 * profile.std_dev]
                if len(outliers) > 0:
                    profile.quality_issues.append(f"{len(outliers)} potential outliers detected")

            # Check for negative costs
            if any(hint in name.lower() for hint in ['cost', 'amount', 'price', 'total']):
                negatives = (non_null < 0).sum()
                if negatives > 0:
                    profile.quality_issues.append(f"{negatives} negative values in cost column")

        # String analysis
        elif pd.api.types.is_object_dtype(series) or pd.api.types.is_string_dtype(series):
            str_series = non_null.astype(str)
            lengths = str_series.str.len()
            profile.min_length = int(lengths.min()) if len(lengths) > 0 else None
            profile.max_length = int(lengths.max()) if len(lengths) > 0 else None
            profile.avg_length = float(lengths.mean()) if len(lengths) > 0 else None

            # Detect patterns
            profile.common_patterns = self._detect_patterns(str_series)

        # Top values
        if len(non_null) > 0:
            value_counts = non_null.value_counts().head(5)
            profile.top_values = list(zip(value_counts.index.tolist(), value_counts.values.tolist()))

        # Quality checks
        if null_percentage > 50:
            profile.quality_issues.append("High null rate (>50%)")
        if uniqueness_ratio == 1.0 and total_count > 100:
            profile.quality_issues.append("All unique values - possible ID column")
        if uniqueness_ratio < 0.01 and unique_count > 1:
            profile.quality_issues.append("Low cardinality - possible category")

        return profile

    def _infer_construction_type(self, series: pd.Series, name: str) -> str:
        """Infer construction-specific data type."""
        name_lower = name.lower()

        # Check column name hints
        for type_name, hints in self.COLUMN_TYPE_HINTS.items():
            if any(hint in name_lower for hint in hints):
                return type_name

        # Check data patterns
        non_null = series.dropna().astype(str)
        if len(non_null) == 0:
            return "unknown"

        sample = non_null.head(100)

        for pattern_name, pattern in self.CONSTRUCTION_PATTERNS.items():
            matches = sample.str.match(pattern, na=False).sum()
            if matches / len(sample) > 0.8:
                return pattern_name

        # Default to pandas dtype
        if pd.api.types.is_numeric_dtype(series):
            return "numeric"
        elif pd.api.types.is_datetime64_any_dtype(series):
            return "datetime"
        else:
            return "text"

    def _detect_patterns(self, str_series: pd.Series) -> List[str]:
        """Detect common patterns in string data."""
        patterns_found = []

        sample = str_series.head(1000)

        for pattern_name, pattern in self.CONSTRUCTION_PATTERNS.items():
            matches = sample.str.match(pattern, na=False).sum()
            if matches / len(sample) > 0.1:
                patterns_found.append(f"{pattern_name} ({matches/len(sample):.0%})")

        return patterns_found[:3]

    def _calculate_quality_score(self, columns: List[ColumnProfile]) -> float:
        """Calculate overall data quality score (0-100)."""
        if not columns:
            return 0.0

        scores = []

        for col in columns:
            col_score = 100

            # Penalize for nulls
            col_score -= min(col.null_percentage, 50)

            # Penalize for quality issues
            col_score -= len(col.quality_issues) * 10

            scores.append(max(col_score, 0))

        return round(sum(scores) / len(scores), 1)

    def _generate_recommendations(self, columns: List[ColumnProfile], df: pd.DataFrame) -> List[str]:
        """Generate recommendations based on profile."""
        recommendations = []

        # High null columns
        high_null = [c for c in columns if c.null_percentage > 30]
        if high_null:
            recommendations.append(
                f"Review {len(high_null)} columns with >30% null values: "
                f"{', '.join(c.name for c in high_null[:3])}"
            )

        # Potential ID columns without uniqueness
        for col in columns:
            if 'id' in col.name.lower() and col.uniqueness_ratio < 1.0:
                recommendations.append(
                    f"Column '{col.name}' appears to be an ID but has duplicate values"
                )

        # Date columns that should be datetime
        for col in columns:
            if col.inferred_type in ['date_iso', 'date_us'] and col.data_type == 'object':
                recommendations.append(
                    f"Convert '{col.name}' to datetime type for better analysis"
                )

        # Cost columns that are strings
        for col in columns:
            if col.inferred_type == 'currency' and col.data_type == 'object':
                recommendations.append(
                    f"Convert '{col.name}' to numeric type (remove $ and commas)"
                )

        return recommendations

    def profile_to_dict(self, profile: DataProfile) -> Dict:
        """Convert profile to dictionary for JSON export."""
        return {
            'source_name': profile.source_name,
            'row_count': profile.row_count,
            'column_count': profile.column_count,
            'duplicate_rows': profile.duplicate_rows,
            'memory_usage': profile.memory_usage,
            'profiled_at': profile.profiled_at.isoformat(),
            'quality_score': profile.quality_score,
            'recommendations': profile.recommendations,
            'columns': [
                {
                    'name': c.name,
                    'data_type': c.data_type,
                    'inferred_type': c.inferred_type,
                    'null_percentage': c.null_percentage,
                    'unique_count': c.unique_count,
                    'quality_issues': c.quality_issues,
                    'top_values': c.top_values[:3]
                }
                for c in profile.columns
            ]
        }

    def generate_profile_report(self, profile: DataProfile) -> str:
        """Generate markdown profile report."""
        report = [f"# Data Profile: {profile.source_name}", ""]
        report.append(f"**Profiled At:** {profile.profiled_at.strftime('%Y-%m-%d %H:%M')}")
        report.append(f"**Quality Score:** {profile.quality_score}/100")
        report.append("")

        # Summary
        report.append("## Summary")
        report.append(f"- **Rows:** {profile.row_count:,}")
        report.append(f"- **Columns:** {profile.column_count}")
        report.append(f"- **Duplicate Rows:** {profile.duplicate_rows:,}")
        report.append(f"- **Memory Usage:** {profile.memory_usage}")
        report.append("")

        # Recommendations
        if profile.recommendations:
            report.append("## Recommendations")
            for rec in profile.recommendations:
                report.append(f"- {rec}")
            report.append("")

        # Column Details
        report.append("## Column Details")
        report.append("")
        report.append("| Column | Type | Inferred | Nulls | Unique | Issues |")
        report.append("|--------|------|----------|-------|--------|--------|")

        for col in profile.columns:
            issues = len(col.quality_issues)
            report.append(
                f"| {col.name} | {col.data_type} | {col.inferred_type} | "
                f"{col.null_percentage:.1f}% | {col.unique_count:,} | {issues} |"
            )

        # Detailed column profiles
        report.append("")
        report.append("## Detailed Column Profiles")

        for col in profile.columns:
            report.append(f"\n### {col.name}")
            report.append(f"- **Type:** {col.data_type} (inferred: {col.inferred_type})")
            report.append(f"- **Nulls:** {col.null_count:,} ({col.null_percentage:.1f}%)")
            report.append(f"- **Unique Values:** {col.unique_count:,} ({col.uniqueness_ratio:.1%})")

            if col.min_value is not None:
                report.append(f"- **Range:** {col.min_value:,.2f} to {col.max_value:,.2f}")
                report.append(f"- **Mean:** {col.mean_value:,.2f}, Median: {col.median_value:,.2f}")

            if col.min_length is not None:
                report.append(f"- **Length:** {col.min_length} to {col.max_length} (avg: {col.avg_length:.1f})")

            if col.top_values:
                report.append(f"- **Top Values:** {col.top_values[:3]}")

            if col.common_patterns:
                report.append(f"- **Patterns:** {', '.join(col.common_patterns)}")

            if col.quality_issues:
                report.append(f"- **Issues:** {', '.join(col.quality_issues)}")

        return "\n".join(report)

    def compare_profiles(self, profile1: DataProfile, profile2: DataProfile) -> Dict:
        """Compare two profiles to detect schema changes or data drift."""
        comparison = {
            'profiles': [profile1.source_name, profile2.source_name],
            'row_count_change': profile2.row_count - profile1.row_count,
            'quality_change': profile2.quality_score - profile1.quality_score,
            'new_columns': [],
            'removed_columns': [],
            'type_changes': [],
            'null_rate_changes': []
        }

        cols1 = {c.name: c for c in profile1.columns}
        cols2 = {c.name: c for c in profile2.columns}

        # Find new/removed columns
        comparison['new_columns'] = [n for n in cols2 if n not in cols1]
        comparison['removed_columns'] = [n for n in cols1 if n not in cols2]

        # Compare common columns
        for name in cols1:
            if name in cols2:
                c1, c2 = cols1[name], cols2[name]

                if c1.data_type != c2.data_type:
                    comparison['type_changes'].append({
                        'column': name,
                        'from': c1.data_type,
                        'to': c2.data_type
                    })

                null_change = c2.null_percentage - c1.null_percentage
                if abs(null_change) > 10:
                    comparison['null_rate_changes'].append({
                        'column': name,
                        'change': null_change
                    })

        return comparison

Quick Start

import pandas as pd

# Load construction data
df = pd.read_excel("project_costs.xlsx")

# Profile the data
profiler = ConstructionDataProfiler()
profile = profiler.profile_dataframe(df, "Project Costs 2025")

# Generate report
report = profiler.generate_profile_report(profile)
print(report)

# Export to JSON
profile_dict = profiler.profile_to_dict(profile)
with open("profile.json", "w") as f:
    json.dump(profile_dict, f, indent=2)

# Compare with previous profile
old_profile = profiler.profile_dataframe(old_df, "Project Costs 2024")
comparison = profiler.compare_profiles(old_profile, profile)
print(f"Quality changed by: {comparison['quality_change']}")

Common Use Cases

1. Pre-ETL Analysis: Profile source data before building pipelines
2. Quality Monitoring: Track data quality over time
3. Schema Validation: Detect unexpected changes in data structure
4. Anomaly Detection: Find outliers and data quality issues

Dependencies

pip install pandas numpy

Resources

• Data Profiling Best Practices: DAMA DMBOK
• Construction Data Standards: CSI MasterFormat, UniFormat