Repository: hsbc/pyratings (language: python)

Issue:
Reduce coupling of functions in consolidate.py
Currently, when computing a portfolio average short-term rating, we :

1. clean the ratings using _get_pure_ratings_
2. consolidate the ratings (i.e. convert ST ratings into AEV scores, identify the proper number - highest for worst rating, lowest for best rating... - and translate it back to ST rating) using _get_worst_ratings_, _get_best_ratings_  or _get_second_best_ratings_
3. get AEV scores from the consolidated ratings using _get_scores_from_ratings_
4. calculate the AEV score average using _get_weighted_average_
5. go back to ST ratings using _get_ratings_from_scores_

The fact of going back and forth from AEV scores to ratings in steps 2 and 3 may introduce a bias. Let's consider a security rated P-1 by Moody's and F1+ by Fitch. It translates into AEV scores of respectively 3.5 and 3. The worst score would then be **3.5**. 
But directly using _get_worst_ratings_ (with the default 'S&P' value for the 'rating_provider_output' parameter) on the P-1 and F1+ ratings, we end up with  A-1+ which would translate in **2.5** using _get_scores_from_ratings_ .


A solution would be to introduce _get_worst_scores_, _get_best_scores_  or _get_second_best_scores_ which only convert ST ratings into AEV scores and identify the proper number but would replace steps 2 and 3.

What do you think?


Discussion / hints:
This is a valid point. It is also true for computing the average portfolio long-term rating.

We are translating ratings into scores to identify the lowest/second lowest/highest score when computing the best/second best/worst rating, Subsequently, we are translating back into a human-readable rating.  
When computing the average rating, we are again translating the same ratings back into scores and vice versa. This involves unnecessary computations.

The functions ``get_best_ratings``, ``get_second_best_ratigs``, and ``get_worst_ratings`` all make use of ``get_scores_from_ratings``.
https://github.com/hsbc/pyratings/blob/5434d9333b8e958d0b7eae628caf18d95f65470a/src/pyratings/consolidate.py#L83-L86

Your suggestion to introduce new functions ``get_best_scores``, ``get_second_best_scores``, and ``get_worst_scores`` will indeed reduce coupling between these individual functions.

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 290840a..2f6397e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,7 +8,12 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
 - Option to choose between three different strategies to translate scores into 
   short-term ratings ([#24](https://github.com/hsbc/pyratings/pull/24)).
 - Functionality to remove prefix '(P)' when cleaning ratings
-  ([#27](https://github.com/hsbc/pyratings/issues/27))
+  ([#27](https://github.com/hsbc/pyratings/issues/27)).
+- Functions to directly compute the "best", "second-best", and "worst" rating score 
+  for individual securities ([#37](https://github.com/hsbc/pyratings/issues/37)).
+- Function ``consolidate_ratings``: Wrapper function for ``get_best_ratings``, 
+  ``get_second_best_ratings``, and ``get_worst_ratings``
+  ([#37](https://github.com/hsbc/pyratings/issues/37)).
 
 ### Changed
 - BREAKING CHANGE: Automatic column naming
diff --git a/src/pyratings/__init__.py b/src/pyratings/__init__.py
index 8590976..5292f0a 100644
--- a/src/pyratings/__init__.py
+++ b/src/pyratings/__init__.py
@@ -15,9 +15,13 @@
 from pyratings.aggregate import get_weighted_average
 from pyratings.clean import get_pure_ratings
 from pyratings.consolidate import (
+    consolidate_ratings,
     get_best_ratings,
+    get_best_scores,
     get_second_best_ratings,
+    get_second_best_scores,
     get_worst_ratings,
+    get_worst_scores,
 )
 from pyratings.get_ratings import get_ratings_from_scores, get_ratings_from_warf
 from pyratings.get_scores import get_scores_from_ratings, get_scores_from_warf
@@ -29,6 +33,10 @@
 __all__ = [
     "_extract_rating_provider",
     "_get_translation_dict",
+    "consolidate_ratings",
+    "get_best_scores",
+    "get_second_best_scores",
+    "get_worst_scores",
     "get_best_ratings",
     "get_pure_ratings",
     "get_ratings_from_scores",
diff --git a/src/pyratings/consolidate.py b/src/pyratings/consolidate.py
index 78833ab..3db2b1d 100644
--- a/src/pyratings/consolidate.py
+++ b/src/pyratings/consolidate.py
@@ -14,20 +14,316 @@
 
 """Module contains functions to consolidate ratings from different rating agencies."""
 
-from typing import List
+from typing import Literal
 
 import pandas as pd
 
 from pyratings.get_ratings import get_ratings_from_scores
 from pyratings.get_scores import get_scores_from_ratings
-from pyratings.utils import _extract_rating_provider, valid_rtg_agncy
+
+
+def get_best_scores(
+    ratings: pd.DataFrame,
+    rating_provider_input: list[str] = None,
+    tenor: Literal["long-term", "short-term"] = "long-term",
+) -> pd.Series:
+    """Compute the best rating scores on a security level basis across rating agencies.
+
+    Parameters
+    ----------
+    ratings
+        Dataframe consisting of clean ratings (i.e. stripped off of watches/outlooks)
+    rating_provider_input
+        Indicates rating providers within `ratings`. Should contain any valid rating
+        provider out of {"Fitch", "Moody's", "S&P", "Bloomberg", "DBRS"}.
+
+        If None, `rating_provider_input` will be inferred from the dataframe column
+        names.
+    tenor
+        Should contain any valid tenor out of {"long-term", "short-term"}
+
+    Returns
+    -------
+    pd.Series
+        Best rating scores on a security level basis.
+
+    Examples
+    --------
+    >>> import pandas as pd
+
+    >>> ratings_df = pd.DataFrame(
+    ...     data=(
+    ...         {
+    ...             "rating_S&P": ['AAA', 'AA-', 'AA+', 'BB-', 'C'],
+    ...             "rating_Moody's": ['Aa1', 'Aa3', 'Aa2', 'Ba3', 'Ca'],
+    ...             "rating_Fitch": ['AA-', 'AA-', 'AA-', 'B+', 'C'],
+    ...         }
+    ...     )
+    ... )
+    >>> get_best_scores(
+    ...     ratings=ratings_df,
+    ...     rating_provider_input=["S&P", "Moody", "Fitch"]
+    ... )
+    0     1
+    1     4
+    2     2
+    3    13
+    4    20
+    Name: best_scores, dtype: int64
+
+    """
+    rating_scores_df = get_scores_from_ratings(
+        ratings=ratings, rating_provider=rating_provider_input, tenor=tenor
+    )
+    rating_scores_series = rating_scores_df.min(axis=1)
+    rating_scores_series.name = "best_scores"
+
+    return rating_scores_series
+
+
+def get_second_best_scores(
+    ratings: pd.DataFrame,
+    rating_provider_input: list[str] = None,
+    tenor: Literal["long-term", "short-term"] = "long-term",
+) -> pd.Series:
+    """Compute the second-best scores on a security level basis across rating agencies.
+
+    Parameters
+    ----------
+    ratings
+        Dataframe consisting of clean ratings (i.e. stripped off of watches/outlooks)
+    rating_provider_input
+        Indicates rating providers within `ratings`. Should contain any valid rating
+        provider out of {"Fitch", "Moody's", "S&P", "Bloomberg", "DBRS"}.
+
+        If None, `rating_provider_input` will be inferred from the dataframe column
+        names.
+    tenor
+        Should contain any valid tenor out of {"long-term", "short-term"}
+
+    Returns
+    -------
+    pd.Series
+        Second-best scores on a security level basis.
+
+    Examples
+    --------
+    >>> import pandas as pd
+
+    >>> ratings_df = pd.DataFrame(
+    ...     data=(
+    ...         {
+    ...             "rating_S&P": ['AAA', 'AA-', 'AA+', 'BB-', 'C'],
+    ...             "rating_Moody's": ['Aa1', 'Aa3', 'Aa2', 'Ba3', 'Ca'],
+    ...             "rating_Fitch": ['AA-', 'AA-', 'AA-', 'B+', 'C'],
+    ...         }
+    ...     )
+    ... )
+    >>> get_second_best_scores(
+    ...     ratings_df, rating_provider_input=["S&P", "Moody", "Fitch"]
+    ... )
+    0     2.0
+    1     4.0
+    2     3.0
+    3    13.0
+    4    21.0
+    Name: second_best_scores, dtype: float64
+
+    """
+    rating_scores_df = get_scores_from_ratings(
+        ratings=ratings, rating_provider=rating_provider_input, tenor=tenor
+    )
+
+    # rank scores per security (axis=1)
+    scores_ranked_df = rating_scores_df.rank(axis=1, method="first", numeric_only=False)
+
+    # get column with rank of 2, if available, otherwise get column with rank 1
+    rating_scores_ranked_series = rating_scores_df[scores_ranked_df <= 2].max(axis=1)
+
+    rating_scores_ranked_series.name = "second_best_scores"
+
+    return rating_scores_ranked_series
+
+
+def get_worst_scores(
+    ratings: pd.DataFrame,
+    rating_provider_input: list[str] = None,
+    tenor: Literal["long-term", "short-term"] = "long-term",
+) -> pd.Series:
+    """Compute the worst scores on a security level basis across rating agencies.
+
+    Parameters
+    ----------
+    ratings
+        Dataframe consisting of clean ratings (i.e. stripped off of watches/outlooks)
+    rating_provider_input
+        Indicates rating providers within `ratings`. Should contain any valid rating
+        provider out of {"Fitch", "Moody's", "S&P", "Bloomberg", "DBRS"}.
+
+        If None, `rating_provider_innput` will be inferred from the dataframe column
+        names.
+    tenor
+        Should contain any valid tenor out of {"long-term", "short-term"}
+
+    Returns
+    -------
+    pd.Series
+        Worst scores on a security level basis.
+
+    Examples
+    --------
+    >>> import pandas as pd
+
+    >>> ratings_df = pd.DataFrame(
+    ...     data=(
+    ...         {
+    ...             "rating_S&P": ['AAA', 'AA-', 'AA+', 'BB-', 'C'],
+    ...             "rating_Moody's": ['Aa1', 'Aa3', 'Aa2', 'Ba3', 'Ca'],
+    ...             "rating_Fitch": ['AA-', 'AA-', 'AA-', 'B+', 'C'],
+    ...         }
+    ...     )
+    ... )
+    >>> get_worst_scores(ratings_df, rating_provider_input=["S&P", "Moody", "Fitch"])
+    0     4
+    1     4
+    2     4
+    3    14
+    4    21
+    Name: worst_scores, dtype: int64
+
+    """
+    rating_scores_df = get_scores_from_ratings(
+        ratings=ratings, rating_provider=rating_provider_input, tenor=tenor
+    )
+    rating_scores_series = rating_scores_df.max(axis=1)
+    rating_scores_series.name = "worst_scores"
+
+    return rating_scores_series
+
+
+def consolidate_ratings(
+    ratings: pd.DataFrame,
+    method: Literal["best", "second_best", "worst"] = "worst",
+    rating_provider_input: list[str] = None,
+    rating_provider_output: Literal[
+        "Fitch", "Moody", "S&P", "Bloomberg", "DBRS"
+    ] = "S&P",
+    tenor: Literal["long-term", "short-term"] = "long-term",
+) -> pd.Series:
+    """Consolidate ratings on a security level basis across rating agencies .
+
+    Parameters
+    ----------
+    ratings
+        Dataframe consisting of clean ratings (i.e. stripped off of watches/outlooks)
+    method
+        Defines the method that will be used in order to consolidate the ratings on a
+        security level basis across rating agencies.
+        Valid methods are {"best", "second_best", "worst"}.
+    rating_provider_input
+        Indicates rating providers within `ratings`. Should contain any valid rating
+        provider out of {"Fitch", "Moody's", "S&P", "Bloomberg", "DBRS"}.
+
+        If None, `rating_provider_input` will be inferred from the dataframe column
+        names.
+    rating_provider_output
+        Indicates which rating scale will be used for output results.
+        Should contain any valid rating provider out of
+        {"Fitch", "Moody's", "S&P", "Bloomberg", "DBRS"}.
+    tenor
+        Should contain any valid tenor out of {"long-term", "short-term"}
+
+    Returns
+    -------
+    pd.Series
+        Consolidated ratings on a security level basis.
+
+    Examples
+    --------
+    >>> import pandas as pd
+
+    >>> ratings_df = pd.DataFrame(
+    ...     data=(
+    ...         {
+    ...             "rating_S&P": ['AAA', 'AA-', 'AA+', 'BB-', 'C'],
+    ...             "rating_Moody's": ['Aa1', 'Aa3', 'Aa2', 'Ba3', 'Ca'],
+    ...             "rating_Fitch": ['AA-', 'AA-', 'AA-', 'B+', 'C'],
+    ...         }
+    ...     )
+    ... )
+
+    Identify the best ratings:
+
+    >>> consolidate_ratings(
+    ...     ratings=ratings_df,
+    ...     method="best",
+    ...     rating_provider_input=["S&P", "Moody", "Fitch"],
+    ...     rating_provider_output="Moody",
+    ... )
+    0    Aaa
+    1    Aa3
+    2    Aa1
+    3    Ba3
+    4    Ca
+    Name: best_rtg, dtype: object
+
+    Identify the second-best ratings:
+
+    >>> consolidate_ratings(
+    ...     ratings=ratings_df,
+    ...     method="second_best",
+    ...     rating_provider_input=["S&P", "Moody", "Fitch"],
+    ...     rating_provider_output="DBRS",
+    ... )
+    0    AAH
+    1    AAL
+    2    AA
+    3    BBL
+    4    C
+    Name: second_best_rtg, dtype: object
+
+    Identify the worst ratings:
+
+    >>> consolidate_ratings(
+    ...     ratings=ratings_df,
+    ...     method="worst",
+    ...     rating_provider_input=["S&P", "Moody", "Fitch"]
+    ... )
+    0    AA-
+    1    AA-
+    2    AA-
+    3    B+
+    4    C
+    Name: worst_rtg, dtype: object
+
+    """
+    func = {
+        "best": get_best_scores,
+        "second_best": get_second_best_scores,
+        "worst": get_worst_scores,
+    }
+
+    # translate ratings -> scores
+    rating_scores_series = func[method](
+        ratings, rating_provider_input=rating_provider_input, tenor=tenor
+    )
+    # convert back to ratings
+    ratings_series = get_ratings_from_scores(
+        rating_scores=rating_scores_series,
+        rating_provider=rating_provider_output,
+        tenor=tenor,
+    )
+    ratings_series.name = f"{method}_rtg"
+    return ratings_series
 
 
 def get_best_ratings(
     ratings: pd.DataFrame,
-    rating_provider_input: List[str] = None,
-    rating_provider_output: str = "S&P",
-    tenor: str = "long-term",
+    rating_provider_input: list[str] = None,
+    rating_provider_output: Literal[
+        "Fitch", "Moody", "S&P", "Bloomberg", "DBRS"
+    ] = "S&P",
+    tenor: Literal["long-term", "short-term"] = "long-term",
 ) -> pd.Series:
     """Compute the best rating on a security level basis across rating agencies.
 
@@ -75,32 +371,24 @@ def get_best_ratings(
     Name: best_rtg, dtype: object
 
     """
-    rating_provider_output = _extract_rating_provider(
-        rating_provider=rating_provider_output,
-        valid_rtg_provider=valid_rtg_agncy[tenor],
-    )
-
-    # translate ratings -> scores
-    rating_scores_df = get_scores_from_ratings(
-        ratings, rating_provider=rating_provider_input, tenor=tenor
-    )
-
-    # determine the lowest ratings score (indicates best rating) and convert to ratings
-    best_ratings_series = get_ratings_from_scores(
-        rating_scores=rating_scores_df.min(axis=1),
-        rating_provider=rating_provider_output,
+    ratings_series = consolidate_ratings(
+        method="best",
+        ratings=ratings,
+        rating_provider_input=rating_provider_input,
+        rating_provider_output=rating_provider_output,
         tenor=tenor,
     )
-    best_ratings_series.name = "best_rtg"
 
-    return best_ratings_series
+    return ratings_series
 
 
 def get_second_best_ratings(
     ratings: pd.DataFrame,
-    rating_provider_input: List[str] = None,
-    rating_provider_output: str = "S&P",
-    tenor: str = "long-term",
+    rating_provider_input: list[str] = None,
+    rating_provider_output: Literal[
+        "Fitch", "Moody", "S&P", "Bloomberg", "DBRS"
+    ] = "S&P",
+    tenor: Literal["long-term", "short-term"] = "long-term",
 ) -> pd.Series:
     """Compute the second-best rating on a security level basis across rating agencies.
 
@@ -150,38 +438,24 @@ def get_second_best_ratings(
     Name: second_best_rtg, dtype: object
 
     """
-    rating_provider_output = _extract_rating_provider(
-        rating_provider=rating_provider_output,
-        valid_rtg_provider=valid_rtg_agncy[tenor],
-    )
-
-    # translate ratings -> scores
-    rating_scores_df = get_scores_from_ratings(
-        ratings, rating_provider=rating_provider_input, tenor=tenor
-    )
-
-    # rank scores per security (axis=1)
-    scores_ranked_df = rating_scores_df.rank(axis=1, method="first", numeric_only=False)
-
-    # get column with rank of 2, if available, otherwise get column with rank 1
-    scores_ranked_ser = rating_scores_df[scores_ranked_df <= 2].max(axis=1)
-
-    # determine the lowest ratings score (indicates best rating) and convert to ratings
-    second_best_ratings_series = get_ratings_from_scores(
-        rating_scores=scores_ranked_ser,
-        rating_provider=rating_provider_output,
+    ratings_series = consolidate_ratings(
+        method="second_best",
+        ratings=ratings,
+        rating_provider_input=rating_provider_input,
+        rating_provider_output=rating_provider_output,
         tenor=tenor,
     )
-    second_best_ratings_series.name = "second_best_rtg"
 
-    return second_best_ratings_series
+    return ratings_series
 
 
 def get_worst_ratings(
     ratings: pd.DataFrame,
-    rating_provider_input: List[str] = None,
-    rating_provider_output: str = "S&P",
-    tenor: str = "long-term",
+    rating_provider_input: list[str] = None,
+    rating_provider_output: Literal[
+        "Fitch", "Moody", "S&P", "Bloomberg", "DBRS"
+    ] = "S&P",
+    tenor: Literal["long-term", "short-term"] = "long-term",
 ) -> pd.Series:
     """Compute the worst rating on a security level basis across rating agencies.
 
@@ -229,23 +503,12 @@ def get_worst_ratings(
     Name: worst_rtg, dtype: object
 
     """
-    rating_provider_output = _extract_rating_provider(
-        rating_provider=rating_provider_output,
-        valid_rtg_provider=valid_rtg_agncy[tenor],
-    )
-
-    # translate ratings -> scores
-    rating_scores_df = get_scores_from_ratings(
-        ratings, rating_provider=rating_provider_input, tenor=tenor
-    )
-
-    # determine the highest ratings score (indicates worst rating) and convert to
-    # ratings
-    worst_ratings_series = get_ratings_from_scores(
-        rating_scores=rating_scores_df.max(axis=1),
-        rating_provider=rating_provider_output,
+    ratings_series = consolidate_ratings(
+        method="worst",
+        ratings=ratings,
+        rating_provider_input=rating_provider_input,
+        rating_provider_output=rating_provider_output,
         tenor=tenor,
     )
-    worst_ratings_series.name = "worst_rtg"
 
-    return worst_ratings_series
+    return ratings_series


Test patch:
diff --git a/tests/test_consolidate.py b/tests/test_consolidate.py
index 56e3518..5f30838 100644
--- a/tests/test_consolidate.py
+++ b/tests/test_consolidate.py
@@ -13,6 +13,7 @@
 #    limitations under the License.
 
 """Module contains unit tests for consolidation of ratings."""
+from __future__ import annotations
 
 import numpy as np
 import pandas as pd
@@ -46,6 +47,117 @@ def rtg_inputs_shortterm() -> pd.DataFrame:
     )
 
 
+@pytest.mark.parametrize("rating_provider_input", (["SP", "Moody", "Fitch"], None))
+def test_get_best_rating_scores_long_term(
+    rtg_inputs_longterm: pd.DataFrame, rating_provider_input: list[str] | None
+) -> None:
+    """It returns the best long-term scores on a security (line-by-line) basis."""
+    actual = rtg.get_best_scores(
+        ratings=rtg_inputs_longterm,
+        rating_provider_input=rating_provider_input,
+        tenor="long-term",
+    )
+
+    expectations = pd.Series(
+        data=[1.0, 4.0, 2.0, 13.0, 20, np.nan, 8, 3], name="best_scores"
+    )
+
+    pd.testing.assert_series_equal(actual, expectations)
+
+
+@pytest.mark.parametrize("rating_provider_input", (["SP", "Moody", "Fitch"], None))
+def test_get_best_rating_scores_shortterm(
+    rtg_inputs_shortterm: pd.DataFrame,
+    rating_provider_input: list[str] | None,
+) -> None:
+    """It returns the best short-term scores on a security (line-by-line) basis."""
+    actual = rtg.get_best_scores(
+        rtg_inputs_shortterm,
+        rating_provider_input=rating_provider_input,
+        tenor="short-term",
+    )
+
+    expectations = pd.Series(
+        data=[5.5, 10.0, 2.5, 9.5, 9.5, np.nan, 8.0, 9.5], name="best_scores"
+    )
+
+    pd.testing.assert_series_equal(actual, expectations)
+
+
+@pytest.mark.parametrize("rating_provider_input", (["SP", "Moody", "Fitch"], None))
+def test_get_second_best_rating_scores_long_term(
+    rtg_inputs_longterm: pd.DataFrame, rating_provider_input: list[str] | None
+) -> None:
+    """It returns the second-best lt scores on a security (line-by-line) basis."""
+    actual = rtg.get_second_best_scores(
+        ratings=rtg_inputs_longterm,
+        rating_provider_input=rating_provider_input,
+        tenor="long-term",
+    )
+
+    expectations = pd.Series(
+        data=[2.0, 4.0, 3.0, 13.0, 21, np.nan, 8, 3], name="second_best_scores"
+    )
+
+    pd.testing.assert_series_equal(actual, expectations)
+
+
+@pytest.mark.parametrize("rating_provider_input", (["SP", "Moody", "Fitch"], None))
+def test_get_second_best_rating_scores_shortterm(
+    rtg_inputs_shortterm: pd.DataFrame,
+    rating_provider_input: list[str] | None,
+) -> None:
+    """It returns the best short-term scores on a security (line-by-line) basis."""
+    actual = rtg.get_second_best_scores(
+        rtg_inputs_shortterm,
+        rating_provider_input=rating_provider_input,
+        tenor="short-term",
+    )
+
+    expectations = pd.Series(
+        data=[6.5, 16.5, 3.5, 16.5, 9.5, np.nan, 8.0, 9.5], name="second_best_scores"
+    )
+
+    pd.testing.assert_series_equal(actual, expectations)
+
+
+@pytest.mark.parametrize("rating_provider_input", (["SP", "Moody", "Fitch"], None))
+def test_get_worst_rating_scores_long_term(
+    rtg_inputs_longterm: pd.DataFrame, rating_provider_input: list[str] | None
+) -> None:
+    """It returns the worst long-term scores on a security (line-by-line) basis."""
+    actual = rtg.get_worst_scores(
+        ratings=rtg_inputs_longterm,
+        rating_provider_input=rating_provider_input,
+        tenor="long-term",
+    )
+
+    expectations = pd.Series(
+        data=[4.0, 4.0, 4.0, 14.0, 21, np.nan, 8, 3], name="worst_scores"
+    )
+
+    pd.testing.assert_series_equal(actual, expectations)
+
+
+@pytest.mark.parametrize("rating_provider_input", (["SP", "Moody", "Fitch"], None))
+def test_get_worst_rating_scores_shortterm(
+    rtg_inputs_shortterm: pd.DataFrame,
+    rating_provider_input: list[str] | None,
+) -> None:
+    """It returns the worst short-term scores on a security (line-by-line) basis."""
+    actual = rtg.get_worst_scores(
+        rtg_inputs_shortterm,
+        rating_provider_input=rating_provider_input,
+        tenor="short-term",
+    )
+
+    expectations = pd.Series(
+        data=[7.5, 16.5, 6.5, 22.0, 13.5, np.nan, 8.0, 10.0], name="worst_scores"
+    )
+
+    pd.testing.assert_series_equal(actual, expectations)
+
+
 def test_get_best_rating_longterm_with_explicit_rating_provider(
     rtg_inputs_longterm: pd.DataFrame,
 ) -> None: