pizza_classifier.py

"""
Random Acts of Pizza Classification - Modern Implementation

This module provides modernized utilities for classifying pizza requests
using machine learning techniques with updated libraries and best practices.
"""

from pathlib import Path
from typing import Tuple, List, Dict, Optional, Union
import json
import re
import warnings

import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
from sklearn.model_selection import train_test_split, GridSearchCV, cross_val_score
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.naive_bayes import BernoulliNB
from sklearn.metrics import (
    accuracy_score,
    roc_auc_score,
    classification_report,
    confusion_matrix,
    roc_curve,
)
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.pipeline import Pipeline
from sklearn.decomposition import TruncatedSVD
import matplotlib.pyplot as plt
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

warnings.filterwarnings('ignore')


class PizzaDataLoader:
    """Load and preprocess pizza request data."""

    def __init__(self, data_dir: Union[str, Path] = "data"):
        """
        Initialize data loader.

        Args:
            data_dir: Directory containing train.json and test.json files
        """
        self.data_dir = Path(data_dir)

    def load_data(
        self,
        train_file: str = "train.json",
        test_file: str = "test.json",
        dev_split: int = 1000
    ) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
        """
        Load training and test data from JSON files.

        Args:
            train_file: Name of training data file
            test_file: Name of test data file
            dev_split: Number of samples to use for dev set

        Returns:
            Tuple of (training_data, dev_data, test_data) DataFrames
        """
        # Load training data
        train_path = self.data_dir / train_file
        with open(train_path, 'r', encoding='utf-8') as f:
            train_data = json.load(f)

        # Load test data
        test_path = self.data_dir / test_file
        with open(test_path, 'r', encoding='utf-8') as f:
            test_data = json.load(f)

        # Normalize JSON to DataFrame
        train_df = pd.json_normalize(train_data)
        test_df = pd.json_normalize(test_data)

        # Split dev set from training data
        dev_df = train_df.iloc[:dev_split].copy()
        train_df = train_df.iloc[dev_split:].copy()

        return train_df, dev_df, test_df


class TextPreprocessor:
    """Modern text preprocessing utilities."""

    @staticmethod
    def clean_text(
        text: str,
        remove_special_chars: bool = True,
        convert_digits: bool = True,
        max_word_length: Optional[int] = None
    ) -> str:
        """
        Clean and preprocess text data.

        Args:
            text: Input text to clean
            remove_special_chars: Whether to remove special characters
            convert_digits: Whether to convert digits to 'number' token
            max_word_length: Maximum word length (truncates longer words)

        Returns:
            Cleaned text string
        """
        if remove_special_chars:
            # Remove non-alphanumeric characters
            text = re.sub(r'[?|$|.|!|@|\n|(|)|<|>|_|\-|,|\']', ' ', text)

        if convert_digits:
            # Convert all digits to 'number' token
            text = re.sub(r'\d+', 'number', text)

        # Normalize whitespace
        text = re.sub(r'\s+', ' ', text).strip()

        if max_word_length:
            # Truncate long words
            words = text.split()
            words = [word[:max_word_length] for word in words]
            text = ' '.join(words)

        return text

    @staticmethod
    def create_text_corpus(
        df: pd.DataFrame,
        text_columns: List[str] = None
    ) -> List[str]:
        """
        Create text corpus from DataFrame columns.

        Args:
            df: DataFrame containing text columns
            text_columns: List of column names to combine

        Returns:
            List of combined text strings
        """
        if text_columns is None:
            text_columns = ['request_title', 'request_text']

        corpus = []
        for _, row in df.iterrows():
            combined_text = ' '.join([
                str(row.get(col, '')) for col in text_columns
            ])
            corpus.append(combined_text)

        return corpus


class FeatureEngineering:
    """Feature engineering for pizza request classification."""

    def __init__(self):
        """Initialize feature engineering tools."""
        self.sentiment_analyzer = SentimentIntensityAnalyzer()
        self.scaler = MinMaxScaler()

    def extract_temporal_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Extract temporal features from Unix timestamps.

        Args:
            df: DataFrame with unix_timestamp_of_request_utc column

        Returns:
            DataFrame with added temporal features
        """
        df = df.copy()

        # Convert Unix timestamp to datetime
        df['datetime'] = pd.to_datetime(
            df['unix_timestamp_of_request_utc'],
            unit='s'
        )

        # Extract temporal features
        df['hour'] = df['datetime'].dt.hour
        df['day_of_week'] = df['datetime'].dt.dayofweek
        df['month'] = df['datetime'].dt.month
        df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)

        # Create time buckets (morning, afternoon, evening, night)
        df['time_bucket'] = pd.cut(
            df['hour'],
            bins=[0, 6, 12, 18, 24],
            labels=['night', 'morning', 'afternoon', 'evening'],
            include_lowest=True
        )

        return df

    def extract_sentiment_features(
        self,
        df: pd.DataFrame,
        text_column: str = 'request_title'
    ) -> pd.DataFrame:
        """
        Extract sentiment features using VADER.

        Args:
            df: DataFrame with text column
            text_column: Name of column containing text

        Returns:
            DataFrame with added sentiment features
        """
        df = df.copy()

        sentiments = []
        for text in df[text_column]:
            scores = self.sentiment_analyzer.polarity_scores(str(text))
            sentiments.append(scores['compound'])

        df[f'{text_column}_sentiment'] = sentiments

        return df

    def extract_user_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """
        Extract and engineer user-related features.

        Args:
            df: DataFrame with user feature columns

        Returns:
            DataFrame with engineered user features
        """
        df = df.copy()

        # Binary feature: whether giver username is known
        df['has_giver'] = (
            df['giver_username_if_known'] != 'N/A'
        ).astype(int)

        # Engagement metrics
        df['vote_diff_change'] = (
            df['requester_upvotes_minus_downvotes_at_retrieval'] -
            df['requester_upvotes_minus_downvotes_at_request']
        )

        # Activity ratios
        df['comment_to_post_ratio'] = np.where(
            df['requester_number_of_posts_at_request'] > 0,
            df['requester_number_of_comments_at_request'] /
            df['requester_number_of_posts_at_request'],
            0
        )

        return df

    def create_feature_matrix(
        self,
        df: pd.DataFrame,
        feature_columns: Optional[List[str]] = None
    ) -> np.ndarray:
        """
        Create normalized feature matrix from DataFrame.

        Args:
            df: DataFrame with feature columns
            feature_columns: List of columns to include (None = use defaults)

        Returns:
            Normalized feature matrix
        """
        if feature_columns is None:
            feature_columns = [
                'number_of_downvotes_of_request_at_retrieval',
                'number_of_upvotes_of_request_at_retrieval',
                'request_number_of_comments_at_retrieval',
                'requester_account_age_in_days_at_request',
                'requester_number_of_comments_at_request',
                'requester_number_of_posts_at_request',
                'requester_number_of_subreddits_at_request',
                'hour',
                'month',
            ]

        feature_matrix = df[feature_columns].fillna(0).values

        # Normalize features
        feature_matrix = self.scaler.fit_transform(feature_matrix)

        return feature_matrix


class PizzaClassifier:
    """Modern pizza request classifier with multiple model options."""

    def __init__(
        self,
        model_type: str = 'logistic',
        use_tfidf: bool = True,
        ngram_range: Tuple[int, int] = (1, 2),
        use_pca: bool = False,
        n_components: int = 600
    ):
        """
        Initialize classifier.

        Args:
            model_type: Type of model ('logistic', 'random_forest', 'gradient_boosting')
            use_tfidf: Whether to use TF-IDF (vs CountVectorizer)
            ngram_range: N-gram range for text vectorization
            use_pca: Whether to use dimensionality reduction
            n_components: Number of PCA components if use_pca=True
        """
        self.model_type = model_type
        self.use_tfidf = use_tfidf
        self.ngram_range = ngram_range
        self.use_pca = use_pca
        self.n_components = n_components

        # Initialize vectorizer
        VectorizerClass = TfidfVectorizer if use_tfidf else CountVectorizer
        self.vectorizer = VectorizerClass(
            min_df=2,
            max_df=0.95,
            lowercase=True,
            stop_words='english',
            strip_accents='unicode',
            ngram_range=ngram_range
        )

        # Initialize model
        self.model = self._get_model()
        self.pca = TruncatedSVD(n_components=n_components) if use_pca else None

    def _get_model(self):
        """Get the specified model type."""
        if self.model_type == 'logistic':
            return LogisticRegression(
                penalty='l2',
                C=0.1,
                max_iter=1000,
                random_state=42
            )
        elif self.model_type == 'random_forest':
            return RandomForestClassifier(
                n_estimators=100,
                max_depth=None,
                min_samples_split=2,
                random_state=42,
                n_jobs=-1
            )
        elif self.model_type == 'gradient_boosting':
            return GradientBoostingClassifier(
                n_estimators=100,
                learning_rate=0.1,
                max_depth=3,
                random_state=42
            )
        else:
            raise ValueError(f"Unknown model type: {self.model_type}")

    def fit(
        self,
        text_data: List[str],
        additional_features: Optional[np.ndarray],
        labels: np.ndarray
    ):
        """
        Fit the classifier.

        Args:
            text_data: List of text strings
            additional_features: Additional feature matrix (optional)
            labels: Target labels
        """
        # Vectorize text
        text_features = self.vectorizer.fit_transform(text_data)

        # Apply PCA if specified
        if self.use_pca:
            text_features = self.pca.fit_transform(text_features)

        # Combine with additional features if provided
        if additional_features is not None:
            if hasattr(text_features, 'toarray'):
                text_features = text_features.toarray()
            features = np.hstack([text_features, additional_features])
        else:
            features = text_features

        # Fit model
        self.model.fit(features, labels)

    def predict(
        self,
        text_data: List[str],
        additional_features: Optional[np.ndarray] = None
    ) -> np.ndarray:
        """
        Make predictions.

        Args:
            text_data: List of text strings
            additional_features: Additional feature matrix (optional)

        Returns:
            Predicted labels
        """
        # Vectorize text
        text_features = self.vectorizer.transform(text_data)

        # Apply PCA if specified
        if self.use_pca:
            text_features = self.pca.transform(text_features)

        # Combine with additional features if provided
        if additional_features is not None:
            if hasattr(text_features, 'toarray'):
                text_features = text_features.toarray()
            features = np.hstack([text_features, additional_features])
        else:
            features = text_features

        return self.model.predict(features)

    def predict_proba(
        self,
        text_data: List[str],
        additional_features: Optional[np.ndarray] = None
    ) -> np.ndarray:
        """Get prediction probabilities."""
        # Vectorize text
        text_features = self.vectorizer.transform(text_data)

        # Apply PCA if specified
        if self.use_pca:
            text_features = self.pca.transform(text_features)

        # Combine with additional features if provided
        if additional_features is not None:
            if hasattr(text_features, 'toarray'):
                text_features = text_features.toarray()
            features = np.hstack([text_features, additional_features])
        else:
            features = text_features

        return self.model.predict_proba(features)

    def evaluate(
        self,
        text_data: List[str],
        additional_features: Optional[np.ndarray],
        labels: np.ndarray
    ) -> Dict[str, float]:
        """
        Evaluate model performance.

        Args:
            text_data: List of text strings
            additional_features: Additional feature matrix (optional)
            labels: True labels

        Returns:
            Dictionary of evaluation metrics
        """
        predictions = self.predict(text_data, additional_features)
        probabilities = self.predict_proba(text_data, additional_features)

        metrics = {
            'accuracy': accuracy_score(labels, predictions),
            'roc_auc': roc_auc_score(labels, probabilities[:, 1]),
        }

        print(f"\n{self.model_type.upper()} Model Evaluation")
        print("=" * 75)
        print(f"Accuracy: {metrics['accuracy']:.4f}")
        print(f"ROC AUC: {metrics['roc_auc']:.4f}")
        print("\nClassification Report:")
        print(classification_report(labels, predictions))
        print("=" * 75)

        return metrics

    @staticmethod
    def plot_roc_curve(y_true: np.ndarray, y_pred_proba: np.ndarray):
        """
        Plot ROC curve.

        Args:
            y_true: True labels
            y_pred_proba: Predicted probabilities
        """
        fpr, tpr, thresholds = roc_curve(y_true, y_pred_proba)
        auc = roc_auc_score(y_true, y_pred_proba)

        plt.figure(figsize=(8, 6))
        plt.plot(fpr, tpr, 'r-', linewidth=2, label=f'ROC (AUC = {auc:.3f})')
        plt.plot([0, 1], [0, 1], 'b--', linewidth=1, label='Random')
        plt.xlabel('False Positive Rate', fontsize=12)
        plt.ylabel('True Positive Rate', fontsize=12)
        plt.title('ROC Curve', fontsize=14)
        plt.legend(loc='lower right')
        plt.grid(alpha=0.3)
        plt.tight_layout()
        plt.show()


def main():
    """Example usage of the pizza classifier."""
    # Load data
    print("Loading data...")
    loader = PizzaDataLoader()
    train_df, dev_df, test_df = loader.load_data()

    # Feature engineering
    print("Engineering features...")
    fe = FeatureEngineering()

    # Process training data
    train_df = fe.extract_temporal_features(train_df)
    train_df = fe.extract_sentiment_features(train_df)
    train_df = fe.extract_user_features(train_df)

    # Process dev data
    dev_df = fe.extract_temporal_features(dev_df)
    dev_df = fe.extract_sentiment_features(dev_df)
    dev_df = fe.extract_user_features(dev_df)

    # Create text corpus
    preprocessor = TextPreprocessor()
    train_corpus = preprocessor.create_text_corpus(train_df)
    dev_corpus = preprocessor.create_text_corpus(dev_df)

    # Get labels
    train_labels = train_df['requester_received_pizza'].values
    dev_labels = dev_df['requester_received_pizza'].values

    # Create feature matrices
    train_features = fe.create_feature_matrix(train_df)
    dev_features = fe.create_feature_matrix(dev_df)

    # Train and evaluate models
    print("\n" + "=" * 75)
    print("TRAINING AND EVALUATING MODELS")
    print("=" * 75)

    for model_type in ['logistic', 'random_forest', 'gradient_boosting']:
        print(f"\n\nTraining {model_type} model...")
        classifier = PizzaClassifier(model_type=model_type)
        classifier.fit(train_corpus, train_features, train_labels)

        print(f"\nEvaluating {model_type} model on dev set...")
        metrics = classifier.evaluate(dev_corpus, dev_features, dev_labels)

        # Plot ROC curve
        probabilities = classifier.predict_proba(dev_corpus, dev_features)
        classifier.plot_roc_curve(dev_labels, probabilities[:, 1])


if __name__ == '__main__':
    main()