"""Idea 2: a training-validated blend of frozen candidate-one and its base model. The candidate-one implementation is copied unchanged. Blend validation is outside the original model's relationship selection: each outer model is fitted from scratch, including all supervised transforms, selection and calibration. """ import gc import hashlib import json from pathlib import Path from time import perf_counter import numpy as np from sklearn.base import BaseEstimator, ClassifierMixin from sklearn.model_selection import StratifiedKFold from sklearn.preprocessing import LabelEncoder from sklearn.utils.validation import check_is_fitted from budgeted_pulsar import BudgetedInteractionPULSAR, head_probabilities, validate_X def canonical(value): return json.loads(json.dumps(value)) def digest(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def training_digest(X, labels): value = hashlib.sha256(np.ascontiguousarray(X).tobytes()) value.update(np.ascontiguousarray(labels, dtype=np.int64).tobytes()) return value.hexdigest() def blend_probabilities(base, augmented, weight): """The weight is the contribution of the relationship-augmented model.""" return (1.0-weight)*base + weight*augmented def choose_blend_weight(base, augmented, labels, weights=(0., .25, .5, .75, 1.)): """Maximise pooled held-out training accuracy; ties prefer the smaller weight.""" table = [] for weight in sorted(set(map(float, weights))): pred = blend_probabilities(base, augmented, weight).argmax(axis=1) correct = int(np.sum(pred == labels)) table.append({"weight": weight, "correct": correct, "cases": len(labels), "accuracy": correct/len(labels)}) best = max(table, key=lambda row: (row["correct"], -row["weight"])) return best["weight"], table def make_blend_folds(labels, n_splits=3, random_state=42): """Keep >=2 examples per class in every outer training subset. Classes with only two examples cannot be held out while retaining the frozen model's training contract. Their examples remain in every training subset and do not contribute validation scores. Eligible examples are held out once for the default three folds. Generic two-fold use may additionally return examples to training to preserve the same minimum. """ labels = np.asarray(labels) counts = np.bincount(labels) if counts.min() < 2: raise ValueError("At least two training examples per class are required.") eligible = np.flatnonzero(counts[labels] >= 3) if not len(eligible): return [], np.arange(len(labels)) actual_folds = min(n_splits, int(counts[counts >= 3].min())) splitter = StratifiedKFold(actual_folds, shuffle=True, random_state=random_state) folds, covered = [], np.zeros(len(labels), dtype=bool) for _, valid_local in splitter.split(eligible, labels[eligible]): proposed = eligible[valid_local] valid = np.concatenate([ proposed[labels[proposed] == c][:max(0, int(count)-2)] for c, count in enumerate(counts)]) valid = np.sort(valid) if not len(valid): continue train = np.setdiff1d(np.arange(len(labels)), valid) if covered[valid].any() or np.bincount(labels[train], minlength=len(counts)).min() < 2: raise AssertionError("Invalid outer validation split.") covered[valid] = True folds.append((train, valid)) return folds, np.flatnonzero(~covered) def verify_probabilities(P, n_cases, n_classes): if P.shape != (n_cases, n_classes) or not np.isfinite(P).all() or (P < 0).any(): raise AssertionError("Invalid saved validation probabilities.") np.testing.assert_allclose(P.sum(axis=1), 1., atol=1e-10) class ValidationBlendedPULSAR(ClassifierMixin, BaseEstimator): """Choose a single branch weight from nested held-out training predictions.""" def __init__(self, blend_folds=3, weights=(0., .25, .5, .75, 1.), random_state=42, model_params=None, verbose=False): self.blend_folds, self.weights = blend_folds, weights self.random_state, self.model_params, self.verbose = random_state, model_params, verbose def _log(self, message): if self.verbose: print(message, flush=True) def underlying_parameters(self): parameters = BudgetedInteractionPULSAR( random_state=self.random_state, fit_controls=True, verbose=self.verbose).get_params() parameters.update(self.model_params or {}) if not parameters["fit_controls"] or parameters["random_state"] != self.random_state: raise ValueError("The branch model must fit controls and use the wrapper's seed.") return parameters def _checkpoint_key(self, train, valid): return hashlib.sha256(json.dumps({ "training_sha256": self.training_sha256_, "wrapper_source_sha256": digest(__file__), "candidate_source_sha256": digest(Path(__file__).with_name("budgeted_pulsar.py")), "parameters": self.underlying_parameters(), "train": train.tolist(), "validation": valid.tolist(), }, sort_keys=True).encode()).hexdigest() def fit_blend(self, X, y, checkpoint_dir=None): """Fit the weight without fitting or loading the full-training final model.""" X, y = validate_X(X), np.asarray(y) if y.ndim != 1 or len(y) != len(X): raise ValueError("Provide exactly one label per recording.") if (isinstance(self.blend_folds, bool) or not isinstance(self.blend_folds, (int, np.integer)) or self.blend_folds < 2): raise ValueError("blend_folds must be an integer >= 2.") weights = np.asarray(self.weights, dtype=float) if (weights.ndim != 1 or not np.isfinite(weights).all() or (weights < 0).any() or (weights > 1).any() or 0. not in weights or 1. not in weights): raise ValueError("Weights must be finite, lie in [0,1], and include 0 and 1.") self.encoder_ = LabelEncoder().fit(y) self.classes_ = self.encoder_.classes_ if len(self.classes_) < 2: raise ValueError("At least two classes are required.") labels = self.encoder_.transform(y) self.shape_ = X.shape[1:] self.training_sha256_ = training_digest(X, labels) folds, excluded = make_blend_folds(labels, self.blend_folds, self.random_state) self.fold_indices_ = [{"train": t.tolist(), "validation": v.tolist()} for t, v in folds] self.excluded_validation_indices_ = excluded.tolist() self.n_blend_folds_ = len(folds) self.fold_records_ = [] self.oof_base_ = np.full((len(X), len(self.classes_)), np.nan) self.oof_augmented_ = np.full_like(self.oof_base_, np.nan) self.oof_covered_ = np.zeros(len(X), dtype=bool) started = perf_counter() checkpoint_dir = Path(checkpoint_dir) if checkpoint_dir is not None else None for index, (train, valid) in enumerate(folds): key = self._checkpoint_key(train, valid) folder = checkpoint_dir/f"fold_{index+1}" if checkpoint_dir else None if folder is not None: folder.mkdir(parents=True, exist_ok=True) metadata_file = folder/"validation.json" if folder else None probability_file = folder/"probabilities.npz" if folder else None if metadata_file is not None and metadata_file.exists(): record = json.loads(metadata_file.read_text()) assert record["checkpoint_key"] == key, "Checkpoint training data or source changed." assert digest(probability_file) == record["probabilities_sha256"] with np.load(probability_file) as saved: np.testing.assert_array_equal(saved["classes"], self.classes_) np.testing.assert_array_equal(saved["validation_indices"], valid) base, augmented = saved["base"], saved["augmented"] self._log(f"Using verified blend-fold checkpoint {index+1}/{len(folds)}") else: self._log(f"Fitting complete candidate one inside blend fold {index+1}/{len(folds)}; " f"training={len(train)}, validation={len(valid)}") model = BudgetedInteractionPULSAR(**self.underlying_parameters()) tick = perf_counter() model.fit(X[train], y[train]) fit_seconds = perf_counter()-tick consistency = model.verify_training_transform() tick = perf_counter() _, probability, _ = model.predict_all(X[valid]) prediction_seconds = perf_counter()-tick np.testing.assert_array_equal(model.classes_, self.classes_) base, augmented = probability["channelwise"], probability["candidate"] record = { "checkpoint_key": key, "fold": index+1, "train_indices": train.tolist(), "validation_indices": valid.tolist(), "minimum_training_class_count": int(np.bincount(labels[train]).min()), "fit_seconds_with_controls": fit_seconds, "validation_prediction_seconds": prediction_seconds, "training_transform_consistent": consistency, "underlying_parameters": canonical(model.get_params()), "relationship_selection_folds_local": model.fold_indices_, "selected_groups": model.selected_groups_, "selection_trace": model.selection_trace_, "initial_selection_losses": getattr(model, "initial_validation_losses_", []), "final_selection_losses": getattr(model, "final_validation_losses_", []), "base_features": model.n_base_features_, "relationship_features": model.n_relationship_features_, "total_features": model.n_features_out_, "candidate_groups": len(model.bank_.groups_), } if folder is not None: np.savez_compressed(probability_file, classes=self.classes_, base=base, augmented=augmented, validation_indices=valid) record["probabilities_sha256"] = digest(probability_file) metadata_file.write_text(json.dumps(record, indent=2, allow_nan=False)+"\n") del model, probability gc.collect() assert record["train_indices"] == train.tolist() assert record["validation_indices"] == valid.tolist() assert record["training_transform_consistent"] verify_probabilities(base, len(valid), len(self.classes_)) verify_probabilities(augmented, len(valid), len(self.classes_)) for inner in record["relationship_selection_folds_local"]: inner_train, inner_valid = train[inner["train"]], train[inner["validation"]] assert not np.intersect1d(inner_train, valid).size assert not np.intersect1d(inner_valid, valid).size assert not np.intersect1d(inner_train, inner_valid).size self.oof_base_[valid], self.oof_augmented_[valid] = base, augmented self.oof_covered_[valid] = True self.fold_records_.append(record) self._log(f"Completed blend fold {index+1}/{len(folds)}") covered = self.oof_covered_ if covered.any(): self.weight_, self.weight_validation_ = choose_blend_weight( self.oof_base_[covered], self.oof_augmented_[covered], labels[covered], weights) self.weight_selection_reason_ = "highest_pooled_heldout_training_accuracy_ties_prefer_base" else: self.weight_, self.weight_validation_ = 0., [] self.weight_selection_reason_ = "insufficient_training_counts_for_honest_validation" self.weight_fit_seconds_ = perf_counter()-started self._log(f"Training validation chose augmented weight={self.weight_:.2f}; " f"coverage={int(covered.sum())}/{len(X)}") return self def attach_fitted(self, model, X, y): """Reuse unchanged full-training branches with externally verified provenance.""" check_is_fitted(self, "weight_") X = validate_X(X) labels = self.encoder_.transform(y) if training_digest(X, labels) != self.training_sha256_: raise ValueError("The final model must use the same training data as blend validation.") if canonical(model.get_params()) != canonical(self.underlying_parameters()): raise ValueError("Pretrained branch parameters differ from the validation model.") np.testing.assert_array_equal(model.classes_, self.classes_) np.testing.assert_array_equal(model.training_sample_, X[:len(model.training_sample_)]) model.verify_training_transform() self.final_model_ = model return self def fit(self, X, y, pretrained_model=None, checkpoint_dir=None): self.fit_blend(X, y, checkpoint_dir=checkpoint_dir) model = (pretrained_model if pretrained_model is not None else BudgetedInteractionPULSAR(**self.underlying_parameters()).fit(X, y)) return self.attach_fitted(model, X, y) def predict_all(self, X): check_is_fitted(self, "final_model_") _, original, timings = self.final_model_.predict_all(X) probability = { "blend": blend_probabilities(original["channelwise"], original["candidate"], self.weight_), "original_candidate": original["candidate"], "channelwise": original["channelwise"], "matched_features": original["matched_features"], } predictions = {name: self.classes_[p.argmax(axis=1)] for name, p in probability.items()} return predictions, probability, timings def predict_proba(self, X): check_is_fitted(self, "final_model_") X = self.final_model_._check_X(X) if self.weight_ == 1.: return self.final_model_.predict_proba(X) output = [] for start in range(0, len(X), self.final_model_.prediction_batch_size): batch = X[start:start+self.final_model_.prediction_batch_size] if self.weight_ == 0.: ranked = self.final_model_.base_.transform(batch) output.append(head_probabilities(self.final_model_.control_heads_, ranked[:, :self.final_model_.n_base_features_])) else: ranked, combined = self.final_model_._features(batch) base = head_probabilities(self.final_model_.control_heads_, ranked[:, :self.final_model_.n_base_features_]) augmented = head_probabilities(self.final_model_.heads_, combined) output.append(blend_probabilities(base, augmented, self.weight_)) return np.vstack(output) def predict(self, X): return self.classes_[self.predict_proba(X).argmax(axis=1)]