stack_proportional -- weight models by relative score, not winner-take-all

CHANGELOG.md

@@ -13,6 +13,7 @@ All notable changes to this project will be documented in this file.
 - Replicate averaging in `run_adaptive()` and `screen()`/`sobol_indices()` (`n_reps`, #122): the same `run_grid(..., n_reps=N)` convention (#112), now applied consistently across every entry point that repeatedly evaluates a simulator/`run_fn`. `run_adaptive` detects an opt-in `rep` keyword on `Simulator.generate` and averages each trial's objective(s) over `n_reps` draws before Optuna sees them; `screen()`/`sobol_indices()` detect the same convention on the bare `run_fn` callable they take instead. Without this, adaptive optimization or sensitivity screening against a stochastic simulator could select a "best" config or report an "important" factor that's actually just a single lucky/unlucky data draw. Default `n_reps=1` preserves prior behavior.
 - `recommend_bucketed_config()`: the discrete counterpart to `RegimeSurrogate` (#123). For a handful of named regimes too sparse (and too far outside any existing training data) for a surrogate to extrapolate across sensibly, runs `run_adaptive` independently per regime, picks each regime's best-found config by a primary objective, groups regimes into named buckets, and aggregates each bucket's per-regime best configs (median for continuous/discrete factors, mode for categorical) into one recommended config per bucket.
 - `recommend_bucketed_config()` split into `recommend_per_regime()` (the expensive adaptive search) and `aggregate_bucketed_config()` (cheap post-processing), with `recommend_bucketed_config()` now a thin wrapper composing them. Lets a caller experiment with different `bucket_fn` groupings against the same search results without re-running `run_adaptive` for every attempt -- found necessary in practice deriving VBPCApy buckets, where a first grouping choice performed poorly and needed re-grouping without repeating an hours-long search.
+- `stack_proportional()`: weights models in direct proportion to their mean score, instead of `stack_scores()`'s linear-program optimum (which puts *all* weight on the single best model for any nonzero gap, however small -- a real problem in practice when two models are near-tied and the "winner" flips between runs on noise well within measurement uncertainty). Falls back to a uniform split when every model scores identically.
 
 ### Fixed
 

src/trade_study/__init__.py

@@ -36,7 +36,7 @@
 )
 from .runner import run_adaptive, run_grid, run_hyperband, run_successive_halving
 from .sensitivity import TableSensitivity, sensitivity_from_table
-from .stacking import ensemble_predict, stack_bayesian, stack_scores
+from .stacking import ensemble_predict, stack_bayesian, stack_proportional, stack_scores
 from .study import (
     Phase,
     Study,
@@ -95,6 +95,7 @@
     "sensitivity_from_table",
     "sobol_indices",
     "stack_bayesian",
+    "stack_proportional",
     "stack_scores",
     "top_k_pareto_filter",
     "weighted_sum_filter",

src/trade_study/stacking.py

@@ -78,6 +78,48 @@ def objective(w: NDArray[np.floating[Any]]) -> float:
     return np.asarray(result.x, dtype=np.float64)
 
 
+def stack_proportional(
+    score_matrix: NDArray[np.floating[Any]],
+    *,
+    maximize: bool = False,
+) -> NDArray[np.floating[Any]]:
+    """Weight models in direct proportion to their mean score.
+
+    Unlike :func:`stack_scores` (a linear program that puts *all* weight
+    on the single best-performing model whenever there's any nonzero gap,
+    even a noise-scale one), this scales smoothly with relative
+    performance -- useful when "how much better" should matter, not just
+    "which one is best". Two near-tied models get near-equal weights
+    here; under :func:`stack_scores` the same tiny gap can flip the
+    result between an even split and 100% on one model, since a linear
+    objective over a simplex has no reason to split weight once any
+    model is even infinitesimally ahead.
+
+    Args:
+        score_matrix: Array of shape (n_models, n_test_points) where each
+            entry is the score of model i on test point j.
+        maximize: If True, higher scores are better. If False, lower
+            scores are better; scores are inverted internally
+            (``max - score``) rather than divided, so a score of exactly
+            0 doesn't produce an unbounded weight.
+
+    Returns:
+        Array of weights, shape (n_models,), summing to 1, each >= 0.
+        Falls back to a uniform split if every model scores identically
+        (nothing to distinguish them by).
+    """
+    mean_scores = np.mean(score_matrix, axis=1)
+    if not maximize:
+        mean_scores = np.max(mean_scores) - mean_scores
+    mean_scores = np.clip(mean_scores, a_min=0.0, a_max=None)
+
+    total = mean_scores.sum()
+    n_models = score_matrix.shape[0]
+    if total <= 0.0:
+        return np.full(n_models, 1.0 / n_models, dtype=np.float64)
+    return np.asarray(mean_scores / total, dtype=np.float64)
+
+
 def ensemble_predict(
     predictions: list[NDArray[np.floating[Any]]],
     weights: NDArray[np.floating[Any]],

tests/test_stacking.py

@@ -7,7 +7,12 @@
 import numpy as np
 import pytest
 
-from trade_study.stacking import ensemble_predict, stack_bayesian, stack_scores
+from trade_study.stacking import (
+    ensemble_predict,
+    stack_bayesian,
+    stack_proportional,
+    stack_scores,
+)
 
 RNG = np.random.default_rng(42)
 
@@ -124,6 +129,90 @@ def test_stack_scores_dtype() -> None:
     assert weights.dtype == np.float64
 
 
+# ---------------------------------------------------------------------------
+# stack_proportional
+# ---------------------------------------------------------------------------
+
+
+def test_stack_proportional_weights_sum_to_one() -> None:
+    scores = RNG.standard_normal((3, 50))
+    weights = stack_proportional(scores)
+    assert np.sum(weights) == pytest.approx(1.0)
+
+
+def test_stack_proportional_weights_non_negative() -> None:
+    scores = RNG.standard_normal((3, 50))
+    weights = stack_proportional(scores)
+    assert np.all(weights >= 0.0)
+
+
+def test_stack_proportional_near_tie_splits_smoothly() -> None:
+    """A tiny gap between two models shouldn't collapse to winner-take-all.
+
+    Unlike stack_scores (a linear program that snaps to the single best
+    model for any nonzero gap), stack_proportional should give both
+    near-tied models comparable weight.
+    """
+    scores = np.array([
+        [0.70, 0.70, 0.70, 0.70],
+        [0.71, 0.71, 0.71, 0.71],  # barely better
+        [0.10, 0.10, 0.10, 0.10],  # clearly worse
+    ])
+    weights = stack_proportional(scores, maximize=True)
+    assert weights[0] == pytest.approx(weights[1], abs=0.05)
+    assert weights[2] < weights[0]
+
+
+def test_stack_proportional_dominant_model_gets_most_weight_minimize() -> None:
+    scores = np.array([
+        [0.1, 0.2, 0.1, 0.15],  # best (lowest)
+        [5.0, 6.0, 5.5, 5.2],
+        [9.0, 8.0, 9.5, 8.8],
+    ])
+    weights = stack_proportional(scores, maximize=False)
+    assert weights[0] > weights[1] > weights[2]
+
+
+def test_stack_proportional_dominant_model_gets_most_weight_maximize() -> None:
+    scores = np.array([
+        [9.0, 8.5, 9.2, 8.8],  # best (highest)
+        [1.0, 1.5, 1.2, 1.1],
+        [0.1, 0.2, 0.15, 0.1],
+    ])
+    weights = stack_proportional(scores, maximize=True)
+    assert weights[0] > weights[1] > weights[2]
+
+
+def test_stack_proportional_uniform_when_all_equal() -> None:
+    """No model distinguishable by score -> falls back to a uniform split."""
+    scores = np.full((4, 10), 0.5)
+    weights = stack_proportional(scores, maximize=True)
+    np.testing.assert_allclose(weights, np.full(4, 0.25))
+
+
+def test_stack_proportional_zero_score_does_not_blow_up() -> None:
+    """A model scoring exactly 0 (minimize) shouldn't produce inf/nan weights."""
+    scores = np.array([
+        [0.0, 0.0, 0.0],
+        [1.0, 1.0, 1.0],
+    ])
+    weights = stack_proportional(scores, maximize=False)
+    assert np.all(np.isfinite(weights))
+    assert weights[0] > weights[1]
+
+
+def test_stack_proportional_shape() -> None:
+    scores = RNG.standard_normal((4, 20))
+    weights = stack_proportional(scores)
+    assert weights.shape == (4,)
+
+
+def test_stack_proportional_dtype() -> None:
+    scores = RNG.standard_normal((2, 10))
+    weights = stack_proportional(scores)
+    assert weights.dtype == np.float64
+
+
 # ---------------------------------------------------------------------------
 # ensemble_predict (#17)
 # ---------------------------------------------------------------------------