Upload 8 files

.gitattributes

@@ -1,2 +1,4 @@
 creditcard_original.csv filter=lfs diff=lfs merge=lfs -text
 data/creditcard_original.csv filter=lfs diff=lfs merge=lfs -text
+isolation_forest_model.pkl filter=lfs diff=lfs merge=lfs -text
+lof_model.pkl filter=lfs diff=lfs merge=lfs -text

explain.py

@@ -0,0 +1,159 @@
+import pandas as pd
+import numpy as np
+from typing import Dict, List, Any
+
+
+class AnomalyExplainer:
+    """Generate explanations for detected anomalies."""
+    
+    def __init__(self):
+        self.thresholds = {
+            'amount_ratio': 3.0,  # 3x user average
+            'z_score': 2.5,       # 2.5 standard deviations
+            'hour_distance': 6,  # 6 hours from common hour
+            'time_since_last': 48,  # 48 hours since last transaction
+            'night_transaction': True,  # Transaction at night
+            'weekend_ratio': 2.0  # Weekend transaction ratio
+        }
+    
+    def explain_anomaly(self, transaction: pd.Series, user_stats: Dict = None) -> List[str]:
+        """Generate explanation for a single anomalous transaction."""
+        explanations = []
+        
+        # Check amount deviation
+        if 'AmountRatio_Mean' in transaction and not pd.isna(transaction['AmountRatio_Mean']):
+            ratio = transaction['AmountRatio_Mean']
+            if ratio > self.thresholds['amount_ratio']:
+                explanations.append(f"Transaction amount is {ratio:.1f}x higher than user's average")
+            elif ratio > self.thresholds['amount_ratio'] * 0.5:
+                explanations.append(f"Transaction amount is {ratio:.1f}x higher than user's average")
+        
+        if 'Amount_ZScore' in transaction and not pd.isna(transaction['Amount_ZScore']):
+            z_score = abs(transaction['Amount_ZScore'])
+            if z_score > self.thresholds['z_score']:
+                explanations.append(f"Transaction amount deviates by {z_score:.1f} standard deviations from user's normal spending")
+        
+        # Check time-based anomalies
+        if 'Hour' in transaction and not pd.isna(transaction['Hour']):
+            hour = transaction['Hour']
+            if hour >= 22 or hour <= 5:
+                explanations.append(f"Unusual transaction time: {hour}:00 (night hours)")
+        
+        if 'Hour_Distance' in transaction and not pd.isna(transaction['Hour_Distance']):
+            hour_dist = transaction['Hour_Distance']
+            if hour_dist > self.thresholds['hour_distance']:
+                explanations.append(f"Transaction time is {hour_dist:.0f} hours away from user's typical transaction hour")
+        
+        if 'IsWeekend' in transaction and transaction['IsWeekend'] == 1:
+            explanations.append("Transaction occurred on a weekend")
+        
+        # Check frequency anomalies
+        if 'TimeSinceLastTx' in transaction and not pd.isna(transaction['TimeSinceLastTx']):
+            time_since = transaction['TimeSinceLastTx']
+            if time_since > self.thresholds['time_since_last']:
+                explanations.append(f"Unusual transaction pattern: {time_since:.0f} hours since last transaction")
+            elif time_since < 1:
+                explanations.append("Rapid succession: multiple transactions within 1 hour")
+        
+        # Check category anomalies
+        if 'Merchant Category' in transaction:
+            category = transaction['Merchant Category']
+            explanations.append(f"Merchant category: {category}")
+        
+        if 'Category_Entropy' in transaction and not pd.isna(transaction['Category_Entropy']):
+            entropy = transaction['Category_Entropy']
+            if entropy < 1.0:
+                explanations.append("User typically has low category diversity - this transaction may be unusual")
+        
+        # If no specific explanations found, provide general one
+        if not explanations:
+            explanations.append("Anomaly detected based on combined feature analysis")
+        
+        return explanations
+    
+    def explain_batch(self, df: pd.DataFrame, user_stats: Dict = None) -> pd.DataFrame:
+        """Generate explanations for a batch of transactions."""
+        df = df.copy()
+        
+        explanations = []
+        for idx, row in df.iterrows():
+            explanation = self.explain_anomaly(row, user_stats)
+            explanations.append('; '.join(explanation))
+        
+        df['Explanation'] = explanations
+        return df
+    
+    def get_feature_importance(self, transaction: pd.Series) -> Dict[str, float]:
+        """Calculate feature importance for the anomaly."""
+        importance = {}
+        
+        # Amount importance
+        if 'Amount_ZScore' in transaction and not pd.isna(transaction['Amount_ZScore']):
+            importance['Amount'] = min(abs(transaction['Amount_ZScore']) / 5.0, 1.0)
+        
+        # Time importance
+        if 'Hour_Distance' in transaction and not pd.isna(transaction['Hour_Distance']):
+            importance['Time'] = min(transaction['Hour_Distance'] / 12.0, 1.0)
+        
+        # Frequency importance
+        if 'TimeSinceLastTx' in transaction and not pd.isna(transaction['TimeSinceLastTx']):
+            importance['Frequency'] = min(transaction['TimeSinceLastTx'] / 72.0, 1.0)
+        
+        # Category importance
+        if 'Category_Entropy' in transaction and not pd.isna(transaction['Category_Entropy']):
+            importance['Category'] = max(0, 1.0 - transaction['Category_Entropy'] / 3.0)
+        
+        # Normalize importance scores
+        total = sum(importance.values()) if importance else 1
+        if total > 0:
+            importance = {k: v / total for k, v in importance.items()}
+        
+        return importance
+    
+    def generate_radar_data(self, df: pd.DataFrame) -> Dict[str, List]:
+        """Generate data for radar chart visualization."""
+        if df.empty:
+            return {'labels': [], 'datasets': []}
+        
+        # Calculate metrics for radar chart
+        metrics = {
+            'Amount': df['Amount'].mean() if 'Amount' in df.columns else 0,
+            'Frequency': df.get('TxCount_Window', pd.Series([1])).mean(),
+            'Time Variance': df.get('Hour_Variance', pd.Series([0])).mean(),
+            'Category Diversity': df.get('Category_Entropy', pd.Series([0])).mean()
+        }
+        
+        # Normalize to 0-100 scale
+        max_vals = {
+            'Amount': df['Amount'].max() if 'Amount' in df.columns else 1,
+            'Frequency': metrics['Frequency'] * 2,
+            'Time Variance': 50,
+            'Category Diversity': 3
+        }
+        
+        normalized = {
+            'Amount': (metrics['Amount'] / max_vals['Amount'] * 100) if max_vals['Amount'] > 0 else 50,
+            'Frequency': (metrics['Frequency'] / max_vals['Frequency'] * 100) if max_vals['Frequency'] > 0 else 50,
+            'Time Variance': (metrics['Time Variance'] / max_vals['Time Variance'] * 100),
+            'Category Diversity': (metrics['Category Diversity'] / max_vals['Category Diversity'] * 100)
+        }
+        
+        return {
+            'labels': list(normalized.keys()),
+            'values': [normalized[k] for k in normalized.keys()]
+        }
+    
+    def compare_with_user_baseline(self, transaction: pd.Series, user_baseline: Dict) -> Dict[str, Any]:
+        """Compare transaction with user's baseline behavior."""
+        comparison = {}
+        
+        if 'Amount' in transaction and 'mean' in user_baseline:
+            comparison['amount_vs_avg'] = transaction['Amount'] / user_baseline['mean']
+        
+        if 'Hour' in transaction:
+            comparison['hour'] = transaction['Hour']
+        
+        if 'Merchant Category' in transaction:
+            comparison['category'] = transaction['Merchant Category']
+        
+        return comparison

isolation_forest_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e83f211f7ee754655cb43cf77877a801febbc72b3e36ea27728e848c5e922948
+size 1263449

lof_model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:8b0a408c1618194e4662b2d3d3e41335a70923d38906fac79e7ec5d0856e92e4
+size 1854630

predict.py

@@ -0,0 +1,129 @@
+import pandas as pd
+import numpy as np
+import joblib
+import os
+from typing import Dict, List, Tuple, Any
+from model.train import AnomalyDetectorTrainer
+
+
+class AnomalyPredictor:
+    """Predict anomalies in credit card transactions."""
+    
+    def __init__(self, model_dir: str = 'model'):
+        self.model_dir = model_dir
+        self.trainer = AnomalyDetectorTrainer(model_dir)
+        self.current_model = None
+        self.current_scaler = None
+        self.feature_columns = []
+    
+    def load_model(self, model_type: str = 'isolation_forest') -> bool:
+        """Load the trained model."""
+        success = self.trainer.load_model(model_type)
+        if success:
+            self.current_model = self.trainer.models[model_type]
+            self.current_scaler = self.trainer.scalers[model_type]
+            self.feature_columns = self.trainer.feature_columns
+        return success
+    
+    def prepare_features(self, df: pd.DataFrame) -> np.ndarray:
+        """Prepare features for prediction."""
+        df = df.copy()
+        
+        # Select available feature columns
+        available_features = [col for col in self.feature_columns if col in df.columns]
+        
+        if not available_features:
+            # Fallback to Amount if available
+            if 'Amount' in df.columns:
+                available_features = ['Amount']
+            else:
+                raise ValueError("No valid features found in dataframe")
+        
+        # Fill missing values with median
+        for col in available_features:
+            if col in df.columns:
+                df[col] = df[col].fillna(df[col].median())
+        
+        X = df[available_features].values
+        
+        # Scale features
+        if self.current_scaler is not None:
+            X_scaled = self.current_scaler.transform(X)
+        else:
+            X_scaled = X
+        
+        return X_scaled, available_features
+    
+    def predict(self, df: pd.DataFrame) -> pd.DataFrame:
+        """Predict anomalies for transactions."""
+        if self.current_model is None:
+            raise ValueError("No model loaded. Call load_model() first.")
+        
+        df = df.copy()
+        
+        # Prepare features
+        X_scaled, used_features = self.prepare_features(df)
+        
+        # Get predictions
+        predictions = self.current_model.predict(X_scaled)
+        
+        # Get anomaly scores
+        if hasattr(self.current_model, 'decision_function'):
+            scores = self.current_model.decision_function(X_scaled)
+        elif hasattr(self.current_model, 'score_samples'):
+            scores = self.current_model.score_samples(X_scaled)
+        else:
+            scores = np.zeros(len(X_scaled))
+        
+        # Convert predictions to labels (-1 = anomaly, 1 = normal)
+        df['Prediction'] = predictions
+        df['Anomaly_Score'] = scores
+        df['Status'] = df['Prediction'].apply(lambda x: 'Anomalous' if x == -1 else 'Normal')
+        df['Confidence'] = np.abs(scores) / (np.max(np.abs(scores)) + 1e-8)
+        
+        return df
+    
+    def predict_single(self, transaction: Dict[str, Any]) -> Dict[str, Any]:
+        """Predict anomaly for a single transaction."""
+        df = pd.DataFrame([transaction])
+        result = self.predict(df)
+        
+        return {
+            'transaction_id': transaction.get('Transaction ID', 'Unknown'),
+            'prediction': result.iloc[0]['Status'],
+            'anomaly_score': float(result.iloc[0]['Anomaly_Score']),
+            'confidence': float(result.iloc[0]['Confidence'])
+        }
+    
+    def get_anomalies(self, df: pd.DataFrame) -> pd.DataFrame:
+        """Get only the anomalous transactions."""
+        predictions = self.predict(df)
+        anomalies = predictions[predictions['Status'] == 'Anomalous']
+        return anomalies
+    
+    def get_statistics(self, df: pd.DataFrame) -> Dict[str, Any]:
+        """Get prediction statistics."""
+        predictions = self.predict(df)
+        
+        total = len(predictions)
+        anomalies = len(predictions[predictions['Status'] == 'Anomalous'])
+        normal = total - anomalies
+        
+        return {
+            'total_transactions': total,
+            'anomalies_detected': anomalies,
+            'normal_transactions': normal,
+            'detection_rate': (anomalies / total * 100) if total > 0 else 0,
+            'average_anomaly_score': float(predictions[predictions['Status'] == 'Anomalous']['Anomaly_Score'].mean()) if anomalies > 0 else 0
+        }
+    
+    def batch_predict(self, df: pd.DataFrame, batch_size: int = 1000) -> pd.DataFrame:
+        """Predict in batches for large datasets."""
+        results = []
+        
+        for i in range(0, len(df), batch_size):
+            batch = df.iloc[i:i + batch_size]
+            batch_result = self.predict(batch)
+            results.append(batch_result)
+        
+        return pd.concat(results, ignore_index=True)

train.py

@@ -0,0 +1,193 @@
+import pandas as pd
+import numpy as np
+from sklearn.ensemble import IsolationForest
+from sklearn.neighbors import LocalOutlierFactor
+from sklearn.preprocessing import StandardScaler
+import joblib
+import os
+from typing import Dict, Tuple, Any
+
+
+class AnomalyDetectorTrainer:
+    """Train anomaly detection models for credit card transactions."""
+    
+    def __init__(self, model_dir: str = 'model'):
+        self.model_dir = model_dir
+        self.models = {}
+        self.scalers = {}
+        self.feature_columns = []
+        
+        # Create model directory if it doesn't exist
+        os.makedirs(model_dir, exist_ok=True)
+    
+    def prepare_features(self, df: pd.DataFrame) -> Tuple[np.ndarray, list]:
+        """Prepare features for training."""
+        df = df.copy()
+        
+        # Select numerical features
+        feature_cols = []
+        
+        # Core features
+        if 'Amount' in df.columns:
+            feature_cols.append('Amount')
+        if 'Amount_ZScore' in df.columns:
+            feature_cols.append('Amount_ZScore')
+        if 'AmountRatio_Mean' in df.columns:
+            feature_cols.append('AmountRatio_Mean')
+        
+        # Time features
+        if 'Hour' in df.columns:
+            feature_cols.append('Hour')
+        if 'DayOfWeek' in df.columns:
+            feature_cols.append('DayOfWeek')
+        if 'IsWeekend' in df.columns:
+            feature_cols.append('IsWeekend')
+        if 'IsNight' in df.columns:
+            feature_cols.append('IsNight')
+        if 'Hour_Distance' in df.columns:
+            feature_cols.append('Hour_Distance')
+        
+        # Frequency features
+        if 'TimeSinceLastTx' in df.columns:
+            feature_cols.append('TimeSinceLastTx')
+        if 'TxCount_Window' in df.columns:
+            feature_cols.append('TxCount_Window')
+        
+        # Category features
+        if 'Category_Entropy' in df.columns:
+            feature_cols.append('Category_Entropy')
+        if 'Merchant_Category_Encoded' in df.columns:
+            feature_cols.append('Merchant_Category_Encoded')
+        
+        # Filter to only columns that exist
+        available_features = [col for col in feature_cols if col in df.columns]
+        
+        if not available_features:
+            # Fallback to basic features
+            available_features = ['Amount']
+        
+        # Fill missing values
+        for col in available_features:
+            df[col] = df[col].fillna(df[col].median())
+        
+        self.feature_columns = available_features
+        X = df[available_features].values
+        
+        return X, available_features
+    
+    def train_isolation_forest(
+        self, 
+        X: np.ndarray, 
+        contamination: float = 0.1,
+        n_estimators: int = 100,
+        random_state: int = 42
+    ) -> IsolationForest:
+        """Train Isolation Forest model."""
+        model = IsolationForest(
+            contamination=contamination,
+            n_estimators=n_estimators,
+            random_state=random_state,
+            n_jobs=-1
+        )
+        model.fit(X)
+        return model
+    
+    def train_lof(
+        self,
+        X: np.ndarray,
+        contamination: float = 0.1,
+        n_neighbors: int = 20
+    ) -> LocalOutlierFactor:
+        """Train Local Outlier Factor model."""
+        model = LocalOutlierFactor(
+            contamination=contamination,
+            n_neighbors=n_neighbors,
+            novelty=True
+        )
+        model.fit(X)
+        return model
+    
+    def train_models(
+        self,
+        df: pd.DataFrame,
+        model_type: str = 'isolation_forest',
+        contamination: float = 0.1
+    ) -> Dict[str, Any]:
+        """Train selected anomaly detection model."""
+        # Prepare features
+        X, feature_cols = self.prepare_features(df)
+        
+        # Scale features
+        scaler = StandardScaler()
+        X_scaled = scaler.fit_transform(X)
+        
+        # Train model based on type
+        if model_type == 'isolation_forest':
+            model = self.train_isolation_forest(X_scaled, contamination=contamination)
+        elif model_type == 'lof':
+            model = self.train_lof(X_scaled, contamination=contamination)
+        else:
+            raise ValueError(f"Unknown model type: {model_type}")
+        
+        # Store model and scaler
+        self.models[model_type] = model
+        self.scalers[model_type] = scaler
+        
+        # Save models
+        self.save_model(model_type)
+        
+        return {
+            'model_type': model_type,
+            'contamination': contamination,
+            'feature_columns': feature_cols,
+            'n_samples': len(X),
+            'n_features': X.shape[1]
+        }
+    
+    def save_model(self, model_type: str):
+        """Save trained model and scaler to disk."""
+        model_path = os.path.join(self.model_dir, f'{model_type}_model.pkl')
+        scaler_path = os.path.join(self.model_dir, f'{model_type}_scaler.pkl')
+        
+        joblib.dump(self.models[model_type], model_path)
+        joblib.dump(self.scalers[model_type], scaler_path)
+        joblib.dump(self.feature_columns, os.path.join(self.model_dir, 'feature_columns.pkl'))
+    
+    def load_model(self, model_type: str):
+        """Load trained model and scaler from disk."""
+        model_path = os.path.join(self.model_dir, f'{model_type}_model.pkl')
+        scaler_path = os.path.join(self.model_dir, f'{model_type}_scaler.pkl')
+        
+        if os.path.exists(model_path) and os.path.exists(scaler_path):
+            self.models[model_type] = joblib.load(model_path)
+            self.scalers[model_type] = joblib.load(scaler_path)
+            self.feature_columns = joblib.load(os.path.join(self.model_dir, 'feature_columns.pkl'))
+            return True
+        return False
+    
+    def get_model_info(self, model_type: str) -> Dict[str, Any]:
+        """Get information about the trained model."""
+        if model_type not in self.models:
+            return {}
+        
+        model = self.models[model_type]
+        
+        info = {
+            'model_type': model_type,
+            'feature_columns': self.feature_columns,
+            'is_trained': True
+        }
+        
+        if model_type == 'isolation_forest':
+            info.update({
+                'n_estimators': model.n_estimators,
+                'contamination': model.contamination,
+                'max_samples': model.max_samples
+            })
+        elif model_type == 'lof':
+            info.update({
+                'n_neighbors': model.n_neighbors,
+                'contamination': model.contamination
+            })
+        
+        return info