Skip to content

openscm_calibration.cost.scmdata#

Cost calculations for models that return [scmdata.run.BaseScmRun][scmdata.run.BaseScmRun] objects

Classes:

Name Description
AlignmentError

Raised when our data's metadata does not align as expected
OptCostCalculatorSSE

Cost calculator based on sum of squared errors

AlignmentError #

Bases: ValueError

Raised when our data's metadata does not align as expected

Methods:

Name Description
__init__

Initialise the error
Source code in src/openscm_calibration/cost/scmdata.py 
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
class AlignmentError(ValueError):
    """
    Raised when our data's metadata does not align as expected
    """

    def __init__(
        self,
        name_left: str,
        val_left: pd.DataFrame | pd.Series[Any],
        name_right: str,
        val_right: pd.DataFrame | pd.Series[Any],
        extra_context: str | None = None,
    ) -> None:
        """
        Initialise the error

        Parameters
        ----------
        name_left
            The name of the first thing being referenced
            (variable, attribute etc.)

        val_left
            The values referred to by ``name_left``

        name_right
            The name of the other thing being referenced (variable, attribute
            etc.)

        name_right
            The values referred to by ``name_left``

        extra_context
            Any extra context to include in the message
        """
        error_msg = (
            f"Please check.\n"
            f"{name_left}:\n{val_left}\n"
            f"{name_right}:\n{val_right}\n"
        )

        if extra_context:
            error_msg = f"{extra_context}. {error_msg}"

        super().__init__(error_msg)

__init__ #

__init__(
    name_left: str,
    val_left: DataFrame | Series[Any],
    name_right: str,
    val_right: DataFrame | Series[Any],
    extra_context: str | None = None,
) -> None

Initialise the error

Parameters:

Name Type Description Default
name_left str

The name of the first thing being referenced (variable, attribute etc.)

 required
val_left DataFrame | Series[Any]

The values referred to by name_left

 required
name_right str

The name of the other thing being referenced (variable, attribute etc.)

 required
name_right str

The values referred to by name_left

 required
extra_context str | None

Any extra context to include in the message

 None
Source code in src/openscm_calibration/cost/scmdata.py 
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
def __init__(
    self,
    name_left: str,
    val_left: pd.DataFrame | pd.Series[Any],
    name_right: str,
    val_right: pd.DataFrame | pd.Series[Any],
    extra_context: str | None = None,
) -> None:
    """
    Initialise the error

    Parameters
    ----------
    name_left
        The name of the first thing being referenced
        (variable, attribute etc.)

    val_left
        The values referred to by ``name_left``

    name_right
        The name of the other thing being referenced (variable, attribute
        etc.)

    name_right
        The values referred to by ``name_left``

    extra_context
        Any extra context to include in the message
    """
    error_msg = (
        f"Please check.\n"
        f"{name_left}:\n{val_left}\n"
        f"{name_right}:\n{val_right}\n"
    )

    if extra_context:
        error_msg = f"{extra_context}. {error_msg}"

    super().__init__(error_msg)

OptCostCalculatorSSE #

Cost calculator based on sum of squared errors

This is a convenience class. We may want to refactor it in future to provide greater flexibility for other cost calculations.

Methods:

Name Description
calculate_cost

Calculate cost function based on model results
calculate_negative_log_likelihood

Calculate the negative log likelihood of a given set of results
from_series_normalisation

Initialise from a series that defines normalisation for each timeseries.
from_unit_normalisation

Initialise assuming unit normalisation for each timeseries.

Attributes:

Name Type Description
model_col str

Column which contains the name of the model.
normalisation BaseScmRun

Normalisation values
target BaseScmRun

Target timeseries
Source code in src/openscm_calibration/cost/scmdata.py 
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
@define
class OptCostCalculatorSSE:
    """
    Cost calculator based on sum of squared errors

    This is a convenience class.
    We may want to refactor it in future
    to provide greater flexibility for other cost calculations.
    """

    target: scmdata.run.BaseScmRun
    """Target timeseries"""

    model_col: str = field(validator=[_is_meta_in_target])
    """
    Column which contains the name of the model.

    This is used when subtracting the model results from the target
    """

    normalisation: scmdata.run.BaseScmRun = field(validator=[_works_with_self_target])
    """
    Normalisation values

    Should have same timeseries as target. See the class methods for helpers.
    """

    @classmethod
    def from_unit_normalisation(
        cls, target: scmdata.run.BaseScmRun, model_col: str
    ) -> OptCostCalculatorSSE:
        """
        Initialise assuming unit normalisation for each timeseries.

        This is a convenience method, but is not recommended for any serious
        work as unit normalisation is unlikely to be a good choice for most
        problems.

        Parameters
        ----------
        target
            Target timeseries

        model_col
            Column which contains of the model in ``target``

        Returns
        -------
            :obj:`OptCostCalculatorSSE` such that the normalisation is 1 for
            all timepoints (with the units defined by whatever the units of
            each timeseries are in ``target``)
        """
        norm = target.timeseries()
        norm.loc[:, :] = 1
        norm_cast = type(target)(norm)

        return cls(target=target, normalisation=norm_cast, model_col=model_col)

    @classmethod
    def from_series_normalisation(
        cls,
        target: scmdata.run.BaseScmRun,
        model_col: str,
        normalisation_series: pd.Series[float],
    ) -> OptCostCalculatorSSE:
        """
        Initialise from a series that defines normalisation for each timeseries.

        The series is broadcast to match the timeseries in target, using the
        same value for all timepoints in each timeseries.

        Parameters
        ----------
        target
            Target timeseries

        model_col
            Column which contains of the model in ``target``

        normalisation_series
            Series to broadcast to create the desired normalisation

        Returns
        -------
            Initialised :obj:`OptCostCalculatorSSE`
        """
        required_columns = {"variable", "unit"}
        available_cols = set(normalisation_series.index.names)
        missing_cols = required_columns - available_cols
        if missing_cols:
            raise MissingValueError(
                "normalisation_series.index.names",
                vals=sorted(available_cols),
                missing_vals=sorted(missing_cols),
            )

        target_ts_no_unit = target.timeseries().reset_index("unit", drop=True)

        # This is basically what pandas does internally when doing ops:
        # align and then broadcast
        norm_series_aligned, _ = normalisation_series.align(target_ts_no_unit)

        if norm_series_aligned.isna().any():
            raise AlignmentError(
                name_left="target_ts_no_unit",
                val_left=target_ts_no_unit,
                name_right="norm_series_aligned",
                val_right=norm_series_aligned,
                extra_context="Even after aligning, there are still nan values",
            )

        if norm_series_aligned.size != target_ts_no_unit.shape[0]:
            raise AlignmentError(
                name_left="target_ts_no_unit",
                val_left=target_ts_no_unit,
                name_right="norm_series_aligned",
                val_right=norm_series_aligned,
                extra_context=(
                    "After aligning, there are more rows in the normalisation "
                    "than in the target"
                ),
            )

        norm_series_aligned = type(target_ts_no_unit)(
            np.broadcast_to(norm_series_aligned.values, target_ts_no_unit.T.shape).T,  # type: ignore # mypy/np confused
            index=norm_series_aligned.index,
            columns=target_ts_no_unit.columns,
        )

        normalisation = type(target)(norm_series_aligned)

        return cls(target=target, normalisation=normalisation, model_col=model_col)

    def calculate_cost(self, model_results: scmdata.run.BaseScmRun) -> float:
        """
        Calculate cost function based on model results

        Parameters
        ----------
        model_results
            Model results of which to calculate the cost

        Returns
        -------
            Cost
        """
        diff = model_results.subtract(  # type: ignore # error in scmdata types
            self.target, op_cols={self.model_col: "res - target"}
        ).divide(
            self.normalisation,
            op_cols={self.model_col: "(res - target) / normalisation"},
        )

        cost = float((diff.convert_unit("1") ** 2).values.sum().sum())

        return cost

    def calculate_negative_log_likelihood(
        self,
        model_results: scmdata.run.BaseScmRun,
    ) -> float:
        """
        Calculate the negative log likelihood of a given set of results

        Parameters
        ----------
        model_results
            Model results for which to calculate the negative log likelihood

        Returns
        -------
        :
            Negative log likelihood (up to an additive constant)
        """
        sses = self.calculate_cost(model_results)
        # TODO: find the proof of this
        negative_log_likelihood = -sses / 2

        return negative_log_likelihood

model_col class-attribute instance-attribute #

model_col: str = field(validator=[_is_meta_in_target])

Column which contains the name of the model.

This is used when subtracting the model results from the target

normalisation class-attribute instance-attribute #

normalisation: BaseScmRun = field(
    validator=[_works_with_self_target]
)

Normalisation values

Should have same timeseries as target. See the class methods for helpers.

target instance-attribute #

target: BaseScmRun

Target timeseries

calculate_cost #

calculate_cost(model_results: BaseScmRun) -> float

Calculate cost function based on model results

Parameters:

Name Type Description Default
model_results BaseScmRun

Model results of which to calculate the cost

 required

Returns:

Type Description
Cost
Source code in src/openscm_calibration/cost/scmdata.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def calculate_cost(self, model_results: scmdata.run.BaseScmRun) -> float:
    """
    Calculate cost function based on model results

    Parameters
    ----------
    model_results
        Model results of which to calculate the cost

    Returns
    -------
        Cost
    """
    diff = model_results.subtract(  # type: ignore # error in scmdata types
        self.target, op_cols={self.model_col: "res - target"}
    ).divide(
        self.normalisation,
        op_cols={self.model_col: "(res - target) / normalisation"},
    )

    cost = float((diff.convert_unit("1") ** 2).values.sum().sum())

    return cost

calculate_negative_log_likelihood #

calculate_negative_log_likelihood(
    model_results: BaseScmRun,
) -> float

Calculate the negative log likelihood of a given set of results

Parameters:

Name Type Description Default
model_results BaseScmRun

Model results for which to calculate the negative log likelihood

 required

Returns:

Type Description
float

Negative log likelihood (up to an additive constant)
Source code in src/openscm_calibration/cost/scmdata.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def calculate_negative_log_likelihood(
    self,
    model_results: scmdata.run.BaseScmRun,
) -> float:
    """
    Calculate the negative log likelihood of a given set of results

    Parameters
    ----------
    model_results
        Model results for which to calculate the negative log likelihood

    Returns
    -------
    :
        Negative log likelihood (up to an additive constant)
    """
    sses = self.calculate_cost(model_results)
    # TODO: find the proof of this
    negative_log_likelihood = -sses / 2

    return negative_log_likelihood

from_series_normalisation classmethod #

from_series_normalisation(
    target: BaseScmRun,
    model_col: str,
    normalisation_series: Series[float],
) -> OptCostCalculatorSSE

Initialise from a series that defines normalisation for each timeseries.

The series is broadcast to match the timeseries in target, using the same value for all timepoints in each timeseries.

Parameters:

Name Type Description Default
target BaseScmRun

Target timeseries

 required
model_col str

Column which contains of the model in target

 required
normalisation_series Series[float]

Series to broadcast to create the desired normalisation

 required

Returns:

Type Description
Initialised :obj:`OptCostCalculatorSSE`
Source code in src/openscm_calibration/cost/scmdata.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
@classmethod
def from_series_normalisation(
    cls,
    target: scmdata.run.BaseScmRun,
    model_col: str,
    normalisation_series: pd.Series[float],
) -> OptCostCalculatorSSE:
    """
    Initialise from a series that defines normalisation for each timeseries.

    The series is broadcast to match the timeseries in target, using the
    same value for all timepoints in each timeseries.

    Parameters
    ----------
    target
        Target timeseries

    model_col
        Column which contains of the model in ``target``

    normalisation_series
        Series to broadcast to create the desired normalisation

    Returns
    -------
        Initialised :obj:`OptCostCalculatorSSE`
    """
    required_columns = {"variable", "unit"}
    available_cols = set(normalisation_series.index.names)
    missing_cols = required_columns - available_cols
    if missing_cols:
        raise MissingValueError(
            "normalisation_series.index.names",
            vals=sorted(available_cols),
            missing_vals=sorted(missing_cols),
        )

    target_ts_no_unit = target.timeseries().reset_index("unit", drop=True)

    # This is basically what pandas does internally when doing ops:
    # align and then broadcast
    norm_series_aligned, _ = normalisation_series.align(target_ts_no_unit)

    if norm_series_aligned.isna().any():
        raise AlignmentError(
            name_left="target_ts_no_unit",
            val_left=target_ts_no_unit,
            name_right="norm_series_aligned",
            val_right=norm_series_aligned,
            extra_context="Even after aligning, there are still nan values",
        )

    if norm_series_aligned.size != target_ts_no_unit.shape[0]:
        raise AlignmentError(
            name_left="target_ts_no_unit",
            val_left=target_ts_no_unit,
            name_right="norm_series_aligned",
            val_right=norm_series_aligned,
            extra_context=(
                "After aligning, there are more rows in the normalisation "
                "than in the target"
            ),
        )

    norm_series_aligned = type(target_ts_no_unit)(
        np.broadcast_to(norm_series_aligned.values, target_ts_no_unit.T.shape).T,  # type: ignore # mypy/np confused
        index=norm_series_aligned.index,
        columns=target_ts_no_unit.columns,
    )

    normalisation = type(target)(norm_series_aligned)

    return cls(target=target, normalisation=normalisation, model_col=model_col)

from_unit_normalisation classmethod #

from_unit_normalisation(
    target: BaseScmRun, model_col: str
) -> OptCostCalculatorSSE

Initialise assuming unit normalisation for each timeseries.

This is a convenience method, but is not recommended for any serious work as unit normalisation is unlikely to be a good choice for most problems.

Parameters:

Name Type Description Default
target BaseScmRun

Target timeseries

 required
model_col str

Column which contains of the model in target

 required

Returns:

Type Description
:obj:`OptCostCalculatorSSE` such that the normalisation is 1 for

all timepoints (with the units defined by whatever the units of each timeseries are in target)
Source code in src/openscm_calibration/cost/scmdata.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@classmethod
def from_unit_normalisation(
    cls, target: scmdata.run.BaseScmRun, model_col: str
) -> OptCostCalculatorSSE:
    """
    Initialise assuming unit normalisation for each timeseries.

    This is a convenience method, but is not recommended for any serious
    work as unit normalisation is unlikely to be a good choice for most
    problems.

    Parameters
    ----------
    target
        Target timeseries

    model_col
        Column which contains of the model in ``target``

    Returns
    -------
        :obj:`OptCostCalculatorSSE` such that the normalisation is 1 for
        all timepoints (with the units defined by whatever the units of
        each timeseries are in ``target``)
    """
    norm = target.timeseries()
    norm.loc[:, :] = 1
    norm_cast = type(target)(norm)

    return cls(target=target, normalisation=norm_cast, model_col=model_col)