Validation Set

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

import numpy as np
from sklearn.model_selection import train_test_split
from typing import Tuple, Optional
import pandas as pd
 
def train_val_test_split(
    X: np.ndarray,
    y: np.ndarray,
    train_size: float = 0.7,
    val_size: float = 0.15,
    test_size: float = 0.15,
    random_state: int = 42,
    stratify: Optional[np.ndarray] = None
) -> Tuple[np.ndarray, ...]:
    """
    Create train/validation/test splits with proper stratification.
    
    Parameters:
    -----------
    X : Feature matrix
    y : Target vector
    train_size, val_size, test_size : Split proportions (must sum to 1.0)
    random_state : Random seed for reproducibility
    stratify : Array to stratify by (typically y for classification)
    
    Returns:
    --------
    X_train, X_val, X_test, y_train, y_val, y_test
    """
    # Validate proportions
    total = train_size + val_size + test_size
    if not np.isclose(total, 1.0):
        raise ValueError(f"Split proportions must sum to 1.0, got {total}")
    
    # First split: separate test set
    # Calculate the proportion of remaining data for validation
    val_prop_of_remaining = val_size / (train_size + val_size)
    
    X_temp, X_test, y_temp, y_test = train_test_split(
        X, y,
        test_size=test_size,
        random_state=random_state,
        stratify=stratify
    )
    
    # Second split: separate training and validation
    stratify_temp = y_temp if stratify is not None else None
    X_train, X_val, y_train, y_val = train_test_split(
        X_temp, y_temp,
        test_size=val_prop_of_remaining,
        random_state=random_state,
        stratify=stratify_temp
    )
    
    # Verify sizes
    n = len(X)
    print(f"Split verification:")
    print(f"  Training:   {len(X_train):,} samples ({len(X_train)/n:.1%})")
    print(f"  Validation: {len(X_val):,} samples ({len(X_val)/n:.1%})")
    print(f"  Test:       {len(X_test):,} samples ({len(X_test)/n:.1%})")
    
    return X_train, X_val, X_test, y_train, y_val, y_test
 
 
# ============================================
# Complete Workflow with Hyperparameter Tuning
# ============================================
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
 
def complete_train_val_test_workflow(X, y, param_grid):
    """
    Demonstrates the complete workflow:
    1. Split into train/val/test
    2. Tune hyperparameters on validation
    3. Retrain on train+val
    4. Final evaluation on test
    """
    # Step 1: Split data
    X_train, X_val, X_test, y_train, y_val, y_test = train_val_test_split(
        X, y,
        train_size=0.7,
        val_size=0.15,
        test_size=0.15,
        random_state=42,
        stratify=y
    )
    
    # Step 2: Hyperparameter search (using validation set)
    best_val_score = -np.inf
    best_params = None
    
    for params in param_grid:
        model = RandomForestClassifier(random_state=42, **params)
        model.fit(X_train, y_train)
        
        val_pred = model.predict(X_val)
        val_score = f1_score(y_val, val_pred, average='weighted')
        
        if val_score > best_val_score:
            best_val_score = val_score
            best_params = params
        
        print(f"Params {params}: Validation F1 = {val_score:.4f}")
    
    print(f"
Best params: {best_params}")
    print(f"Best validation F1: {best_val_score:.4f}")
    
    # Step 3: Retrain on train + validation with best params
    X_train_full = np.vstack([X_train, X_val])
    y_train_full = np.concatenate([y_train, y_val])
    
    final_model = RandomForestClassifier(random_state=42, **best_params)
    final_model.fit(X_train_full, y_train_full)
    
    # Step 4: Final evaluation on test set (ONE TIME ONLY)
    test_pred = final_model.predict(X_test)
    test_score = f1_score(y_test, test_pred, average='weighted')
    
    print(f"
{'='*50}")
    print(f"FINAL TEST F1 SCORE: {test_score:.4f}")
    print(f"{'='*50}")
    print("WARNING: This test score should not be used for further model selection.")
    
    return final_model, test_score
 
 
# Example usage
param_grid = [
    {'n_estimators': 50, 'max_depth': 5},
    {'n_estimators': 100, 'max_depth': 5},
    {'n_estimators': 100, 'max_depth': 10},
    {'n_estimators': 200, 'max_depth': 10},
]
 
# X, y = your_data_loading_function()
# model, final_score = complete_train_val_test_workflow(X, y, param_grid)

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

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import hashlib
 
@dataclass
class DatasetSplit:
    """Immutable record of a train/val/test split for auditing."""
    split_id: str
    created_at: datetime
    train_indices: List[int]
    val_indices: List[int]
    test_indices: List[int]
    random_seed: int
    stratify_column: Optional[str]
    notes: str
 
class ProductionSplitManager:
    """
    Manages train/val/test splits for production ML.
    Key features:
    - Immutable splits with versioning
    - Hash-based data fingerprinting
    - Audit trail for regulatory compliance
    """
    
    def __init__(self, experiment_tracker):
        self.experiment_tracker = experiment_tracker
        self.splits: Dict[str, DatasetSplit] = {}
    
    def create_split(
        self,
        X,
        y,
        train_size: float = 0.7,
        val_size: float = 0.15,
        random_seed: int = 42,
        stratify_col: Optional[str] = None,
        notes: str = ""
    ) -> DatasetSplit:
        """
        Create and register a new immutable split.
        """
        # Create data fingerprint for verification
        data_hash = self._compute_data_hash(X, y)
        
        # Perform the split
        # (using previously defined train_val_test_split logic)
        indices = np.arange(len(X))
        stratify = y if stratify_col else None
        
        idx_train, idx_val, idx_test, _, _, _ = train_val_test_split(
            indices, y,
            train_size=train_size,
            val_size=val_size,
            test_size=1 - train_size - val_size,
            random_state=random_seed,
            stratify=stratify
        )
        
        # Create split record
        split = DatasetSplit(
            split_id=f"{data_hash[:8]}_{random_seed}_{datetime.now().strftime('%Y%m%d')}",
            created_at=datetime.now(),
            train_indices=list(idx_train),
            val_indices=list(idx_val),
            test_indices=list(idx_test),
            random_seed=random_seed,
            stratify_column=stratify_col,
            notes=notes
        )
        
        # Register with experiment tracker
        self.experiment_tracker.log_artifact('split_config', split)
        self.splits[split.split_id] = split
        
        return split
    
    def _compute_data_hash(self, X, y) -> str:
        """Compute deterministic hash of dataset for verification."""
        combined = np.concatenate([X.flatten(), y.flatten()])
        return hashlib.sha256(combined.tobytes()).hexdigest()
    
    def verify_no_test_leakage(self, split_id: str, accessed_indices: List[int]) -> bool:
        """
        Verify that accessed indices don't include test data.
        Use this to gate any data access during development.
        """
        split = self.splits[split_id]
        test_set = set(split.test_indices)
        accessed_set = set(accessed_indices)
        
        overlap = test_set & accessed_set
        if overlap:
            raise ValueError(
                f"TEST DATA LEAK DETECTED! {len(overlap)} test indices accessed: "
                f"{list(overlap)[:10]}..."
            )
        return True
 
 
class RollingTestSetManager:
    """
    Manages test sets for ongoing production evaluation.
    New data becomes test data; old data can be recycled.
    """
    
    def __init__(self, holdout_days: int = 30):
        self.holdout_days = holdout_days
        self.data_log = []
    
    def add_data(self, data_batch, timestamp: datetime):
        """Add new data batch with timestamp."""
        self.data_log.append({
            'data': data_batch,
            'timestamp': timestamp,
            'used_for_training': False
        })
    
    def get_current_splits(self, as_of: datetime):
        """
        Get train/test split as of a given date.
        - Training: data more than holdout_days old
        - Test: data from last holdout_days
        """
        cutoff = as_of - timedelta(days=self.holdout_days)
        
        train_data = [
            batch['data'] for batch in self.data_log
            if batch['timestamp'] < cutoff
        ]
        
        test_data = [
            batch['data'] for batch in self.data_log
            if cutoff <= batch['timestamp'] < as_of
        ]
        
        return train_data, test_data