jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-11-06 01:12:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

integration tests for all extensions

Browse Source
This commit is contained in:
hjoaquim
2023-09-25 16:59:53 +01:00
parent 6dab67e83c
commit 1bfd131fad
10 changed files with 1768 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
openbb_sdk/extensions/crypto/integration/test_crypto_api.py Normal file
View File

@@ -0,0 +1,65 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"multiplier": 1,
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"provider": "polygon",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"provider": "yfinance",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_crypto_load(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/crypto/load?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

115
openbb_sdk/extensions/econometrics/integration/test_econometrics_api.py Normal file
View File

@@ -0,0 +1,115 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[({"data": ""})],
)
def test_econometrics_corr(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/corr?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "y_column": "", "x_columns": ""})],
)
def test_econometrics_ols_summary(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/ols_summary?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "y_column": "", "x_columns": ""})],
)
def test_econometrics_dwat(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/dwat?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "y_column": "", "x_columns": "", "lags": ""})],
)
def test_econometrics_bgot(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/bgot?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "columns": ""})],
)
def test_econometrics_coint(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/coint?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "y_column": "", "x_column": "", "lag": ""})],
)
def test_econometrics_granger(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/granger?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "column": "", "regression": ""})],
)
def test_econometrics_unitroot(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/econometrics/unitroot?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

287
openbb_sdk/extensions/economy/integration/test_economy_api.py Normal file
View File

@@ -0,0 +1,287 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[({"index": "dowjones"})],
)
def test_economy_const(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/const?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"countries": ["portugal", "spain"],
"units": "growth_same",
"frequency": "monthly",
"harmonized": True,
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
)
],
)
def test_economy_cpi(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/cpi?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"prepost": True,
"rounding": True,
"provider": "yfinance",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_economy_index(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/index?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_economy_european_index(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/european_index?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"symbol": "AAPL"})],
)
def test_economy_european_index_constituents(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/european_index_constituents?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({}), ({"europe": True, "provider": "cboe"})],
)
def test_economy_available_indices(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/available_indices?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({})],
)
def test_economy_risk(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/risk?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"query": "TEST_STRING", "symbol": "AAPL"}),
(
{
"europe": True,
"provider": "cboe",
"query": "TEST_STRING",
"symbol": "AAPL",
}
),
],
)
def test_economy_index_search(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/index_search?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"region": "US"})],
)
def test_economy_index_snapshots(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/index_snapshots?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"query": "TEST_STRING"})],
)
def test_economy_cot_search(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/cot_search?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({}),
(
{
"code": "13874P",
"data_type": "FO",
"legacy_format": True,
"report_type": "ALL",
"measure": "CR",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"transform": "diff",
"provider": "quandl",
}
),
],
)
def test_economy_cot(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/cot?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"series_name": "PE Ratio by Month",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"collapse": "monthly",
"transform": "diff",
}
)
],
)
def test_economy_sp500_multiples(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/economy/sp500_multiples?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

193
openbb_sdk/extensions/fixedincome/integration/test_fixedincome_api.py Normal file
View File

@@ -0,0 +1,193 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[({"start_date": "2023-01-01", "end_date": "2023-06-06"})],
)
def test_fixedincome_treasury(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/treasury?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"date": "2023-01-01", "inflation_adjusted": True})],
)
def test_fixedincome_ycrv(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/ycrv?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"period": "overnight",
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_fixedincome_sofr(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/sofr?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"parameter": "volume_weighted_trimmed_mean_rate",
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_fixedincome_estr(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/estr?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"parameter": "rate",
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_fixedincome_sonia(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/sonia?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"parameter": "overnight",
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_fixedincome_ameribor(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/ameribor?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"parameter": "weekly",
"provider": "fred",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_fixedincome_fed(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/fed?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({}), ({"long_run": True, "provider": "fred"})],
)
def test_fixedincome_projections(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/projections?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"start_date": "2023-01-01", "end_date": "2023-06-06"})],
)
def test_fixedincome_iorb(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/fixedincome/iorb?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

93
openbb_sdk/extensions/forex/integration/test_forex_api.py Normal file
View File

@@ -0,0 +1,93 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[
({}),
(
{
"symbol": "AAPL",
"date": datetime.date(2023, 9, 25),
"search": "TEST_STRING",
"active": True,
"order": "asc",
"sort": "ticker",
"limit": 1000,
"provider": "polygon",
}
),
],
)
def test_forex_pairs(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/forex/pairs?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"multiplier": 1,
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"provider": "polygon",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"provider": "yfinance",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_forex_load(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/forex/load?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

68
openbb_sdk/extensions/futures/integration/test_futures_api.py Normal file
View File

@@ -0,0 +1,68 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[
(
{
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"expiration": "TEST_STRING",
}
),
(
{
"interval": "1d",
"period": "1d",
"prepost": True,
"adjust": True,
"back_adjust": True,
"provider": "yfinance",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"expiration": "TEST_STRING",
}
),
],
)
def test_futures_load(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/futures/load?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"symbol": "AAPL", "date": "2023-01-01"})],
)
def test_futures_curve(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/futures/curve?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

53
openbb_sdk/extensions/news/integration/test_news_api.py Normal file
View File

@@ -0,0 +1,53 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[
({"limit": 20}),
(
{
"display": "full",
"date": "2023-01-01",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"updated_since": 1,
"published_since": 1,
"sort": "created",
"order": "desc",
"isin": "TEST_STRING",
"cusip": "TEST_STRING",
"channels": "TEST_STRING",
"topics": "TEST_STRING",
"authors": "TEST_STRING",
"content_types": "TEST_STRING",
"provider": "benzinga",
"limit": 20,
}
),
],
)
def test_news_globalnews(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/news/globalnews?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

157
openbb_sdk/extensions/qa/integration/test_qa_api.py Normal file
View File

@@ -0,0 +1,157 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[({"data": "", "target": ""})],
)
def test_qa_normality(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/normality?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": ""})],
)
def test_qa_capm(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/capm?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "threshold_start": "", "threshold_end": ""})],
)
def test_qa_om(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/om?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "window": ""})],
)
def test_qa_kurtosis(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/kurtosis?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "fuller_reg": "", "kpss_reg": ""})],
)
def test_qa_unitroot(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/unitroot?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "rfr": "", "window": ""})],
)
def test_qa_sh(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/sh?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "target_return": "", "window": "", "adjusted": ""})],
)
def test_qa_so(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/so?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "window": ""})],
)
def test_qa_skew(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/skew?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "window": "", "quantile_pct": ""})],
)
def test_qa_quantile(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/quantile?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": ""})],
)
def test_qa_summary(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/qa/summary?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

460
openbb_sdk/extensions/stocks/integration/test_stocks_api.py
View File

@@ -7,6 +7,7 @@ def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@@ -19,32 +20,31 @@ def headers():
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "period": "annual", "limit": 12}),
(
{
"provider": "",
"symbol": "",
"period": "",
"limit": "",
"type": "",
"year": "",
"company_name": "",
"company_name_search": "",
"sic": "",
"filing_date": "",
"filing_date_lt": "",
"filing_date_lte": "",
"filing_date_gt": "",
"filing_date_gte": "",
"period_of_report_date": "",
"period_of_report_date_lt": "",
"period_of_report_date_lte": "",
"period_of_report_date_gt": "",
"period_of_report_date_gte": "",
"include_sources": "",
"order": "",
"sort": "",
"type": "reported",
"year": 1,
"provider": "intrinio",
"symbol": "AAPL",
"period": "annual",
"limit": 12,
}
)
),
(
{
"company_name": "TEST_STRING",
"company_name_search": "TEST_STRING",
"sic": "TEST_STRING",
"include_sources": True,
"order": "asc",
"sort": "filing_date",
"provider": "polygon",
"symbol": "AAPL",
"period": "annual",
"limit": 12,
}
),
],
)
def test_stocks_fa_balance(params, headers):
@@ -52,35 +52,35 @@ def test_stocks_fa_balance(params, headers):
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/balance?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "limit": ""})],
[({"symbol": "AAPL", "limit": 10})],
)
def test_stocks_fa_balance_growth(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/balance_growth?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "start_date": "", "end_date": ""})],
[({"start_date": "2023-01-01", "end_date": "2023-06-06"})],
)
def test_stocks_fa_cal(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/cal?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -88,32 +88,31 @@ def test_stocks_fa_cal(params, headers):
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "period": "annual", "limit": 12}),
(
{
"provider": "",
"symbol": "",
"period": "",
"limit": "",
"type": "",
"year": "",
"company_name": "",
"company_name_search": "",
"sic": "",
"filing_date": "",
"filing_date_lt": "",
"filing_date_lte": "",
"filing_date_gt": "",
"filing_date_gte": "",
"period_of_report_date": "",
"period_of_report_date_lt": "",
"period_of_report_date_lte": "",
"period_of_report_date_gt": "",
"period_of_report_date_gte": "",
"include_sources": "",
"order": "",
"sort": "",
"type": "reported",
"year": 1,
"provider": "intrinio",
"symbol": "AAPL",
"period": "annual",
"limit": 12,
}
)
),
(
{
"company_name": "TEST_STRING",
"company_name_search": "TEST_STRING",
"sic": "TEST_STRING",
"include_sources": True,
"order": "asc",
"sort": "filing_date",
"provider": "polygon",
"symbol": "AAPL",
"period": "annual",
"limit": 12,
}
),
],
)
def test_stocks_fa_cash(params, headers):
@@ -121,105 +120,105 @@ def test_stocks_fa_cash(params, headers):
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/cash?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "limit": ""})],
[({"symbol": "AAPL", "limit": 10})],
)
def test_stocks_fa_cash_growth(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/cash_growth?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_comp(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/comp?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "start_date": "", "end_date": ""})],
[({"start_date": "2023-01-01", "end_date": "2023-06-06"})],
)
def test_stocks_fa_comsplit(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/comsplit?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_divs(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/divs?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "limit": ""})],
[({"symbol": "AAPL", "limit": 50})],
)
def test_stocks_fa_earning(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/earning?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_emp(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/emp?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "period": "", "limit": ""})],
[({"symbol": "AAPL", "period": "annual", "limit": 30})],
)
def test_stocks_fa_est(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/est?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -227,32 +226,31 @@ def test_stocks_fa_est(params, headers):
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "period": "annual", "limit": 12}),
(
{
"provider": "",
"symbol": "",
"period": "",
"limit": "",
"type": "",
"year": "",
"company_name": "",
"company_name_search": "",
"sic": "",
"filing_date": "",
"filing_date_lt": "",
"filing_date_lte": "",
"filing_date_gt": "",
"filing_date_gte": "",
"period_of_report_date": "",
"period_of_report_date_lt": "",
"period_of_report_date_lte": "",
"period_of_report_date_gt": "",
"period_of_report_date_gte": "",
"include_sources": "",
"order": "",
"sort": "",
"type": "reported",
"year": 1,
"provider": "intrinio",
"symbol": "AAPL",
"period": "annual",
"limit": 12,
}
)
),
(
{
"company_name": "TEST_STRING",
"company_name_search": "TEST_STRING",
"sic": "TEST_STRING",
"include_sources": True,
"order": "asc",
"sort": "filing_date",
"provider": "polygon",
"symbol": "AAPL",
"period": "annual",
"limit": 12,
}
),
],
)
def test_stocks_fa_income(params, headers):
@@ -260,21 +258,21 @@ def test_stocks_fa_income(params, headers):
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/income?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "limit": "", "period": ""})],
[({"symbol": "AAPL", "limit": 10, "period": "annual"})],
)
def test_stocks_fa_income_growth(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/income_growth?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -284,12 +282,11 @@ def test_stocks_fa_income_growth(params, headers):
[
(
{
"provider": "",
"symbol": "",
"transactionType": "",
"reportingCik": "",
"companyCik": "",
"page": "",
"symbol": "AAPL",
"transactionType": ["P-Purchase"],
"reportingCik": 1,
"companyCik": 1,
"page": 1,
}
)
],
@@ -299,231 +296,217 @@ def test_stocks_fa_ins(params, headers):
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/ins?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "include_current_quarter": "", "date": ""})],
[({"symbol": "AAPL", "include_current_quarter": True, "date": "2023-01-01"})],
)
def test_stocks_fa_ins_own(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/ins_own?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "period": "", "limit": ""})],
[({"symbol": "AAPL", "period": "annual", "limit": 100})],
)
def test_stocks_fa_metrics(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/metrics?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_mgmt(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/mgmt?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_overview(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/overview?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "date": "", "page": ""})],
[({"symbol": "AAPL", "date": "2023-01-01", "page": 1})],
)
def test_stocks_fa_own(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/own?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_pt(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/pt?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "with_grade": ""})],
[({"symbol": "AAPL"}), ({"with_grade": True, "provider": "fmp", "symbol": "AAPL"})],
)
def test_stocks_fa_pta(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/pta?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "period": "", "limit": ""})],
[({"symbol": "AAPL", "period": "annual", "limit": 12})],
)
def test_stocks_fa_ratios(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/ratios?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "period": "", "structure": ""})],
[({"symbol": "AAPL", "period": "annual", "structure": "flat"})],
)
def test_stocks_fa_revgeo(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/revgeo?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "period": "", "structure": ""})],
[({"symbol": "AAPL", "period": "annual", "structure": "flat"})],
)
def test_stocks_fa_revseg(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/revseg?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "type": "", "page": "", "limit": ""})],
[({"symbol": "AAPL", "type": "1", "page": 1, "limit": 100})],
)
def test_stocks_fa_sec(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/sec?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_shrs(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/shrs?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_fa_split(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/split?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "year": "", "quarter": ""})],
[({"symbol": "AAPL", "year": 1, "quarter": 1})],
)
def test_stocks_fa_transcript(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/fa/transcript?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_ca_peers(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/ca/peers?{query_str}"
result = requests.get(url, headers=headers)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "date": ""})],
)
def test_stocks_options_chains(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/options/chains?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -531,40 +514,100 @@ def test_stocks_options_chains(params, headers):
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL"}),
({"date": "2023-01-01", "provider": "intrinio", "symbol": "AAPL"}),
],
)
def test_stocks_options_chains(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/options/chains?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
({"symbol": "AAPL", "start_date": "2023-01-01", "end_date": "2023-06-06"}),
(
{
"period": "daily",
"interval": "60min",
"adjusted": True,
"extended_hours": True,
"month": "TEST_STRING",
"outputsize": "full",
"provider": "alpha_vantage",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "AAPL",
"start_date": "",
"end_date": "",
"period": "",
"interval": "",
"adjusted": "",
"extended_hours": "",
"month": "",
"outputsize": "",
"timeseries": "",
"timezone": "",
"source": "",
"start_time": "",
"end_time": "",
"interval_size": "",
"multiplier": "",
"timespan": "",
"sort": "",
"limit": "",
"prepost": "",
"actions": "",
"auto_adjust": "",
"back_adjust": "",
"progress": "",
"ignore_tz": "",
"rounding": "",
"repair": "",
"keepna": "",
"group_by": "",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
)
),
(
{
"timezone": "UTC",
"source": "realtime",
"interval_size": "60m",
"provider": "intrinio",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"multiplier": 1,
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"provider": "polygon",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"prepost": True,
"actions": True,
"auto_adjust": True,
"back_adjust": True,
"progress": True,
"ignore_tz": True,
"rounding": True,
"repair": True,
"keepna": True,
"group_by": "column",
"provider": "yfinance",
"symbol": "AAPL",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
],
)
def test_stocks_load(params, headers):
@@ -572,10 +615,7 @@ def test_stocks_load(params, headers):
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/load?{query_str}"
print(url)
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@@ -583,28 +623,37 @@ def test_stocks_load(params, headers):
@pytest.mark.parametrize(
"params",
[
({"symbols": "AAPL,MSFT", "limit": 20}),
(
{
"provider": "",
"symbols": "",
"limit": "",
"display": "",
"date": "",
"start_date": "",
"end_date": "",
"updated_since": "",
"published_since": "",
"sort": "",
"order": "",
"isin": "",
"cusip": "",
"channels": "",
"topics": "",
"authors": "",
"content_types": "",
"published_utc": "",
"display": "full",
"date": "2023-01-01",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"updated_since": 1,
"published_since": 1,
"sort": "created",
"order": "desc",
"isin": "TEST_STRING",
"cusip": "TEST_STRING",
"channels": "TEST_STRING",
"topics": "TEST_STRING",
"authors": "TEST_STRING",
"content_types": "TEST_STRING",
"provider": "benzinga",
"symbols": "AAPL,MSFT",
"limit": 20,
}
)
),
(
{
"published_utc": "TEST_STRING",
"order": "desc",
"provider": "polygon",
"symbols": "AAPL,MSFT",
"limit": 20,
}
),
],
)
def test_stocks_news(params, headers):
@@ -612,62 +661,65 @@ def test_stocks_news(params, headers):
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/news?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "limit": ""})],
[({"symbol": "AAPL", "limit": 100})],
)
def test_stocks_multiples(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/multiples?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "query": "", "ticker": ""})],
[({"query": "TEST_STRING", "ticker": True})],
)
def test_stocks_search(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/search?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": "", "source": ""})],
[
({"symbol": "AAPL"}),
({"source": "iex", "provider": "intrinio", "symbol": "AAPL"}),
],
)
def test_stocks_quote(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/quote?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"provider": "", "symbol": ""})],
[({"symbol": "AAPL"})],
)
def test_stocks_info(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/stocks/info?{query_str}"
result = requests.get(url, headers=headers)
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200

481
openbb_sdk/extensions/ta/integration/test_ta_api.py Normal file
View File

@@ -0,0 +1,481 @@
import pytest
import requests
from openbb_provider.utils.helpers import get_querystring
def get_token():
return requests.post(
"http://0.0.0.0:8000/api/v1/account/token",
data={"username": "openbb", "password": "openbb"},
timeout=5,
)
@pytest.fixture(scope="session")
def headers():
access_token = get_token().json()["access_token"]
return {"Authorization": f"Bearer {access_token}"}
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"length": "",
"mamode": "",
"drift": "",
"offset": "",
}
)
],
)
def test_ta_atr(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/atr?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"close_column": "",
"period": "",
"start_date": "",
"end_date": "",
}
)
],
)
def test_ta_fib(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/fib?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "offset": ""})],
)
def test_ta_obv(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/obv?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "length": "", "signal": ""})],
)
def test_ta_fisher(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/fisher?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "fast": "", "slow": "", "offset": ""})],
)
def test_ta_adosc(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/adosc?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "",
"index": "",
"length": "",
"std": "",
"mamode": "",
"offset": "",
}
)
],
)
def test_ta_bbands(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/bbands?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "index": "", "length": "", "offset": ""})],
)
def test_ta_zlma(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/zlma?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "length": "", "scalar": ""})],
)
def test_ta_aroon(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/aroon?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "index": "", "length": "", "offset": ""})],
)
def test_ta_sma(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/sma?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"target": "",
"show_all": "",
"asint": "",
"offset": "",
}
)
],
)
def test_ta_demark(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/demark?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "anchor": "", "offset": ""})],
)
def test_ta_vwap(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/vwap?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "index": "", "fast": "", "slow": "", "signal": ""})],
)
def test_ta_macd(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/macd?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "index": "", "length": "", "offset": ""})],
)
def test_ta_hma(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/hma?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "lower_length": "", "upper_length": "", "offset": ""})],
)
def test_ta_donchian(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/donchian?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"conversion": "",
"base": "",
"lagging": "",
"offset": "",
"lookahead": "",
}
)
],
)
def test_ta_ichimoku(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ichimoku?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "target": "", "period": ""})],
)
def test_ta_clenow(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/clenow?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "offset": ""})],
)
def test_ta_ad(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ad?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "length": "", "scalar": "", "drift": ""})],
)
def test_ta_adx(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/adx?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "index": "", "length": "", "offset": ""})],
)
def test_ta_wma(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/wma?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "length": "", "scalar": ""})],
)
def test_ta_cci(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/cci?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"target": "",
"index": "",
"length": "",
"scalar": "",
"drift": "",
}
)
],
)
def test_ta_rsi(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/rsi?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"fast_k_period": "",
"slow_d_period": "",
"slow_k_period": "",
}
)
],
)
def test_ta_stoch(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/stoch?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"length": "",
"scalar": "",
"mamode": "",
"offset": "",
}
)
],
)
def test_ta_kc(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/kc?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "index": "", "length": ""})],
)
def test_ta_cg(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/cg?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[
(
{
"data": "",
"index": "",
"lower_q": "",
"upper_q": "",
"model": "",
"is_crypto": "",
}
)
],
)
def test_ta_cones(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/cones?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@pytest.mark.parametrize(
"params",
[({"data": "", "target": "", "index": "", "length": "", "offset": ""})],
)
def test_ta_ema(params, headers):
params = {p: v for p, v in params.items() if v}
query_str = get_querystring(params, [])
url = f"http://0.0.0.0:8000/api/v1/ta/ema?{query_str}"
result = requests.get(url, headers=headers, timeout=5)
assert isinstance(result, requests.Response)
assert result.status_code == 200