data_processor.py

# data_processor.py

import pandas as pd
import numpy as np
from typing import Union, Dict, List, Tuple
from datetime import datetime, timedelta

class DataProcessor:
    """Handles ERP sales data processing for concentration analysis"""
    
    def __init__(self):
        """Initialize the data processor"""
        self.supported_formats = ['csv', 'xlsx', 'xls']
        self.date_columns = ['date', 'transaction_date', 'invoice_date', 'order_date']
        
    def load_data(self, file) -> pd.DataFrame:
        """Load data from uploaded file"""
        if file.name.endswith('.csv'):
            df = pd.read_csv(file)
        elif file.name.endswith(('.xlsx', '.xls')):
            df = pd.read_excel(file)
        else:
            raise ValueError(f"Unsupported file format. Please upload {', '.join(self.supported_formats)}")
        
        return df
    
    def process_erp_data(self, df: pd.DataFrame, 
                       date_range: Tuple[str, str] = None,
                       customer_col: str = None,
                       amount_col: str = None) -> pd.DataFrame:
        """Process ERP sales data for concentration analysis"""
        # If columns not specified, try to auto-detect
        if customer_col is None:
            customer_col = self._find_customer_column(df)
        
        if amount_col is None:
            amount_col = self._find_amount_column(df)
        
        # Standardize date column
        date_col = self._find_date_column(df)
        if date_col:
            df[date_col] = pd.to_datetime(df[date_col], errors='coerce')
            
            # Filter by date range if specified
            if date_range:
                start_date, end_date = date_range
                start_date = pd.to_datetime(start_date)
                end_date = pd.to_datetime(end_date)
                df = df[(df[date_col] >= start_date) & (df[date_col] <= end_date)]
        
        # Handle transaction types if present
        transaction_type_col = self._find_transaction_type_column(df)
        if transaction_type_col:
            # Identify credits/returns and adjust amounts
            credit_mask = df[transaction_type_col].str.lower().isin(['credit', 'return', 'refund', 'credit memo'])
            if credit_mask.any():
                df.loc[credit_mask, amount_col] = -abs(df.loc[credit_mask, amount_col])
        
        # Group by customer to get total revenue
        customer_summary = df.groupby(customer_col)[amount_col].sum().reset_index()
        
        # Rename columns to standardized format
        customer_summary.columns = ['customer', 'revenue']
        
        # Clean the data
        customer_summary['revenue'] = self._clean_revenue_data(customer_summary['revenue'])
        
        # Remove customers with zero or negative total revenue
        customer_summary = customer_summary[customer_summary['revenue'] > 0].copy()
        
        # Sort by revenue (descending)
        customer_summary = customer_summary.sort_values('revenue', ascending=False).reset_index(drop=True)
        
        return customer_summary
    
    def generate_sample_erp_data(self, n_transactions: int = 5000) -> pd.DataFrame:
        """Generate realistic sample ERP sales data"""
        np.random.seed(42)
        
        # Generate customer base with different sizes
        customers = []
        
        # Large enterprise customers
        for i in range(5):
            customers.append({
                'customer_id': f'ENT{i+1:03d}',
                'customer_name': f'Enterprise Corp {i+1}',
                'size_factor': np.random.uniform(8, 15)  # Large transaction factor
            })
        
        # Medium businesses
        for i in range(20):
            customers.append({
                'customer_id': f'MID{i+1:03d}',
                'customer_name': f'Midsize Inc {i+1}',
                'size_factor': np.random.uniform(3, 8)   # Medium transaction factor
            })
        
        # Small businesses
        for i in range(75):
            customers.append({
                'customer_id': f'SML{i+1:03d}',
                'customer_name': f'Small Company {i+1}',
                'size_factor': np.random.uniform(0.5, 3) # Small transaction factor
            })
        
        # Create transactions
        transactions = []
        start_date = datetime(2023, 1, 1)
        end_date = datetime(2023, 12, 31)
        date_range = (end_date - start_date).days
        
        products = [
            {'id': 'PRD001', 'name': 'Server License', 'base_price': 8000},
            {'id': 'PRD002', 'name': 'Desktop License', 'base_price': 350},
            {'id': 'PRD003', 'name': 'Cloud Service Basic', 'base_price': 1200},
            {'id': 'PRD004', 'name': 'Cloud Service Premium', 'base_price': 4500},
            {'id': 'PRD005', 'name': 'Support Contract', 'base_price': 2000},
            {'id': 'PRD006', 'name': 'Training Package', 'base_price': 3000},
            {'id': 'PRD007', 'name': 'Hardware Module A', 'base_price': 1500},
            {'id': 'PRD008', 'name': 'Hardware Module B', 'base_price': 2500},
        ]
        
        sales_reps = [
            'John Smith', 'Sarah Johnson', 'Michael Brown', 'Jennifer Davis',
            'Robert Wilson', 'Lisa Moore', 'David Taylor', 'Jessica Anderson'
        ]
        
        regions = ['North America', 'Europe', 'Asia Pacific', 'Latin America']
        
        channels = ['Direct', 'Partner', 'Distributor', 'Online']
        
        for i in range(n_transactions):
            # Select random customer with weighted probability based on size
            customer_idx = np.random.choice(
                range(len(customers)), 
                p=[c['size_factor']/sum(c['size_factor'] for c in customers) for c in customers]
            )
            customer = customers[customer_idx]
            
            # Random transaction date
            transaction_date = start_date + timedelta(days=np.random.randint(0, date_range))
            
            # Random product
            product = products[np.random.randint(0, len(products))]
            
            # Random quantity (larger customers tend to order more)
            quantity = max(1, int(np.random.normal(customer['size_factor'], customer['size_factor']/2)))
            
            # Calculate amount (with some random variation)
            unit_price = product['base_price'] * np.random.uniform(0.9, 1.1)  # Some price variation
            amount = unit_price * quantity
            
            # Occasionally create returns/credits (5% of transactions)
            transaction_type = 'Sale'
            if np.random.random() < 0.05:
                transaction_type = 'Return'
                amount = -amount * np.random.uniform(0.1, 1.0)  # Partial return in most cases
            
            # Other random attributes
            sales_rep = np.random.choice(sales_reps)
            region = np.random.choice(regions)
            channel = np.random.choice(channels)
            
            # Create transaction record
            transactions.append({
                'transaction_id': f'TRX{i+1:06d}',
                'transaction_date': transaction_date,
                'customer_id': customer['customer_id'],
                'customer_name': customer['customer_name'],
                'product_id': product['id'],
                'product_name': product['name'],
                'quantity': quantity,
                'unit_price': unit_price,
                'amount': amount,
                'transaction_type': transaction_type,
                'sales_rep': sales_rep,
                'region': region,
                'channel': channel,
                'invoice_number': f'INV-{transaction_date.strftime("%Y%m")}-{i+1:04d}'
            })
        
        return pd.DataFrame(transactions)
    
    def _find_customer_column(self, df: pd.DataFrame) -> str:
        """Find the customer name column"""
        possible_names = ['customer', 'customer name', 'customer_name', 'client', 
                        'client name', 'client_name', 'company', 'company name']
        
        for col in df.columns:
            if col.lower() in possible_names:
                return col
        
        # Try to find columns containing 'customer' or 'client'
        for col in df.columns:
            if 'customer' in col.lower() or 'client' in col.lower():
                return col
        
        raise ValueError("Could not identify customer column")
    
    def _find_amount_column(self, df: pd.DataFrame) -> str:
        """Find the transaction amount column"""
        possible_names = ['amount', 'total', 'revenue', 'sales', 'price', 
                        'total_amount', 'extended_price', 'net_amount', 'line_total']
        
        for col in df.columns:
            if col.lower() in possible_names:
                return col
        
        # Try to find columns containing 'amount', 'total' or 'price'
        for col in df.columns:
            if 'amount' in col.lower() or 'total' in col.lower() or 'price' in col.lower():
                # Check if column is numeric
                if np.issubdtype(df[col].dtype, np.number):
                    return col
        
        # If still not found, look for any numeric columns
        numeric_cols = df.select_dtypes(include=[np.number]).columns
        if len(numeric_cols) > 0:
            return numeric_cols[0]
        
        raise ValueError("Could not identify amount column")
    
    def _find_date_column(self, df: pd.DataFrame) -> str:
        """Find the transaction date column"""
        for col in df.columns:
            if col.lower() in self.date_columns:
                return col
        
        # Try to find columns containing 'date'
        for col in df.columns:
            if 'date' in col.lower():
                return col
        
        # If not found, check if any column has datetime-like values
        for col in df.columns:
            try:
                pd.to_datetime(df[col].iloc[0])
                return col
            except:
                continue
        
        return None  # It's possible there is no date column
    
    def _find_transaction_type_column(self, df: pd.DataFrame) -> str:
        """Find the transaction type column (for identifying returns/credits)"""
        possible_names = ['transaction_type', 'type', 'document_type', 'invoice_type']
        
        for col in df.columns:
            if col.lower() in possible_names:
                return col
        
        # Try to find columns containing 'type'
        for col in df.columns:
            if 'type' in col.lower():
                # Check if it has typical transaction type values
                if df[col].astype(str).str.lower().isin(['sale', 'credit', 'return', 'invoice']).any():
                    return col
        
        return None  # It's possible there is no transaction type column
    
    def _clean_revenue_data(self, revenue_series: pd.Series) -> pd.Series:
        """Clean and standardize revenue data"""
        # If string, remove currency symbols and convert
        if revenue_series.dtype == 'object':
            revenue_series = revenue_series.astype(str).str.replace(',', '')
            revenue_series = pd.to_numeric(revenue_series, errors='coerce')
        
        return revenue_series.fillna(0)