Add KGE Score (#700)

* Add KGE scores
* Added KGE notebook to tutorial gallery, and adjusted capitalisation

---------

Signed-off-by: Tennessee Leeuwenburg <[email protected]>
Co-authored-by: durgals <[email protected]>
Co-authored-by: shr015 <[email protected]>
Co-authored-by: Stephanie Chong <[email protected]>
Co-authored-by: Nicholas Loveday <[email protected]>

docs/api.md

@@ -23,6 +23,7 @@
 .. autofunction:: scores.continuous.correlation.pearsonr
 .. autofunction:: scores.continuous.multiplicative_bias
 .. autofunction:: scores.continuous.pbias
+.. autofunction:: scores.continuous.kge
 .. autofunction:: scores.continuous.isotonic_fit
 .. autofunction:: scores.continuous.consistent_expectile_score
 .. autofunction:: scores.continuous.consistent_quantile_score

docs/included.md

@@ -52,7 +52,11 @@
 * - Isotonic Regression (Isotonic Fit, Reliability Diagram)
   - [API](api.md#scores.continuous.isotonic_fit)
   - [Tutorial](project:./tutorials/Isotonic_Regression_And_Reliability_Diagrams.md)
-  - [de Leeuw et al. (2009)](https://doi.org/10.18637/jss.v032.i05); [Dimitriadis et al. (2020)](https://doi.org/10.1073/pnas.2016191118); [Jordan et al. (2020), version 2](https://doi.org/10.48550/arXiv.1904.04761)   
+  - [de Leeuw et al. (2009)](https://doi.org/10.18637/jss.v032.i05); [Dimitriadis et al. (2020)](https://doi.org/10.1073/pnas.2016191118); [Jordan et al. (2020), version 2](https://doi.org/10.48550/arXiv.1904.04761) 
+* - Kling–Gupta Efficiency (KGE)
+  - [API](api.md#scores.continuous.kge)
+  - [Tutorial](project:./tutorials/Kling_Gupta_Efficiency.md)
+  - [Gupta et al. (2009)](https://doi.org/10.1016/j.jhydrol.2009.08.003); [Knoben et al. (2019)](https://doi.org/10.5194/hess-23-4323-2019)    
 * - Mean Absolute Error (MAE)
   - [API](api.md#scores.continuous.mae)
   - [Tutorial](project:./tutorials/Mean_Absolute_Error.md)

src/scores/continuous/__init__.py

@@ -16,6 +16,7 @@
 from scores.continuous.quantile_loss_impl import quantile_score
 from scores.continuous.standard_impl import (
     additive_bias,
+    kge,
     mae,
     mean_error,
     mse,
@@ -46,6 +47,7 @@
     "mean_error",
     "multiplicative_bias",
     "pbias",
+    "kge",
     "isotonic_fit",
     "consistent_expectile_score",
     "consistent_huber_score",

src/scores/continuous/standard_impl.py

@@ -2,8 +2,9 @@
 This module contains standard methods which may be used for continuous scoring
 """
 
-from typing import Optional
+from typing import Optional, Union
 
+import numpy as np
 import xarray as xr
 
 import scores.functions
@@ -418,3 +419,145 @@ def pbias(
 
     _pbias = 100 * error.mean(dim=reduce_dims) / obs.mean(dim=reduce_dims)
     return _pbias
+
+
+def kge(
+    fcst: xr.DataArray,
+    obs: xr.DataArray,
+    *,
+    reduce_dims: Optional[FlexibleDimensionTypes] = None,
+    preserve_dims: Optional[FlexibleDimensionTypes] = None,
+    scaling_factors: Optional[Union[list[float], np.ndarray]] = None,
+    return_components: bool = False,
+) -> XarrayLike:
+    # pylint: disable=too-many-locals
+    """
+    Calculate the Kling-Gupta Efficiency (KGE) between observed and simulated (or forecast) values.
+
+    KGE is a performance metric that decomposes the error into three components:
+    correlation, variability, and bias.
+    It is computed as:
+
+    .. math::
+        \\text{KGE} = 1 - \\sqrt{\\left[s_\\rho \\cdot (\\rho - 1)\\right]^2 +
+        \\left[s_\\alpha \\cdot (\\alpha - 1)\\right]^2 + \\left[s_\\beta \\cdot (\\beta - 1)\\right]^2}
+
+    .. math::
+        \\alpha = \\frac{\\sigma_x}{\\sigma_y}
+
+    .. math::
+        \\beta = \\frac{\\mu_x}{\\mu_y}
+
+    where:
+        - :math:`\\rho`  = Pearson's correlation coefficient between observed and forecast values as defined in :py:func:`scores.continuous.correlation.pearsonr`
+        - :math:`\\alpha` is the ratio of the standard deviations (variability ratio)
+        - :math:`\\beta` is the ratio of the means (bias)
+        - :math:`x` and :math:`y` are forecast and observed values, respectively
+        - :math:`\\mu_x` and :math:`\\mu_y` are the means of forecast and observed values, respectively
+        - :math:`\\sigma_x` and :math:`\\sigma_y` are the standard deviations of forecast and observed values, respectively
+        - :math:`s_\\rho`, :math:`s_\\alpha` and :math:`s_\\beta` are the scaling factors for the correlation coefficient :math:`\\rho`,
+            the variability term :math:`\\alpha` and the bias term :math:`\\beta`
+
+    Args:
+        fcst: Forecast or predicted variables.
+        obs: Observed variables.
+        reduce_dims: Optionally specify which dimensions to reduce when
+            calculating the KGE. All other dimensions will be preserved.
+        preserve_dims: Optionally specify which dimensions to preserve when
+            calculating the KGE. All other dimensions will be reduced. As a
+            special case, 'all' will allow all dimensions to be preserved. In
+            this case, the result will be all NaN with the same shape/dimensionality
+            as the forecast because the standard deviation is zero for a single point.
+        scaling_factors : A 3-element vector or list describing the weights for each term in the KGE.
+            Defined by: scaling_factors = [:math:`s_\\rho`, :math:`s_\\alpha`, :math:`s_\\beta`] to apply to the correlation term :math:`\\rho`,
+            the variability term :math:`\\alpha` and the bias term :math:`\\beta` respectively. Defaults to (1.0, 1.0, 1.0). (*See
+            equation 10 in Gupta et al. (2009) for definitions of them*).
+        return_components (bool | False): If True, the function also returns the individual terms contributing to the KGE score.
+
+    Returns:
+        The Kling-Gupta Efficiency (KGE) score as an xarray DataArray.
+
+        If ``return_components`` is True, the function returns ``xarray.Dataset`` kge_s with the following variables:
+
+        - `kge`: The KGE score.
+        - `rho`: The Pearson correlation coefficient.
+        - `alpha`: The variability ratio.
+        - `beta`: The bias term.
+
+    Notes:
+        - Statistics are calculated only from values for which both observations and
+          simulations are not null values.
+        - This function isn't set up to take weights.
+        - Currently this function is working only on xr.DataArray.
+        - When preserve_dims is set to 'all', the function returns NaN,
+          similar to the Pearson correlation coefficient calculation for a single data point
+          because the standard deviation is zero for a single point.
+
+    References:
+        -   Gupta, H. V., Kling, H., Yilmaz, K. K., & Martinez, G. F. (2009). Decomposition of the mean squared error and
+            NSE performance criteria: Implications for improving hydrological modeling. Journal of Hydrology, 377(1-2), 80-91.
+            https://doi.org/10.1016/j.jhydrol.2009.08.003.
+        -   Knoben, W. J. M., Freer, J. E., & Woods, R. A. (2019). Technical note: Inherent benchmark or not?
+            Comparing Nash-Sutcliffe and Kling-Gupta efficiency scores. Hydrology and Earth System Sciences, 23(10), 4323-4331.
+            https://doi.org/10.5194/hess-23-4323-2019.
+
+
+    Examples:
+        >>> kge_s = kge(forecasts, obs,preserve_dims='lat')  # if data is of dimension {lat,time}, kge value is computed across the time dimension
+        >>> kge_s = kge(forecasts, obs,reduce_dims="time")  # if data is of dimension {lat,time}, reduce_dims="time" is same as preserve_dims='lat'
+        >>> kge_s = kge(forecasts, obs, return_components=True) # kge_s is dataset of all three components and kge value itself
+        >>> kge_s_weighted = kge(forecasts, obs, scaling_factors=(0.5, 1.0, 2.0)) # with scaling factors
+
+
+
+    """
+
+    # Type checks as xrray.corr can only handle xr.DataArray
+    if not isinstance(fcst, xr.DataArray):
+        raise TypeError("kge: fcst must be an xarray.DataArray")
+    if not isinstance(obs, xr.DataArray):
+        raise TypeError("kge: obs must be an xarray.DataArray")
+    if scaling_factors is not None:
+        if isinstance(scaling_factors, (list, np.ndarray)):
+            # Check if the input has exactly 3 elements
+            if len(scaling_factors) != 3:
+                raise ValueError("kge: scaling_factors must contain exactly 3 elements")
+        else:
+            raise TypeError("kge: scaling_factors must be a list of floats or a numpy array")
+    else:
+        scaling_factors = [1.0, 1.0, 1.0]
+
+    s_rho, s_alpha, s_beta = scaling_factors
+    reduce_dims = scores.utils.gather_dimensions(
+        fcst.dims, obs.dims, reduce_dims=reduce_dims, preserve_dims=preserve_dims
+    )
+    # Need to broadcast and match NaNs so that the fcst and obs are for the
+    # same points
+    fcst, obs = broadcast_and_match_nan(fcst, obs)
+    # compute linear correlation coefficient r between fcst and obs
+    rho = xr.corr(fcst, obs, reduce_dims)  # type: ignore
+
+    # compute alpha (sigma_sim / sigma_obs)
+    sigma_fcst = fcst.std(reduce_dims)
+    sigma_obs = obs.std(reduce_dims)
+    alpha = sigma_fcst / sigma_obs
+
+    # compute beta (mu_sim / mu_obs)
+    mu_fcst = fcst.mean(reduce_dims)
+    mu_obs = obs.mean(reduce_dims)
+    beta = mu_fcst / mu_obs
+
+    # compute Euclidian distance from the ideal point in the scaled space
+    ed_s = np.sqrt((s_rho * (rho - 1)) ** 2 + (s_alpha * (alpha - 1)) ** 2 + (s_beta * (beta - 1)) ** 2)
+    kge_s = 1 - ed_s
+    if return_components:
+        # Create dataset of all components
+        kge_s = xr.Dataset(
+            {
+                "kge": kge_s,
+                "rho": rho,
+                "alpha": alpha,
+                "beta": beta,
+            }
+        )
+    return kge_s  # type: ignore

src/scores/probability/crps_impl.py

@@ -1182,7 +1182,7 @@ def interval_tw_crps_for_ensemble(
         >>> interval_tw_crps_for_ensemble(fcst, obs, 'ensemble', -20, 10)
     """
     if isinstance(lower_threshold, xr.DataArray) or isinstance(upper_threshold, xr.DataArray):
-        if (lower_threshold >= upper_threshold).any().values.item():
+        if (lower_threshold >= upper_threshold).any().values.item():  # type: ignore  # mypy is wrong, I think
             raise ValueError("`lower_threshold` must be less than `upper_threshold`")
     elif lower_threshold >= upper_threshold:
         raise ValueError("`lower_threshold` must be less than `upper_threshold`")

tests/continuous/test_standard.py

@@ -689,6 +689,91 @@ def test_rmse_angular():
 
 EXP_DS_PBIAS1 = xr.Dataset({"a": EXP_PBIAS1, "b": EXP_PBIAS2})
 
+## for KGE
+DA1_KGE = xr.DataArray(
+    np.array([[1, 2, 3], [0, 1, 0], [0.5, -0.5, 0.5], [3, 6, 3]]),
+    dims=("space", "time"),
+    coords=[
+        ("space", ["w", "x", "y", "z"]),
+        ("time", [1, 2, 3]),
+    ],
+)
+
+DA2_KGE = xr.DataArray(
+    np.array([[2, 4, 6], [6, 5, 6], [3, 4, 5], [3, np.nan, 3]]),
+    dims=("space", "time"),
+    coords=[
+        ("space", ["w", "x", "y", "z"]),
+        ("time", [1, 2, 3]),
+    ],
+)
+
+DA3_KGE = xr.DataArray(
+    np.array([[1, 2, 3], [3, 2.5, 3], [1.5, 2, 2.5], [1.5, np.nan, 1.5]]),
+    dims=("space", "time"),
+    coords=[
+        ("space", ["w", "x", "y", "z"]),
+        ("time", [1, 2, 3]),
+    ],
+)
+DA4_KGE = xr.DataArray(
+    np.array([[1, 3, 7], [2, 2, 8], [3, 1, 7]]),
+    dims=("space", "time"),
+    coords=[
+        ("space", ["x", "y", "z"]),
+        ("time", [1, 2, 3]),
+    ],
+)
+DA5_KGE = xr.DataArray(
+    np.array([1, 2, 3]),
+    dims=("space"),
+    coords=[("space", ["x", "y", "z"])],
+)
+
+## Expected KGE values
+EXP_KGE_KEEP_SPACE_DIM = xr.DataArray(
+    np.array([0.2928932188134524, -1.2103875562418747, -0.44811448882050064, np.nan]),
+    dims=("space"),
+    coords=[("space", ["w", "x", "y", "z"])],
+)
+EXP_KGE_REDUCE_ALL = xr.DataArray(0.2928932188134524)
+
+EXP_KGE_rho_returns_components = xr.DataArray(1.0)
+EXP_KGE_alpha_returns_components = xr.DataArray(0.5)
+EXP_KGE_beta_returns_components = xr.DataArray(0.5)
+
+EXP_KGE_returns_components = xr.Dataset(
+    {
+        "kge": EXP_KGE_REDUCE_ALL,
+        "rho": EXP_KGE_rho_returns_components,
+        "alpha": EXP_KGE_alpha_returns_components,
+        "beta": EXP_KGE_beta_returns_components,
+    }
+)
+
+
+EXP_KGE_Scaling_Factors = xr.DataArray(
+    1 - np.sqrt((0.5 * (1 - 1)) ** 2 + (1.0 * (0.5 - 1)) ** 2 + (2 * (0.5 - 1)) ** 2)
+)
+
+
+EXP_KGE_DIFF_SIZE = xr.DataArray(
+    np.array([1.0, -1.0, -1.8791915368841288]),
+    dims=("time"),
+    coords=[("time", [1, 2, 3])],
+)
+
+## Parametrized test for kge function to check various incorrect types and sizes
+Incorrect_Input_KGE = [1, 2, 3]
+Incorrect_SFactors_Type_KGE = "incorrect_type"
+Incorrect_SFactors_List_KGE = [1, 2]
+Incorrect_SFactors_Numpy_KGE = np.array([1, 2, 3, 4])
+
+EXP_KGE_message1 = "kge: fcst must be an xarray.DataArray"
+EXP_KGE_message2 = "kge: obs must be an xarray.DataArray"
+EXP_KGE_message3 = "kge: scaling_factors must be a list of floats or a numpy array"
+EXP_KGE_message4 = "kge: scaling_factors must contain exactly 3 elements"
+
 
 @pytest.mark.parametrize(
     ("fcst", "obs", "reduce_dims", "preserve_dims", "weights", "expected"),
@@ -813,3 +898,71 @@ def test_pbias_dask():
     result = result.compute()
     assert isinstance(result.data, np.ndarray)
     xr.testing.assert_equal(result, EXP_PBIAS3)
+
+
+@pytest.mark.parametrize(
+    ("fcst", "obs", "reduce_dims", "preserve_dims", "return_components", "scaling_factors", "expected"),
+    [
+        # Check reduce dim arg
+        (DA1_KGE, DA2_KGE, None, "space", False, None, EXP_KGE_KEEP_SPACE_DIM),
+        # Check preserve dim arg
+        (DA1_KGE, DA2_KGE, "time", None, False, None, EXP_KGE_KEEP_SPACE_DIM),
+        # Check reduce all
+        (DA3_KGE, DA2_KGE, None, None, False, None, EXP_KGE_REDUCE_ALL),
+        # returning components
+        (DA3_KGE, DA2_KGE, None, None, True, None, EXP_KGE_returns_components),
+        # Check scaling_factors
+        (DA3_KGE, DA2_KGE, None, None, False, [0.5, 1.0, 2.0], EXP_KGE_Scaling_Factors),
+        # Check different size arrays as input
+        (DA4_KGE, DA5_KGE, "space", None, False, None, EXP_KGE_DIFF_SIZE),
+    ],
+)
+def test_kge(fcst, obs, reduce_dims, preserve_dims, return_components, scaling_factors, expected):
+    """
+    Tests continuous.kge
+    """
+    result = scores.continuous.kge(
+        fcst,
+        obs,
+        reduce_dims=reduce_dims,
+        preserve_dims=preserve_dims,
+        return_components=return_components,
+        scaling_factors=scaling_factors,
+    )
+    xr.testing.assert_allclose(result, expected, rtol=1e-10, atol=1e-10)
+
+
+def test_kge_dask():
+    """
+    Tests that continuous.kge works with Dask
+    """
+    fcst = DA3_KGE.chunk()
+    obs = DA2_KGE.chunk()
+    result = scores.continuous.kge(fcst, obs)
+    assert isinstance(result.data, dask.array.Array)  # type: ignore
+    result = result.compute()  # type: ignore
+    assert isinstance(result.data, (np.ndarray, np.generic))
+    xr.testing.assert_equal(result, EXP_KGE_REDUCE_ALL)
+
+
+@pytest.mark.parametrize(
+    "fcst, obs, scaling_factors, expected_exception, expected_message",
+    [
+        # Test case for fcst with incorrect type (list instead of xr.DataArray)
+        (Incorrect_Input_KGE, DA2_KGE, None, TypeError, EXP_KGE_message1),
+        # Test case for obs with incorrect type (list instead of xr.DataArray)
+        (DA1_KGE, Incorrect_Input_KGE, None, TypeError, EXP_KGE_message2),
+        # Test case for scaling_factors with incorrect type (string instead of list or np.ndarray)
+        (DA1_KGE, DA2_KGE, Incorrect_SFactors_Type_KGE, TypeError, EXP_KGE_message3),
+        # Test case for scaling_factors with incorrect number of elements (list with 2 elements)
+        (DA1_KGE, DA2_KGE, Incorrect_SFactors_List_KGE, ValueError, EXP_KGE_message4),
+        # Test case for scaling_factors with incorrect number of elements (numpy array with 4 elements)
+        (DA1_KGE, DA2_KGE, Incorrect_SFactors_Numpy_KGE, ValueError, EXP_KGE_message4),
+    ],
+)
+def test_kge_errors(fcst, obs, scaling_factors, expected_exception, expected_message):
+    """
+    Test continuous.kge raises error with an incorrect type and sizes
+    """
+    with pytest.raises(expected_exception, match=expected_message):
+        scores.continuous.kge(fcst, obs, scaling_factors=scaling_factors)  # type: ignore