-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Rename econometrics commands to be more verbose (#5702)
- Loading branch information
Showing
3 changed files
with
65 additions
and
292 deletions.
There are no files selected for viewing
256 changes: 14 additions & 242 deletions
256
openbb_platform/extensions/econometrics/integration/test_econometrics_api.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,270 +1,42 @@ | ||
| import base64 | ||
| import json | ||
| import random | ||
| from typing import Literal | ||
|
|
||
| import pytest | ||
| import requests | ||
| from openbb_core.env import Env | ||
| from openbb_provider.utils.helpers import get_querystring | ||
|
|
||
| data: dict = {} | ||
|
|
||
|
|
||
| def get_headers(): | ||
| if "headers" in data: | ||
| return data["headers"] | ||
|
|
||
| userpass = f"{Env().API_USERNAME}:{Env().API_PASSWORD}" | ||
| userpass_bytes = userpass.encode("ascii") | ||
| base64_bytes = base64.b64encode(userpass_bytes) | ||
|
|
||
| data["headers"] = {"Authorization": f"Basic {base64_bytes.decode('ascii')}"} | ||
| return data["headers"] | ||
|
|
||
|
|
||
| def request_data(menu: str, symbol: str, provider: str): | ||
| """Randomly pick a symbol and a provider and get data from the selected menu.""" | ||
| url = f"http://0.0.0.0:8000/api/v1/{menu}/load?symbol={symbol}&provider={provider}" | ||
| result = requests.get(url, headers=get_headers(), timeout=10) | ||
| return result.json()["results"] | ||
|
|
||
|
|
||
| def get_stocks_data(): | ||
| if "stocks_data" in data: | ||
| return data["stocks_data"] | ||
|
|
||
| symbol = random.choice(["AAPL", "NVDA", "MSFT", "TSLA", "AMZN", "V"]) # noqa: S311 | ||
| provider = random.choice(["fmp", "polygon", "yfinance"]) # noqa: S311 | ||
|
|
||
| data["stocks_data"] = request_data("stocks", symbol=symbol, provider=provider) | ||
| return data["stocks_data"] | ||
|
|
||
|
|
||
| def get_crypto_data(): | ||
| if "crypto_data" in data: | ||
| return data["crypto_data"] | ||
|
|
||
| # TODO : add more crypto providers and symbols | ||
| symbol = random.choice(["BTC"]) # noqa: S311 | ||
| provider = random.choice(["fmp"]) # noqa: S311 | ||
|
|
||
| data["crypto_data"] = request_data( | ||
| menu="crypto", | ||
| symbol=symbol, | ||
| provider=provider, | ||
| ) | ||
| return data["crypto_data"] | ||
|
|
||
|
|
||
| def get_data(menu: Literal["stocks", "crypto"]): | ||
| funcs = {"stocks": get_stocks_data, "crypto": get_crypto_data} | ||
| return funcs[menu]() | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [ | ||
| ({"data": ""}, "stocks"), | ||
| ({"data": ""}, "crypto"), | ||
| ], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_corr(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
|
|
||
| body = json.dumps(get_data(data_type)) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/corr?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=10, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_corr(): | ||
| ... | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [ | ||
| ( | ||
| {"data": "", "y_column": "close", "x_columns": ["high"]}, | ||
| "stocks", | ||
| ), | ||
| ( | ||
| {"data": "", "y_column": "close", "x_columns": ["high"]}, | ||
| "crypto", | ||
| ), | ||
| ], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_ols_summary(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
|
|
||
| body = json.dumps( | ||
| { | ||
| "data": get_data(data_type), | ||
| "x_columns": params.pop("x_columns"), | ||
| } | ||
| ) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/ols_summary?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=20, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_ols_summary(): | ||
| ... | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [ | ||
| ( | ||
| {"data": "", "y_column": "volume", "x_columns": ["close"]}, | ||
| "stocks", | ||
| ), | ||
| ( | ||
| {"data": "", "y_column": "volume", "x_columns": ["close"]}, | ||
| "crypto", | ||
| ), | ||
| ], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_dwat(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
|
|
||
| body = json.dumps( | ||
| { | ||
| "data": get_data(data_type), | ||
| "x_columns": params.pop("x_columns"), | ||
| } | ||
| ) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/dwat?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=10, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_dwat(): | ||
| ... | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [ | ||
| ( | ||
| { | ||
| "data": "", | ||
| "y_column": "volume", | ||
| "x_columns": ["close"], | ||
| "lags": "", | ||
| }, | ||
| "stocks", | ||
| ), | ||
| ( | ||
| { | ||
| "data": "", | ||
| "y_column": "volume", | ||
| "x_columns": ["close"], | ||
| "lags": "2", | ||
| }, | ||
| "crypto", | ||
| ), | ||
| ], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_bgot(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
|
|
||
| body = json.dumps( | ||
| { | ||
| "data": get_data(data_type), | ||
| "x_columns": params.pop("x_columns"), | ||
| } | ||
| ) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/bgot?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=10, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_bgot(): | ||
| ... | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [ | ||
| ( | ||
| { | ||
| "data": "", | ||
| "columns": ["close", "volume"], | ||
| }, | ||
| "stocks", | ||
| ), | ||
| ( | ||
| { | ||
| "data": "", | ||
| "columns": ["close", "volume"], | ||
| }, | ||
| "crypto", | ||
| ), | ||
| ], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_coint(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
|
|
||
| body = json.dumps( | ||
| { | ||
| "data": get_data(data_type), | ||
| "columns": params.pop("columns"), | ||
| } | ||
| ) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/coint?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=10, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_coint(): | ||
| ... | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [ | ||
| ( | ||
| {"data": "", "y_column": "volume", "x_column": "close", "lag": ""}, | ||
| "stocks", | ||
| ), | ||
| ( | ||
| {"data": "", "y_column": "volume", "x_column": "close", "lag": "2"}, | ||
| "crypto", | ||
| ), | ||
| ], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_granger(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
| body = json.dumps(get_data(data_type)) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/granger?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=10, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_granger(): | ||
| ... | ||
|
|
||
|
|
||
| @pytest.mark.skip(reason="econometrics is a python only extensions so far") | ||
| @pytest.mark.parametrize( | ||
| "params, data_type", | ||
| [({"data": "", "column": "high", "regression": "c"}, "stocks")], | ||
| ) | ||
| @pytest.mark.integration | ||
| def test_econometrics_unitroot(params, data_type): | ||
| params = {p: v for p, v in params.items() if v} | ||
| body = json.dumps(get_data(data_type)) | ||
|
|
||
| query_str = get_querystring(params, []) | ||
| url = f"http://0.0.0.0:8000/api/v1/econometrics/unitroot?{query_str}" | ||
| result = requests.post(url, headers=get_headers(), timeout=10, data=body) | ||
| assert isinstance(result, requests.Response) | ||
| assert result.status_code == 200 | ||
| def test_econometrics_unitroot(): | ||
| ... |
Oops, something went wrong.