diff --git a/openbb_platform/extensions/quantitative/integration/test_quantitative_api.py b/openbb_platform/extensions/quantitative/integration/test_quantitative_api.py index 6bcaa5da5e06..10e9432d1f8e 100644 --- a/openbb_platform/extensions/quantitative/integration/test_quantitative_api.py +++ b/openbb_platform/extensions/quantitative/integration/test_quantitative_api.py @@ -127,12 +127,12 @@ def test_quantitative_capm(params, data_type): ], ) @pytest.mark.integration -def test_quantitative_omega_ratio(params, data_type): +def test_quantitative_performance_omega_ratio(params, data_type): params = {p: v for p, v in params.items() if v} data = json.dumps(get_data(data_type)) query_str = get_querystring(params, []) - url = f"http://0.0.0.0:8000/api/v1/quantitative/omega_ratio?{query_str}" + url = f"http://0.0.0.0:8000/api/v1/quantitative/performance/omega_ratio?{query_str}" result = requests.post(url, headers=get_headers(), timeout=10, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 @@ -146,12 +146,12 @@ def test_quantitative_omega_ratio(params, data_type): ], ) @pytest.mark.integration -def test_quantitative_kurtosis(params, data_type): +def test_quantitative_rolling_kurtosis(params, data_type): params = {p: v for p, v in params.items() if v} data = json.dumps(get_data(data_type)) query_str = get_querystring(params, []) - url = f"http://0.0.0.0:8000/api/v1/quantitative/kurtosis?{query_str}" + url = f"http://0.0.0.0:8000/api/v1/quantitative/rolling/kurtosis?{query_str}" result = requests.post(url, headers=get_headers(), timeout=10, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 @@ -212,12 +212,14 @@ def test_quantitative_unitroot_test(params, data_type): ], ) @pytest.mark.integration -def test_quantitative_sharpe_ratio(params, data_type): +def test_quantitative_performance_sharpe_ratio(params, data_type): params = {p: v for p, v in params.items() if v} data = json.dumps(get_data(data_type)) query_str = get_querystring(params, []) - url = f"http://0.0.0.0:8000/api/v1/quantitative/sharpe_ratio?{query_str}" + url = ( + f"http://0.0.0.0:8000/api/v1/quantitative/performance/sharpe_ratio?{query_str}" + ) result = requests.post(url, headers=get_headers(), timeout=10, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 @@ -251,12 +253,14 @@ def test_quantitative_sharpe_ratio(params, data_type): ], ) @pytest.mark.integration -def test_quantitative_sortino_ratio(params, data_type): +def test_quantitative_performance_sortino_ratio(params, data_type): params = {p: v for p, v in params.items() if v} data = json.dumps(get_data(data_type)) query_str = get_querystring(params, []) - url = f"http://0.0.0.0:8000/api/v1/quantitative/sortino_ratio?{query_str}" + url = ( + f"http://0.0.0.0:8000/api/v1/quantitative/performance/sortino_ratio?{query_str}" + ) result = requests.post(url, headers=get_headers(), timeout=10, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 @@ -269,12 +273,66 @@ def test_quantitative_sortino_ratio(params, data_type): ], ) @pytest.mark.integration -def test_quantitative_skewness(params, data_type): +def test_quantitative_rolling_skew(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/rolling/skew?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "window": "220", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_rolling_variance(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/rolling/variance?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "window": "220", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_rolling_stdev(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/rolling/stdev?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "window": "220", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_rolling_mean(params, data_type): params = {p: v for p, v in params.items() if v} data = json.dumps(get_data(data_type)) query_str = get_querystring(params, []) - url = f"http://0.0.0.0:8000/api/v1/quantitative/skewness?{query_str}" + url = f"http://0.0.0.0:8000/api/v1/quantitative/rolling/mean?{query_str}" result = requests.post(url, headers=get_headers(), timeout=60, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 @@ -306,12 +364,12 @@ def test_quantitative_skewness(params, data_type): ], ) @pytest.mark.integration -def test_quantitative_quantile(params, data_type): +def test_quantitative_rolling_quantile(params, data_type): params = {p: v for p, v in params.items() if v} data = json.dumps(get_data(data_type)) query_str = get_querystring(params, []) - url = f"http://0.0.0.0:8000/api/v1/quantitative/quantile?{query_str}" + url = f"http://0.0.0.0:8000/api/v1/quantitative/rolling/quantile?{query_str}" result = requests.post(url, headers=get_headers(), timeout=10, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 @@ -334,3 +392,133 @@ def test_quantitative_summary(params, data_type): result = requests.post(url, headers=get_headers(), timeout=10, data=data) assert isinstance(result, requests.Response) assert result.status_code == 200 + + +############ +# quantitative/stats +############ + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_skew(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/stats/skew?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_kurtosis(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/stats/kurtosis?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_mean(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/stats/mean?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_stdev(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/stats/stdev?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close", "index": "date"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_variance(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/stats/variance?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=60, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 + + +@parametrize( + "params, data_type", + [ + ( + { + "data": "", + "target": "close", + "quantile_pct": "", + "index": "date", + }, + "equity", + ), + ( + { + "data": "", + "target": "high", + "quantile_pct": "0.6", + "index": "date", + }, + "crypto", + ), + ], +) +@pytest.mark.integration +def test_quantitative_stats_quantile(params, data_type): + params = {p: v for p, v in params.items() if v} + data = json.dumps(get_data(data_type)) + + query_str = get_querystring(params, []) + url = f"http://0.0.0.0:8000/api/v1/quantitative/stats/quantile?{query_str}" + result = requests.post(url, headers=get_headers(), timeout=10, data=data) + assert isinstance(result, requests.Response) + assert result.status_code == 200 diff --git a/openbb_platform/extensions/quantitative/integration/test_quantitative_python.py b/openbb_platform/extensions/quantitative/integration/test_quantitative_python.py index f41f26613b17..5c55621a8ae0 100644 --- a/openbb_platform/extensions/quantitative/integration/test_quantitative_python.py +++ b/openbb_platform/extensions/quantitative/integration/test_quantitative_python.py @@ -117,11 +117,11 @@ def test_quantitative_capm(params, data_type, obb): ], ) @pytest.mark.integration -def test_quantitative_omega_ratio(params, data_type, obb): +def test_quantitative_performance_omega_ratio(params, data_type, obb): params = {p: v for p, v in params.items() if v} params["data"] = get_data(data_type) - result = obb.quantitative.omega_ratio(**params) + result = obb.quantitative.performance.omega_ratio(**params) assert result assert isinstance(result, OBBject) @@ -134,11 +134,11 @@ def test_quantitative_omega_ratio(params, data_type, obb): ], ) @pytest.mark.integration -def test_quantitative_kurtosis(params, data_type, obb): +def test_quantitative_rolling_kurtosis(params, data_type, obb): params = {p: v for p, v in params.items() if v} params["data"] = get_data(data_type) - result = obb.quantitative.kurtosis(**params) + result = obb.quantitative.rolling.kurtosis(**params) assert result assert isinstance(result, OBBject) assert len(result.results) > 0 @@ -203,11 +203,11 @@ def test_quantitative_unitroot_test(params, data_type, obb): ], ) @pytest.mark.integration -def test_quantitative_sharpe_ratio(params, data_type, obb): +def test_quantitative_performance_sharpe_ratio(params, data_type, obb): params = {p: v for p, v in params.items() if v} params["data"] = get_data(data_type) - result = obb.quantitative.sharpe_ratio(**params) + result = obb.quantitative.performance.sharpe_ratio(**params) assert result assert isinstance(result, OBBject) @@ -240,11 +240,11 @@ def test_quantitative_sharpe_ratio(params, data_type, obb): ], ) @pytest.mark.integration -def test_quantitative_sortino_ratio(params, data_type, obb): +def test_quantitative_performance_sortino_ratio(params, data_type, obb): params = {p: v for p, v in params.items() if v} params["data"] = get_data(data_type) - result = obb.quantitative.sortino_ratio(**params) + result = obb.quantitative.performance.sortino_ratio(**params) assert result assert isinstance(result, OBBject) @@ -256,11 +256,11 @@ def test_quantitative_sortino_ratio(params, data_type, obb): ], ) @pytest.mark.integration -def test_quantitative_skewness(params, data_type, obb): +def test_quantitative_rolling_skew(params, data_type, obb): params = {p: v for p, v in params.items() if v} params["data"] = get_data(data_type) - result = obb.quantitative.skewness(**params) + result = obb.quantitative.rolling.skew(**params) assert result assert isinstance(result, OBBject) assert len(result.results) > 0 @@ -292,11 +292,11 @@ def test_quantitative_skewness(params, data_type, obb): ], ) @pytest.mark.integration -def test_quantitative_quantile(params, data_type, obb): +def test_quantitative_rolling_quantile(params, data_type, obb): params = {p: v for p, v in params.items() if v} params["data"] = get_data(data_type) - result = obb.quantitative.quantile(**params) + result = obb.quantitative.rolling.quantile(**params) assert result assert isinstance(result, OBBject) assert len(result.results) > 0 @@ -317,3 +317,228 @@ def test_quantitative_summary(params, data_type, obb): result = obb.quantitative.summary(**params) assert result assert isinstance(result, OBBject) + + +@parametrize( + "params, data_type", + [ + ( + { + "data": "", + "target": "close", + "window": "10", + "quantile_pct": "", + "index": "date", + }, + "equity", + ), + ( + { + "data": "", + "target": "high", + "window": "50", + "quantile_pct": "0.6", + "index": "date", + }, + "crypto", + ), + ], +) +@pytest.mark.integration +def test_quantitative_rolling_stdev(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.rolling.stdev(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ( + { + "data": "", + "target": "close", + "window": "10", + "quantile_pct": "", + "index": "date", + }, + "equity", + ), + ( + { + "data": "", + "target": "high", + "window": "50", + "quantile_pct": "0.6", + "index": "date", + }, + "crypto", + ), + ], +) +@pytest.mark.integration +def test_quantitative_rolling_mean(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.rolling.mean(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ( + { + "data": "", + "target": "close", + "window": "10", + "quantile_pct": "", + "index": "date", + }, + "equity", + ), + ( + { + "data": "", + "target": "high", + "window": "50", + "quantile_pct": "0.6", + "index": "date", + }, + "crypto", + ), + ], +) +@pytest.mark.integration +def test_quantitative_rolling_variance(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.rolling.variance(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_skew(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.stats.skew(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_kurtosis(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.stats.kurtosis(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_variance(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.stats.variance(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_stdev(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.stats.stdev(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ({"data": "", "target": "close"}, "equity"), + ], +) +@pytest.mark.integration +def test_quantitative_stats_mean(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.stats.mean(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 + + +@parametrize( + "params, data_type", + [ + ( + { + "data": "", + "target": "close", + "quantile_pct": "", + }, + "equity", + ), + ( + { + "data": "", + "target": "close", + "quantile_pct": "0.6", + }, + "crypto", + ), + ], +) +@pytest.mark.integration +def test_quantitative_stats_quantile(params, data_type, obb): + params = {p: v for p, v in params.items() if v} + params["data"] = get_data(data_type) + + result = obb.quantitative.stats.quantile(**params) + assert result + assert isinstance(result, OBBject) + assert len(result.results) > 0 diff --git a/openbb_platform/extensions/quantitative/openbb_quantitative/performance/performance_router.py b/openbb_platform/extensions/quantitative/openbb_quantitative/performance/performance_router.py new file mode 100644 index 000000000000..92bd611a3802 --- /dev/null +++ b/openbb_platform/extensions/quantitative/openbb_quantitative/performance/performance_router.py @@ -0,0 +1,203 @@ +from typing import List + +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, +) +from openbb_core.provider.abstract.data import Data +from openbb_quantitative.helpers import validate_window +from openbb_quantitative.models import ( + OmegaModel, +) +from pydantic import PositiveInt + +router = Router(prefix="/performance") + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.omega_ratio(data=returns, target="close")', + ], +) +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. + """ + 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"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.sharpe_ratio(data=returns, target="close")', + ], +) +def sharpe_ratio( + data: List[Data], + target: str, + rfr: float = 0.0, + window: PositiveInt = 252, + index: str = "date", +) -> OBBject[List[Data]]: + """Get Rolling 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 + index : str, optional + + Returns + ------- + OBBject[List[Data]] + Sharpe ratio. + """ + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + validate_window(series_target, window) + series_target.name = f"sharpe_{window}" + returns = series_target.pct_change().dropna().rolling(window).sum() + std = series_target.rolling(window).std() / np.sqrt(window) + results = ((returns - rfr) / std).dropna().reset_index(drop=False) + + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + '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)', + ], +) +def sortino_ratio( + data: List[Data], + target: str, + target_return: float = 0.0, + window: PositiveInt = 252, + adjusted: bool = False, + index: str = "date", +) -> OBBject[List[Data]]: + """Get rolling 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 + index:str + Index column for input data + Returns + ------- + OBBject[List[Data]] + Sortino ratio. + """ + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + validate_window(series_target, window) + 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() + .reset_index(drop=False) + ) + + if adjusted: + results = results / np.sqrt(2) + + results_ = df_to_basemodel(results) + + return OBBject(results=results_) diff --git a/openbb_platform/extensions/quantitative/openbb_quantitative/quantitative_router.py b/openbb_platform/extensions/quantitative/openbb_quantitative/quantitative_router.py index a381da937cb3..bede1e9ac84e 100644 --- a/openbb_platform/extensions/quantitative/openbb_quantitative/quantitative_router.py +++ b/openbb_platform/extensions/quantitative/openbb_quantitative/quantitative_router.py @@ -2,32 +2,37 @@ 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, validate_window +from openbb_quantitative.performance.performance_router import ( + router as performance_router, +) +from openbb_quantitative.rolling.rolling_router import router as rolling_router +from openbb_quantitative.stats.stats_router import router as stats_router + +from .helpers import get_fama_raw from .models import ( ADFTestModel, CAPMModel, KPSSTestModel, NormalityModel, - OmegaModel, SummaryModel, TestModel, UnitRootModel, ) router = Router(prefix="") +router.include_router(rolling_router) +router.include_router(stats_router) +router.include_router(performance_router) @router.command( @@ -137,120 +142,7 @@ def capm(data: List[Data], target: str) -> OBBject[CAPMModel]: return OBBject(results=results) -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "obb.quantitative.omega_ratio(data=stock_data, target='close')", - ], -) -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. - """ - 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"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "obb.quantitative.kurtosis(data=stock_data, target='close', window=252)", - ], -) -def kurtosis( - data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" -) -> OBBject[List[Data]]: - """Get the rolling 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. - index : str, optional - Index column name, by default "date" - - Returns - ------- - OBBject[List[Data]] - Kurtosis. - """ - import pandas_ta as ta # pylint: disable=import-outside-toplevel # type: ignore - - df = basemodel_to_df(data, index=index) - series_target = get_target_column(df, target) - validate_window(series_target, window) - results = ( - ta.kurtosis(close=series_target, length=window).dropna().reset_index(drop=False) - ) - results = df_to_basemodel(results) - - return OBBject(results=results) - - -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "obb.quantitative.unitroot_test(data=stock_data, target='close')", - 'obb.quantitative.unitroot_test(data=stock_data, target="close", fuller_reg="ct", kpss_reg="ct")', - ], -) +@router.command(methods=["POST"]) def unitroot_test( data: List[Data], target: str, @@ -309,238 +201,7 @@ def unitroot_test( return OBBject(results=unitroot_summary) -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "obb.quantitative.sharpe_ratio(data=stock_data, target='close')", - "obb.quantitative.sharpe_ratio(data=stock_data, target='close', rfr=0.01, window=126)", - ], -) -def sharpe_ratio( - data: List[Data], - target: str, - rfr: float = 0.0, - window: PositiveInt = 252, - index: str = "date", -) -> OBBject[List[Data]]: - """Get Rolling 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 - index : str, optional - - Returns - ------- - OBBject[List[Data]] - Sharpe ratio. - """ - df = basemodel_to_df(data, index=index) - series_target = get_target_column(df, target) - validate_window(series_target, window) - series_target.name = f"sharpe_{window}" - returns = series_target.pct_change().dropna().rolling(window).sum() - std = series_target.rolling(window).std() / np.sqrt(window) - results = ((returns - rfr) / std).dropna().reset_index(drop=False) - - results = df_to_basemodel(results) - - return OBBject(results=results) - - -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "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)", - ], -) -def sortino_ratio( - data: List[Data], - target: str, - target_return: float = 0.0, - window: PositiveInt = 252, - adjusted: bool = False, - index: str = "date", -) -> OBBject[List[Data]]: - """Get rolling 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 - index:str - Index column for input data - Returns - ------- - OBBject[List[Data]] - Sortino ratio. - """ - df = basemodel_to_df(data, index=index) - series_target = get_target_column(df, target) - validate_window(series_target, window) - 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() - .reset_index(drop=False) - ) - - if adjusted: - results = results / np.sqrt(2) - - results_ = df_to_basemodel(results) - - return OBBject(results=results_) - - -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "obb.quantitative.skewness(data=stock_data, target='close', window=252)", - ], -) -def skewness( - data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" -) -> OBBject[List[Data]]: - """Get Rolling 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. - index : str, optional - Index column name, by default "date" - Returns - ------- - OBBject[List[Data]] - Skewness. - """ - import pandas_ta as ta # pylint: disable=import-outside-toplevel # type: ignore - - df = basemodel_to_df(data, index=index) - series_target = get_target_column(df, target) - validate_window(series_target, window) - results = ( - ta.skew(close=series_target, length=window).dropna().reset_index(drop=False) - ) - results = df_to_basemodel(results) - - return OBBject(results=results) - - -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - '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)", - ], -) -def quantile( - data: List[Data], - target: str, - window: PositiveInt = 21, - quantile_pct: NonNegativeFloat = 0.5, - index: str = "date", -) -> OBBject[List[Data]]: - """Get Rolling 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 - Quantile to get - Returns - ------- - OBBject[List[Data]] - Quantile. - """ - import pandas_ta as ta # pylint: disable=import-outside-toplevel # type: ignore - - df = basemodel_to_df(data, index=index) - series_target = get_target_column(df, target) - validate_window(series_target, window) - 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().reset_index(drop=False) - ) - - results_ = df_to_basemodel(results) - - return OBBject(results=results_) - - -@router.command( - methods=["POST"], - examples=[ - "stock_data = obb.equity.price.historical(symbol='TSLA', start_date='2023-01-01', provider='fmp').to_df()", - "obb.quantitative.volatility(data=stock_data, target='close')", - ], -) +@router.command(methods=["POST"]) def summary(data: List[Data], target: str) -> OBBject[SummaryModel]: """Get Summary Statistics. diff --git a/openbb_platform/extensions/quantitative/openbb_quantitative/rolling/rolling_router.py b/openbb_platform/extensions/quantitative/openbb_quantitative/rolling/rolling_router.py new file mode 100644 index 000000000000..6e464083aea4 --- /dev/null +++ b/openbb_platform/extensions/quantitative/openbb_quantitative/rolling/rolling_router.py @@ -0,0 +1,319 @@ +"""Rolling submenu of quantitative models for rolling statistics.""" + +from typing import List + +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, +) +from openbb_core.provider.abstract.data import Data +from openbb_quantitative.helpers import validate_window +from openbb_quantitative.statistics import ( + kurtosis_, + mean_, + skew_, + std_dev_, + var_, +) +from pydantic import NonNegativeFloat, PositiveInt + +router = Router(prefix="/rolling") + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.rolling.skew(data=returns, target="close")', + ], +) +def skew( + data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" +) -> OBBject[List[Data]]: + """Get Rolling Skew. + + Skew 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. + index : str, optional + Index column name, by default "date" + + Returns + ------- + OBBject[List[Data]] + Rolling skew. + + """ + + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + series_target.name = f"rolling_skew_{window}" + validate_window(series_target, window) + results = ( + series_target.rolling(window).apply(skew_).dropna().reset_index(drop=False) + ) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.rolling.variance(data=returns, target="close", window=252)', + ], +) +def variance( + data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" +) -> OBBject[List[Data]]: + """ + Calculate the rolling variance of a target column within a given window size. + + Variance measures the dispersion of a set of data points around their mean. It is a key metric for + assessing the volatility and stability of financial returns or other time series data over a specified rolling window. + + Parameters: + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate variance. + window: PositiveInt + The number of observations used for calculating the rolling measure. + index: str, optional + The name of the index column, default is "date". + + Returns: + ------- + OBBject[List[Data]] + An object containing the rolling variance values. + """ + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + series_target.name = f"rolling_var_{window}" + validate_window(series_target, window) + results = series_target.rolling(window).apply(var_).dropna().reset_index(drop=False) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.rolling.stdev(data=returns, target="close", window=252)', + ], +) +def stdev( + data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" +) -> OBBject[List[Data]]: + """ + Calculate the rolling standard deviation of a target column within a given window size. + + Standard deviation is a measure of the amount of variation or dispersion of a set of values. + It is widely used to assess the risk and volatility of financial returns or other time series data + over a specified rolling window. It is the square root of the variance. + + Parameters: + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate standard deviation. + window: PositiveInt + The number of observations used for calculating the rolling measure. + index: str, optional + The name of the index column, default is "date". + + Returns: + ------- + OBBject[List[Data]] + An object containing the rolling standard deviation values. + """ + + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + series_target.name = f"rolling_stdev_{window}" + validate_window(series_target, window) + results = ( + series_target.rolling(window).apply(std_dev_).dropna().reset_index(drop=False) + ) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.rolling.kurtosis(data=returns, target="close", window=252)', + ], +) +def kurtosis( + data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" +) -> OBBject[List[Data]]: + """ + Calculate the rolling kurtosis of a target column within a given window size. + + Kurtosis measures the "tailedness" of the probability distribution of a real-valued random variable. + High kurtosis indicates a distribution with heavy tails (outliers), suggesting a higher risk of extreme outcomes. + Low kurtosis indicates a distribution with lighter tails (less outliers), suggesting less risk of extreme outcomes. + This function helps in assessing the risk of outliers in financial returns or other time series data over a specified + rolling window. + + Parameters: + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate kurtosis. + window: PositiveInt + The number of observations used for calculating the rolling measure. + index: str, optional + The name of the index column, default is "date". + + Returns: + ------- + OBBject[List[Data]] + An object containing the rolling kurtosis values. + """ + + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + series_target.name = f"rolling_kurtosis_{window}" + validate_window(series_target, window) + results = ( + series_target.rolling(window).apply(kurtosis_).dropna().reset_index(drop=False) + ) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.rolling.quantile(data=returns, target="close", window=252, quantile_pct=0.25)', + 'obb.quantitative.rolling.quantile(data=returns, target="close", window=252, quantile_pct=0.75)', + ], +) +def quantile( + data: List[Data], + target: str, + window: PositiveInt = 21, + quantile_pct: NonNegativeFloat = 0.5, + index: str = "date", +) -> OBBject[List[Data]]: + """ + Calculate the rolling quantile of a target column within a given window size at a specified quantile percentage. + + Quantiles are points dividing the range of a probability distribution into intervals with equal probabilities, + or dividing the sample in the same way. This function is useful for understanding the distribution of data + within a specified window, allowing for analysis of trends, identification of outliers, and assessment of risk.. + + Parameters: + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate the quantile. + window: PositiveInt + The number of observations used for calculating the rolling measure. + quantile_pct: NonNegativeFloat, optional + The quantile percentage to calculate (e.g., 0.5 for median), default is 0.5. + index: str, optional + The name of the index column, default is "date". + + Returns: + ------- + OBBject[List[Data]] + An object containing the rolling quantile values with the median. + """ + + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + validate_window(series_target, window) + roll = series_target.rolling(window) + df_median = roll.median() + df_quantile = roll.quantile(quantile_pct) + results = ( + pd.concat( + [df_median, df_quantile], + axis=1, + keys=[ + f"rolling_median_{window}", + f"rolling_quantile_{quantile_pct}_{window}", + ], + ) + .dropna() + .reset_index(drop=False) + ) + + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.rolling.mean(data=returns, target="close", window=252)', + ], +) +def mean( + data: List[Data], target: str, window: PositiveInt = 21, index: str = "date" +) -> OBBject[List[Data]]: + """ + Calculate the rolling mean (average) of a target column within a given window size. + + The rolling mean is a simple moving average that calculates the average of a target variable over a specified window. + This function is widely used in financial analysis to smooth short-term fluctuations and highlight longer-term trends + or cycles in time series data. + + Parameters: + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate the mean. + window: PositiveInt + The number of observations used for calculating the rolling measure. + index: str, optional + The name of the index column, default is "date". + + Returns: + OBBject[List[Data]] + An object containing the rolling mean values. + """ + + df = basemodel_to_df(data, index=index) + series_target = get_target_column(df, target) + series_target.name = f"rolling_mean_{window}" + validate_window(series_target, window) + results = ( + series_target.rolling(window).apply(mean_).dropna().reset_index(drop=False) + ) + results = df_to_basemodel(results) + + return OBBject(results=results) diff --git a/openbb_platform/extensions/quantitative/openbb_quantitative/statistics.py b/openbb_platform/extensions/quantitative/openbb_quantitative/statistics.py new file mode 100644 index 000000000000..db137ac77cfc --- /dev/null +++ b/openbb_platform/extensions/quantitative/openbb_quantitative/statistics.py @@ -0,0 +1,41 @@ +"""Statistics Functions""" + +from typing import Union + +from numpy import ( + mean as mean_np, + ndarray, + std, + var as var_np, +) +from pandas import DataFrame, Series +from scipy import stats + +# Because python is weird and these being the same name as the fastapi router functions +# which overwrites the function signature, we add the _ after the function name + + +def kurtosis_(data: Union[DataFrame, Series, ndarray]) -> float: + """Kurtosis is a measure of the "tailedness" of the probability distribution of a real-valued random variable.""" + return stats.kurtosis(data) + + +def skew_(data: Union[DataFrame, Series, ndarray]) -> float: + """Skewness is a measure of the asymmetry of the probability distribution of a + real-valued random variable about its mean.""" + return stats.skew(data) + + +def mean_(data: Union[DataFrame, Series, ndarray]) -> float: + """Mean is the average of the numbers.""" + return mean_np(data) + + +def std_dev_(data: Union[DataFrame, Series, ndarray]) -> float: + """Standard deviation is a measure of the amount of variation or dispersion of a set of values.""" + return std(data) + + +def var_(data: Union[DataFrame, Series, ndarray]) -> float: + """Variance is a measure of the amount of variation or dispersion of a set of values.""" + return var_np(data) diff --git a/openbb_platform/extensions/quantitative/openbb_quantitative/stats/stats_router.py b/openbb_platform/extensions/quantitative/openbb_quantitative/stats/stats_router.py new file mode 100644 index 000000000000..deff80471b1d --- /dev/null +++ b/openbb_platform/extensions/quantitative/openbb_quantitative/stats/stats_router.py @@ -0,0 +1,258 @@ +"""Rolling submenu of quantitative models for rolling statistics.""" + +from typing import List + +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, +) +from openbb_core.provider.abstract.data import Data +from openbb_quantitative.statistics import ( + kurtosis_, + mean_, + skew_, + std_dev_, + var_, +) +from pydantic import NonNegativeFloat + +router = Router(prefix="/stats") + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.stats.skew(data=returns, target="close")', + ], +) +def skew( + data: List[Data], + target: str, +) -> OBBject[List[Data]]: + """Get the skew. of the data set + + Skew 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. + + Returns + ------- + OBBject[List[Data]] + Rolling skew. + """ + + df = basemodel_to_df(data) + series_target = get_target_column(df, target) + results = pd.DataFrame([skew_(series_target)], columns=["skew"]) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.stats.variance(data=returns, target="close")', + ], +) +def variance(data: List[Data], target: str) -> OBBject[List[Data]]: + """ + Calculate the variance of a target column. + + Variance measures the dispersion of a set of data points around their mean. It is a key metric for + assessing the volatility and stability of financial returns or other time series data. + + Parameters + ---------- + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate variance. + + Returns + ------- + OBBject[List[Data]] + An object containing the rolling variance values. + """ + df = basemodel_to_df(data) + series_target = get_target_column(df, target) + results = pd.DataFrame([var_(series_target)], columns=["variance"]) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.stats.stdev(data=returns, target="close")', + ], +) +def stdev(data: List[Data], target: str) -> OBBject[List[Data]]: + """ + Calculate the rolling standard deviation of a target column. + + Standard deviation is a measure of the amount of variation or dispersion of a set of values. + It is widely used to assess the risk and volatility of financial returns or other time series data + It is the square root of the variance. + + Parameters + ---------- + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate standard deviation. + + Returns + ------- + OBBject[List[Data]] + An object containing the rolling standard deviation values. + """ + + df = basemodel_to_df(data) + series_target = get_target_column(df, target) + results = pd.DataFrame([std_dev_(series_target)], columns=["stdev"]) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.stats.kurtosis(data=returns, target="close")', + ], +) +def kurtosis(data: List[Data], target) -> OBBject[List[Data]]: + """ + Calculate the rolling kurtosis of a target column. + + Kurtosis measures the "tailedness" of the probability distribution of a real-valued random variable. + High kurtosis indicates a distribution with heavy tails (outliers), suggesting a higher risk of extreme outcomes. + Low kurtosis indicates a distribution with lighter tails (less outliers), suggesting less risk of extreme outcomes. + This function helps in assessing the risk of outliers in financial returns or other time series data. + + Parameters + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate kurtosis. + + Returns + ------ + OBBject[List[Data]] + An object containing the kurtosis value + """ + + df = basemodel_to_df(data) + series_target = get_target_column(df, target) + results = pd.DataFrame([kurtosis_(series_target)], columns=["kurtosis"]) + results = df_to_basemodel(results) + + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.stats.quantile(data=returns, target="close", quantile_pct=0.75)', + ], +) +def quantile( + data: List[Data], + target: str, + quantile_pct: NonNegativeFloat = 0.5, +) -> OBBject[List[Data]]: + """ + Calculate the quantile of a target column at a specified quantile percentage. + + Quantiles are points dividing the range of a probability distribution into intervals with equal probabilities, + or dividing the sample in the same way. + + Parameters + ---------- + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate the quantile. + quantile_pct: NonNegativeFloat, optional + The quantile percentage to calculate (e.g., 0.5 for median), default is 0.5. + + Returns + ------- + OBBject[List[Data]] + An object containing the rolling quantile values with the median. + """ + + df = basemodel_to_df( + data, + ) + series_target = get_target_column(df, target) + results = pd.DataFrame( + [series_target.quantile(quantile_pct)], columns=[f"{quantile_pct}_quantile"] + ) + results = df_to_basemodel(results) + return OBBject(results=results) + + +@router.command( + methods=["POST"], + examples=[ + 'stock_data = obb.equity.price.historical(symbol="TSLA", start_date="2023-01-01", provider="fmp").to_df()', + 'returns = stock_data["close"].pct_change().dropna()', + 'obb.quantitative.stats.mean(data=returns, target="close")', + ], +) +def mean( + data: List[Data], + target: str, +) -> OBBject[List[Data]]: + """ + Calculate the mean (average) of a target column. + + The rolling mean is a simple moving average that calculates the average of a target variable. + This function is widely used in financial analysis to smooth short-term fluctuations and highlight longer-term trends + or cycles in time series data. + + Parameters + ---------- + data: List[Data] + The time series data as a list of data points. + target: str + The name of the column for which to calculate the mean. + + Returns + ------- + OBBject[List[Data]] + An object containing the mean value. + """ + + df = basemodel_to_df(data) + series_target = get_target_column(df, target) + results = pd.DataFrame([mean_(series_target)], columns=["mean"]) + results = df_to_basemodel(results) + + return OBBject(results=results) diff --git a/openbb_platform/extensions/quantitative/tests/test_statistics.py b/openbb_platform/extensions/quantitative/tests/test_statistics.py new file mode 100644 index 000000000000..11c1ce2ec439 --- /dev/null +++ b/openbb_platform/extensions/quantitative/tests/test_statistics.py @@ -0,0 +1,25 @@ +import pandas as pd +import pytest +from openbb_quantitative.statistics import kurtosis_, mean_, skew_, std_dev_, var_ + +test_data = pd.Series([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]) + + +def test_kurtosis(): + assert kurtosis_(test_data) == pytest.approx(-1.224, abs=1e-3) + + +def test_skew(): + assert skew_(test_data) == pytest.approx(0.0, abs=1e-3) + + +def test_std_dev(): + assert std_dev_(test_data) == pytest.approx(2.872, abs=1e-3) + + +def test_mean(): + assert mean_(test_data) == pytest.approx(5.5, abs=1e-3) + + +def test_var(): + assert var_(test_data) == pytest.approx(8.25, abs=1e-3) diff --git a/openbb_platform/extensions/tests/test_integration_tests_python.py b/openbb_platform/extensions/tests/test_integration_tests_python.py index 110034fa82f4..8940db9821e6 100644 --- a/openbb_platform/extensions/tests/test_integration_tests_python.py +++ b/openbb_platform/extensions/tests/test_integration_tests_python.py @@ -59,13 +59,13 @@ def test_charting_extension_function_coverage() -> None: assert missing_items == [], "\n".join(missing_items) -def test_missing_api_integration_tests() -> None: +def test_missing_python_integration_tests() -> None: """Check if there are missing tests.""" missing = check_missing_integration_tests(test_type="python") assert not missing, "\n".join(missing) -def test_outdated_api_integration_tests() -> None: +def test_outdated_python_integration_tests() -> None: """Check if there are outdated tests.""" outdated = check_outdated_integration_tests(test_type="python") assert not outdated, "\n".join(outdated) diff --git a/openbb_platform/extensions/tests/utils/router_testers.py b/openbb_platform/extensions/tests/utils/router_testers.py index ed8127fe8d36..bbc9ff3b7ec5 100644 --- a/openbb_platform/extensions/tests/utils/router_testers.py +++ b/openbb_platform/extensions/tests/utils/router_testers.py @@ -132,7 +132,7 @@ def check_router_model_functions_signature() -> List[str]: ) if expected_return_type not in str(function.__annotations__["return"]): missing_return_type.append( - f"{function.__name__} in {router_name}" + f"{function.__name__} in {router_name} " f"doesn't have the expected return type: {expected_return_type}" )