FerroxLabs Data Scrubber
Core Philosophy
Data cleaning is the unglamorous but critical foundation of any data-driven system. Raw data is messy: missing values, inconsistent formats, duplicates, encoding errors, and outliers. A systematic data cleaning pipeline transforms raw chaos into reliable, analysis-ready data. The goal is not perfection -- it is fitness for purpose. Every cleaning decision should be documented, reversible, and auditable.
Data Cleaning Pipeline Design
Pipeline Architecture
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any
import pandas as pd
@dataclass
class CleaningReport:
"""Track all changes made during cleaning."""
total_rows_in: int = 0
total_rows_out: int = 0
steps: list[dict] = field(default_factory=list)
def add_step(self, name: str, rows_affected: int, details: str = ""):
self.steps.append({
"step": name,
"rows_affected": rows_affected,
"details": details,
})
# ... (condensed) ...
pipeline.add_step(NormalizeText(columns=['name', 'city']))
pipeline.add_step(ParseDates(columns=['created_at', 'updated_at']))
pipeline.add_step(DetectOutliers(column='amount', method='iqr'))
pipeline.add_step(ValidateConstraints())
clean_df, report = pipeline.run(raw_df)
print(report.summary())
Missing Value Strategies
Detection
import pandas as pd
import numpy as np
def analyze_missing_values(df: pd.DataFrame) -> pd.DataFrame:
"""Generate a missing value report for each column."""
missing = df.isnull().sum()
percent = (missing / len(df)) * 100
dtypes = df.dtypes
report = pd.DataFrame({
'column': missing.index,
'missing_count': missing.values,
'missing_pct': percent.values.round(2),
'dtype': dtypes.values,
}).sort_values('missing_pct', ascending=False)
return report[report['missing_count'] > 0]
# Example output:
# column missing_count missing_pct dtype
# phone 2340 23.40 object
# address 890 8.90 object
# age 120 1.20 float64
Strategies by Data Type
class HandleMissingValues(CleaningStep):
def name(self) -> str:
return "Handle Missing Values"
def execute(self, df: pd.DataFrame, report: CleaningReport) -> pd.DataFrame:
total_fixed = 0
for col in df.columns:
missing = df[col].isnull().sum()
if missing == 0:
continue
pct_missing = missing / len(df) * 100
if pct_missing > 50:
# Drop column if >50% missing
df = df.drop(columns=[col])
report.add_step(self.name(), missing, f"Dropped column '{col}' ({pct_missing:.1f}% missing)")
# ... (condensed) ...
if remaining > 0:
df = df.dropna(subset=[col])
total_fixed += missing
report.add_step(self.name(), total_fixed, "Total missing values handled")
return df
Advanced Imputation
from sklearn.impute import KNNImputer
from sklearn.experimental import enable_iterative_imputer
from sklearn.impute import IterativeImputer
def impute_numeric_columns(df: pd.DataFrame, method: str = 'knn') -> pd.DataFrame:
"""Impute missing numeric values using ML-based methods."""
numeric_cols = df.select_dtypes(include=[np.number]).columns
if method == 'knn':
imputer = KNNImputer(n_neighbors=5, weights='distance')
elif method == 'iterative':
imputer = IterativeImputer(max_iter=10, random_state=42)
else:
raise ValueError(f"Unknown method: {method}")
df[numeric_cols] = imputer.fit_transform(df[numeric_cols])
return df
Outlier Detection
Statistical Methods
class DetectOutliers(CleaningStep):
def __init__(self, column: str, method: str = 'iqr', action: str = 'flag'):
self.column = column
self.method = method
self.action = action # 'flag', 'remove', 'cap'
def name(self) -> str:
return f"Outlier Detection ({self.column})"
def execute(self, df: pd.DataFrame, report: CleaningReport) -> pd.DataFrame:
if self.method == 'iqr':
outlier_mask = self._iqr_method(df)
elif self.method == 'zscore':
outlier_mask = self._zscore_method(df)
elif self.method == 'modified_zscore':
outlier_mask = self._modified_zscore_method(df)
else:
raise ValueError(f"Unknown method: {self.method}")
# ... (condensed) ...
Q1 = df[self.column].quantile(0.25)
Q3 = df[self.column].quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - 1.5 * IQR
upper = Q3 + 1.5 * IQR
df[self.column] = df[self.column].clip(lower=lower, upper=upper)
return df
Duplicate Detection
class RemoveDuplicates(CleaningStep):
def __init__(self, subset: list[str] | None = None, strategy: str = 'exact'):
self.subset = subset
self.strategy = strategy
def name(self) -> str:
return "Remove Duplicates"
def execute(self, df: pd.DataFrame, report: CleaningReport) -> pd.DataFrame:
before = len(df)
if self.strategy == 'exact':
df = df.drop_duplicates(subset=self.subset, keep='first')
elif self.strategy == 'fuzzy':
df = self._fuzzy_dedup(df)
removed = before - len(df)
report.add_step(self.name(), removed,
# ... (condensed) ...
for j in range(i + 1, len(values)):
if j in to_remove:
continue
if fuzz.ratio(str(values[i]).lower(), str(values[j]).lower()) > 90:
to_remove.add(j)
return df.drop(index=list(to_remove)).reset_index(drop=True)
Text Normalization
import re
import unicodedata
class NormalizeText(CleaningStep):
def __init__(self, columns: list[str]):
self.columns = columns
def name(self) -> str:
return "Normalize Text"
def execute(self, df: pd.DataFrame, report: CleaningReport) -> pd.DataFrame:
total_modified = 0
for col in self.columns:
if col not in df.columns:
continue
original = df[col].copy()
df[col] = df[col].apply(self._normalize)
modified = (original != df[col]).sum()
# ... (condensed) ...
return None
local, domain = email.rsplit('@', 1)
# Remove dots in Gmail local part
if domain in ('gmail.com', 'googlemail.com'):
local = local.replace('.', '').split('+')[0]
domain = 'gmail.com'
return f"{local}@{domain}"
Date Parsing
from dateutil import parser as dateparser
class ParseDates(CleaningStep):
COMMON_FORMATS = [
'%Y-%m-%d',
'%Y-%m-%dT%H:%M:%S',
'%Y-%m-%dT%H:%M:%SZ',
'%Y-%m-%dT%H:%M:%S%z',
'%m/%d/%Y',
'%d/%m/%Y',
'%m-%d-%Y',
'%d-%m-%Y',
'%B %d, %Y',
'%b %d, %Y',
'%d %B %Y',
'%Y%m%d',
]
# ... (condensed) ...
except (ValueError, TypeError):
continue
# Fallback to dateutil parser (slower but handles more formats)
try:
return pd.Timestamp(dateparser.parse(value, dayfirst=self.dayfirst))
except (ValueError, TypeError):
return None
Encoding Fixes
import chardet
def detect_and_fix_encoding(file_path: str) -> pd.DataFrame:
"""Detect file encoding and read with correct encoding."""
# Detect encoding
with open(file_path, 'rb') as f:
raw_data = f.read(100000) # Read first 100KB for detection
detected = chardet.detect(raw_data)
encoding = detected['encoding']
confidence = detected['confidence']
print(f"Detected encoding: {encoding} (confidence: {confidence:.0%})")
# Try detected encoding, fall back to common alternatives
encodings_to_try = [encoding, 'utf-8', 'latin-1', 'cp1252', 'iso-8859-1']
for enc in encodings_to_try:
try:
df = pd.read_csv(file_path, encoding=enc)
# ... (condensed) ...
'ö': 'o', 'ü': 'u', 'ñ': 'n', 'ç': 'c',
'’': "'", '“': '"', 'â€\x9d': '"', 'â€"': '-',
'â€"': '--', '…': '...',
}
for bad, good in replacements.items():
text = text.replace(bad, good)
return text
Validation Rules
class ValidateConstraints(CleaningStep):
def name(self) -> str:
return "Validate Constraints"
def execute(self, df: pd.DataFrame, report: CleaningReport) -> pd.DataFrame:
violations = []
# Email format
if 'email' in df.columns:
email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
invalid_emails = ~df['email'].str.match(email_pattern, na=False)
count = invalid_emails.sum()
if count > 0:
violations.append(f"Invalid emails: {count}")
df.loc[invalid_emails, 'email'] = None
# Numeric ranges
if 'age' in df.columns:
# ... (condensed) ...
missing = df[col].isnull().sum()
if missing > 0:
violations.append(f"Missing required '{col}': {missing}")
report.add_step(self.name(), len(violations),
"; ".join(violations) if violations else "All constraints satisfied")
return df
Data Quality Reporting
def generate_quality_report(df: pd.DataFrame) -> dict:
"""Generate a comprehensive data quality report."""
return {
"overview": {
"total_rows": len(df),
"total_columns": len(df.columns),
"total_cells": len(df) * len(df.columns),
"total_missing": df.isnull().sum().sum(),
"completeness_pct": round((1 - df.isnull().sum().sum() / (len(df) * len(df.columns))) * 100, 2),
},
"columns": {
col: {
"dtype": str(df[col].dtype),
"non_null": int(df[col].notna().sum()),
"null_count": int(df[col].isnull().sum()),
"null_pct": round(df[col].isnull().sum() / len(df) * 100, 2),
"unique_count": int(df[col].nunique()),
"unique_pct": round(df[col].nunique() / max(df[col].notna().sum(), 1) * 100, 2),
"sample_values": df[col].dropna().head(3).tolist(),
}
for col in df.columns
},
}
Best Practices
- Never modify raw data in place: Always work on copies, keep originals
- Document every cleaning decision: Why was this value removed/changed?
- Make cleaning reproducible: Scripts, not manual edits
- Generate quality reports before AND after cleaning: Measure improvement
- Handle edge cases explicitly: Empty strings, whitespace-only, special characters
- Validate after cleaning: Ensure constraints are satisfied
- Use appropriate methods per data type: Median for numeric, mode for categorical
- Be conservative with outlier removal: Flag first, remove only when justified
- Test cleaning pipeline on sample data: Verify behavior before full run
- Version control cleaning scripts: Track changes to cleaning logic
When to Use
Use this skill when:
- Designing or implementing data scrubber solutions
- Reviewing or improving existing data scrubber approaches
- Making architectural or implementation decisions about data scrubber
- Learning data scrubber patterns and best practices
- Troubleshooting data scrubber-related issues
Do NOT use this skill when:
- The question is about a fundamentally different technology domain
- A more specific sibling skill covers the exact topic needed
- The user needs a complete hands-on tutorial rather than expert guidance
Output Format
# Data Scrubber Analysis
## Context Assessment
[Situation summary and constraints]
## Recommended Approach
[Primary recommendation with rationale]
## Implementation Steps
1. [Step with specific details]
2. [Step with specific details]
3. [Step with specific details]
## Trade-offs and Considerations
- [Key trade-off 1]
- [Key trade-off 2]
## Next Steps
- [Immediate action item]
- [Follow-up action item]
Example
Input: "Help me implement data scrubber for a medium-scale production application"
Output: A structured analysis covering current state assessment, recommended data scrubber approach with specific patterns, implementation roadmap with milestones, and risk mitigation strategies tailored to the application scale and constraints.
Edge Cases
- Legacy system integration: When data scrubber must coexist with legacy approaches, provide a gradual migration path rather than a complete rewrite
- Scale mismatch: When the solution complexity exceeds the project scale, recommend a simpler approach and note when to revisit
- Team skill gaps: When the team lacks experience with the recommended approach, include learning resources and simpler alternatives
- Conflicting requirements: When constraints conflict (e.g., performance vs. maintainability), explicitly state the trade-off and recommend based on stated priorities