OpenBB/openbb_platform/extensions/quantitative/openbb_quantitative/quantitative_router.py

"""Quantitative Analysis Router."""

from typing import List, Literal

import numpy as np
import pandas as pd
from openbb_core.app.model.obbject import OBBject
from openbb_core.app.router import Router
from openbb_core.app.utils import (
    basemodel_to_df,
    df_to_basemodel,
    get_target_column,
    get_target_columns,
)
from openbb_core.provider.abstract.data import Data
from pydantic import NonNegativeFloat, PositiveInt

from .helpers import get_fama_raw
from .models import (
    ADFTestModel,
    CAPMModel,
    KPSSTestModel,
    NormalityModel,
    OmegaModel,
    SummaryModel,
    TestModel,
    UnitRootModel,
)

router = Router(prefix="")


@router.command(methods=["POST"])
def normality(data: List[Data], target: str) -> OBBject[NormalityModel]:
    """Get Normality Statistics.

    - **Kurtosis**: whether the kurtosis of a sample differs from the normal distribution.
    - **Skewness**: whether the skewness of a sample differs from the normal distribution.
    - **Jarque-Bera**: whether the sample data has the skewness and kurtosis matching a normal distribution.
    - **Shapiro-Wilk**: whether a random sample comes from a normal distribution.
    - **Kolmogorov-Smirnov**: whether two underlying one-dimensional probability distributions differ.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.

    Returns
    -------
    OBBject[NormalityModel]
        Normality tests summary. See qa_models.NormalityModel for details.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.normality(data=stock_data, target="close")
    """
    from scipy import stats  # pylint: disable=import-outside-toplevel

    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    kt_statistic, kt_pvalue = stats.kurtosistest(series_target)
    sk_statistic, sk_pvalue = stats.skewtest(series_target)
    jb_statistic, jb_pvalue = stats.jarque_bera(series_target)
    sh_statistic, sh_pvalue = stats.shapiro(series_target)
    ks_statistic, ks_pvalue = stats.kstest(series_target, "norm")

    norm_summary = NormalityModel(
        kurtosis=TestModel(statistic=kt_statistic, p_value=kt_pvalue),
        skewness=TestModel(statistic=sk_statistic, p_value=sk_pvalue),
        jarque_bera=TestModel(statistic=jb_statistic, p_value=jb_pvalue),
        shapiro_wilk=TestModel(statistic=sh_statistic, p_value=sh_pvalue),
        kolmogorov_smirnov=TestModel(statistic=ks_statistic, p_value=ks_pvalue),
    )

    return OBBject(results=norm_summary)


@router.command(methods=["POST"])
def capm(data: List[Data], target: str) -> OBBject[CAPMModel]:
    """Get Capital Asset Pricing Model (CAPM).

    CAPM offers a streamlined way to assess the expected return on an investment while accounting for its risk relative
    to the market. It's a cornerstone of modern financial theory that helps investors understand the trade-off between
    risk and return, guiding more informed investment choices.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.

    Returns
    -------
    OBBject[CAPMModel]
        CAPM model summary.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.capm(data=stock_data, target="close")
    """
    import statsmodels.api as sm  # pylint: disable=import-outside-toplevel # type: ignore

    df = basemodel_to_df(data)

    df_target = get_target_columns(df, ["date", target])
    df_target = df_target.set_index("date")
    df_target.loc[:, "return"] = df_target.pct_change()
    df_target = df_target.dropna()
    df_target.index = pd.to_datetime(df_target.index)
    start_date = df_target.index.min().strftime("%Y-%m-%d")
    end_date = df_target.index.max().strftime("%Y-%m-%d")
    df_fama = get_fama_raw(start_date, end_date)
    df_target = df_target.merge(df_fama, left_index=True, right_index=True)
    df_target["excess_return"] = df_target["return"] - df_target["RF"]
    df_target["excess_mkt"] = df_target["MKT-RF"] - df_target["RF"]
    df_target = df_target.dropna()

    y = df_target[["excess_return"]]
    x = df_target["excess_mkt"]
    x = sm.add_constant(x)
    model = sm.OLS(y, x).fit()

    results = CAPMModel(
        market_risk=model.params["excess_mkt"],
        systematic_risk=model.rsquared,
        idiosyncratic_risk=1 - model.rsquared,
    )

    return OBBject(results=results)


@router.command(methods=["POST"])
def omega_ratio(
    data: List[Data],
    target: str,
    threshold_start: float = 0.0,
    threshold_end: float = 1.5,
) -> OBBject[List[OmegaModel]]:
    """Calculate the Omega Ratio.

    The Omega Ratio is a sophisticated metric that goes beyond traditional performance measures by considering the
    probability of achieving returns above a given threshold. It offers a more nuanced view of risk and reward,
    focusing on the likelihood of success rather than just average outcomes.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    threshold_start : float, optional
        Start threshold, by default 0.0
    threshold_end : float, optional
        End threshold, by default 1.5

    Returns
    -------
    OBBject[List[OmegaModel]]
        Omega ratios.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.omega_ratio(data=stock_data, target="close")
    """
    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    epsilon = 1e-6  # to avoid division by zero

    def get_omega_ratio(df_target: pd.Series, threshold: float) -> float:
        """Get omega ratio."""
        daily_threshold = (threshold + 1) ** np.sqrt(1 / 252) - 1
        excess = df_target - daily_threshold
        numerator = excess[excess > 0].sum()
        denominator = -excess[excess < 0].sum() + epsilon

        return numerator / denominator

    threshold = np.linspace(threshold_start, threshold_end, 50)
    results = []
    for i in threshold:
        omega_ = get_omega_ratio(series_target, i)
        results.append(OmegaModel(threshold=i, omega=omega_))

    return OBBject(results=results)


@router.command(methods=["POST"])
def kurtosis(data: List[Data], target: str, window: PositiveInt) -> OBBject[List[Data]]:
    """Get the Kurtosis.

    Kurtosis provides insights into the shape of the data's distribution, particularly the heaviness of its tails.
    Kurtosis is a statistical measure that reveals whether the data points tend to cluster around the mean or if
    outliers are more common than a normal distribution would suggest. A higher kurtosis indicates more data points are
    found in the tails, suggesting potential volatility or risk.
    This analysis is crucial for understanding the underlying characteristics of your data, helping to anticipate
    extreme events or anomalies over a specified window of time.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    window : PositiveInt
        Window size.

    Returns
    -------
    OBBject[List[Data]]
        Kurtosis.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.kurtosis(data=stock_data, target="close", window=252)
    """
    import pandas_ta as ta  # pylint: disable=import-outside-toplevel # type: ignore

    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)
    results = ta.kurtosis(close=series_target, length=window).dropna()
    results = df_to_basemodel(results)

    return OBBject(results=results)


@router.command(methods=["POST"])
def unitroot_test(
    data: List[Data],
    target: str,
    fuller_reg: Literal["c", "ct", "ctt", "nc", "c"] = "c",
    kpss_reg: Literal["c", "ct"] = "c",
) -> OBBject[UnitRootModel]:
    """Get Unit Root Test.

    This function applies two renowned tests to assess whether your data series is stationary or if it contains a unit
    root, indicating it may be influenced by time-based trends or seasonality. The Augmented Dickey-Fuller (ADF) test
    helps identify the presence of a unit root, suggesting that the series could be non-stationary and potentially
    unpredictable over time. On the other hand, the Kwiatkowski-Phillips-Schmidt-Shin (KPSS) test checks for the
    stationarity of the series, where failing to reject the null hypothesis indicates a stable, stationary series.
    Together, these tests provide a comprehensive view of your data's time series properties, essential for
    accurate modeling and forecasting.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    fuller_reg : Literal["c", "ct", "ctt", "nc", "c"]
        Regression type for ADF test.
    kpss_reg : Literal["c", "ct"]
        Regression type for KPSS test.

    Returns
    -------
    OBBject[UnitRootModel]
        Unit root tests summary.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.unitroot_test(data=stock_data, target="close")
    >>> obb.quantitative.unitroot_test(data=stock_data, target="close", fuller_reg="ct", kpss_reg="ct")
    """
    # pylint: disable=import-outside-toplevel
    from statsmodels.tsa import stattools  # type: ignore

    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    adf = stattools.adfuller(series_target, regression=fuller_reg)
    kpss = stattools.kpss(series_target, regression=kpss_reg, nlags="auto")

    unitroot_summary = UnitRootModel(
        adf=ADFTestModel(
            statistic=adf[0],
            p_value=adf[1],
            nlags=adf[2] if isinstance(adf[2], int) else 0,
            nobs=adf[3] if isinstance(adf[3], int) else 0,
            icbest=adf[5] if isinstance(adf[5], float) else 0.0,  # type: ignore
        ),
        kpss=KPSSTestModel(
            statistic=kpss[0],
            p_value=kpss[1],
            nlags=kpss[2],
        ),
    )
    return OBBject(results=unitroot_summary)


@router.command(methods=["POST"])
def sharpe_ratio(
    data: List[Data], target: str, rfr: float = 0.0, window: PositiveInt = 252
) -> OBBject[List[Data]]:
    """Get Sharpe Ratio.

    This function calculates the Sharpe Ratio, a metric used to assess the return of an investment compared to its risk.
    By factoring in the risk-free rate, it helps you understand how much extra return you're getting for the extra
    volatility that you endure by holding a riskier asset. The Sharpe Ratio is essential for investors looking to
    compare the efficiency of different investments, providing a clear picture of potential rewards in relation to their
    risks over a specified period. Ideal for gauging the effectiveness of investment strategies, it offers insights into
    optimizing your portfolio for maximum return on risk.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    rfr : float, optional
        Risk-free rate, by default 0.0
    window : PositiveInt, optional
        Window size, by default 252

    Returns
    -------
    OBBject[List[Data]]
        Sharpe ratio.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.sharpe_ratio(data=stock_data, target="close")
    >>> obb.quantitative.sharpe_ratio(data=stock_data, target="close", rfr=0.01, window=126)
    """
    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    returns = series_target.pct_change().dropna().rolling(window).sum()
    std = series_target.rolling(window).std() / np.sqrt(window)
    results = ((returns - rfr) / std).dropna()

    results = df_to_basemodel(results)

    return OBBject(results=results)


@router.command(methods=["POST"])
def sortino_ratio(
    data: List[Data],
    target: str,
    target_return: float = 0.0,
    window: PositiveInt = 252,
    adjusted: bool = False,
) -> OBBject[List[Data]]:
    """Get Sortino Ratio.

    The Sortino Ratio enhances the evaluation of investment returns by distinguishing harmful volatility
    from total volatility. Unlike other metrics that treat all volatility as risk, this command specifically assesses
    the volatility of negative returns relative to a target or desired return.
    It's particularly useful for investors who are more concerned with downside risk than with overall volatility.
    By calculating the Sortino Ratio, investors can better understand the risk-adjusted return of their investments,
    focusing on the likelihood and impact of negative returns.
    This approach offers a more nuanced tool for portfolio optimization, especially in strategies aiming
    to minimize the downside.

    For method & terminology see:
    http://www.redrockcapital.com/Sortino__A__Sharper__Ratio_Red_Rock_Capital.pdf

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    target_return : float, optional
        Target return, by default 0.0
    window : PositiveInt, optional
        Window size, by default 252
    adjusted : bool, optional
        Adjust sortino ratio to compare it to sharpe ratio, by default False

    Returns
    -------
    OBBject[List[Data]]
        Sortino ratio.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.sortino_ratio(data=stock_data, target="close")
    >>> obb.quantitative.sortino_ratio(data=stock_data, target="close", target_return=0.01, window=126, adjusted=True)
    """
    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    returns = series_target.pct_change().dropna().rolling(window).sum()
    downside_deviation = returns.rolling(window).apply(
        lambda x: (x.values[x.values < 0]).std() / np.sqrt(252) * 100
    )
    results = ((returns - target_return) / downside_deviation).dropna()

    if adjusted:
        results = results / np.sqrt(2)

    results_ = df_to_basemodel(results)

    return OBBject(results=results_)


@router.command(methods=["POST"])
def skewness(data: List[Data], target: str, window: PositiveInt) -> OBBject[List[Data]]:
    """Get Skewness.

    Skewness is a statistical measure that reveals the degree of asymmetry of a distribution around its mean.
    Positive skewness indicates a distribution with an extended tail to the right, while negative skewness shows a tail
    that stretches left. Understanding skewness can provide insights into potential biases in data and help anticipate
    the nature of future data points. It's particularly useful for identifying the likelihood of extreme outcomes in
    financial returns, enabling more informed decision-making based on the distribution's shape over a specified period.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    window : PositiveInt
        Window size.

    Returns
    -------
    OBBject[List[Data]]
        Skewness.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.skewness(data=stock_data, target="close", window=252)
    """
    import pandas_ta as ta  # pylint: disable=import-outside-toplevel # type: ignore

    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)
    results = ta.skew(close=series_target, length=window).dropna()
    results = df_to_basemodel(results)

    return OBBject(results=results)


@router.command(methods=["POST"])
def quantile(
    data: List[Data],
    target: str,
    window: PositiveInt,
    quantile_pct: NonNegativeFloat = 0.5,
) -> OBBject[List[Data]]:
    """Get Quantile.

    Quantile is a statistical measure that identifies the value below which a given percentage of observations in a
    group of data falls. By setting the quantile percentage, you can determine any point in the distribution, not just
    the median. Whether you're interested in the median, quartiles, or any other position within your data's range,
    this tool offers a precise way to understand the distribution's characteristics.
    It's especially useful for identifying outliers, understanding dispersion, and setting thresholds for
    decision-making based on the distribution of data over a specified period.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.
    window : PositiveInt
        Window size.
    quantile_pct : NonNegativeFloat, optional

    Returns
    -------
    OBBject[List[Data]]
        Quantile.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.quantile(data=stock_data, target="close", window=252, quantile_pct=0.25)
    >>> obb.quantitative.quantile(data=stock_data, target="close", window=252, quantile_pct=0.75)
    """
    import pandas_ta as ta  # pylint: disable=import-outside-toplevel # type: ignore

    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    df_median = ta.median(close=series_target, length=window).to_frame()
    df_quantile = ta.quantile(series_target, length=window, q=quantile_pct).to_frame()
    results = pd.concat([df_median, df_quantile], axis=1).dropna()

    results_ = df_to_basemodel(results)

    return OBBject(results=results_)


@router.command(methods=["POST"])
def summary(data: List[Data], target: str) -> OBBject[SummaryModel]:
    """Get Summary Statistics.

    The summary that offers a snapshot of its central tendencies, variability, and distribution.
    This command calculates essential statistics, including mean, standard deviation, variance,
    and specific percentiles, to provide a detailed profile of your target column. B
    y examining these metrics, you gain insights into the data's overall behavior, helping to identify patterns,
    outliers, or anomalies. The summary table is an invaluable tool for initial data exploration,
    ensuring you have a solid foundation for further analysis or reporting.

    Parameters
    ----------
    data : List[Data]
        Time series data.
    target : str
        Target column name.

    Returns
    -------
    OBBject[SummaryModel]
        Summary table.

    Examples
    --------
    >>> from openbb import obb
    >>> stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp")
    >>> obb.quantitative.summary(data=stock_data, target="close")
    """
    df = basemodel_to_df(data)
    series_target = get_target_column(df, target)

    df_stats = series_target.describe(percentiles=[0.1, 0.25, 0.5, 0.75, 0.9])
    df_stats.loc["var"] = df_stats.loc["std"] ** 2
    results = SummaryModel(
        count=df_stats.loc["count"],
        mean=df_stats.loc["mean"],
        std=df_stats.loc["std"],
        var=df_stats.loc["var"],
        min=df_stats.loc["min"],
        p_25=df_stats.loc["25%"],
        p_50=df_stats.loc["50%"],
        p_75=df_stats.loc["75%"],
        max=df_stats.loc["max"],
    )

    return OBBject(results=results)