--- 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()

...