Datavision / backend /core /autonomous_data_ops.py
DataVision CI/CD Bot
release: clean production build for HuggingFace Space
ee7d7b9
Raw
History Blame Contribute Delete
21.2 kB
"""
πŸ€– AUTONOMOUS DATA OPERATIONS - Auto-Fix & Enhancement Engine
===============================================================
Automatically fixes and enhances data quality:
- Missing value imputation (smart strategies)
- Outlier detection and handling
- Data type corrections
- Duplicate removal
- Auto-enrichment (date features, currency, geolocation)
- Schema validation and evolution
This module runs autonomously on data upload to ensure clean, ready-to-analyze data.
"""
import pandas as pd
import numpy as np
import logging
import re
from datetime import datetime
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class DataFixResult:
"""Result from an autonomous data fix operation"""
operation: str
column: Optional[str]
rows_affected: int
description: str
before_value: Optional[Any] = None
after_value: Optional[Any] = None
@dataclass
class AutoFixReport:
"""Complete report of all autonomous fixes applied"""
original_rows: int
original_cols: int
final_rows: int
final_cols: int
fixes_applied: List[DataFixResult] = field(default_factory=list)
enrichments_added: List[str] = field(default_factory=list)
quality_score_before: float = 0.0
quality_score_after: float = 0.0
processing_time_ms: int = 0
def to_dict(self) -> Dict:
return {
"original_shape": {"rows": self.original_rows, "cols": self.original_cols},
"final_shape": {"rows": self.final_rows, "cols": self.final_cols},
"fixes_applied": [
{
"operation": f.operation,
"column": f.column,
"rows_affected": f.rows_affected,
"description": f.description
} for f in self.fixes_applied
],
"enrichments_added": self.enrichments_added,
"quality_improvement": {
"before": round(self.quality_score_before, 2),
"after": round(self.quality_score_after, 2),
"improvement": round(self.quality_score_after - self.quality_score_before, 2)
},
"processing_time_ms": self.processing_time_ms
}
class AutonomousDataOps:
"""
πŸ€– Autonomous Data Operations Engine
Automatically detects and fixes data quality issues:
1. Missing Value Imputation
2. Outlier Detection & Handling
3. Data Type Corrections
4. Duplicate Removal
5. Data Enrichment
"""
def __init__(self):
self.fix_history: List[DataFixResult] = []
def auto_fix(
self,
df: pd.DataFrame,
fix_missing: bool = True,
fix_outliers: bool = True,
fix_duplicates: bool = True,
fix_types: bool = True,
enrich_dates: bool = True,
aggressive: bool = False
) -> Tuple[pd.DataFrame, AutoFixReport]:
"""
πŸš€ Main entry point - Automatically fix all data issues.
Args:
df: Input DataFrame
fix_missing: Impute missing values
fix_outliers: Cap/handle outliers
fix_duplicates: Remove duplicate rows
fix_types: Correct data types
enrich_dates: Add date-based features
aggressive: Apply more aggressive fixes
Returns:
Tuple of (fixed DataFrame, AutoFixReport)
"""
start_time = datetime.now()
# Initialize report
report = AutoFixReport(
original_rows=len(df),
original_cols=len(df.columns),
quality_score_before=self._calculate_quality_score(df)
)
# Make a copy to avoid modifying original
df_fixed = df.copy()
logger.info(f"πŸ€– Starting Autonomous Data Fix ({len(df)} rows, {len(df.columns)} cols)")
# Step 1: Remove duplicates
if fix_duplicates:
df_fixed, dup_fixes = self._fix_duplicates(df_fixed)
report.fixes_applied.extend(dup_fixes)
# Step 2: Fix data types
if fix_types:
df_fixed, type_fixes = self._fix_data_types(df_fixed)
report.fixes_applied.extend(type_fixes)
# Step 3: Handle missing values
if fix_missing:
df_fixed, missing_fixes = self._fix_missing_values(df_fixed, aggressive)
report.fixes_applied.extend(missing_fixes)
# Step 4: Handle outliers
if fix_outliers:
df_fixed, outlier_fixes = self._fix_outliers(df_fixed, aggressive)
report.fixes_applied.extend(outlier_fixes)
# Step 5: Enrich data (add derived features)
if enrich_dates:
df_fixed, enrichments = self._enrich_data(df_fixed)
report.enrichments_added.extend(enrichments)
# Calculate final metrics
report.final_rows = len(df_fixed)
report.final_cols = len(df_fixed.columns)
report.quality_score_after = self._calculate_quality_score(df_fixed)
report.processing_time_ms = int((datetime.now() - start_time).total_seconds() * 1000)
logger.info(f"βœ… Auto-fix complete: {len(report.fixes_applied)} fixes, "
f"quality {report.quality_score_before:.0%} β†’ {report.quality_score_after:.0%}")
return df_fixed, report
# =========================================================================
# DUPLICATE HANDLING
# =========================================================================
def _fix_duplicates(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, List[DataFixResult]]:
"""Remove duplicate rows"""
fixes = []
n_duplicates = df.duplicated().sum()
if n_duplicates > 0:
df = df.drop_duplicates().reset_index(drop=True)
fixes.append(DataFixResult(
operation="remove_duplicates",
column=None,
rows_affected=n_duplicates,
description=f"Removed {n_duplicates} duplicate rows"
))
logger.info(f" πŸ—‘οΈ Removed {n_duplicates} duplicate rows")
return df, fixes
# =========================================================================
# DATA TYPE CORRECTIONS
# =========================================================================
def _fix_data_types(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, List[DataFixResult]]:
"""Automatically correct data types"""
fixes = []
for col in df.columns:
original_dtype = str(df[col].dtype)
# Try to detect and convert types
if df[col].dtype == 'object':
# Try numeric conversion
numeric_converted = self._try_convert_numeric(df[col])
if numeric_converted is not None:
df[col] = numeric_converted
fixes.append(DataFixResult(
operation="convert_to_numeric",
column=col,
rows_affected=len(df),
description=f"Converted '{col}' from {original_dtype} to numeric"
))
continue
# Try datetime conversion
date_converted = self._try_convert_datetime(df[col])
if date_converted is not None:
df[col] = date_converted
fixes.append(DataFixResult(
operation="convert_to_datetime",
column=col,
rows_affected=len(df),
description=f"Converted '{col}' from {original_dtype} to datetime"
))
continue
# Try boolean conversion
bool_converted = self._try_convert_boolean(df[col])
if bool_converted is not None:
df[col] = bool_converted
fixes.append(DataFixResult(
operation="convert_to_boolean",
column=col,
rows_affected=len(df),
description=f"Converted '{col}' from {original_dtype} to boolean"
))
return df, fixes
def _try_convert_numeric(self, series: pd.Series) -> Optional[pd.Series]:
"""Try to convert a series to numeric"""
try:
# Remove common formatting (currency, commas)
cleaned = series.astype(str).str.replace(r'[$,€£Β₯β‚Ή%]', '', regex=True)
cleaned = cleaned.str.strip()
# Convert to numeric
numeric = pd.to_numeric(cleaned, errors='coerce')
# Only convert if >80% are valid numbers
valid_ratio = numeric.notna().sum() / len(series)
if valid_ratio > 0.8 and series.notna().sum() > 0:
return numeric
except:
pass
return None
def _try_convert_datetime(self, series: pd.Series) -> Optional[pd.Series]:
"""Try to convert a series to datetime"""
try:
# Check if column name suggests date
col_name = series.name.lower() if series.name else ""
date_keywords = ['date', 'time', 'created', 'updated', 'timestamp', 'dt']
if not any(kw in col_name for kw in date_keywords):
return None
# Try conversion
dates = pd.to_datetime(series, errors='coerce', infer_datetime_format=True)
# Only convert if >70% are valid dates
valid_ratio = dates.notna().sum() / len(series)
if valid_ratio > 0.7:
return dates
except:
pass
return None
def _try_convert_boolean(self, series: pd.Series) -> Optional[pd.Series]:
"""Try to convert a series to boolean"""
try:
unique_vals = set(series.dropna().astype(str).str.lower().str.strip())
true_vals = {'true', 'yes', 'y', '1', 'on', 'active'}
false_vals = {'false', 'no', 'n', '0', 'off', 'inactive'}
if unique_vals.issubset(true_vals | false_vals) and len(unique_vals) <= 2:
return series.astype(str).str.lower().str.strip().isin(true_vals)
except:
pass
return None
# =========================================================================
# MISSING VALUE HANDLING
# =========================================================================
def _fix_missing_values(
self,
df: pd.DataFrame,
aggressive: bool = False
) -> Tuple[pd.DataFrame, List[DataFixResult]]:
"""Smart imputation of missing values"""
fixes = []
for col in df.columns:
n_missing = df[col].isna().sum()
if n_missing == 0:
continue
missing_pct = n_missing / len(df)
# Skip if too many missing (>50% for non-aggressive, >80% for aggressive)
threshold = 0.8 if aggressive else 0.5
if missing_pct > threshold:
if aggressive:
# Drop column entirely
df = df.drop(columns=[col])
fixes.append(DataFixResult(
operation="drop_column",
column=col,
rows_affected=len(df),
description=f"Dropped column '{col}' ({missing_pct:.0%} missing)"
))
continue
# Choose imputation strategy based on data type
if pd.api.types.is_numeric_dtype(df[col]):
# Numeric: use median (robust to outliers)
fill_value = df[col].median()
df[col] = df[col].fillna(fill_value)
strategy = "median"
elif pd.api.types.is_datetime64_any_dtype(df[col]):
# Datetime: forward fill or use mode
df[col] = df[col].fillna(method='ffill').fillna(method='bfill')
strategy = "forward/backward fill"
else:
# Categorical/text: use mode (most frequent)
mode_val = df[col].mode()
fill_value = mode_val.iloc[0] if len(mode_val) > 0 else "Unknown"
df[col] = df[col].fillna(fill_value)
strategy = "mode"
fixes.append(DataFixResult(
operation="impute_missing",
column=col,
rows_affected=n_missing,
description=f"Imputed {n_missing} missing values in '{col}' using {strategy}"
))
logger.info(f" πŸ”§ Imputed {n_missing} missing in '{col}' ({strategy})")
return df, fixes
# =========================================================================
# OUTLIER HANDLING
# =========================================================================
def _fix_outliers(
self,
df: pd.DataFrame,
aggressive: bool = False
) -> Tuple[pd.DataFrame, List[DataFixResult]]:
"""Detect and handle outliers in numeric columns"""
fixes = []
numeric_cols = df.select_dtypes(include=[np.number]).columns
for col in numeric_cols:
# Skip columns with too few unique values (likely categorical encoded as numeric)
if df[col].nunique() < 10:
continue
# Calculate IQR bounds
q1 = df[col].quantile(0.25)
q3 = df[col].quantile(0.75)
iqr = q3 - q1
if iqr == 0:
continue
lower_bound = q1 - 1.5 * iqr
upper_bound = q3 + 1.5 * iqr
# Find outliers
outliers = (df[col] < lower_bound) | (df[col] > upper_bound)
n_outliers = outliers.sum()
if n_outliers > 0 and n_outliers < len(df) * 0.1: # Cap if <10% are outliers
# Winsorize (cap at percentiles)
if aggressive:
# Remove outlier rows
df = df[~outliers]
fixes.append(DataFixResult(
operation="remove_outliers",
column=col,
rows_affected=n_outliers,
description=f"Removed {n_outliers} outlier rows from '{col}'"
))
else:
# Cap outliers at bounds (winsorizing)
p1, p99 = df[col].quantile([0.01, 0.99])
df[col] = df[col].clip(lower=p1, upper=p99)
fixes.append(DataFixResult(
operation="cap_outliers",
column=col,
rows_affected=n_outliers,
description=f"Capped {n_outliers} outliers in '{col}' to [1st, 99th] percentile"
))
logger.info(f" πŸ“Š Capped {n_outliers} outliers in '{col}'")
return df, fixes
# =========================================================================
# DATA ENRICHMENT
# =========================================================================
def _enrich_data(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, List[str]]:
"""Add derived features to enrich the data"""
enrichments = []
# Enrich datetime columns
datetime_cols = df.select_dtypes(include=['datetime64']).columns
for col in datetime_cols:
base_name = col.replace('_date', '').replace('date_', '').replace('Date', '')
# Extract year, month, day, weekday
if f"{base_name}_year" not in df.columns:
df[f"{base_name}_year"] = df[col].dt.year
enrichments.append(f"{base_name}_year")
if f"{base_name}_month" not in df.columns:
df[f"{base_name}_month"] = df[col].dt.month
enrichments.append(f"{base_name}_month")
if f"{base_name}_day_of_week" not in df.columns:
df[f"{base_name}_day_of_week"] = df[col].dt.dayofweek
enrichments.append(f"{base_name}_day_of_week")
if f"{base_name}_is_weekend" not in df.columns:
df[f"{base_name}_is_weekend"] = df[col].dt.dayofweek >= 5
enrichments.append(f"{base_name}_is_weekend")
logger.info(f" ✨ Enriched datetime column '{col}' with {len(enrichments)} features")
# Detect and create interaction features for important columns
numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
# Create ratios for common patterns
if 'price' in [c.lower() for c in numeric_cols] and 'quantity' in [c.lower() for c in numeric_cols]:
price_col = [c for c in numeric_cols if 'price' in c.lower()][0]
qty_col = [c for c in numeric_cols if 'quantity' in c.lower()][0]
if 'total_value' not in df.columns:
df['total_value'] = df[price_col] * df[qty_col]
enrichments.append('total_value')
return df, enrichments
# =========================================================================
# QUALITY SCORING
# =========================================================================
def _calculate_quality_score(self, df: pd.DataFrame) -> float:
"""Calculate data quality score (0-1)"""
if len(df) == 0 or len(df.columns) == 0:
return 0.0
scores = []
# Completeness: % non-missing
total_cells = len(df) * len(df.columns)
missing_cells = df.isna().sum().sum()
completeness = 1 - (missing_cells / total_cells)
scores.append(completeness)
# Uniqueness: % non-duplicate rows
n_duplicates = df.duplicated().sum()
uniqueness = 1 - (n_duplicates / len(df))
scores.append(uniqueness)
# Consistency: % of columns with consistent types
type_consistent = sum(1 for col in df.columns
if df[col].apply(type).nunique() <= 2)
consistency = type_consistent / len(df.columns)
scores.append(consistency)
# Weighted average
return np.average(scores, weights=[0.4, 0.3, 0.3])
# =========================================================================
# UTILITY METHODS
# =========================================================================
def detect_issues(self, df: pd.DataFrame) -> Dict[str, List[str]]:
"""Detect data quality issues without fixing them"""
issues = {
"missing_values": [],
"duplicates": [],
"outliers": [],
"type_issues": []
}
# Check missing values
for col in df.columns:
missing_pct = df[col].isna().mean()
if missing_pct > 0.05:
issues["missing_values"].append(f"{col}: {missing_pct:.1%} missing")
# Check duplicates
n_dups = df.duplicated().sum()
if n_dups > 0:
issues["duplicates"].append(f"{n_dups} duplicate rows ({n_dups/len(df):.1%})")
# Check outliers in numeric columns
for col in df.select_dtypes(include=[np.number]).columns:
if df[col].nunique() > 10:
q1, q3 = df[col].quantile([0.25, 0.75])
iqr = q3 - q1
if iqr > 0:
outliers = ((df[col] < q1 - 1.5*iqr) | (df[col] > q3 + 1.5*iqr)).sum()
if outliers > len(df) * 0.01:
issues["outliers"].append(f"{col}: {outliers} outliers ({outliers/len(df):.1%})")
return issues
def get_fix_recommendation(self, df: pd.DataFrame) -> str:
"""Get human-readable fix recommendations"""
issues = self.detect_issues(df)
recommendations = []
if issues["duplicates"]:
recommendations.append(f"πŸ—‘οΈ Remove {issues['duplicates'][0]}")
if issues["missing_values"]:
recommendations.append(f"πŸ”§ Fill missing values in {len(issues['missing_values'])} columns")
if issues["outliers"]:
recommendations.append(f"πŸ“Š Handle outliers in {len(issues['outliers'])} columns")
if not recommendations:
return "βœ… Data looks clean! No major issues detected."
return "\n".join(recommendations)
# Global instance
autonomous_data_ops = AutonomousDataOps()