name: "data-quality-check" description: "Assess construction data quality using completeness, accuracy, consistency, timeliness, and validity metrics. Automated validation with regex patterns, thresholds, and reporting." homepage: "https://datadrivenconstruction.io" metadata: {"openclaw": {"emoji": "✔️", "os": ["win32"], "homepage": "https://datadrivenconstruction.io", "requires": {"bins": ["python3"]}}}

Data Quality Check for Construction

Overview

Based on DDC methodology (Chapter 2.6), this skill provides comprehensive data quality assessment for construction projects. Poor data quality leads to poor decisions - validate early, validate often.

Book Reference: "Требования к качеству данных и его обеспечение" / "Data Quality Requirements"

"Качество данных определяется пятью ключевыми метриками: полнота, точность, согласованность, своевременность и достоверность." — DDC Book, Chapter 2.6

Quick Start

import pandas as pd

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

# Quick quality check
quality_score = {
    'completeness': (1 - df.isnull().sum().sum() / df.size) * 100,
    'unique_ids': df['ElementId'].nunique() == len(df),
    'valid_volumes': (df['Volume_m3'] >= 0).all()
}

print(f"Completeness: {quality_score['completeness']:.1f}%")
print(f"Unique IDs: {quality_score['unique_ids']}")
print(f"Valid volumes: {quality_score['valid_volumes']}")

Data Quality Dimensions

The 5 Quality Metrics

import pandas as pd
import numpy as np
import re
from datetime import datetime, timedelta

class DataQualityChecker:
    """Comprehensive data quality assessment for construction data"""

    def __init__(self, df):
        self.df = df.copy()
        self.results = {}
        self.issues = []

    def check_completeness(self, required_columns=None):
        """Check for missing values (Полнота)"""
        if required_columns is None:
            required_columns = self.df.columns.tolist()

        completeness = {}
        for col in required_columns:
            if col in self.df.columns:
                non_null = self.df[col].notna().sum()
                total = len(self.df)
                completeness[col] = (non_null / total) * 100
            else:
                completeness[col] = 0
                self.issues.append(f"Missing required column: {col}")

        overall = np.mean(list(completeness.values()))

        self.results['completeness'] = {
            'by_column': completeness,
            'overall': overall,
            'threshold': 95,
            'passed': overall >= 95
        }

        return self.results['completeness']

    def check_accuracy(self, rules=None):
        """Check data accuracy against rules (Точность)"""
        if rules is None:
            # Default construction data rules
            rules = {
                'Volume_m3': {'min': 0, 'max': 10000},
                'Area_m2': {'min': 0, 'max': 100000},
                'Weight_kg': {'min': 0, 'max': 1000000},
                'Cost': {'min': 0, 'max': 100000000}
            }

        accuracy = {}
        for col, bounds in rules.items():
            if col in self.df.columns:
                valid = self.df[col].between(
                    bounds.get('min', -np.inf),
                    bounds.get('max', np.inf)
                ).sum()
                total = self.df[col].notna().sum()
                accuracy[col] = (valid / total * 100) if total > 0 else 100

                # Log invalid values
                invalid_count = total - valid
                if invalid_count > 0:
                    self.issues.append(
                        f"{col}: {invalid_count} values outside range [{bounds.get('min')}, {bounds.get('max')}]"
                    )

        overall = np.mean(list(accuracy.values())) if accuracy else 100

        self.results['accuracy'] = {
            'by_column': accuracy,
            'overall': overall,
            'threshold': 98,
            'passed': overall >= 98
        }

        return self.results['accuracy']

    def check_consistency(self, unique_cols=None, relationship_rules=None):
        """Check data consistency (Согласованность)"""
        consistency = {}

        # Check unique columns
        if unique_cols is None:
            unique_cols = ['ElementId']

        for col in unique_cols:
            if col in self.df.columns:
                is_unique = self.df[col].nunique() == len(self.df)
                consistency[f'{col}_unique'] = 100 if is_unique else \
                    (self.df[col].nunique() / len(self.df) * 100)

                if not is_unique:
                    duplicates = self.df[self.df[col].duplicated()][col].unique()
                    self.issues.append(f"Duplicate {col}: {len(duplicates)} duplicates found")

        # Check cross-field relationships
        if relationship_rules is None:
            relationship_rules = [
                ('End_Date', '>=', 'Start_Date'),
                ('Gross_Volume', '>=', 'Net_Volume')
            ]

        for col1, op, col2 in relationship_rules:
            if col1 in self.df.columns and col2 in self.df.columns:
                if op == '>=':
                    valid = (self.df[col1] >= self.df[col2]).sum()
                elif op == '>':
                    valid = (self.df[col1] > self.df[col2]).sum()
                elif op == '==':
                    valid = (self.df[col1] == self.df[col2]).sum()

                total = self.df[[col1, col2]].notna().all(axis=1).sum()
                consistency[f'{col1}_{op}_{col2}'] = (valid / total * 100) if total > 0 else 100

        overall = np.mean(list(consistency.values())) if consistency else 100

        self.results['consistency'] = {
            'checks': consistency,
            'overall': overall,
            'threshold': 99,
            'passed': overall >= 99
        }

        return self.results['consistency']

    def check_timeliness(self, date_col='Modified_Date', max_age_days=30):
        """Check data timeliness (Своевременность)"""
        if date_col not in self.df.columns:
            self.results['timeliness'] = {
                'overall': None,
                'message': f'Column {date_col} not found'
            }
            return self.results['timeliness']

        dates = pd.to_datetime(self.df[date_col], errors='coerce')
        cutoff = datetime.now() - timedelta(days=max_age_days)

        recent = (dates >= cutoff).sum()
        total = dates.notna().sum()
        timeliness_pct = (recent / total * 100) if total > 0 else 0

        oldest = dates.min()
        newest = dates.max()
        avg_age = (datetime.now() - dates.mean()).days if dates.notna().any() else None

        self.results['timeliness'] = {
            'recent_percentage': timeliness_pct,
            'oldest_record': oldest,
            'newest_record': newest,
            'average_age_days': avg_age,
            'threshold': 80,
            'passed': timeliness_pct >= 80
        }

        return self.results['timeliness']

    def check_validity(self, patterns=None):
        """Check data validity with regex patterns (Достоверность)"""
        if patterns is None:
            patterns = {
                'ElementId': r'^[A-Z]{1,3}\d{3,6}$',  # e.g., W001, FL12345
                'Level': r'^Level\s*\d+$|^L\d+$|^Уровень\s*\d+$',
                'Email': r'^[\w\.-]+@[\w\.-]+\.\w+$',
                'Phone': r'^\+?\d{10,15}$'
            }

        validity = {}
        for col, pattern in patterns.items():
            if col in self.df.columns:
                non_null = self.df[col].dropna()
                if len(non_null) > 0:
                    matches = non_null.astype(str).str.match(pattern).sum()
                    validity[col] = (matches / len(non_null) * 100)

                    invalid = len(non_null) - matches
                    if invalid > 0:
                        self.issues.append(f"{col}: {invalid} values don't match pattern")
                else:
                    validity[col] = 100

        overall = np.mean(list(validity.values())) if validity else 100

        self.results['validity'] = {
            'by_column': validity,
            'overall': overall,
            'threshold': 95,
            'passed': overall >= 95
        }

        return self.results['validity']

    def run_full_check(self):
        """Run all quality checks"""
        self.check_completeness()
        self.check_accuracy()
        self.check_consistency()
        self.check_timeliness()
        self.check_validity()

        # Calculate overall score
        scores = []
        for metric in ['completeness', 'accuracy', 'consistency', 'validity']:
            if metric in self.results and self.results[metric].get('overall'):
                scores.append(self.results[metric]['overall'])

        self.results['overall_score'] = np.mean(scores) if scores else 0
        self.results['grade'] = self._calculate_grade(self.results['overall_score'])
        self.results['issues'] = self.issues

        return self.results

    def _calculate_grade(self, score):
        """Calculate quality grade"""
        if score >= 98:
            return 'A+'
        elif score >= 95:
            return 'A'
        elif score >= 90:
            return 'B'
        elif score >= 80:
            return 'C'
        elif score >= 70:
            return 'D'
        else:
            return 'F'

    def generate_report(self):
        """Generate quality report"""
        if not self.results:
            self.run_full_check()

        report = []
        report.append("=" * 60)
        report.append("DATA QUALITY REPORT")
        report.append("=" * 60)
        report.append(f"Records analyzed: {len(self.df)}")
        report.append(f"Columns: {len(self.df.columns)}")
        report.append("")
        report.append(f"OVERALL SCORE: {self.results['overall_score']:.1f}% (Grade: {self.results['grade']})")
        report.append("")
        report.append("-" * 60)

        # Detail by dimension
        for metric in ['completeness', 'accuracy', 'consistency', 'validity', 'timeliness']:
            if metric in self.results:
                r = self.results[metric]
                passed = '✓' if r.get('passed', False) else '✗'
                overall = r.get('overall', r.get('recent_percentage', 'N/A'))
                if isinstance(overall, (int, float)):
                    report.append(f"{metric.upper():15s}: {overall:>6.1f}% {passed}")
                else:
                    report.append(f"{metric.upper():15s}: {overall}")

        report.append("-" * 60)

        if self.issues:
            report.append("")
            report.append("ISSUES FOUND:")
            for issue in self.issues[:10]:  # Show first 10
                report.append(f"  • {issue}")
            if len(self.issues) > 10:
                report.append(f"  ... and {len(self.issues) - 10} more issues")

        report.append("")
        report.append("=" * 60)

        return "\n".join(report)

Validation Rules Builder

Custom Validation Rules

class ValidationRulesBuilder:
    """Build custom validation rules for construction data"""

    def __init__(self):
        self.rules = []

    def add_not_null(self, column):
        """Column must not have null values"""
        self.rules.append({
            'type': 'not_null',
            'column': column,
            'check': lambda df, col=column: df[col].notna().all()
        })
        return self

    def add_unique(self, column):
        """Column must have unique values"""
        self.rules.append({
            'type': 'unique',
            'column': column,
            'check': lambda df, col=column: df[col].nunique() == len(df)
        })
        return self

    def add_range(self, column, min_val=None, max_val=None):
        """Column values must be within range"""
        self.rules.append({
            'type': 'range',