Exploring AI Use Cases in 2025: A Comprehensive Guide

import tensorflow as tf
from tensorflow.keras import layers, models
import numpy as np

# Simple CNN for medical image classification
def create_medical_image_classifier():
    model = models.Sequential([
        layers.Conv2D(32, (3, 3), activation='relu', input_shape=(224, 224, 3)),
        layers.MaxPooling2D((2, 2)),
        layers.Conv2D(64, (3, 3), activation='relu'),
        layers.MaxPooling2D((2, 2)),
        layers.Conv2D(128, (3, 3), activation='relu'),
        layers.Flatten(),
        layers.Dense(128, activation='relu'),
        layers.Dropout(0.5),
        layers.Dense(2, activation='softmax')  # Binary classification
    ])

    model.compile(optimizer='adam',
                  loss='categorical_crossentropy',
                  metrics=['accuracy'])
    return model

# Usage example
model = create_medical_image_classifier()
# model.fit(X_train, y_train, epochs=10, validation_data=(X_val, y_val))

import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.feature_extraction.text import TfidfVectorizer

class ProductRecommendationEngine:
    def __init__(self):
        self.vectorizer = TfidfVectorizer(max_features=5000)
        self.product_features = None

    def fit(self, products_df):
        """Train the recommendation engine on product data"""
        # Combine product features into a single text field
        products_df['combined_features'] = (
            products_df['title'] + ' ' + 
            products_df['category'] + ' ' + 
            products_df['description']
        )

        # Create TF-IDF matrix
        self.product_features = self.vectorizer.fit_transform(
            products_df['combined_features']
        )
        self.products_df = products_df

    def get_recommendations(self, product_id, n_recommendations=5):
        """Get product recommendations based on similarity"""
        # Find product index
        idx = self.products_df[self.products_df['id'] == product_id].index[0]

        # Calculate similarity scores
        sim_scores = cosine_similarity(
            self.product_features[idx], 
            self.product_features
        ).flatten()

        # Get top similar products
        similar_indices = sim_scores.argsort()[-n_recommendations-1:-1][::-1]

        return self.products_df.iloc[similar_indices][['id', 'title', 'price']]

# Example usage
engine = ProductRecommendationEngine()
# engine.fit(products_dataframe)
# recommendations = engine.get_recommendations(product_id=123)

import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import pandas as pd

class FraudDetectionSystem:
    def __init__(self, contamination=0.01):
        self.scaler = StandardScaler()
        self.model = IsolationForest(contamination=contamination, random_state=42)

    def extract_features(self, transaction):
        """Extract relevant features from transaction data"""
        features = {
            'amount': transaction['amount'],
            'merchant_risk_score': transaction['merchant_risk_score'],
            'time_since_last_transaction': transaction['time_since_last'],
            'location_risk': transaction['location_risk'],
            'velocity_hour': transaction['transactions_last_hour'],
            'velocity_day': transaction['transactions_last_day'],
            'amount_deviation': abs(transaction['amount'] - transaction['avg_amount']),
            'merchant_frequency': transaction['merchant_frequency']
        }
        return pd.DataFrame([features])

    def train(self, historical_transactions):
        """Train the fraud detection model"""
        X = self.extract_features_batch(historical_transactions)
        X_scaled = self.scaler.fit_transform(X)
        self.model.fit(X_scaled)

    def predict_fraud(self, transaction):
        """Predict if a transaction is fraudulent"""
        features = self.extract_features(transaction)
        features_scaled = self.scaler.transform(features)

        # Returns -1 for anomalies (fraud), 1 for normal transactions
        prediction = self.model.predict(features_scaled)[0]
        anomaly_score = self.model.score_samples(features_scaled)[0]

        return {
            'is_fraud': prediction == -1,
            'risk_score': -anomaly_score,  # Higher score = higher risk
            'action': 'BLOCK' if prediction == -1 else 'ALLOW'
        }

# Example usage
detector = FraudDetectionSystem()
# detector.train(historical_data)
# result = detector.predict_fraud(new_transaction)

import numpy as np
from scipy.optimize import linprog
import pandas as pd

class SupplyChainOptimizer:
    def __init__(self):
        self.demand_forecast_model = None
        self.route_optimizer = None

    def optimize_inventory_allocation(self, warehouses, stores, 
                                    current_inventory, forecasted_demand):
        """
        Optimize inventory allocation across multiple locations
        """
        n_warehouses = len(warehouses)
        n_stores = len(stores)

        # Cost matrix (transportation costs from warehouses to stores)
        costs = self._calculate_transport_costs(warehouses, stores)

        # Flatten cost matrix for linear programming
        c = costs.flatten()

        # Constraints: Supply constraints for each warehouse
        A_supply = np.zeros((n_warehouses, n_warehouses * n_stores))
        for i in range(n_warehouses):
            A_supply[i, i*n_stores:(i+1)*n_stores] = 1
        b_supply = current_inventory

        # Constraints: Demand constraints for each store
        A_demand = np.zeros((n_stores, n_warehouses * n_stores))
        for j in range(n_stores):
            for i in range(n_warehouses):
                A_demand[j, i*n_stores + j] = -1
        b_demand = -forecasted_demand

        # Combine constraints
        A = np.vstack([A_supply, A_demand])
        b = np.hstack([b_supply, b_demand])

        # Solve optimization problem
        result = linprog(c, A_ub=A, b_ub=b, bounds=(0, None), method='highs')

        # Reshape solution back to matrix form
        allocation = result.x.reshape(n_warehouses, n_stores)

        return {
            'allocation_matrix': allocation,
            'total_cost': result.fun,
            'success': result.success
        }

    def predict_demand(self, historical_data, external_factors):
        """
        Predict future demand using time series analysis and external factors
        """
        # Simplified example using moving average with seasonality
        seasonal_period = 7  # Weekly seasonality
        forecast_horizon = 14  # Two weeks ahead

        # Extract trend and seasonality
        rolling_mean = historical_data.rolling(window=seasonal_period).mean()
        seasonal_pattern = historical_data.groupby(
            historical_data.index % seasonal_period
        ).mean()

        # Apply external factor adjustments
        weather_impact = external_factors.get('weather_severity', 1.0)
        event_impact = external_factors.get('local_events', 1.0)

        base_forecast = rolling_mean.iloc[-1] * weather_impact * event_impact

        # Generate forecast with seasonal adjustment
        forecast = []
        for day in range(forecast_horizon):
            seasonal_factor = seasonal_pattern[day % seasonal_period]
            daily_forecast = base_forecast * (seasonal_factor / seasonal_pattern.mean())
            forecast.append(daily_forecast)

        return pd.Series(forecast, name='demand_forecast')

# Example usage
optimizer = SupplyChainOptimizer()
# allocation = optimizer.optimize_inventory_allocation(
#     warehouses, stores, inventory, demand
# )

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split

class CustomerLifetimeValuePredictor:
    def __init__(self):
        self.model = RandomForestRegressor(n_estimators=100, random_state=42)
        self.feature_importance = None

    def engineer_features(self, customer_data):
        """
        Create features for CLV prediction
        """
        features = pd.DataFrame()

        # Recency, Frequency, Monetary (RFM) features
        features['days_since_first_purchase'] = (
            pd.Timestamp.now() - customer_data['first_purchase_date']
        ).dt.days

        features['days_since_last_purchase'] = (
            pd.Timestamp.now() - customer_data['last_purchase_date']
        ).dt.days

        features['purchase_frequency'] = (
            customer_data['total_purchases'] / 
            features['days_since_first_purchase']
        ) * 365  # Annualized

        features['avg_order_value'] = (
            customer_data['total_revenue'] / 
            customer_data['total_purchases']
        )

        # Engagement features
        features['email_open_rate'] = customer_data['emails_opened'] / customer_data['emails_sent']
        features['website_visit_frequency'] = customer_data['website_visits'] / features['days_since_first_purchase']
        features['product_categories_purchased'] = customer_data['unique_categories']

        # Customer segment features
        features['is_premium'] = customer_data['customer_tier'] == 'premium'
        features['acquisition_channel_encoded'] = pd.Categorical(
            customer_data['acquisition_channel']
        ).codes

        return features

    def train(self, historical_customers):
        """
        Train the CLV prediction model
        """
        # Engineer features
        X = self.engineer_features(historical_customers)
        y = historical_customers['lifetime_value']

        # Split data
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )

        # Train model
        self.model.fit(X_train, y_train)

        # Store feature importance
        self.feature_importance = pd.DataFrame({
            'feature': X.columns,
            'importance': self.model.feature_importances_
        }).sort_values('importance', ascending=False)

        # Return performance metrics
        train_score = self.model.score(X_train, y_train)
        test_score = self.model.score(X_test, y_test)

        return {
            'train_r2': train_score,
            'test_r2': test_score,
            'top_features': self.feature_importance.head(5)
        }

    def predict_clv(self, customer_data, time_horizon_days=365):
        """
        Predict CLV for new customers
        """
        features = self.engineer_features(customer_data)
        predicted_clv = self.model.predict(features)

        # Adjust for time horizon
        daily_value = predicted_clv / 365
        horizon_clv = daily_value * time_horizon_days

        return {
            'predicted_clv': horizon_clv[0],
            'confidence_interval': self._calculate_confidence_interval(features),
            'key_drivers': self._identify_key_drivers(features)
        }

# Example usage
clv_predictor = CustomerLifetimeValuePredictor()
# clv_predictor.train(historical_customer_data)
# prediction = clv_predictor.predict_clv(new_customer_data)

from transformers import pipeline
import pandas as pd

class SentimentAnalyzer:
    def __init__(self, model_name="bert-base-uncased"):
        self.classifier = pipeline("sentiment-analysis", model=model_name)

    def analyze_customer_feedback(self, reviews_df):
        """
        Analyze sentiment of customer reviews
        """
        results = []

        for idx, review in reviews_df.iterrows():
            # Analyze sentiment
            sentiment = self.classifier(review['text'])[0]

            # Extract aspects mentioned in review
            aspects = self._extract_aspects(review['text'])

            results.append({
                'review_id': review['id'],
                'sentiment': sentiment['label'],
                'confidence': sentiment['score'],
                'aspects': aspects,
                'actionable': self._is_actionable(sentiment, aspects)
            })

        return pd.DataFrame(results)

    def _extract_aspects(self, text):
        """Extract product/service aspects from text"""
        # Simplified example - in practice, use aspect-based sentiment analysis
        aspects = {
            'price': any(word in text.lower() for word in ['price', 'cost', 'expensive']),
            'quality': any(word in text.lower() for word in ['quality', 'durable', 'reliable']),
            'service': any(word in text.lower() for word in ['service', 'support', 'help'])
        }
        return [k for k, v in aspects.items() if v]

# Usage example
analyzer = SentimentAnalyzer()
# sentiment_results = analyzer.analyze_customer_feedback(reviews_dataframe)

import cv2
import torch
from torchvision import models, transforms

class InventoryTracker:
    def __init__(self):
        self.model = models.detection.fasterrcnn_resnet50_fpn(pretrained=True)
        self.model.eval()
        self.transform = transforms.Compose([transforms.ToTensor()])

    def count_products(self, image_path):
        """
        Count products in shelf image
        """
        # Load and preprocess image
        image = cv2.imread(image_path)
        image_rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
        image_tensor = self.transform(image_rgb)

        # Detect objects
        with torch.no_grad():
            predictions = self.model([image_tensor])

        # Filter and count products
        products = []
        for idx, label in enumerate(predictions[0]['labels']):
            if predictions[0]['scores'][idx] > 0.8:  # Confidence threshold
                box = predictions[0]['boxes'][idx]
                products.append({
                    'class': self._get_class_name(label),
                    'confidence': predictions[0]['scores'][idx].item(),
                    'location': box.tolist()
                })

        return {
            'total_products': len(products),
            'product_details': products,
            'shelf_occupancy': self._calculate_occupancy(products, image.shape)
        }

# Usage example
tracker = InventoryTracker()
# inventory_count = tracker.count_products('shelf_image.jpg')

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.neural_network import MLPClassifier
from sklearn.ensemble import VotingClassifier
import numpy as np

class PredictiveMaintenanceSystem:
    def __init__(self):
        # Create ensemble of different models
        self.rf_model = RandomForestClassifier(n_estimators=100)
        self.gb_model = GradientBoostingClassifier(n_estimators=100)
        self.nn_model = MLPClassifier(hidden_layer_sizes=(100, 50))

        self.ensemble = VotingClassifier(
            estimators=[
                ('rf', self.rf_model),
                ('gb', self.gb_model),
                ('nn', self.nn_model)
            ],
            voting='soft'
        )

    def predict_failure(self, sensor_data):
        """
        Predict equipment failure from sensor readings
        """
        # Extract features from raw sensor data
        features = self._extract_features(sensor_data)

        # Get probability of failure
        failure_prob = self.ensemble.predict_proba(features)[0, 1]

        # Determine maintenance action
        if failure_prob > 0.8:
            action = 'IMMEDIATE_MAINTENANCE'
        elif failure_prob > 0.5:
            action = 'SCHEDULE_MAINTENANCE'
        else:
            action = 'MONITOR'

        return {
            'failure_probability': failure_prob,
            'recommended_action': action,
            'confidence': self._calculate_confidence(features),
            'contributing_factors': self._identify_factors(features)
        }

    def _extract_features(self, sensor_data):
        """Extract relevant features from sensor data"""
        features = []

        # Temperature statistics
        features.extend([
            sensor_data['temperature'].mean(),
            sensor_data['temperature'].std(),
            sensor_data['temperature'].max()
        ])

        # Vibration analysis
        features.extend([
            sensor_data['vibration'].mean(),
            np.percentile(sensor_data['vibration'], 95)
        ])

        # Operating hours
        features.append(sensor_data['operating_hours'])

        return np.array(features).reshape(1, -1)

# Usage example
maintenance_system = PredictiveMaintenanceSystem()
# maintenance_system.train(historical_sensor_data, failure_labels)
# prediction = maintenance_system.predict_failure(current_sensor_data)

import numpy as np

class DifferentialPrivacyEngine:
    def __init__(self, epsilon=1.0):
        """
        Initialize with privacy budget epsilon
        Lower epsilon = more privacy, less accuracy
        """
        self.epsilon = epsilon

    def add_laplace_noise(self, data, sensitivity):
        """
        Add Laplace noise to preserve privacy
        """
        scale = sensitivity / self.epsilon
        noise = np.random.laplace(0, scale, data.shape)
        return data + noise

    def private_mean(self, data, lower_bound, upper_bound):
        """
        Calculate differentially private mean
        """
        # Clip data to bounds
        clipped_data = np.clip(data, lower_bound, upper_bound)

        # Calculate sensitivity
        sensitivity = (upper_bound - lower_bound) / len(data)

        # Add noise to mean
        true_mean = np.mean(clipped_data)
        private_mean = self.add_laplace_noise(true_mean, sensitivity)

        return private_mean

    def private_count(self, data, condition):
        """
        Count with differential privacy
        """
        true_count = np.sum(condition(data))
        sensitivity = 1  # Each person contributes at most 1

        private_count = self.add_laplace_noise(true_count, sensitivity)
        return max(0, int(private_count))  # Ensure non-negative

# Usage example
privacy_engine = DifferentialPrivacyEngine(epsilon=0.5)
# private_average = privacy_engine.private_mean(salaries, 0, 200000)

from sklearn.metrics import confusion_matrix
import pandas as pd

class FairnessAuditor:
    def __init__(self):
        self.metrics = {}

    def audit_model_fairness(self, model, X_test, y_test, sensitive_features):
        """
        Audit model for bias across sensitive features
        """
        predictions = model.predict(X_test)

        for feature in sensitive_features:
            # Calculate metrics for each group
            groups = X_test[feature].unique()
            group_metrics = {}

            for group in groups:
                mask = X_test[feature] == group
                group_pred = predictions[mask]
                group_true = y_test[mask]

                # Calculate fairness metrics
                tn, fp, fn, tp = confusion_matrix(group_true, group_pred).ravel()

                group_metrics[group] = {
                    'accuracy': (tp + tn) / (tp + tn + fp + fn),
                    'true_positive_rate': tp / (tp + fn) if (tp + fn) > 0 else 0,
                    'false_positive_rate': fp / (fp + tn) if (fp + tn) > 0 else 0,
                    'selection_rate': (tp + fp) / len(group_pred)
                }

            # Calculate disparate impact
            rates = [m['selection_rate'] for m in group_metrics.values()]
            disparate_impact = min(rates) / max(rates) if max(rates) > 0 else 0

            self.metrics[feature] = {
                'group_metrics': group_metrics,
                'disparate_impact': disparate_impact,
                'fair': disparate_impact > 0.8  # 80% rule
            }

        return self.generate_report()

    def generate_report(self):
        """Generate fairness audit report"""
        report = []

        for feature, metrics in self.metrics.items():
            report.append(f"\nFairness Analysis for {feature}:")
            report.append(f"Disparate Impact: {metrics['disparate_impact']:.3f}")
            report.append(f"Passes 80% Rule: {'Yes' if metrics['fair'] else 'No'}")

            for group, group_metrics in metrics['group_metrics'].items():
                report.append(f"\n{group}:")
                for metric, value in group_metrics.items():
                    report.append(f"  {metric}: {value:.3f}")

        return '\n'.join(report)

# Usage example
auditor = FairnessAuditor()
# fairness_report = auditor.audit_model_fairness(
#     model, X_test, y_test, ['gender', 'race', 'age_group']
# )

class MLOpsPipeline:
    def __init__(self):
        self.models = {}
        self.metrics = {}
        self.deployment_history = []

    def train_model(self, model_name, training_data, hyperparameters):
        """
        Train model with automatic versioning and tracking
        """
        # Version control
        version = f"{model_name}_v{len(self.models.get(model_name, [])) + 1}"

        # Training with monitoring
        model = self._train_with_monitoring(
            training_data, hyperparameters
        )

        # Automated testing
        test_results = self._run_tests(model, version)

        # Store model and metadata
        self.models[model_name] = self.models.get(model_name, [])
        self.models[model_name].append({
            'version': version,
            'model': model,
            'hyperparameters': hyperparameters,
            'test_results': test_results,
            'timestamp': pd.Timestamp.now()
        })

        return version

    def deploy_model(self, model_name, version, deployment_config):
        """
        Deploy model with canary release and monitoring
        """
        # Validate model exists
        model_info = self._get_model(model_name, version)

        # Canary deployment
        deployment = {
            'model': model_name,
            'version': version,
            'config': deployment_config,
            'status': 'canary',
            'traffic_percentage': 10,
            'timestamp': pd.Timestamp.now()
        }

        # Monitor performance
        self._setup_monitoring(deployment)

        # Gradual rollout based on metrics
        self.deployment_history.append(deployment)

        return deployment

    def monitor_performance(self, model_name, metrics):
        """
        Continuous monitoring and alerting
        """
        current_deployment = self._get_current_deployment(model_name)

        # Check for model drift
        drift_detected = self._detect_drift(metrics)

        # Performance degradation check
        performance_issues = self._check_performance(metrics)

        if drift_detected or performance_issues:
            self._trigger_retraining(model_name)
            self._send_alert(model_name, metrics)

        # Log metrics
        self.metrics[model_name] = self.metrics.get(model_name, [])
        self.metrics[model_name].append({
            'timestamp': pd.Timestamp.now(),
            'metrics': metrics,
            'drift': drift_detected,
            'issues': performance_issues
        })

# Usage example
mlops = MLOpsPipeline()
# version = mlops.train_model('fraud_detector', training_data, hyperparams)
# mlops.deploy_model('fraud_detector', version, deployment_config)