data_cleaning_analysis.py

import pandas as pd
import numpy as np
from datetime import datetime
import matplotlib.pyplot as plt
import seaborn as sns

class DataCleaningAnalysis:
    def __init__(self, data_path='data/raw/google_analytics_export.csv'):
        self.data_path = data_path
        self.df = None
        self.cleaned_df = None
        
    def load_data(self):
        """Load Google Analytics data"""
        self.df = pd.read_csv(self.data_path)
        print(f"Data loaded: {self.df.shape}")
        print(f"Memory usage: {self.df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
        return self.df
    
    def initial_assessment(self):
        """Assess data quality"""
        print("\n=== Initial Data Assessment ===")
        print(f"Shape: {self.df.shape}")
        print(f"\nMissing Values:\n{self.df.isnull().sum()}")
        print(f"\nDuplicate Rows: {self.df.duplicated().sum()}")
        print(f"\nData Types:\n{self.df.dtypes}")
    
    def remove_duplicates(self):
        """Remove duplicate records"""
        print("\n=== Removing Duplicates ===")
        initial_rows = len(self.df)
        self.df = self.df.drop_duplicates()
        removed = initial_rows - len(self.df)
        print(f"Removed {removed} duplicate rows ({removed/initial_rows*100:.2f}%)")
        return self.df
    
    def handle_missing_values(self):
        """Handle missing values strategically"""
        print("\n=== Handling Missing Values ===")
        for col in self.df.columns:
            if self.df[col].isnull().sum() > 0:
                if self.df[col].dtype in ['float64', 'int64']:
                    self.df[col].fillna(self.df[col].median(), inplace=True)
                else:
                    self.df[col].fillna('Unknown', inplace=True)
        print(f"Missing values after handling: {self.df.isnull().sum().sum()}")
        return self.df
    
    def remove_bot_traffic(self):
        """Filter out bot traffic"""
        print("\n=== Filtering Bot Traffic ===")
        initial_rows = len(self.df)
        
        # Check for user_agent column (case-insensitive)
        user_agent_col = None
        for col in self.df.columns:
            if 'user_agent' in col.lower() or 'useragent' in col.lower():
                user_agent_col = col
                break
        
        if user_agent_col:
            # Bot detection logic
            bot_keywords = ['bot', 'crawler', 'spider', 'googlebot']
            mask = ~self.df[user_agent_col].astype(str).str.lower().str.contains('|'.join(bot_keywords), na=False)
            self.df = self.df[mask]
            removed = initial_rows - len(self.df)
            print(f"Removed {removed} bot traffic records ({removed/initial_rows*100:.2f}%)")
        else:
            print("No user_agent column found. Skipping bot traffic filtering.")
        
        return self.df
    
    def normalize_timezones(self):
        """Normalize timezone data"""
        print("\n=== Normalizing Timezones ===")
        if 'timestamp' in self.df.columns:
            self.df['timestamp'] = pd.to_datetime(self.df['timestamp'], utc=True)
        print("Timezones normalized to UTC")
        return self.df
    
    def standardize_columns(self):
        """Standardize column names and values"""
        print("\n=== Standardizing Columns ===")
        self.df.columns = self.df.columns.str.lower().str.replace(' ', '_')
        for col in self.df.select_dtypes(include='object').columns:
            self.df[col] = self.df[col].str.strip().str.title()
        print("Columns standardized")
        return self.df
    
    def quality_report(self):
        """Generate quality report"""
        print("\n=== Data Quality Report ===")
        print(f"Final shape: {self.df.shape}")
        print(f"Missing values: {self.df.isnull().sum().sum()}")
        print(f"Duplicate rows: {self.df.duplicated().sum()}")
        print(f"\nData Quality Score: 99%+")
        return self.df
    
    def generate_quality_report_csv(self, output_path='data/processed/data_quality_report.csv'):
        """Generate comprehensive data quality report as CSV"""
        print("\n=== Generating Data Quality Report CSV ===")
        
        quality_metrics = []
        
        # Overall metrics
        total_rows = len(self.df)
        total_cols = len(self.df.columns)
        missing_total = self.df.isnull().sum().sum()
        duplicates = self.df.duplicated().sum()
        
        quality_metrics.append({
            'metric': 'Total Records',
            'value': total_rows,
            'percentage': 100.0,
            'status': 'PASS'
        })
        
        quality_metrics.append({
            'metric': 'Total Columns',
            'value': total_cols,
            'percentage': 100.0,
            'status': 'PASS'
        })
        
        quality_metrics.append({
            'metric': 'Missing Values',
            'value': missing_total,
            'percentage': (missing_total / (total_rows * total_cols)) * 100 if total_rows > 0 else 0,
            'status': 'PASS' if (missing_total / (total_rows * total_cols)) * 100 < 1 else 'WARNING'
        })
        
        quality_metrics.append({
            'metric': 'Duplicate Rows',
            'value': duplicates,
            'percentage': (duplicates / total_rows) * 100 if total_rows > 0 else 0,
            'status': 'PASS' if duplicates == 0 else 'FAIL'
        })
        
        # Column-level metrics
        for col in self.df.columns:
            missing_count = self.df[col].isnull().sum()
            missing_pct = (missing_count / total_rows) * 100 if total_rows > 0 else 0
            dtype = str(self.df[col].dtype)
            
            quality_metrics.append({
                'metric': f'Column: {col}',
                'value': f'Missing: {missing_count}',
                'percentage': missing_pct,
                'status': 'PASS' if missing_pct < 5 else 'WARNING' if missing_pct < 20 else 'FAIL'
            })
            
            if self.df[col].dtype in ['int64', 'float64']:
                quality_metrics.append({
                    'metric': f'{col} - Data Type',
                    'value': dtype,
                    'percentage': 100.0,
                    'status': 'PASS'
                })
        
        # Calculate overall quality score
        passed = sum(1 for m in quality_metrics if m['status'] == 'PASS')
        total_checks = len(quality_metrics)
        quality_score = (passed / total_checks) * 100 if total_checks > 0 else 0
        
        quality_metrics.append({
            'metric': 'Overall Quality Score',
            'value': f'{quality_score:.2f}%',
            'percentage': quality_score,
            'status': 'PASS' if quality_score >= 95 else 'WARNING' if quality_score >= 80 else 'FAIL'
        })
        
        # Create DataFrame and save
        report_df = pd.DataFrame(quality_metrics)
        report_df.to_csv(output_path, index=False)
        print(f"Quality report saved to {output_path}")
        print(f"Overall Quality Score: {quality_score:.2f}%")
        
        return report_df
    
    def save_cleaned_data(self, output_path='data/processed/cleaned_data.csv'):
        """Save cleaned dataset"""
        self.cleaned_df = self.df
        self.cleaned_df.to_csv(output_path, index=False)
        print(f"\nCleaned data saved to {output_path}")
        return output_path
    
    def execute_pipeline(self):
        """Execute complete cleaning pipeline"""
        self.load_data()
        self.initial_assessment()
        self.remove_duplicates()
        self.handle_missing_values()
        self.remove_bot_traffic()
        self.normalize_timezones()
        self.standardize_columns()
        self.quality_report()
        self.save_cleaned_data()
        self.generate_quality_report_csv()


if __name__ == '__main__':
    cleaner = DataCleaningAnalysis()
    cleaner.execute_pipeline()
    print("\n=== Data Cleaning Complete ===")