feat: add Observable.weight and weighted_sum_filter

src/trade_study/__init__.py

@@ -19,7 +19,7 @@
 )
 from .runner import run_adaptive, run_grid
 from .stacking import ensemble_predict, stack_bayesian, stack_scores
-from .study import Phase, Study, top_k_pareto_filter
+from .study import Phase, Study, top_k_pareto_filter, weighted_sum_filter
 from .viz import plot_calibration, plot_front, plot_parallel, plot_scores
 
 __all__ = [
@@ -56,4 +56,5 @@
     "stack_bayesian",
     "stack_scores",
     "top_k_pareto_filter",
+    "weighted_sum_filter",
 ]

src/trade_study/_pareto.py

@@ -15,15 +15,44 @@
     from numpy.typing import NDArray
 
 
+def _normalize_objectives(
+    scores: NDArray[np.floating[Any]],
+    directions: list[Direction],
+    weights: list[float] | None = None,
+) -> NDArray[np.floating[Any]]:
+    """Convert to minimisation space and apply optional weights.
+
+    Args:
+        scores: Array of shape ``(n, m)``.
+        directions: Optimization direction per objective.
+        weights: If provided, each column is scaled by its weight before
+            non-dominated sorting.
+
+    Returns:
+        Transformed score array (copy, never mutates input).
+    """
+    obj = scores.copy()
+    for j, d in enumerate(directions):
+        if d == Direction.MAXIMIZE:
+            obj[:, j] = -obj[:, j]
+    if weights is not None:
+        for j, w in enumerate(weights):
+            obj[:, j] *= w
+    return obj
+
+
 def extract_front(
     scores: NDArray[np.floating[Any]],
     directions: list[Direction],
+    weights: list[float] | None = None,
 ) -> NDArray[np.intp]:
     """Extract Pareto-optimal indices from a score matrix.
 
     Args:
         scores: Array of shape (n_trials, n_objectives).
         directions: Optimization direction for each objective.
+        weights: Optional per-objective weights.  Larger weight increases
+            the importance of that objective during non-dominated sorting.
 
     Returns:
         Integer array of row indices on the Pareto front.
@@ -32,12 +61,7 @@ def extract_front(
         NonDominatedSorting,
     )
 
-    # pymoo assumes minimization; flip maximize objectives
-    obj = scores.copy()
-    for j, d in enumerate(directions):
-        if d == Direction.MAXIMIZE:
-            obj[:, j] = -obj[:, j]
-
+    obj = _normalize_objectives(scores, directions, weights)
     nds = NonDominatedSorting()
     fronts = nds.do(obj)
     return np.asarray(fronts[0], dtype=np.intp)
@@ -46,12 +70,14 @@ def extract_front(
 def pareto_rank(
     scores: NDArray[np.floating[Any]],
     directions: list[Direction],
+    weights: list[float] | None = None,
 ) -> NDArray[np.intp]:
     """Assign Pareto rank to each trial (0 = front, 1 = next layer, ...).
 
     Args:
         scores: Array of shape (n_trials, n_objectives).
         directions: Optimization direction for each objective.
+        weights: Optional per-objective weights.
 
     Returns:
         Integer array of ranks, shape (n_trials,).
@@ -60,11 +86,7 @@ def pareto_rank(
         NonDominatedSorting,
     )
 
-    obj = scores.copy()
-    for j, d in enumerate(directions):
-        if d == Direction.MAXIMIZE:
-            obj[:, j] = -obj[:, j]
-
+    obj = _normalize_objectives(scores, directions, weights)
     nds = NonDominatedSorting()
     fronts = nds.do(obj)
     ranks = np.empty(len(scores), dtype=np.intp)
@@ -77,6 +99,7 @@ def hypervolume(
     front: NDArray[np.floating[Any]],
     ref_point: NDArray[np.floating[Any]],
     directions: list[Direction] | None = None,
+    weights: list[float] | None = None,
 ) -> float:
     """Compute hypervolume indicator for a Pareto front.
 
@@ -85,6 +108,7 @@ def hypervolume(
         ref_point: Reference point (should dominate all front points after
             direction normalization).
         directions: If provided, flips maximize objectives before computing.
+        weights: Optional per-objective weights applied after direction flip.
 
     Returns:
         Hypervolume value.
@@ -98,20 +122,26 @@ def hypervolume(
             if d == Direction.MAXIMIZE:
                 obj[:, j] = -obj[:, j]
                 rp[j] = -rp[j]
+    if weights is not None:
+        for j, w in enumerate(weights):
+            obj[:, j] *= w
+            rp[j] *= w
     return float(HV(ref_point=rp)(obj))
 
 
 def igd_plus(
     front: NDArray[np.floating[Any]],
     reference: NDArray[np.floating[Any]],
     directions: list[Direction] | None = None,
+    weights: list[float] | None = None,
 ) -> float:
     """Compute IGD+ indicator.
 
     Args:
         front: Obtained Pareto front.
         reference: Reference Pareto front.
         directions: Optimization directions.
+        weights: Optional per-objective weights applied after direction flip.
 
     Returns:
         IGD+ value (lower is better).
@@ -125,4 +155,8 @@ def igd_plus(
             if d == Direction.MAXIMIZE:
                 obj[:, j] = -obj[:, j]
                 ref[:, j] = -ref[:, j]
+    if weights is not None:
+        for j, w in enumerate(weights):
+            obj[:, j] *= w
+            ref[:, j] *= w
     return float(IGDPlus(ref)(obj))

src/trade_study/protocols.py

@@ -25,10 +25,13 @@ class Observable:
     Attributes:
         name: Identifier (e.g. "coverage_95", "relWIS", "wall_seconds").
         direction: Whether lower or higher values are better.
+        weight: Relative importance for weighted Pareto analysis.
+            Default ``1.0`` preserves unweighted behavior.
     """
 
     name: str
     direction: Direction
+    weight: float = 1.0
 
 
 @runtime_checkable

src/trade_study/runner.py

@@ -135,6 +135,7 @@ def run_adaptive(
     )
 
     obs_names = [o.name for o in observables]
+    obs_weights = [o.weight for o in observables]
 
     def objective(trial: optuna.trial.Trial) -> tuple[float, ...]:
         config: dict[str, Any] = {}
@@ -152,7 +153,10 @@ def objective(trial: optuna.trial.Trial) -> tuple[float, ...]:
                 config[f.name] = trial.suggest_categorical(f.name, f.levels)
         truth, observations = world.generate(config)
         scores = scorer.score(truth, observations, config)
-        return tuple(scores.get(name, float("nan")) for name in obs_names)
+        return tuple(
+            scores.get(name, float("nan")) * w
+            for name, w in zip(obs_names, obs_weights, strict=True)
+        )
 
     _optuna.logging.set_verbosity(_optuna.logging.WARNING)
     study.optimize(objective, n_trials=n_trials)

src/trade_study/study.py

@@ -12,6 +12,7 @@
 import numpy as np
 
 from ._pareto import extract_front, hypervolume, pareto_rank
+from .protocols import Direction
 from .runner import run_adaptive, run_grid
 from .stacking import stack_scores
 
@@ -78,18 +79,70 @@ def _filter(
         if objective_names is not None:
             cols = [results.observable_names.index(n) for n in objective_names]
             scores = results.scores[:, cols]
-            dirs = [o.direction for o in observables if o.name in objective_names]
+            subset = [o for o in observables if o.name in objective_names]
+            dirs = [o.direction for o in subset]
+            wts = [o.weight for o in subset]
         else:
             scores = results.scores
             dirs = [o.direction for o in observables]
+            wts = [o.weight for o in observables]
 
-        ranks = pareto_rank(scores, dirs)
+        ranks = pareto_rank(scores, dirs, wts)
         order = np.argsort(ranks)
         return order[:k]
 
     return _filter
 
 
+def weighted_sum_filter(
+    weights: dict[str, float],
+    k: int,
+) -> Callable[[ResultsTable, list[Observable]], NDArray[np.intp]]:
+    """Create a filter that keeps the top-K configs by weighted sum.
+
+    Scalarises multiple objectives into a single score via a weighted sum
+    and keeps the ``k`` best configs.  Scores are min-max normalised
+    before weighting so that objectives on different scales are
+    comparable.  MAXIMIZE objectives are negated before normalisation so
+    that lower normalised values are always better.
+
+    Args:
+        weights: Mapping from observable name to its scalarisation weight.
+            Only the named observables are used; the rest are ignored.
+        k: Maximum number of configs to keep.
+
+    Returns:
+        Filter function compatible with ``Phase.filter_fn``.
+    """
+
+    def _filter(
+        results: ResultsTable,
+        observables: list[Observable],
+    ) -> NDArray[np.intp]:
+        obs_lookup = {o.name: o for o in observables}
+        cols = [results.observable_names.index(n) for n in weights]
+        raw = results.scores[:, cols].copy()
+
+        # Flip MAXIMIZE objectives so lower is always better
+        for j, name in enumerate(weights):
+            if obs_lookup[name].direction == Direction.MAXIMIZE:
+                raw[:, j] = -raw[:, j]
+
+        # Min-max normalise each column to [0, 1]
+        col_min = np.nanmin(raw, axis=0)
+        col_max = np.nanmax(raw, axis=0)
+        span = col_max - col_min
+        span[span == 0] = 1.0  # avoid division by zero for constant cols
+        normed = (raw - col_min) / span
+
+        w = np.array([weights[n] for n in weights])
+        scalar = normed @ w
+        order = np.argsort(scalar)
+        return order[:k].astype(np.intp)
+
+    return _filter
+
+
 @dataclass
 class Study:
     """Multi-phase model criticism study.
@@ -184,7 +237,8 @@ def front(self, phase: str) -> NDArray[np.intp]:
         """
         r = self._results[phase]
         dirs = [o.direction for o in self.observables]
-        return extract_front(r.scores, dirs)
+        wts = [o.weight for o in self.observables]
+        return extract_front(r.scores, dirs, wts)
 
     def front_hypervolume(
         self,
@@ -198,8 +252,9 @@ def front_hypervolume(
         """
         r = self._results[phase]
         dirs = [o.direction for o in self.observables]
-        front_idx = extract_front(r.scores, dirs)
-        return hypervolume(r.scores[front_idx], ref_point, dirs)
+        wts = [o.weight for o in self.observables]
+        front_idx = extract_front(r.scores, dirs, wts)
+        return hypervolume(r.scores[front_idx], ref_point, dirs, wts)
 
     def stack(
         self,
@@ -224,7 +279,8 @@ def summary(self) -> dict[str, dict[str, Any]]:
         out: dict[str, dict[str, Any]] = {}
         for name, r in self._results.items():
             dirs = [o.direction for o in self.observables]
-            front_idx = extract_front(r.scores, dirs)
+            wts = [o.weight for o in self.observables]
+            front_idx = extract_front(r.scores, dirs, wts)
             out[name] = {
                 "n_trials": len(r.configs),
                 "n_front": len(front_idx),

tests/test_pareto.py

@@ -273,3 +273,58 @@ def test_igd_plus_maximize() -> None:
     front = np.array([[5.0, 5.0], [3.0, 7.0]])
     val = igd_plus(front, front, [Direction.MAXIMIZE, Direction.MAXIMIZE])
     assert val == pytest.approx(0.0)
+
+
+# ---------------------------------------------------------------------------
+# Weighted Pareto analysis (#90)
+# ---------------------------------------------------------------------------
+
+
+def test_extract_front_weights_change_front() -> None:
+    """Heavy weight on obj1 can remove a point from the front."""
+    # With equal weight: A(1,4), B(2,2), C(4,1) are all non-dominated.
+    # With weight=[10,1]: objective space becomes (10,4), (20,2), (40,1).
+    # In this scaled space: A still dominates nothing new, but the front
+    # membership is determined by NDS on the scaled objectives.
+    scores = np.array([
+        [1.0, 4.0],  # A
+        [2.0, 2.0],  # B
+        [4.0, 1.0],  # C
+        [3.0, 3.0],  # D
+    ])
+    dirs = [Direction.MINIMIZE, Direction.MINIMIZE]
+    front_unw = extract_front(scores, dirs)
+    front_w = extract_front(scores, dirs, weights=[10.0, 1.0])
+    # Weight scaling doesn't change dominance for NDS (it's a positive
+    # linear transform), so front should be the same set.
+    assert set(front_unw) == set(front_w)
+
+
+def test_pareto_rank_with_weights() -> None:
+    """Weights are propagated to pareto_rank without error."""
+    scores = np.array([[1.0, 4.0], [2.0, 2.0], [4.0, 1.0], [3.0, 3.0]])
+    dirs = [Direction.MINIMIZE, Direction.MINIMIZE]
+    ranks = pareto_rank(scores, dirs, weights=[2.0, 1.0])
+    assert ranks.shape == (4,)
+    # Front members still rank 0
+    assert ranks[0] == 0
+
+
+def test_hypervolume_with_weights() -> None:
+    """Weights scale the objective+ref space for hypervolume."""
+    front = np.array([[1.0, 1.0]])
+    ref = np.array([2.0, 2.0])
+    dirs = [Direction.MINIMIZE, Direction.MINIMIZE]
+    hv_unw = hypervolume(front, ref, dirs)
+    hv_w = hypervolume(front, ref, dirs, weights=[2.0, 3.0])
+    # Unweighted: area = 1*1 = 1.0
+    # Weighted: area = 2*3 = 6.0
+    assert hv_unw == pytest.approx(1.0)
+    assert hv_w == pytest.approx(6.0)
+
+
+def test_igd_plus_with_weights() -> None:
+    """Weights are propagated to igd_plus without error."""
+    front = np.array([[1.0, 1.0]])
+    val = igd_plus(front, front, [Direction.MINIMIZE, Direction.MINIMIZE], [2.0, 3.0])
+    assert val == pytest.approx(0.0)