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