Files
OpenBB/openbb_platform/extensions/index/integration/test_index_api.py
Danglewood fe3fa6af8e [Enhancement]: Add TMX data provider extension. (#5913)
* dev install

* add TMX provider extension

* normalize ytm and coupopn_rate values

* black

* codespell

* ruff

* black again

* pylint

* ftw

* undo what Pylint thinks is best

* more black

* renovate insider_trading standard model

* fix tests

* typo

* tests

* empty string

* pylint unused argument

* nans for some reason

* black

* static files

* static files with openbb-tmx installed this time

* yfinance TypeError..?

* NA -> N/A

* update for version bump

* list exchange-calendars as tmx dependency

* eod chains date -> eod_date

* new line

* not that file

* more black

* one more black

* fix weird merge thing

* ruff

* index constituents params

* insider trading

* fixing things

* black

* black AGAIN

* news integration params

* the other missing param

* ruff

* historical dividends field update

* add json_schema_extra

* gainers

* symbols -> symbol

* company news symbol

* unit test param

* review items

* wrangle a rogue field

* index router example

* company news thing - symbols -> symbol

* field name map

---------

Co-authored-by: Igor Radovanovic <74266147+IgorWounds@users.noreply.github.com>
Co-authored-by: Henrique Joaquim <h.joaquim@campus.fct.unl.pt>
Co-authored-by: James Maslek <jmaslek11@gmail.com>
2024-02-20 14:41:26 +00:00

378 lines
10 KiB
Python

import base64
import pytest
import requests
from extensions.tests.conftest import parametrize
from openbb_core.env import Env
from openbb_core.provider.utils.helpers import get_querystring
@pytest.fixture(scope="session")
def headers():
userpass = f"{Env().API_USERNAME}:{Env().API_PASSWORD}"
userpass_bytes = userpass.encode("ascii")
base64_bytes = base64.b64encode(userpass_bytes)
return {"Authorization": f"Basic {base64_bytes.decode('ascii')}"}
# pylint: disable=redefined-outer-name
@parametrize(
"params",
[
({"symbol": "dowjones", "provider": "fmp"}),
({"symbol": "^TX60", "provider": "tmx", "use_cache": False}),
({"symbol": "BUKBUS", "provider": "cboe"}),
],
)
@pytest.mark.integration
def test_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/index/constituents?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"interval": "1m",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2024-01-01",
"end_date": "2024-02-05",
"use_cache": False,
"sort": None,
"limit": None,
}
),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": False,
"sort": None,
"limit": None,
}
),
(
{
"interval": "1min",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2024-01-01",
"end_date": "2024-02-05",
"timeseries": 1,
"sort": "desc",
"limit": None,
}
),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
"limit": None,
}
),
(
{
"sort": "desc",
"limit": 49999,
"interval": "1m",
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"sort": "desc",
"limit": 49999,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
}
),
(
{
"interval": "1d",
"period": "max",
"prepost": True,
"rounding": True,
"provider": "yfinance",
"symbol": "DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"sort": None,
"limit": None,
}
),
(
{
"interval": "1d",
"provider": "intrinio",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"symbol": "DJI",
"sort": "desc",
"limit": 100,
}
),
],
)
@pytest.mark.integration
def test_index_price_historical(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/index/price/historical?{query_str}"
result = requests.get(url, headers=headers, timeout=20)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-03-03",
"provider": "fmp",
"sort": "desc",
"interval": "1day",
"limit": None,
}
),
(
{
"interval": "1m",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": True,
"limit": None,
"sort": None,
}
),
(
{
"interval": "1d",
"provider": "cboe",
"symbol": "AAVE100",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"use_cache": False,
"limit": None,
"sort": None,
}
),
(
{
"interval": "1min",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
"limit": None,
}
),
(
{
"interval": "1day",
"provider": "fmp",
"symbol": "^DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"timeseries": 1,
"sort": "desc",
"limit": None,
}
),
(
{
"timespan": "minute",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"interval": None,
}
),
(
{
"timespan": "day",
"sort": "desc",
"limit": 49999,
"adjusted": True,
"multiplier": 1,
"provider": "polygon",
"symbol": "NDX",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"interval": None,
}
),
(
{
"interval": "1d",
"period": "max",
"prepost": True,
"rounding": True,
"provider": "yfinance",
"symbol": "DJI",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"limit": None,
"sort": None,
}
),
(
{
"provider": "intrinio",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"symbol": "$DJI",
"tag": "level",
"sort": "desc",
"limit": 100,
"type": None,
"interval": None,
}
),
],
)
@pytest.mark.integration
def test_index_market(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/index/market?{query_str}"
result = requests.get(url, headers=headers, timeout=20)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"provider": "cboe", "use_cache": False}),
({"provider": "fmp"}),
({"provider": "yfinance"}),
({"provider": "tmx", "use_cache": False}),
],
)
@pytest.mark.integration
def test_index_available(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/index/available?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"query": "D", "is_symbol": True, "provider": "cboe", "use_cache": False}),
],
)
@pytest.mark.integration
def test_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/index/search?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"provider": "cboe", "region": "us"}),
({"provider": "tmx", "region": "ca", "use_cache": False}),
],
)
@pytest.mark.integration
def test_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/index/snapshots?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
(
{
"series_name": "pe_month",
"start_date": "2023-01-01",
"end_date": "2023-06-06",
"collapse": "monthly",
"transform": "diff",
"provider": "nasdaq",
}
)
],
)
@pytest.mark.integration
def test_index_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/index/sp500_multiples?{query_str}"
result = requests.get(url, headers=headers, timeout=20)
assert isinstance(result, requests.Response)
assert result.status_code == 200
@parametrize(
"params",
[
({"provider": "tmx", "symbol": "^TX60", "use_cache": False}),
],
)
@pytest.mark.integration
def test_index_sectors(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/index/sectors?{query_str}"
result = requests.get(url, headers=headers, timeout=10)
assert isinstance(result, requests.Response)
assert result.status_code == 200