1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

import numpy as np
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split
from sklearn.svm import SVC
from sklearn.pipeline import Pipeline
 
def demonstrate_scaling_importance():
    """
    Show the impact of feature scaling on SVM performance.
    """
    np.random.seed(42)
    
    # Generate data with features on very different scales
    n_samples = 500
    
    # Feature 1: small scale (e.g., age: 0-100)
    X1 = np.random.randn(n_samples, 1) * 15 + 40
    
    # Feature 2: large scale (e.g., income: 0-1,000,000)
    X2 = np.random.randn(n_samples, 1) * 100000 + 500000
    
    # Feature 3: small scale
    X3 = np.random.randn(n_samples, 1) * 5
    
    X = np.hstack([X1, X2, X3])
    
    # True decision depends on normalized values
    y = (X1.ravel() / 15 + X3.ravel() / 5 > 0).astype(int)
    
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.3, random_state=42
    )
    
    print("Impact of Feature Scaling on SVM")
    print("=" * 50)
    print(f"Feature scales: {X.std(axis=0)}")
    print()
    
    # Without scaling
    svm_unscaled = SVC(kernel='rbf', C=1.0, gamma='scale')
    svm_unscaled.fit(X_train, y_train)
    acc_unscaled = svm_unscaled.score(X_test, y_test)
    
    print(f"Without scaling: {acc_unscaled:.3f} accuracy")
    print(f"  Number of SVs: {svm_unscaled.n_support_.sum()}")
    
    # With StandardScaler
    pipeline_standard = Pipeline([
        ('scaler', StandardScaler()),
        ('svm', SVC(kernel='rbf', C=1.0, gamma='scale'))
    ])
    pipeline_standard.fit(X_train, y_train)
    acc_standard = pipeline_standard.score(X_test, y_test)
    
    print(f"\nWith StandardScaler: {acc_standard:.3f} accuracy")
    print(f"  Number of SVs: {pipeline_standard.named_steps['svm'].n_support_.sum()}")
    
    # With MinMaxScaler
    pipeline_minmax = Pipeline([
        ('scaler', MinMaxScaler()),
        ('svm', SVC(kernel='rbf', C=1.0, gamma='scale'))
    ])
    pipeline_minmax.fit(X_train, y_train)
    acc_minmax = pipeline_minmax.score(X_test, y_test)
    
    print(f"\nWith MinMaxScaler: {acc_minmax:.3f} accuracy")
    print(f"  Number of SVs: {pipeline_minmax.named_steps['svm'].n_support_.sum()}")
    
    print(f"\nImprovement from scaling: {(acc_standard - acc_unscaled) / acc_unscaled * 100:.1f}%")
 
 
def proper_preprocessing_pipeline():
    """
    Demonstrate proper preprocessing pipeline for production.
    """
    from sklearn.impute import SimpleImputer
    from sklearn.compose import ColumnTransformer
    from sklearn.preprocessing import OneHotEncoder
    
    print("\nProduction-Ready Preprocessing Pipeline")
    print("=" * 50)
    
    # Define preprocessing for different column types
    numeric_features = [0, 1, 2]  # indices
    categorical_features = [3, 4]  # indices
    
    numeric_transformer = Pipeline(steps=[
        ('imputer', SimpleImputer(strategy='median')),
        ('scaler', StandardScaler())
    ])
    
    categorical_transformer = Pipeline(steps=[
        ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
        ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False))
    ])
    
    preprocessor = ColumnTransformer(
        transformers=[
            ('num', numeric_transformer, numeric_features),
            ('cat', categorical_transformer, categorical_features)
        ])
    
    # Full pipeline
    full_pipeline = Pipeline(steps=[
        ('preprocessor', preprocessor),
        ('classifier', SVC(kernel='rbf', C=1.0, gamma='scale'))
    ])
    
    print("Pipeline structure:")
    print("  1. Numeric: Impute (median) → StandardScaler")
    print("  2. Categorical: Impute (constant) → OneHotEncoder")
    print("  3. SVC with RBF kernel")
    print()
    print("This pipeline handles missing values, mixed types,")
    print("and ensures proper scaling automatically.")
    
    return full_pipeline
 
 
# Run demonstrations
demonstrate_scaling_importance()
pipeline = proper_preprocessing_pipeline()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

import numpy as np
from sklearn.svm import SVC
from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.datasets import make_classification
from scipy.stats import loguniform
import time
 
def grid_search_example():
    """
    Demonstrate proper grid search for SVM hyperparameters.
    """
    # Generate sample data
    X, y = make_classification(
        n_samples=1000, n_features=20, n_informative=10,
        n_redundant=5, random_state=42
    )
    
    print("SVM Hyperparameter Tuning with Grid Search")
    print("=" * 50)
    
    # Define the pipeline
    pipeline = Pipeline([
        ('scaler', StandardScaler()),
        ('svm', SVC())
    ])
    
    # Define parameter grid
    # Note: Use 'svm__' prefix because SVM is in a pipeline
    param_grid = {
        'svm__C': [0.01, 0.1, 1, 10, 100],
        'svm__gamma': [0.001, 0.01, 0.1, 1, 10],
        'svm__kernel': ['rbf']
    }
    
    print(f"Grid size: {5 * 5 * 1} = 25 configurations")
    print(f"With 5-fold CV: {25 * 5} = 125 fits")
    print()
    
    # Perform grid search
    start_time = time.time()
    grid_search = GridSearchCV(
        pipeline,
        param_grid,
        cv=5,
        scoring='accuracy',
        n_jobs=-1,  # Use all CPUs
        verbose=1
    )
    grid_search.fit(X, y)
    elapsed = time.time() - start_time
    
    print(f"\nGrid search completed in {elapsed:.1f} seconds")
    print(f"\nBest parameters: {grid_search.best_params_}")
    print(f"Best CV score: {grid_search.best_score_:.4f}")
    
    # Show top 5 configurations
    results = grid_search.cv_results_
    sorted_idx = np.argsort(results['mean_test_score'])[::-1]
    
    print("\nTop 5 configurations:")
    for i, idx in enumerate(sorted_idx[:5]):
        print(f"  {i+1}. C={results['param_svm__C'][idx]}, "
              f"γ={results['param_svm__gamma'][idx]}: "
              f"{results['mean_test_score'][idx]:.4f} ± {results['std_test_score'][idx]:.4f}")
    
    return grid_search
 
 
def randomized_search_example():
    """
    Demonstrate randomized search for larger parameter spaces.
    """
    X, y = make_classification(
        n_samples=1000, n_features=20, n_informative=10,
        n_redundant=5, random_state=42
    )
    
    print("\nSVM Hyperparameter Tuning with Randomized Search")
    print("=" * 50)
    
    pipeline = Pipeline([
        ('scaler', StandardScaler()),
        ('svm', SVC())
    ])
    
    # Continuous distributions for parameters
    param_distributions = {
        'svm__C': loguniform(1e-3, 1e3),       # Log-uniform from 0.001 to 1000
        'svm__gamma': loguniform(1e-4, 1e1),   # Log-uniform from 0.0001 to 10
        'svm__kernel': ['rbf', 'poly'],
        'svm__degree': [2, 3, 4],  # Only used for poly kernel
    }
    
    # Sample 30 configurations
    n_iter = 30
    
    start_time = time.time()
    random_search = RandomizedSearchCV(
        pipeline,
        param_distributions,
        n_iter=n_iter,
        cv=5,
        scoring='accuracy',
        n_jobs=-1,
        random_state=42,
        verbose=1
    )
    random_search.fit(X, y)
    elapsed = time.time() - start_time
    
    print(f"\nRandomized search completed in {elapsed:.1f} seconds")
    print(f"  (Sampled {n_iter} configurations)")
    print(f"\nBest parameters: {random_search.best_params_}")
    print(f"Best CV score: {random_search.best_score_:.4f}")
    
    return random_search
 
 
def practical_tuning_strategy():
    """
    Demonstrate a practical multi-stage tuning strategy.
    """
    print("\nPractical Tuning Strategy")
    print("=" * 50)
    
    strategy = """
    Stage 1: Coarse Grid (find approximate region)
    ─────────────────────────────────────────────
    C: [0.01, 0.1, 1, 10, 100]
    γ: [0.001, 0.01, 0.1, 1]
    
    Stage 2: Fine Grid (refine in best region)
    ─────────────────────────────────────────────
    If Stage 1 best is C=10, γ=0.1:
    C: [5, 7, 10, 15, 20]
    γ: [0.05, 0.07, 0.1, 0.15, 0.2]
    
    Stage 3 (optional): Very Fine Tuning
    ─────────────────────────────────────────────
    If needed, narrow further around Stage 2 best.
    Often diminishing returns here.
    
    Pro Tips:
    ─────────────────────────────────────────────
    1. Always use log scale for C and γ
    2. Start coarse to save time
    3. Use 5-fold CV minimum; 10-fold for small data
    4. Monitor for overfit: if best C is extreme, investigate
    5. n_jobs=-1 parallelizes across CPU cores
    """
    print(strategy)
 
 
# Run examples
grid_search_example()
randomized_search_example()
practical_tuning_strategy()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

import numpy as np
import joblib
import time
from sklearn.svm import SVC
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
 
class SVMProductionWrapper:
    """
    Production-ready wrapper for SVM models.
    
    Provides:
    - Input validation
    - Logging
    - Latency monitoring
    - Error handling
    """
    
    def __init__(self, model_path):
        """Load model from disk."""
        self.model = joblib.load(model_path)
        self.prediction_times = []
        self.n_predictions = 0
        
    def validate_input(self, X):
        """Validate input data format and values."""
        # Check type
        if not isinstance(X, np.ndarray):
            X = np.array(X)
        
        # Check dimensions
        if X.ndim == 1:
            X = X.reshape(1, -1)
        
        # Check feature count
        expected_features = self._get_expected_features()
        if X.shape[1] != expected_features:
            raise ValueError(
                f"Expected {expected_features} features, got {X.shape[1]}"
            )
        
        # Check for NaN/Inf
        if np.any(np.isnan(X)) or np.any(np.isinf(X)):
            raise ValueError("Input contains NaN or Inf values")
        
        return X
    
    def _get_expected_features(self):
        """Extract expected feature count from model."""
        # Handle pipeline
        if hasattr(self.model, 'named_steps'):
            svm = self.model.named_steps.get('svm', self.model)
        else:
            svm = self.model
        
        # Get from support vectors shape
        if hasattr(svm, 'support_vectors_'):
            return svm.support_vectors_.shape[1]
        
        return None  # Unknown
    
    def predict(self, X, return_proba=False):
        """
        Make predictions with validation and monitoring.
        """
        start_time = time.time()
        
        try:
            # Validate
            X = self.validate_input(X)
            
            # Predict
            if return_proba and hasattr(self.model, 'predict_proba'):
                result = self.model.predict_proba(X)
            else:
                result = self.model.predict(X)
            
            # Record latency
            elapsed = time.time() - start_time
            self.prediction_times.append(elapsed)
            self.n_predictions += len(X)
            
            return result
            
        except Exception as e:
            # Log error (in production, use proper logging)
            print(f"Prediction error: {e}")
            raise
    
    def get_latency_stats(self):
        """Return latency statistics."""
        if not self.prediction_times:
            return {}
        
        times = np.array(self.prediction_times)
        return {
            'n_predictions': self.n_predictions,
            'n_calls': len(times),
            'mean_latency_ms': np.mean(times) * 1000,
            'p50_latency_ms': np.percentile(times, 50) * 1000,
            'p95_latency_ms': np.percentile(times, 95) * 1000,
            'p99_latency_ms': np.percentile(times, 99) * 1000,
        }
    
    def model_info(self):
        """Return model information."""
        info = {
            'type': type(self.model).__name__,
        }
        
        # Extract SVM-specific info
        if hasattr(self.model, 'named_steps'):
            svm = self.model.named_steps.get('svm')
            if svm and hasattr(svm, 'n_support_'):
                info['n_support_vectors'] = svm.n_support_.sum()
                info['kernel'] = svm.kernel
                info['C'] = svm.C
        
        return info
 
 
def benchmark_prediction_latency():
    """
    Benchmark SVM prediction latency for different model sizes.
    """
    from sklearn.datasets import make_classification
    
    print("SVM Prediction Latency Benchmark")
    print("=" * 60)
    
    scenarios = [
        {'n_train': 1000, 'n_features': 20, 'C': 1.0, 'label': 'Small'},
        {'n_train': 5000, 'n_features': 50, 'C': 1.0, 'label': 'Medium'},
        {'n_train': 10000, 'n_features': 100, 'C': 10.0, 'label': 'Large'},
    ]
    
    for scenario in scenarios:
        # Generate data
        X, y = make_classification(
            n_samples=scenario['n_train'],
            n_features=scenario['n_features'],
            n_informative=scenario['n_features'] // 2,
            random_state=42
        )
        
        # Train model
        pipeline = Pipeline([
            ('scaler', StandardScaler()),
            ('svm', SVC(kernel='rbf', C=scenario['C'], gamma='scale'))
        ])
        pipeline.fit(X, y)
        
        n_sv = pipeline.named_steps['svm'].n_support_.sum()
        
        # Benchmark single predictions
        X_test = np.random.randn(1, scenario['n_features'])
        
        times = []
        for _ in range(100):
            start = time.time()
            _ = pipeline.predict(X_test)
            times.append(time.time() - start)
        
        mean_time = np.mean(times) * 1000
        
        print(f"\n{scenario['label']} Model:")
        print(f"  Training samples: {scenario['n_train']}")
        print(f"  Features: {scenario['n_features']}")
        print(f"  Support vectors: {n_sv}")
        print(f"  Prediction latency: {mean_time:.3f} ms")
        print(f"  Throughput: {1000/mean_time:.0f} predictions/sec")
 
 
benchmark_prediction_latency()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

import numpy as np
from sklearn.svm import SVC
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import cross_val_score
from sklearn.calibration import CalibratedClassifierCV
 
def diagnose_svm_issues(X_train, y_train, X_test, y_test):
    """
    Diagnostic function to identify common SVM issues.
    """
    print("SVM Diagnostic Report")
    print("=" * 60)
    
    # 1. Check feature scaling
    feature_ranges = X_train.max(axis=0) - X_train.min(axis=0)
    max_range_ratio = feature_ranges.max() / (feature_ranges.min() + 1e-10)
    
    print("\n1. Feature Scaling Check:")
    print(f"   Feature range ratio: {max_range_ratio:.1f}")
    if max_range_ratio > 10:
        print("   ⚠️  WARNING: Features on very different scales!")
        print("   → Apply StandardScaler before training")
    else:
        print("   ✓ Feature scales look reasonable")
    
    # 2. Check class balance
    classes, counts = np.unique(y_train, return_counts=True)
    min_count, max_count = counts.min(), counts.max()
    imbalance_ratio = max_count / min_count
    
    print("\n2. Class Balance Check:")
    for c, count in zip(classes, counts):
        print(f"   Class {c}: {count} samples ({count/len(y_train)*100:.1f}%)")
    if imbalance_ratio > 3:
        print(f"   ⚠️  WARNING: Imbalance ratio {imbalance_ratio:.1f}:1")
        print("   → Consider class_weight='balanced'")
    else:
        print("   ✓ Class balance looks reasonable")
    
    # 3. Quick model diagnostics
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_test_scaled = scaler.transform(X_test)
    
    print("\n3. Quick Model Tests:")
    
    # Linear baseline
    svm_linear = SVC(kernel='linear', C=1.0)
    svm_linear.fit(X_train_scaled, y_train)
    linear_acc = svm_linear.score(X_test_scaled, y_test)
    linear_sv_ratio = svm_linear.n_support_.sum() / len(y_train)
    
    print(f"   Linear kernel: {linear_acc:.3f} accuracy, {linear_sv_ratio:.1%} SVs")
    
    # RBF with default settings
    svm_rbf = SVC(kernel='rbf', C=1.0, gamma='scale')
    svm_rbf.fit(X_train_scaled, y_train)
    rbf_acc = svm_rbf.score(X_test_scaled, y_test)
    rbf_sv_ratio = svm_rbf.n_support_.sum() / len(y_train)
    
    print(f"   RBF kernel:    {rbf_acc:.3f} accuracy, {rbf_sv_ratio:.1%} SVs")
    
    # 4. SV ratio warning
    print("\n4. Support Vector Ratio:")
    for name, sv_ratio in [('Linear', linear_sv_ratio), ('RBF', rbf_sv_ratio)]:
        if sv_ratio > 0.6:
            print(f"   ⚠️  {name}: {sv_ratio:.1%} SVs is HIGH")
            print("      → Try reducing C or different kernel")
        elif sv_ratio > 0.4:
            print(f"   ⚠️  {name}: {sv_ratio:.1%} SVs is moderate")
        else:
            print(f"   ✓ {name}: {sv_ratio:.1%} SVs looks good")
    
    # 5. Recommendations
    print("\n5. Recommendations:")
    
    if linear_acc >= rbf_acc - 0.02:
        print("   → Linear kernel performs as well as RBF")
        print("     Consider using LinearSVC for speed")
    
    if rbf_sv_ratio > 0.5:
        print("   → High SV ratio suggests:")
        print("     * Try lower C (e.g., 0.1)")
        print("     * Check data quality")
        print("     * Problem may be inherently hard")
    
    return {
        'linear_acc': linear_acc,
        'rbf_acc': rbf_acc,
        'linear_sv_ratio': linear_sv_ratio,
        'rbf_sv_ratio': rbf_sv_ratio,
    }
 
 
# Example usage (when you have data)
# diagnose_svm_issues(X_train, y_train, X_test, y_test)