Skip to content

Objectives

ObjectiveSpec dataclass

Specification for a single score component with its preference weight.

Attributes:

Name Type Description
component ScoreComponent

The ScoreComponent that computes the metric.
weight float

Non-negative weight indicating the component's importance (>= 0).
Source code in energy_repset/objectives.py 
12
13
14
15
16
17
18
19
20
21
@dataclass(frozen=True)
class ObjectiveSpec:
    """Specification for a single score component with its preference weight.

    Attributes:
        component: The ScoreComponent that computes the metric.
        weight: Non-negative weight indicating the component's importance (>= 0).
    """
    component: ScoreComponent
    weight: float

ObjectiveSet

Pillar O: A collection of weighted score components for evaluating selections.

This class holds multiple ScoreComponents, each with a weight indicating its importance. Components define their optimization direction (min/max), while weights specify preference magnitude. The ObjectiveSet prepares all components with context data and evaluates candidate selections.

Attributes:

Name Type Description
weighted_score_components dict[str, ObjectiveSpec]

Dictionary mapping component names to ObjectiveSpec instances containing the component and its weight.

Examples:

Create an objective set with multiple fidelity metrics:

>>> from energy_repset.objectives import ObjectiveSet
>>> from energy_repset.score_components import (
...     WassersteinFidelity, CorrelationFidelity
... )
>>>
>>> objective_set = ObjectiveSet({
...     'wasserstein': (0.5, WassersteinFidelity()),
...     'correlation': (0.5, CorrelationFidelity()),
... })

With variable-specific weights:

>>> wass = WassersteinFidelity(variable_weights={'demand': 2.0, 'solar': 1.0})
>>> objective_set = ObjectiveSet({
...     'wasserstein': (1.0, wass),
... })
Source code in energy_repset/objectives.py 
 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
class ObjectiveSet:
    """Pillar O: A collection of weighted score components for evaluating selections.

    This class holds multiple ScoreComponents, each with a weight indicating its
    importance. Components define their optimization direction (min/max), while
    weights specify preference magnitude. The ObjectiveSet prepares all components
    with context data and evaluates candidate selections.

    Attributes:
        weighted_score_components: Dictionary mapping component names to ObjectiveSpec
            instances containing the component and its weight.

    Examples:
        Create an objective set with multiple fidelity metrics:

        >>> from energy_repset.objectives import ObjectiveSet
        >>> from energy_repset.score_components import (
        ...     WassersteinFidelity, CorrelationFidelity
        ... )
        >>>
        >>> objective_set = ObjectiveSet({
        ...     'wasserstein': (0.5, WassersteinFidelity()),
        ...     'correlation': (0.5, CorrelationFidelity()),
        ... })

        With variable-specific weights:

        >>> wass = WassersteinFidelity(variable_weights={'demand': 2.0, 'solar': 1.0})
        >>> objective_set = ObjectiveSet({
        ...     'wasserstein': (1.0, wass),
        ... })
    """

    def __init__(
            self,
            weighted_score_components: Dict[str, Tuple[float, ScoreComponent]]
    ) -> None:
        """Initialize ObjectiveSet with weighted score components.

        Args:
            weighted_score_components: Dictionary mapping component names to
                tuples of (weight, ScoreComponent). Weights must be non-negative.

        Raises:
            ValueError: If any weight is negative or if any component lacks a
                'direction' attribute set to 'min' or 'max'.
        """
        self.weighted_score_components: Dict[str, ObjectiveSpec] = {
            name: s if isinstance(s, ObjectiveSpec) else ObjectiveSpec(component=s[1], weight=float(s[0]))
            for name, s in weighted_score_components.items()
        }
        for s in self.weighted_score_components.values():
            if s.weight < 0:
                raise ValueError(f"Weight for {s.component.name} must be >= 0.")
            if getattr(s.component, "direction", None) not in ("min", "max"):
                raise ValueError(f"Component {s.component.name} must declare direction 'min' or 'max'.")

    def prepare(self, context: ProblemContext) -> None:
        """Prepare all score components with context data.

        This method calls prepare() on each component to allow pre-computation
        of reference statistics, duration curves, etc.

        Args:
            context: ProblemContext containing raw data and features.
        """
        for spec in self.weighted_score_components.values():
            spec.component.prepare(context)

    def evaluate(self, combination: SliceCombination, context: ProblemContext) -> Dict[str, float]:
        """Evaluate a candidate selection across all score components.

        Args:
            combination: Tuple of slice labels forming the candidate selection.
            context: ProblemContext for accessing data.

        Returns:
            Dictionary mapping component names to their unweighted scores.

        Note:
            Returns raw scores from components. Weights are applied by SelectionPolicy
            during the selection process.
        """
        return {
           spec.component.name: float(spec.component.score(combination))
           for spec in self.weighted_score_components.values()
       }

    def component_meta(self) -> Dict[str, Dict[str, Any]]:
        """Get metadata for all components.

        Returns:
            Dictionary mapping component names to their metadata containing:
                - 'direction': Optimization direction ('min' or 'max')
                - 'pref': Preference weight (>= 0)
        """
        return {
            spec.component.name: {"direction": spec.component.direction, "pref": float(spec.weight)}
            for spec in self.weighted_score_components.values()
        }

__init__ 

__init__(weighted_score_components: dict[str, tuple[float, ScoreComponent]]) -> None

Initialize ObjectiveSet with weighted score components.

Parameters:

Name Type Description Default
weighted_score_components dict[str, tuple[float, ScoreComponent]]

Dictionary mapping component names to tuples of (weight, ScoreComponent). Weights must be non-negative.

 required

Raises:

Type Description
ValueError

If any weight is negative or if any component lacks a 'direction' attribute set to 'min' or 'max'.
Source code in energy_repset/objectives.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(
        self,
        weighted_score_components: Dict[str, Tuple[float, ScoreComponent]]
) -> None:
    """Initialize ObjectiveSet with weighted score components.

    Args:
        weighted_score_components: Dictionary mapping component names to
            tuples of (weight, ScoreComponent). Weights must be non-negative.

    Raises:
        ValueError: If any weight is negative or if any component lacks a
            'direction' attribute set to 'min' or 'max'.
    """
    self.weighted_score_components: Dict[str, ObjectiveSpec] = {
        name: s if isinstance(s, ObjectiveSpec) else ObjectiveSpec(component=s[1], weight=float(s[0]))
        for name, s in weighted_score_components.items()
    }
    for s in self.weighted_score_components.values():
        if s.weight < 0:
            raise ValueError(f"Weight for {s.component.name} must be >= 0.")
        if getattr(s.component, "direction", None) not in ("min", "max"):
            raise ValueError(f"Component {s.component.name} must declare direction 'min' or 'max'.")

prepare 

prepare(context: ProblemContext) -> None

Prepare all score components with context data.

This method calls prepare() on each component to allow pre-computation of reference statistics, duration curves, etc.

Parameters:

Name Type Description Default
context ProblemContext

ProblemContext containing raw data and features.

 required
Source code in energy_repset/objectives.py 
81
82
83
84
85
86
87
88
89
90
91
def prepare(self, context: ProblemContext) -> None:
    """Prepare all score components with context data.

    This method calls prepare() on each component to allow pre-computation
    of reference statistics, duration curves, etc.

    Args:
        context: ProblemContext containing raw data and features.
    """
    for spec in self.weighted_score_components.values():
        spec.component.prepare(context)

evaluate 

evaluate(combination: SliceCombination, context: ProblemContext) -> dict[str, float]

Evaluate a candidate selection across all score components.

Parameters:

Name Type Description Default
combination SliceCombination

Tuple of slice labels forming the candidate selection.

 required
context ProblemContext

ProblemContext for accessing data.

 required

Returns:

Type Description
dict[str, float]

Dictionary mapping component names to their unweighted scores.
Note

Returns raw scores from components. Weights are applied by SelectionPolicy during the selection process.

Source code in energy_repset/objectives.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def evaluate(self, combination: SliceCombination, context: ProblemContext) -> Dict[str, float]:
    """Evaluate a candidate selection across all score components.

    Args:
        combination: Tuple of slice labels forming the candidate selection.
        context: ProblemContext for accessing data.

    Returns:
        Dictionary mapping component names to their unweighted scores.

    Note:
        Returns raw scores from components. Weights are applied by SelectionPolicy
        during the selection process.
    """
    return {
       spec.component.name: float(spec.component.score(combination))
       for spec in self.weighted_score_components.values()
   }

component_meta 

component_meta() -> dict[str, dict[str, Any]]

Get metadata for all components.

Returns:

Type Description
dict[str, dict[str, Any]]

Dictionary mapping component names to their metadata containing: - 'direction': Optimization direction ('min' or 'max') - 'pref': Preference weight (>= 0)
Source code in energy_repset/objectives.py 
112
113
114
115
116
117
118
119
120
121
122
123
def component_meta(self) -> Dict[str, Dict[str, Any]]:
    """Get metadata for all components.

    Returns:
        Dictionary mapping component names to their metadata containing:
            - 'direction': Optimization direction ('min' or 'max')
            - 'pref': Preference weight (>= 0)
    """
    return {
        spec.component.name: {"direction": spec.component.direction, "pref": float(spec.weight)}
        for spec in self.weighted_score_components.values()
    }

ScoreComponent

Bases: ABC

Protocol for a single metric used in evaluating candidate selections.

ScoreComponents are the building blocks of the ObjectiveSet. Each component computes a scalar score measuring how well a candidate selection performs on a specific criterion (e.g., distribution fidelity, diversity, balance).

Implementations must define
  • A unique name identifying the component
  • An optimization direction ('min' or 'max')
  • A prepare() method to precompute reference data from the full context
  • A score() method to evaluate a candidate selection

Attributes:

Name Type Description
name str

Unique identifier for this component.
direction ScoreComponentDirection

Optimization direction, either "min" or "max".

Examples:

Implementing a simple score component:

>>> from energy_repset.score_components.base_score_component import ScoreComponent
>>> from energy_repset.context import ProblemContext
>>> from energy_repset.types import SliceCombination
>>> import numpy as np
>>>
>>> class SimpleMeanDeviation(ScoreComponent):
...     def __init__(self):
...         self.name = "mean_deviation"
...         self.direction = "min"
...         self.full_mean = None
...
...     def prepare(self, context: ProblemContext) -> None:
...         '''Compute reference mean from full dataset.'''
...         self.full_mean = context.df_raw.mean().mean()
...
...     def score(self, combination: SliceCombination) -> float:
...         '''Measure deviation from reference mean.'''
...         # Get data for selection and compute deviation
...         # (implementation details omitted)
...         return abs(selection_mean - self.full_mean)
Source code in energy_repset/score_components/base_score_component.py 
 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
class ScoreComponent(ABC):
    """Protocol for a single metric used in evaluating candidate selections.

    ScoreComponents are the building blocks of the ObjectiveSet. Each component
    computes a scalar score measuring how well a candidate selection performs
    on a specific criterion (e.g., distribution fidelity, diversity, balance).

    Implementations must define:
        - A unique name identifying the component
        - An optimization direction ('min' or 'max')
        - A prepare() method to precompute reference data from the full context
        - A score() method to evaluate a candidate selection

    Attributes:
        name: Unique identifier for this component.
        direction: Optimization direction, either "min" or "max".

    Examples:
        Implementing a simple score component:

        >>> from energy_repset.score_components.base_score_component import ScoreComponent
        >>> from energy_repset.context import ProblemContext
        >>> from energy_repset.types import SliceCombination
        >>> import numpy as np
        >>>
        >>> class SimpleMeanDeviation(ScoreComponent):
        ...     def __init__(self):
        ...         self.name = "mean_deviation"
        ...         self.direction = "min"
        ...         self.full_mean = None
        ...
        ...     def prepare(self, context: ProblemContext) -> None:
        ...         '''Compute reference mean from full dataset.'''
        ...         self.full_mean = context.df_raw.mean().mean()
        ...
        ...     def score(self, combination: SliceCombination) -> float:
        ...         '''Measure deviation from reference mean.'''
        ...         # Get data for selection and compute deviation
        ...         # (implementation details omitted)
        ...         return abs(selection_mean - self.full_mean)
    """
    name: str
    direction: ScoreComponentDirection

    @abstractmethod
    def prepare(self, context: ProblemContext) -> None:
        """Precompute state needed before scoring selections.

        This method is called once before evaluating any combinations. Use it
        to compute reference statistics, duration curves, or other data derived
        from the full dataset that will be compared against selections.

        Args:
            context: ProblemContext containing raw data, features, and metadata.

        Note:
            This method should store computed state as instance attributes for
            use in score().
        """
        ...

    @abstractmethod
    def score(self, combination: SliceCombination) -> float:
        """Compute the component score for a candidate selection.

        Args:
            combination: Tuple of slice labels forming the candidate selection.

        Returns:
            Scalar score. Lower is better for direction='min', higher is better
            for direction='max'.

        Note:
            This method is called many times during search. Precompute expensive
            operations in prepare() to avoid redundant calculations.
        """
        ...

    @staticmethod
    def _default_weight_normalization(
            weights: Optional[Dict[str, float]],
            keys: List[str]
    ) -> Dict[str, float]:
        """Normalize weight dictionary to match actual keys.

        Normalize weights: None → equal (1.0), specified → use values (missing get 0.0)

        Args:
            weights: User-specified weights, or None for equal weights.
            keys: Actual keys that need weights (e.g., variable names, feature names).

        Returns:
            Dictionary mapping each key to its weight:
            - If weights is None: all keys get 1.0 (equal weights)
            - If weights is provided: specified keys get their value, missing keys get 0.0

        Examples:

            >>> # No weights specified - equal weights
            >>> normalize_weights(None, ['demand', 'solar', 'wind'])
            {'demand': 1.0, 'solar': 1.0, 'wind': 1.0}

            >>> # Partial weights - missing get 0.0
            >>> normalize_weights({'demand': 2.0, 'solar': 1.5}, ['demand', 'solar', 'wind'])
            {'demand': 2.0, 'solar': 1.5, 'wind': 0.0}

            >>> # All weights specified
            >>> normalize_weights({'demand': 2.0, 'solar': 1.0, 'wind': 0.5}, ['demand', 'solar'])
            {'demand': 2.0, 'solar': 1.0, 'wind': 0.5}
        """
        if weights is None:
            return {key: 1.0 for key in keys}
        return {key: weights.get(key, 0.0) for key in keys}

prepare abstractmethod 

prepare(context: ProblemContext) -> None

Precompute state needed before scoring selections.

This method is called once before evaluating any combinations. Use it to compute reference statistics, duration curves, or other data derived from the full dataset that will be compared against selections.

Parameters:

Name Type Description Default
context ProblemContext

ProblemContext containing raw data, features, and metadata.

 required
Note

This method should store computed state as instance attributes for use in score().

Source code in energy_repset/score_components/base_score_component.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@abstractmethod
def prepare(self, context: ProblemContext) -> None:
    """Precompute state needed before scoring selections.

    This method is called once before evaluating any combinations. Use it
    to compute reference statistics, duration curves, or other data derived
    from the full dataset that will be compared against selections.

    Args:
        context: ProblemContext containing raw data, features, and metadata.

    Note:
        This method should store computed state as instance attributes for
        use in score().
    """
    ...

score abstractmethod 

score(combination: SliceCombination) -> float

Compute the component score for a candidate selection.

Parameters:

Name Type Description Default
combination SliceCombination

Tuple of slice labels forming the candidate selection.

 required

Returns:

Type Description
float

Scalar score. Lower is better for direction='min', higher is better
float

for direction='max'.
Note

This method is called many times during search. Precompute expensive operations in prepare() to avoid redundant calculations.

Source code in energy_repset/score_components/base_score_component.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@abstractmethod
def score(self, combination: SliceCombination) -> float:
    """Compute the component score for a candidate selection.

    Args:
        combination: Tuple of slice labels forming the candidate selection.

    Returns:
        Scalar score. Lower is better for direction='min', higher is better
        for direction='max'.

    Note:
        This method is called many times during search. Precompute expensive
        operations in prepare() to avoid redundant calculations.
    """
    ...