lywsvip/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-07-01 01:14:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update the quantitative extension to make more sense (#6087)

Browse Source 
* Split out a rolling submenu for the rolling functions

* Make a performance and a stats submenu.

* Test the statistics functions

* lint

* lint

* dupe test

* pylint

* ruff

* Try tests quick

* black magic signature funcs

* fix my custom tests

* Fix the existing imports/urls

* push the api update

* okay I figured out whats going on

* this should be all  of them

* Correct docstringing examples

---------

Co-authored-by: Igor Radovanovic <74266147+IgorWounds@users.noreply.github.com>
Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>
This commit is contained in:
James Maslek
2024-02-21 12:09:57 -05:00
committed by GitHub
parent 754e14ca71
commit 3cc6025ab4
10 changed files with 1298 additions and 378 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
openbb_platform/extensions/quantitative/integration/test_quantitative_api.py
View File

@@ -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/skewness?{query_str}"
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/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

249
openbb_platform/extensions/quantitative/integration/test_quantitative_python.py
View File

@@ -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

203
openbb_platform/extensions/quantitative/openbb_quantitative/performance/performance_router.py Normal file
View File

@@ -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_)

363
openbb_platform/extensions/quantitative/openbb_quantitative/quantitative_router.py
View File

@@ -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.

319
openbb_platform/extensions/quantitative/openbb_quantitative/rolling/rolling_router.py Normal file
View File

@@ -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)

41
openbb_platform/extensions/quantitative/openbb_quantitative/statistics.py Normal file
View File

@@ -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)

258
openbb_platform/extensions/quantitative/openbb_quantitative/stats/stats_router.py Normal file
View File

@@ -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)

25
openbb_platform/extensions/quantitative/tests/test_statistics.py Normal file
View File

@@ -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)

4
openbb_platform/extensions/tests/test_integration_tests_python.py
View File

@@ -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)

2
openbb_platform/extensions/tests/utils/router_testers.py
View File

@@ -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}"
)