RSK World - Text Classification Dataset - Project Files Browser | RSK World

generate_sample_audio.py train_transformers.py data_quality.py

scripts/train_transformers.py

"""
================================================================================
Text Classification Dataset - Transformer-Based Model Training
================================================================================
Project: Text Classification Dataset
Category: Text Data / NLP

Author: Molla Samser
Designer & Tester: Rima Khatun
Website: https://rskworld.in
Email: help@rskworld.in | support@rskworld.in
Phone: +91 93305 39277

Copyright (c) 2026 RSK World - All Rights Reserved
Content used for educational purposes only.

Requirements:
    pip install transformers torch datasets scikit-learn pandas

Created: December 2026
================================================================================
"""

import os
import argparse
from typing import Dict, List, Optional
from datetime import datetime

import numpy as np
import pandas as pd
import torch
from torch.utils.data import DataLoader, Dataset
from sklearn.metrics import accuracy_score, f1_score, classification_report

# Transformers imports
from transformers import (
    AutoTokenizer,
    AutoModelForSequenceClassification,
    TrainingArguments,
    Trainer,
    EarlyStoppingCallback
)
from datasets import Dataset as HFDataset, DatasetDict

# Project information
__author__ = "Molla Samser"
__website__ = "https://rskworld.in"
__email__ = "help@rskworld.in"


# Category labels
CATEGORY_LABELS = {
    0: 'Technology',
    1: 'Sports',
    2: 'Politics',
    3: 'Entertainment',
    4: 'Business',
    5: 'Science'
}

NUM_LABELS = len(CATEGORY_LABELS)


class TextClassificationDataset(Dataset):
    """
    PyTorch Dataset for text classification.
    
    Author: Molla Samser | RSK World (https://rskworld.in)
    """
    
    def __init__(
        self,
        texts: List[str],
        labels: List[int],
        tokenizer,
        max_length: int = 256
    ):
        """
        Initialize the dataset.
        
        Args:
            texts: List of input texts
            labels: List of labels
            tokenizer: HuggingFace tokenizer
            max_length: Maximum sequence length
        """
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = str(self.texts[idx])
        label = self.labels[idx]
        
        encoding = self.tokenizer(
            text,
            max_length=self.max_length,
            padding='max_length',
            truncation=True,
            return_tensors='pt'
        )
        
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }


def compute_metrics(eval_pred):
    """
    Compute metrics for evaluation.
    
    Args:
        eval_pred: Tuple of (predictions, labels)
        
    Returns:
        Dictionary of metrics
    """
    predictions, labels = eval_pred
    predictions = np.argmax(predictions, axis=1)
    
    accuracy = accuracy_score(labels, predictions)
    f1_weighted = f1_score(labels, predictions, average='weighted')
    f1_macro = f1_score(labels, predictions, average='macro')
    
    return {
        'accuracy': accuracy,
        'f1_weighted': f1_weighted,
        'f1_macro': f1_macro
    }


def load_data(data_dir: str) -> Dict[str, pd.DataFrame]:
    """
    Load train, validation, and test datasets.
    
    Args:
        data_dir: Path to data directory
        
    Returns:
        Dictionary with dataframes
    """
    data = {}
    
    for split in ['train', 'validation', 'test']:
        filepath = os.path.join(data_dir, 'csv', f'{split}.csv')
        data[split] = pd.read_csv(filepath, comment='#')
        print(f"Loaded {split}: {len(data[split])} samples")
    
    return data


def prepare_hf_dataset(data: Dict[str, pd.DataFrame]) -> DatasetDict:
    """
    Prepare HuggingFace DatasetDict from pandas DataFrames.
    
    Args:
        data: Dictionary of DataFrames
        
    Returns:
        HuggingFace DatasetDict
    """
    datasets = {}
    
    for split, df in data.items():
        datasets[split] = HFDataset.from_pandas(df[['text', 'label']])
    
    return DatasetDict(datasets)


def tokenize_function(examples, tokenizer, max_length=256):
    """Tokenize examples for the model."""
    return tokenizer(
        examples['text'],
        padding='max_length',
        truncation=True,
        max_length=max_length
    )


def train_transformer_model(
    model_name: str = 'bert-base-uncased',
    data_dir: str = '../data',
    output_dir: str = './results',
    epochs: int = 3,
    batch_size: int = 16,
    learning_rate: float = 2e-5,
    max_length: int = 256,
    warmup_steps: int = 500,
    weight_decay: float = 0.01,
    use_fp16: bool = True
):
    """
    Train a transformer-based text classification model.
    
    Args:
        model_name: HuggingFace model name
        data_dir: Path to data directory
        output_dir: Output directory for model
        epochs: Number of training epochs
        batch_size: Batch size
        learning_rate: Learning rate
        max_length: Maximum sequence length
        warmup_steps: Warmup steps for scheduler
        weight_decay: Weight decay for optimizer
        use_fp16: Use mixed precision training
        
    Author: Molla Samser | RSK World (https://rskworld.in)
    """
    print(f"\n{'='*60}")
    print("Transformer Model Training - RSK World")
    print(f"Author: {__author__}")
    print(f"Website: {__website__}")
    print(f"Email: {__email__}")
    print(f"{'='*60}\n")
    
    # Check for GPU
    device = 'cuda' if torch.cuda.is_available() else 'cpu'
    print(f"Using device: {device}")
    
    # Disable fp16 on CPU
    if device == 'cpu':
        use_fp16 = False
    
    # Load data
    print("\nLoading dataset...")
    data = load_data(data_dir)
    
    # Load tokenizer
    print(f"\nLoading tokenizer: {model_name}")
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    # Prepare datasets
    print("Preparing datasets...")
    dataset_dict = prepare_hf_dataset(data)
    
    # Tokenize datasets
    tokenized_datasets = dataset_dict.map(
        lambda x: tokenize_function(x, tokenizer, max_length),
        batched=True,
        remove_columns=['text']
    )
    
    # Load model
    print(f"\nLoading model: {model_name}")
    model = AutoModelForSequenceClassification.from_pretrained(
        model_name,
        num_labels=NUM_LABELS,
        id2label=CATEGORY_LABELS,
        label2id={v: k for k, v in CATEGORY_LABELS.items()}
    )
    
    # Training arguments
    training_args = TrainingArguments(
        output_dir=output_dir,
        num_train_epochs=epochs,
        per_device_train_batch_size=batch_size,
        per_device_eval_batch_size=batch_size,
        warmup_steps=warmup_steps,
        weight_decay=weight_decay,
        learning_rate=learning_rate,
        logging_dir=f'{output_dir}/logs',
        logging_steps=100,
        eval_strategy='epoch',
        save_strategy='epoch',
        load_best_model_at_end=True,
        metric_for_best_model='f1_weighted',
        greater_is_better=True,
        fp16=use_fp16,
        report_to='none',  # Disable wandb etc.
        seed=42
    )
    
    # Initialize trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_datasets['train'],
        eval_dataset=tokenized_datasets['validation'],
        tokenizer=tokenizer,
        compute_metrics=compute_metrics,
        callbacks=[EarlyStoppingCallback(early_stopping_patience=3)]
    )
    
    # Train
    print(f"\n{'='*60}")
    print("Starting Training...")
    print(f"{'='*60}\n")
    
    train_result = trainer.train()
    
    # Save model
    print(f"\nSaving model to {output_dir}...")
    trainer.save_model()
    tokenizer.save_pretrained(output_dir)
    
    # Evaluate on test set
    print(f"\n{'='*60}")
    print("Evaluating on Test Set...")
    print(f"{'='*60}\n")
    
    test_results = trainer.evaluate(tokenized_datasets['test'])
    
    print(f"Test Accuracy: {test_results['eval_accuracy']:.4f}")
    print(f"Test F1 (Weighted): {test_results['eval_f1_weighted']:.4f}")
    print(f"Test F1 (Macro): {test_results['eval_f1_macro']:.4f}")
    
    # Generate predictions for detailed report
    predictions = trainer.predict(tokenized_datasets['test'])
    pred_labels = np.argmax(predictions.predictions, axis=1)
    true_labels = predictions.label_ids
    
    print("\nClassification Report:")
    print(classification_report(
        true_labels,
        pred_labels,
        target_names=list(CATEGORY_LABELS.values())
    ))
    
    # Save results
    results = {
        'model_name': model_name,
        'train_loss': train_result.training_loss,
        'test_accuracy': test_results['eval_accuracy'],
        'test_f1_weighted': test_results['eval_f1_weighted'],
        'test_f1_macro': test_results['eval_f1_macro'],
        'epochs': epochs,
        'batch_size': batch_size,
        'learning_rate': learning_rate,
        'timestamp': datetime.now().isoformat(),
        'author': __author__,
        'website': __website__
    }
    
    import json
    with open(f'{output_dir}/training_results.json', 'w') as f:
        json.dump(results, f, indent=2)
    
    print(f"\n{'='*60}")
    print("Training Complete!")
    print(f"Model saved to: {output_dir}")
    print(f"{'='*60}\n")
    
    return results


def predict_text(
    text: str,
    model_dir: str,
    device: str = None
) -> Dict:
    """
    Predict category for a single text.
    
    Args:
        text: Input text
        model_dir: Path to saved model
        device: Device to use (auto-detected if None)
        
    Returns:
        Dictionary with prediction results
    """
    if device is None:
        device = 'cuda' if torch.cuda.is_available() else 'cpu'
    
    # Load model and tokenizer
    tokenizer = AutoTokenizer.from_pretrained(model_dir)
    model = AutoModelForSequenceClassification.from_pretrained(model_dir)
    model.to(device)
    model.eval()
    
    # Tokenize
    inputs = tokenizer(
        text,
        padding=True,
        truncation=True,
        max_length=256,
        return_tensors='pt'
    ).to(device)
    
    # Predict
    with torch.no_grad():
        outputs = model(**inputs)
        probs = torch.softmax(outputs.logits, dim=-1)
        pred_label = torch.argmax(probs, dim=-1).item()
        confidence = probs[0][pred_label].item()
    
    return {
        'text': text[:100] + '...' if len(text) > 100 else text,
        'predicted_label': pred_label,
        'predicted_category': CATEGORY_LABELS[pred_label],
        'confidence': confidence,
        'all_probabilities': {
            CATEGORY_LABELS[i]: probs[0][i].item()
            for i in range(NUM_LABELS)
        }
    }


def main():
    parser = argparse.ArgumentParser(
        description='Train Transformer Text Classification - RSK World'
    )
    parser.add_argument(
        '--model', type=str, default='bert-base-uncased',
        help='HuggingFace model name'
    )
    parser.add_argument(
        '--data-dir', type=str, default='../data',
        help='Data directory path'
    )
    parser.add_argument(
        '--output-dir', type=str, default='./transformer_model',
        help='Output directory'
    )
    parser.add_argument(
        '--epochs', type=int, default=3,
        help='Training epochs'
    )
    parser.add_argument(
        '--batch-size', type=int, default=16,
        help='Batch size'
    )
    parser.add_argument(
        '--learning-rate', type=float, default=2e-5,
        help='Learning rate'
    )
    parser.add_argument(
        '--max-length', type=int, default=256,
        help='Max sequence length'
    )
    
    args = parser.parse_args()
    
    train_transformer_model(
        model_name=args.model,
        data_dir=args.data_dir,
        output_dir=args.output_dir,
        epochs=args.epochs,
        batch_size=args.batch_size,
        learning_rate=args.learning_rate,
        max_length=args.max_length
    )


if __name__ == "__main__":
    main()

454 lines•12.5 KB

python

scripts/data_quality.py

Raw Download

"""
================================================================================
Text Classification Dataset - Data Quality Analyzer
================================================================================
Project: Text Classification Dataset
Category: Text Data / NLP

Author: Molla Samser
Designer & Tester: Rima Khatun
Website: https://rskworld.in
Email: help@rskworld.in | support@rskworld.in
Phone: +91 93305 39277

Copyright (c) 2026 RSK World - All Rights Reserved
Content used for educational purposes only.

Features:
- Duplicate Detection
- Missing Value Analysis
- Class Imbalance Detection
- Text Quality Metrics
- Outlier Detection
- Language Detection
- Noise Detection
- Data Completeness Scoring
- Automated Recommendations

Created: December 2026
================================================================================
"""

import re
import string
from typing import Dict, List, Tuple, Optional, Set
from collections import Counter
from datetime import datetime

import numpy as np
import pandas as pd

# Project information
__author__ = "Molla Samser"
__website__ = "https://rskworld.in"
__email__ = "help@rskworld.in"

# Category mapping
CATEGORIES = {
    0: 'Technology', 1: 'Sports', 2: 'Politics',
    3: 'Entertainment', 4: 'Business', 5: 'Science'
}


class DataQualityAnalyzer:
    """
    Comprehensive data quality analyzer for text classification datasets.
    
    Analyzes:
    - Data completeness
    - Duplicate records
    - Class balance
    - Text quality metrics
    - Potential label errors
    - Outliers
    
    Author: Molla Samser | RSK World (https://rskworld.in)
    """
    
    def __init__(self, verbose: bool = True):
        """
        Initialize the analyzer.
        
        Args:
            verbose: Print detailed analysis
        """
        self.verbose = verbose
        self.report = {}
        self.issues = []
        self.warnings = []
        self.recommendations = []
    
    def analyze(
        self,
        df: pd.DataFrame,
        text_column: str = 'text',
        label_column: str = 'label'
    ) -> Dict:
        """
        Perform comprehensive data quality analysis.
        
        Args:
            df: DataFrame to analyze
            text_column: Name of text column
            label_column: Name of label column
            
        Returns:
            Comprehensive quality report
        """
        if self.verbose:
            print(f"\n{'='*60}")
            print("Data Quality Analysis - RSK World")
            print(f"Author: {__author__} | Website: {__website__}")
            print(f"{'='*60}\n")
        
        # Basic info
        self._analyze_basic_info(df, text_column, label_column)
        
        # Missing values
        self._analyze_missing_values(df, text_column, label_column)
        
        # Duplicates
        self._analyze_duplicates(df, text_column)
        
        # Class balance
        self._analyze_class_balance(df, label_column)
        
        # Text quality
        self._analyze_text_quality(df, text_column)
        
        # Outliers
        self._analyze_outliers(df, text_column)
        
        # Potential label errors
        self._analyze_potential_label_errors(df, text_column, label_column)
        
        # Generate recommendations
        self._generate_recommendations()
        
        # Calculate overall score
        self._calculate_quality_score()
        
        # Print summary
        if self.verbose:
            self._print_summary()
        
        return self.report
    
    def _analyze_basic_info(self, df: pd.DataFrame, text_col: str, label_col: str):
        """Analyze basic dataset information."""
        self.report['basic_info'] = {
            'total_samples': len(df),
            'columns': list(df.columns),
            'text_column': text_col,
            'label_column': label_col,
            'memory_usage_mb': df.memory_usage(deep=True).sum() / 1024**2
        }
        
        if self.verbose:
            print("1. Basic Information")
            print("-" * 40)
            print(f"   Total samples: {len(df):,}")
            print(f"   Columns: {list(df.columns)}")
            print(f"   Memory usage: {self.report['basic_info']['memory_usage_mb']:.2f} MB\n")
    
    def _analyze_missing_values(self, df: pd.DataFrame, text_col: str, label_col: str):
        """Analyze missing values."""
        missing_text = df[text_col].isna().sum()
        missing_label = df[label_col].isna().sum()
        empty_text = (df[text_col].str.strip() == '').sum() if df[text_col].dtype == 'object' else 0
        
        self.report['missing_values'] = {
            'missing_text': int(missing_text),
            'missing_labels': int(missing_label),
            'empty_text': int(empty_text),
            'missing_percentage': float((missing_text + missing_label) / (len(df) * 2) * 100)
        }
        
        if missing_text > 0 or missing_label > 0:
            self.issues.append(f"Found {missing_text + missing_label} missing values")
        if empty_text > 0:
            self.warnings.append(f"Found {empty_text} empty text entries")
        
        if self.verbose:
            print("2. Missing Values Analysis")
            print("-" * 40)
            print(f"   Missing text: {missing_text}")
            print(f"   Missing labels: {missing_label}")
            print(f"   Empty text: {empty_text}\n")
    
    def _analyze_duplicates(self, df: pd.DataFrame, text_col: str):
        """Analyze duplicate records."""
        # Exact duplicates
        exact_duplicates = df.duplicated().sum()
        
        # Text duplicates (same text, possibly different labels)
        text_duplicates = df[text_col].duplicated().sum()
        
        # Near duplicates (simplified check)
        normalized_texts = df[text_col].str.lower().str.strip()
        normalized_duplicates = normalized_texts.duplicated().sum()
        
        self.report['duplicates'] = {
            'exact_duplicates': int(exact_duplicates),
            'text_duplicates': int(text_duplicates),
            'normalized_duplicates': int(normalized_duplicates),
            'duplicate_percentage': float(text_duplicates / len(df) * 100)
        }
        
        if text_duplicates > 0:
            self.warnings.append(f"Found {text_duplicates} duplicate texts")
        
        if self.verbose:
            print("3. Duplicate Analysis")
            print("-" * 40)
            print(f"   Exact duplicates: {exact_duplicates}")
            print(f"   Text duplicates: {text_duplicates}")
            print(f"   After normalization: {normalized_duplicates}\n")
    
    def _analyze_class_balance(self, df: pd.DataFrame, label_col: str):
        """Analyze class distribution."""
        class_counts = df[label_col].value_counts().sort_index()
        total = len(df)
        
        # Calculate imbalance ratio
        max_class = class_counts.max()
        min_class = class_counts.min()
        imbalance_ratio = max_class / min_class if min_class > 0 else float('inf')
        
        # Calculate entropy (higher = more balanced)
        proportions = class_counts / total
        entropy = -sum(p * np.log2(p) for p in proportions if p > 0)
        max_entropy = np.log2(len(class_counts))
        normalized_entropy = entropy / max_entropy if max_entropy > 0 else 0
        
        self.report['class_balance'] = {
            'class_distribution': class_counts.to_dict(),
            'class_percentages': {k: float(v/total*100) for k, v in class_counts.items()},
            'imbalance_ratio': float(imbalance_ratio),
            'entropy': float(entropy),
            'normalized_entropy': float(normalized_entropy),
            'is_balanced': imbalance_ratio < 2.0
        }
        
        if imbalance_ratio > 3.0:
            self.issues.append(f"Severe class imbalance (ratio: {imbalance_ratio:.1f}:1)")
        elif imbalance_ratio > 2.0:
            self.warnings.append(f"Moderate class imbalance (ratio: {imbalance_ratio:.1f}:1)")
        
        if self.verbose:
            print("4. Class Balance Analysis")
            print("-" * 40)
            for label, count in class_counts.items():
                cat_name = CATEGORIES.get(label, f'Class {label}')
                print(f"   {cat_name}: {count} ({count/total*100:.1f}%)")
            print(f"   Imbalance ratio: {imbalance_ratio:.2f}:1")
            print(f"   Balance score: {normalized_entropy*100:.1f}%\n")
    
    def _analyze_text_quality(self, df: pd.DataFrame, text_col: str):
        """Analyze text quality metrics."""
        texts = df[text_col].dropna()
        
        # Length statistics
        word_counts = texts.str.split().str.len()
        char_counts = texts.str.len()
        
        # Special character ratio
        special_char_ratios = texts.apply(
            lambda x: len(re.findall(r'[^a-zA-Z0-9\s]', str(x))) / max(len(str(x)), 1)
        )
        
        # URL count
        url_pattern = r'https?://\S+|www\.\S+'
        url_counts = texts.apply(lambda x: len(re.findall(url_pattern, str(x))))
        
        # Uppercase ratio
        uppercase_ratios = texts.apply(
            lambda x: sum(1 for c in str(x) if c.isupper()) / max(len(str(x)), 1)
        )
        
        self.report['text_quality'] = {
            'word_count': {
                'mean': float(word_counts.mean()),
                'std': float(word_counts.std()),
                'min': int(word_counts.min()),
                'max': int(word_counts.max()),
                'median': float(word_counts.median())
            },
            'char_count': {
                'mean': float(char_counts.mean()),
                'std': float(char_counts.std()),
                'min': int(char_counts.min()),
                'max': int(char_counts.max())
            },
            'special_char_ratio': {
                'mean': float(special_char_ratios.mean()),
                'max': float(special_char_ratios.max())
            },
            'texts_with_urls': int(url_counts.gt(0).sum()),
            'avg_uppercase_ratio': float(uppercase_ratios.mean()),
            'very_short_texts': int(word_counts.lt(5).sum()),
            'very_long_texts': int(word_counts.gt(100).sum())
        }
        
        very_short = word_counts.lt(5).sum()
        if very_short > len(df) * 0.1:
            self.warnings.append(f"{very_short} texts have fewer than 5 words")
        
        if self.verbose:
            print("5. Text Quality Metrics")
            print("-" * 40)
            print(f"   Average word count: {word_counts.mean():.1f}")
            print(f"   Word count std: {word_counts.std():.1f}")
            print(f"   Min/Max words: {word_counts.min()} / {word_counts.max()}")
            print(f"   Texts with URLs: {url_counts.gt(0).sum()}")
            print(f"   Very short texts (<5 words): {very_short}\n")
    
    def _analyze_outliers(self, df: pd.DataFrame, text_col: str):
        """Detect outlier texts."""
        texts = df[text_col].dropna()
        word_counts = texts.str.split().str.len()
        
        # IQR method for outliers
        Q1 = word_counts.quantile(0.25)
        Q3 = word_counts.quantile(0.75)
        IQR = Q3 - Q1
        
        lower_bound = Q1 - 1.5 * IQR
        upper_bound = Q3 + 1.5 * IQR
        
        outliers_low = word_counts.lt(lower_bound).sum()
        outliers_high = word_counts.gt(upper_bound).sum()
        
        self.report['outliers'] = {
            'lower_bound': float(max(0, lower_bound)),
            'upper_bound': float(upper_bound),
            'outliers_low': int(outliers_low),
            'outliers_high': int(outliers_high),
            'total_outliers': int(outliers_low + outliers_high),
            'outlier_percentage': float((outliers_low + outliers_high) / len(df) * 100)
        }
        
        if outliers_high > len(df) * 0.05:
            self.warnings.append(f"{outliers_high} texts are unusually long")
        
        if self.verbose:
            print("6. Outlier Detection")
            print("-" * 40)
            print(f"   Normal range: {max(0, lower_bound):.0f} - {upper_bound:.0f} words")
            print(f"   Outliers (too short): {outliers_low}")
            print(f"   Outliers (too long): {outliers_high}\n")
    
    def _analyze_potential_label_errors(self, df: pd.DataFrame, text_col: str, label_col: str):
        """Detect potential label errors using keyword analysis."""
        # Category keywords
        keywords = {
            0: ['apple', 'google', 'microsoft', 'ai', 'tech', 'software', 'app'],
            1: ['football', 'basketball', 'tennis', 'game', 'team', 'player', 'win'],
            2: ['government', 'president', 'congress', 'election', 'vote', 'policy'],
            3: ['movie', 'film', 'music', 'actor', 'netflix', 'entertainment'],
            4: ['stock', 'market', 'business', 'company', 'economy', 'profit'],
            5: ['scientist', 'research', 'nasa', 'space', 'climate', 'discovery']
        }
        
        potential_errors = []
        
        for idx, row in df.iterrows():
            text = str(row[text_col]).lower()
            label = row[label_col]
            
            # Count keyword matches for each category
            scores = {}
            for cat, kws in keywords.items():
                scores[cat] = sum(1 for kw in kws if kw in text)
            
            # Check if another category has more matches
            if scores:
                best_match = max(scores, key=scores.get)
                if scores[best_match] > 2 and best_match != label and scores.get(label, 0) == 0:
                    potential_errors.append({
                        'index': idx,
                        'current_label': int(label),
                        'suggested_label': int(best_match),
                        'confidence': scores[best_match]
                    })
        
        self.report['potential_label_errors'] = {
            'count': len(potential_errors),
            'percentage': float(len(potential_errors) / len(df) * 100),
            'examples': potential_errors[:10]  # First 10
        }
        
        if len(potential_errors) > len(df) * 0.05:
            self.warnings.append(f"Found {len(potential_errors)} potential label errors")
        
        if self.verbose:
            print("7. Potential Label Errors")
            print("-" * 40)
            print(f"   Potential errors found: {len(potential_errors)}")
            print(f"   Error rate: {len(potential_errors)/len(df)*100:.1f}%\n")
    
    def _generate_recommendations(self):
        """Generate recommendations based on analysis."""
        r = self.report
        
        # Missing values
        if r['missing_values']['missing_text'] > 0 or r['missing_values']['empty_text'] > 0:
            self.recommendations.append("Remove or impute missing/empty text values")
        
        # Duplicates
        if r['duplicates']['text_duplicates'] > 0:
            self.recommendations.append("Consider removing duplicate texts to prevent data leakage")
        
        # Class imbalance
        if r['class_balance']['imbalance_ratio'] > 2.0:
            self.recommendations.append("Apply class balancing techniques (oversampling, undersampling, or class weights)")
        
        # Text quality
        if r['text_quality']['very_short_texts'] > 10:
            self.recommendations.append("Review very short texts - they may lack sufficient information")
        
        # Outliers
        if r['outliers']['total_outliers'] > 10:
            self.recommendations.append("Review outlier texts - consider truncation or removal")
        
        # Label errors
        if r['potential_label_errors']['count'] > 5:
            self.recommendations.append("Review flagged potential label errors for correction")
        
        self.report['recommendations'] = self.recommendations
    
    def _calculate_quality_score(self):
        """Calculate overall data quality score."""
        r = self.report
        
        scores = []
        
        # Missing value score (0-100)
        missing_score = 100 - min(100, r['missing_values']['missing_percentage'] * 10)
        scores.append(('Completeness', missing_score))
        
        # Duplicate score (0-100)
        dup_score = 100 - min(100, r['duplicates']['duplicate_percentage'] * 5)
        scores.append(('Uniqueness', dup_score))
        
        # Balance score (0-100)
        balance_score = r['class_balance']['normalized_entropy'] * 100
        scores.append(('Balance', balance_score))
        
        # Outlier score (0-100)
        outlier_score = 100 - min(100, r['outliers']['outlier_percentage'] * 5)
        scores.append(('Consistency', outlier_score))
        
        # Label quality score (0-100)
        label_score = 100 - min(100, r['potential_label_errors']['percentage'] * 10)
        scores.append(('Label Quality', label_score))
        
        overall_score = sum(s[1] for s in scores) / len(scores)
        
        self.report['quality_scores'] = {
            'dimensions': dict(scores),
            'overall_score': float(overall_score),
            'grade': self._score_to_grade(overall_score)
        }
    
    def _score_to_grade(self, score: float) -> str:
        """Convert score to letter grade."""
        if score >= 90: return 'A'
        elif score >= 80: return 'B'
        elif score >= 70: return 'C'
        elif score >= 60: return 'D'
        else: return 'F'
    
    def _print_summary(self):
        """Print analysis summary."""
        print("=" * 60)
        print("QUALITY SCORE SUMMARY")
        print("=" * 60)
        
        scores = self.report['quality_scores']
        for dim, score in scores['dimensions'].items():
            bar = '█' * int(score // 10) + '░' * (10 - int(score // 10))
            print(f"   {dim:15} [{bar}] {score:.0f}%")
        
        print("-" * 60)
        print(f"   Overall Score: {scores['overall_score']:.1f}% (Grade: {scores['grade']})")
        
        if self.issues:
            print(f"\n🔴 Issues ({len(self.issues)}):")
            for issue in self.issues:
                print(f"   • {issue}")
        
        if self.warnings:
            print(f"\n🟡 Warnings ({len(self.warnings)}):")
            for warning in self.warnings:
                print(f"   • {warning}")
        
        if self.recommendations:
            print(f"\n💡 Recommendations ({len(self.recommendations)}):")
            for rec in self.recommendations:
                print(f"   • {rec}")
        
        print(f"\n{'='*60}")
        print(f"Analysis complete! | Author: {__author__}")
        print(f"{'='*60}")
    
    def export_report(self, output_path: str = 'quality_report.json'):
        """Export report to JSON file."""
        import json
        
        export_data = {
            'metadata': {
                'author': __author__,
                'website': __website__,
                'generated_at': datetime.now().isoformat()
            },
            'report': self.report,
            'issues': self.issues,
            'warnings': self.warnings
        }
        
        with open(output_path, 'w') as f:
            json.dump(export_data, f, indent=2, default=str)
        
        if self.verbose:
            print(f"Report exported to: {output_path}")


def analyze_dataset(csv_path: str) -> Dict:
    """
    Quick function to analyze a CSV dataset.
    
    Args:
        csv_path: Path to CSV file
        
    Returns:
        Quality report dictionary
    """
    df = pd.read_csv(csv_path, comment='#')
    analyzer = DataQualityAnalyzer()
    return analyzer.analyze(df)


if __name__ == "__main__":
    # Demo
    try:
        report = analyze_dataset('../data/csv/train.csv')
    except FileNotFoundError:
        print("Dataset not found. Please ensure train.csv exists in ../data/csv/")

539 lines•20 KB

python

Theme Settings

Color Scheme

Display Options

Font Size