jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-10-28 04:52:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

[V5] openbb-quantitative: Refactor for V5 (#7506)

Browse Source 
* refactor quantitative for v5

* add factors endpoint and add a few charts

---------

Co-authored-by: deeleeramone <>
This commit is contained in:
Danglewood
2026-05-29 00:46:25 -07:00
committed by GitHub
parent 43b14e1746
commit cd7d602b8f
46 changed files with 8394 additions and 4794 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,559 +1,159 @@
"""Test qa extension."""
import random
from typing import Literal
"""Integration tests for the openbb-quantitative Python interface."""
import pytest
from openbb_core.app.model.obbject import OBBject
# pylint:disable=inconsistent-return-statements
@pytest.fixture(scope="session")
def obb(pytestconfig):
"""Fixture to setup obb."""
"""Return the obb application object for the integration session."""
if pytestconfig.getoption("markexpr") != "not integration":
import openbb # pylint:disable=import-outside-toplevel
import openbb
return openbb.obb
return None
# pylint:disable=redefined-outer-name
data: dict = {}
def get_stocks_data():
"""Get stocks data."""
import openbb # pylint:disable=import-outside-toplevel
if "stocks_data" in data:
return data["stocks_data"]
symbol = random.choice(["AAPL", "NVDA", "MSFT", "TSLA", "AMZN", "V"]) # noqa: S311
provider = random.choice(["fmp", "yfinance"]) # noqa: S311
data["stocks_data"] = openbb.obb.equity.price.historical( # type: ignore
symbol=symbol, provider=provider # type: ignore
).results
return data["stocks_data"]
def get_crypto_data():
"""Get crypto data."""
import openbb # pylint:disable=import-outside-toplevel
if "crypto_data" in data:
return data["crypto_data"]
# TODO : add more crypto providers and symbols
symbol = random.choice(["BTCUSD"]) # noqa: S311
provider = random.choice(["fmp"]) # noqa: S311
data["crypto_data"] = openbb.obb.crypto.price.historical(
symbol=symbol, provider=provider
).results
return data["crypto_data"]
def get_data(menu: Literal["equity", "crypto"]):
"""Get data."""
funcs = {"equity": get_stocks_data, "crypto": get_crypto_data}
return funcs[menu]()
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
({"data": "", "target": "high"}, "crypto"),
],
)
@pytest.mark.integration
def test_quantitative_normality(params, data_type, obb):
"""Test normality."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.normality(**params)
assert result
def test_quantitative_normality(obb, prices_data):
result = obb.quantitative.normality(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
({"data": "", "target": "high"}, "crypto"),
],
)
@pytest.mark.integration
def test_quantitative_capm(params, data_type, obb):
"""Test capm."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.capm(**params)
assert result
def test_quantitative_capm(obb, prices_data):
result = obb.quantitative.capm(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"threshold_start": "",
"threshold_end": "",
},
"equity",
),
(
{
"data": "",
"target": "high",
"threshold_start": "0.1",
"threshold_end": "1.6",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_performance_omega_ratio(params, data_type, obb):
"""Test omega ratio."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.performance.omega_ratio(**params)
assert result
def test_quantitative_unitroot_test(obb, prices_data):
result = obb.quantitative.unitroot_test(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close", "window": "5", "index": "date"}, "equity"),
({"data": "", "target": "high", "window": "10", "index": "date"}, "crypto"),
],
)
@pytest.mark.integration
def test_quantitative_rolling_kurtosis(params, data_type, obb):
"""Test rolling kurtosis."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.rolling.kurtosis(**params)
assert result
def test_quantitative_summary(obb, prices_data):
result = obb.quantitative.summary(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"fuller_reg": "c",
"kpss_reg": "ct",
},
"equity",
),
(
{
"data": "",
"target": "high",
"fuller_reg": "ct",
"kpss_reg": "c",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_unitroot_test(params, data_type, obb):
"""Test unitroot test."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.unitroot_test(**params)
assert result
def test_quantitative_rolling_skew(obb, prices_data):
result = obb.quantitative.rolling.skew(data=prices_data, target="close", window=20)
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"rfr": "",
"window": "100",
"index": "date",
},
"equity",
),
(
{
"data": "",
"target": "high",
"rfr": "0.5",
"window": "100",
"index": "date",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_performance_sharpe_ratio(params, data_type, obb):
"""Test sharpe ratio."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.performance.sharpe_ratio(**params)
assert result
def test_quantitative_rolling_variance(obb, prices_data):
result = obb.quantitative.rolling.variance(
data=prices_data, target="close", window=20
)
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"target_return": "",
"window": "100",
"adjusted": "",
"index": "date",
},
"equity",
),
(
{
"data": "",
"target": "close",
"target_return": "",
"window": "100",
"adjusted": "true",
"index": "date",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_performance_sortino_ratio(params, data_type, obb):
"""Test sortino ratio."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.performance.sortino_ratio(**params)
assert result
def test_quantitative_rolling_stdev(obb, prices_data):
result = obb.quantitative.rolling.stdev(data=prices_data, target="close", window=20)
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close", "window": "220", "index": "date"}, "equity"),
],
)
@pytest.mark.integration
def test_quantitative_rolling_skew(params, data_type, obb):
"""Test rolling skew."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.rolling.skew(**params)
assert result
def test_quantitative_rolling_kurtosis(obb, prices_data):
result = obb.quantitative.rolling.kurtosis(
data=prices_data, target="close", window=20
)
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.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_quantile(params, data_type, obb):
"""Test rolling quantile."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.rolling.quantile(**params)
assert result
def test_quantitative_rolling_mean(obb, prices_data):
result = obb.quantitative.rolling.mean(data=prices_data, target="close", window=20)
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
({"data": "", "target": "high"}, "crypto"),
],
)
@pytest.mark.integration
def test_quantitative_summary(params, data_type, obb):
"""Test summary."""
params = {p: v for p, v in params.items() if v}
params["data"] = get_data(data_type)
result = obb.quantitative.summary(**params)
assert result
def test_quantitative_rolling_quantile(obb, prices_data):
result = obb.quantitative.rolling.quantile(
data=prices_data, target="close", window=20, quantile_pct=0.75
)
assert isinstance(result, OBBject)
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"window": "10",
"index": "date",
},
"equity",
),
(
{
"data": "",
"target": "high",
"window": "50",
"index": "date",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_rolling_stdev(params, data_type, obb):
"""Test rolling stdev."""
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
def test_quantitative_stats_skew(obb, prices_data):
result = obb.quantitative.stats.skew(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"window": "10",
"index": "date",
},
"equity",
),
(
{
"data": "",
"target": "high",
"window": "50",
"index": "date",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_rolling_mean(params, data_type, obb):
"""Test rolling mean."""
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
def test_quantitative_stats_variance(obb, prices_data):
result = obb.quantitative.stats.variance(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
(
{
"data": "",
"target": "close",
"window": "10",
"index": "date",
},
"equity",
),
(
{
"data": "",
"target": "high",
"window": "50",
"index": "date",
},
"crypto",
),
],
)
@pytest.mark.integration
def test_quantitative_rolling_variance(params, data_type, obb):
"""Test rolling variance."""
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
def test_quantitative_stats_stdev(obb, prices_data):
result = obb.quantitative.stats.stdev(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
],
)
@pytest.mark.integration
def test_quantitative_stats_skew(params, data_type, obb):
"""Test skew."""
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
def test_quantitative_stats_kurtosis(obb, prices_data):
result = obb.quantitative.stats.kurtosis(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
],
)
@pytest.mark.integration
def test_quantitative_stats_kurtosis(params, data_type, obb):
"""Test kurtosis."""
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
def test_quantitative_stats_mean(obb, prices_data):
result = obb.quantitative.stats.mean(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
],
)
@pytest.mark.integration
def test_quantitative_stats_variance(params, data_type, obb):
"""Test variance."""
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
def test_quantitative_stats_quantile(obb, prices_data):
result = obb.quantitative.stats.quantile(
data=prices_data, target="close", quantile_pct=0.75
)
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
],
)
@pytest.mark.integration
def test_quantitative_stats_stdev(params, data_type, obb):
"""Test stdev."""
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
def test_quantitative_performance_omega_ratio(obb, prices_data):
result = obb.quantitative.performance.omega_ratio(data=prices_data, target="close")
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.parametrize(
"params, data_type",
[
({"data": "", "target": "close"}, "equity"),
],
)
@pytest.mark.integration
def test_quantitative_stats_mean(params, data_type, obb):
"""Test mean."""
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
def test_quantitative_performance_sharpe_ratio(obb, prices_data):
result = obb.quantitative.performance.sharpe_ratio(
data=prices_data, target="close", window=20
)
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results
@pytest.mark.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):
"""Test quantile."""
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
def test_quantitative_performance_sortino_ratio(obb, prices_data):
result = obb.quantitative.performance.sortino_ratio(
data=prices_data, target="close", window=20
)
assert isinstance(result, OBBject)
assert len(result.results) > 0
assert result.results