# 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)