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

intrinio

Browse Source
This commit is contained in:
tehcoderer
2023-11-29 15:01:05 -05:00
parent 6a5d1f9398
commit d41b99b990
28 changed files with 3436 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
openbb_platform/core/openbb_core/provider/utils/client.py
View File

@@ -53,14 +53,6 @@ class ClientResponse(aiohttp.ClientResponse):
return aiohttp.RequestInfo(url, request_info.method, headers, url)
async def get_one(self) -> Dict[str, Any]:
"""Return the first item in the response."""
data = await self.json()
if isinstance(data, list):
return data[0]
return data
async def json(self, **kwargs) -> Union[dict, list]:
"""Return the json response."""
return await super().json(**kwargs)
@@ -100,7 +92,12 @@ class ClientSession(aiohttp.ClientSession):
async def get_one(self, url: str, **kwargs) -> Dict[str, Any]:
"""Send GET request and return first item in json if list."""
response = await self.request("GET", url, **kwargs)
return await response.get_one()
data = await response.json()
if isinstance(data, list):
return data[0]
return data
async def request(
self, *args, raise_for_status: bool = False, **kwargs

2
openbb_platform/core/openbb_core/provider/utils/helpers.py
View File

@@ -149,7 +149,7 @@ async def async_requests(
if isinstance(results[0], list):
return [item for sublist in results for item in sublist if item]
return results
return [item for item in results if item]
finally:
await session.close()

43
openbb_platform/providers/intrinio/openbb_intrinio/models/balance_sheet.py
View File

@@ -1,8 +1,6 @@
"""Intrinio Balance Sheet Model."""
from concurrent.futures import ThreadPoolExecutor
from itertools import repeat
from typing import Any, Dict, List, Optional
from openbb_core.provider.abstract.fetcher import Fetcher
@@ -10,6 +8,7 @@ from openbb_core.provider.standard_models.balance_sheet import (
BalanceSheetData,
BalanceSheetQueryParams,
)
from openbb_core.provider.utils.helpers import ClientResponse, async_requests
from openbb_intrinio.utils.helpers import get_data_one
from pydantic import alias_generators
@@ -54,7 +53,7 @@ class IntrinioBalanceSheetFetcher(
return IntrinioBalanceSheetQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioBalanceSheetQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -64,9 +63,6 @@ class IntrinioBalanceSheetFetcher(
statement_code = "balance_sheet_statement"
period_type = "FY" if query.period == "annual" else "QTR"
fundamentals_data: Dict = {}
data: List[Dict] = []
base_url = "https://api-v2.intrinio.com"
fundamentals_url_params = f"statement_code={statement_code}&type={period_type}"
fundamentals_url = (
@@ -74,34 +70,29 @@ class IntrinioBalanceSheetFetcher(
f"{fundamentals_url_params}&api_key={api_key}"
)
fundamentals_data = get_data_one(fundamentals_url, **kwargs).get(
"fundamentals", []
)
fundamentals_response = await get_data_one(fundamentals_url, **kwargs)
fundamentals_data = fundamentals_response.get("fundamentals", [])
fiscal_periods = [
f"{item['fiscal_year']}-{item['fiscal_period']}"
for item in fundamentals_data
]
fiscal_periods = fiscal_periods[: query.limit]
def get_financial_statement_data(period: str, data: List[Dict]) -> None:
statement_data: Dict = {}
async def callback(response: ClientResponse, _: Any) -> Dict:
"""Return the response."""
statement_data = await response.json()
return {
"date": statement_data["fundamental"]["end_date"],
"period": statement_data["fundamental"]["fiscal_period"],
"financials": statement_data["standardized_financials"],
}
intrinio_id = f"{query.symbol}-{statement_code}-{period}"
statement_url = f"{base_url}/fundamentals/{intrinio_id}/standardized_financials?api_key={api_key}"
statement_data = get_data_one(statement_url, **kwargs)
urls = [
f"{base_url}/fundamentals/{query.symbol}-{statement_code}-{period}/standardized_financials?api_key={api_key}"
for period in fiscal_periods
]
data.append(
{
"date": statement_data["fundamental"]["end_date"],
"period": statement_data["fundamental"]["fiscal_period"],
"financials": statement_data["standardized_financials"],
}
)
with ThreadPoolExecutor() as executor:
executor.map(get_financial_statement_data, fiscal_periods, repeat(data))
return data
return await async_requests(urls, callback, **kwargs)
@staticmethod
def transform_data(

6
openbb_platform/providers/intrinio/openbb_intrinio/models/calendar_ipo.py
View File

@@ -165,7 +165,7 @@ class IntrinioCalendarIpoFetcher(
return IntrinioCalendarIpoQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioCalendarIpoQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -177,7 +177,9 @@ class IntrinioCalendarIpoFetcher(
query_str = get_querystring(query.model_dump(by_alias=True), [])
url = f"{base_url}?{query_str}&api_key={api_key}"
return get_data_one(url, **kwargs)["initial_public_offerings"]
data = await get_data_one(url, **kwargs)
return data.get("initial_public_offerings", [])
@staticmethod
def transform_data(

45
openbb_platform/providers/intrinio/openbb_intrinio/models/cash_flow.py
View File

@@ -1,8 +1,6 @@
"""Intrinio Cash Flow Statement Model."""
from concurrent.futures import ThreadPoolExecutor
from itertools import repeat
from typing import Any, Dict, List, Optional
from openbb_core.provider.abstract.fetcher import Fetcher
@@ -10,6 +8,7 @@ from openbb_core.provider.standard_models.cash_flow import (
CashFlowStatementData,
CashFlowStatementQueryParams,
)
from openbb_core.provider.utils.helpers import ClientResponse, async_requests
from openbb_intrinio.utils.helpers import get_data_one
from pydantic import alias_generators
@@ -70,7 +69,7 @@ class IntrinioCashFlowStatementFetcher(
return IntrinioCashFlowStatementQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioCashFlowStatementQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -80,9 +79,6 @@ class IntrinioCashFlowStatementFetcher(
statement_code = "cash_flow_statement"
period_type = "FY" if query.period == "annual" else "QTR"
fundamentals_data: Dict = {}
data: List[Dict] = []
base_url = "https://api-v2.intrinio.com"
fundamentals_url_params = f"statement_code={statement_code}&type={period_type}"
fundamentals_url = (
@@ -90,34 +86,31 @@ class IntrinioCashFlowStatementFetcher(
f"{fundamentals_url_params}&api_key={api_key}"
)
fundamentals_data = get_data_one(fundamentals_url, **kwargs).get(
"fundamentals", []
)
fundamentals_response = await get_data_one(fundamentals_url, **kwargs)
fundamentals_data = fundamentals_response.get("fundamentals", [])
fiscal_periods = [
f"{item['fiscal_year']}-{item['fiscal_period']}"
for item in fundamentals_data
]
fiscal_periods = fiscal_periods[: query.limit]
def get_financial_statement_data(period: str, data: List[Dict]) -> None:
statement_data: Dict = {}
async def callback(response: ClientResponse, _: Any) -> Dict:
"""Return the response."""
statement_data = await response.json()
return {
"date": statement_data["fundamental"]["end_date"],
"period": statement_data["fundamental"]["fiscal_period"],
"financials": statement_data["standardized_financials"],
}
intrinio_id = f"{query.symbol}-{statement_code}-{period}"
statement_url = f"{base_url}/fundamentals/{intrinio_id}/standardized_financials?api_key={api_key}"
statement_data = get_data_one(statement_url, **kwargs)
intrinio_id = f"{query.symbol}-{statement_code}"
urls = [
f"{base_url}/fundamentals/{intrinio_id}-{period}/standardized_financials?api_key={api_key}"
for period in fiscal_periods
]
data.append(
{
"date": statement_data["fundamental"]["end_date"],
"period": statement_data["fundamental"]["fiscal_period"],
"financials": statement_data["standardized_financials"],
}
)
with ThreadPoolExecutor() as executor:
executor.map(get_financial_statement_data, fiscal_periods, repeat(data))
return data
return await async_requests(urls, callback, **kwargs)
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/company_filings.py
View File

@@ -82,7 +82,7 @@ class IntrinioCompanyFilingsFetcher(
return IntrinioCompanyFilingsQueryParams(**transformed_params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioCompanyFilingsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -93,7 +93,7 @@ class IntrinioCompanyFilingsFetcher(
base_url = "https://api-v2.intrinio.com/companies"
query_str = get_querystring(query.model_dump(by_alias=True), ["symbol"])
url = f"{base_url}/{query.symbol}/filings?{query_str}&api_key={api_key}"
return get_data_many(url, "filings", **kwargs)
return await get_data_many(url, "filings", **kwargs)
@staticmethod
def transform_data(

39
openbb_platform/providers/intrinio/openbb_intrinio/models/company_news.py
View File

@@ -1,7 +1,6 @@
"""Intrinio Company News Model."""
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Any, Dict, List, Optional
@@ -10,8 +9,11 @@ from openbb_core.provider.standard_models.company_news import (
CompanyNewsData,
CompanyNewsQueryParams,
)
from openbb_core.provider.utils.helpers import get_querystring
from openbb_intrinio.utils.helpers import get_data_many
from openbb_core.provider.utils.helpers import (
ClientResponse,
async_requests,
get_querystring,
)
from pydantic import Field, field_validator
@@ -59,28 +61,33 @@ class IntrinioCompanyNewsFetcher(
return IntrinioCompanyNewsQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioCompanyNewsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
results: List[Dict] = []
def get_data(symbol):
base_url = "https://api-v2.intrinio.com/companies"
query_str = get_querystring(query.model_dump(by_alias=True), ["symbols"])
url = f"{base_url}/{symbol}/news?{query_str}&api_key={api_key}"
data = get_data_many(url, "news", **kwargs)
data = [{**d, "symbol": symbol} for d in data]
results.append(data)
base_url = "https://api-v2.intrinio.com/companies"
query_str = get_querystring(query.model_dump(by_alias=True), ["symbols"])
with ThreadPoolExecutor() as executor:
executor.map(get_data, [s.strip() for s in query.symbols.split(",")])
async def callback(response: ClientResponse, _: Any) -> List[Dict]:
"""Return the response."""
if response.status != 200:
return []
results = [item for sublist in results for item in sublist]
return results
symbol = response.url.parts[-2]
data = await response.json()
return [{**d, "symbol": symbol} for d in data.get("news", [])]
urls = [
f"{base_url}/{symbol}/news?{query_str}&api_key={api_key}"
for symbol in [s.strip() for s in query.symbols.split(",")]
]
return await async_requests(urls, callback, **kwargs)
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/currency_pairs.py
View File

@@ -47,7 +47,7 @@ class IntrinioCurrencyPairsFetcher(
return IntrinioCurrencyPairsQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioCurrencyPairsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -58,7 +58,7 @@ class IntrinioCurrencyPairsFetcher(
base_url = "https://api-v2.intrinio.com"
url = f"{base_url}/forex/pairs?api_key={api_key}"
return get_data_many(url, "pairs", **kwargs)
return await get_data_many(url, "pairs", **kwargs)
@staticmethod
def transform_data(

48
openbb_platform/providers/intrinio/openbb_intrinio/models/equity_historical.py
View File

@@ -10,8 +10,12 @@ from openbb_core.provider.standard_models.equity_historical import (
EquityHistoricalQueryParams,
)
from openbb_core.provider.utils.descriptions import QUERY_DESCRIPTIONS
from openbb_core.provider.utils.helpers import get_querystring
from openbb_intrinio.utils.helpers import get_data_one
from openbb_core.provider.utils.helpers import (
ClientResponse,
ClientSession,
async_requests,
get_querystring,
)
from pydantic import Field, PrivateAttr, model_validator
@@ -177,7 +181,7 @@ class IntrinioEquityHistoricalFetcher(
# pylint: disable=protected-access
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioEquityHistoricalQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -186,6 +190,9 @@ class IntrinioEquityHistoricalFetcher(
api_key = credentials.get("intrinio_api_key") if credentials else ""
base_url = f"https://api-v2.intrinio.com/securities/{query.symbol}/prices"
query_str = get_querystring(
query.model_dump(by_alias=True), ["symbol", "interval"]
)
if query._interval_size:
base_url += f"/intervals?interval_size={query._interval_size}"
@@ -194,26 +201,25 @@ class IntrinioEquityHistoricalFetcher(
base_url += f"?frequency={query._frequency}"
data_key = "stock_prices"
query_str = get_querystring(
query.model_dump(by_alias=True), ["symbol", "interval"]
)
async def callback(response: ClientResponse, session: ClientSession) -> list:
"""Return the response."""
init_response = await response.json()
all_data: list = init_response.get(data_key, [])
next_page = init_response.get("next_page", None)
while next_page:
url = response.url.update_query(next_page=next_page).human_repr()
response_data = await session.get_json(url)
all_data.extend(response_data.get(data_key, []))
next_page = response_data.get("next_page", None)
return all_data
url = f"{base_url}&{query_str}&api_key={api_key}"
data = get_data_one(url, **kwargs)
next_page = data.get("next_page", None)
data = data.get(data_key, [])
while next_page:
query_str = get_querystring(
query.model_dump(by_alias=True), ["symbol", "interval"]
)
url = f"{base_url}&{query_str}&next_page={next_page}&api_key={api_key}"
temp_data = get_data_one(url, **kwargs)
next_page = temp_data.get("next_page", None)
data.extend(temp_data.get(data_key, []))
return data
return await async_requests([url], callback, **kwargs)
# pylint: disable=unused-argument
@staticmethod

5
openbb_platform/providers/intrinio/openbb_intrinio/models/equity_info.py
View File

@@ -45,7 +45,7 @@ class IntrinioEquityInfoFetcher(
return IntrinioEquityInfoQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioEquityInfoQueryParams, # pylint: disable=unused-argument
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -54,7 +54,8 @@ class IntrinioEquityInfoFetcher(
api_key = credentials.get("intrinio_api_key") if credentials else ""
base_url = "https://api-v2.intrinio.com"
url = f"{base_url}/companies/{query.symbol}?api_key={api_key}"
return get_data_one(url, **kwargs)
return await get_data_one(url, **kwargs)
@staticmethod
def transform_data(

34
openbb_platform/providers/intrinio/openbb_intrinio/models/equity_quote.py
View File

@@ -1,6 +1,5 @@
"""Intrinio Equity Quote Model."""
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Any, Dict, List, Optional
@@ -9,7 +8,10 @@ from openbb_core.provider.standard_models.equity_quote import (
EquityQuoteData,
EquityQuoteQueryParams,
)
from openbb_intrinio.utils.helpers import get_data_one
from openbb_core.provider.utils.helpers import (
ClientResponse,
async_requests,
)
from openbb_intrinio.utils.references import SOURCES
from pydantic import Field, field_validator
@@ -115,26 +117,32 @@ class IntrinioEquityQuoteFetcher(
return IntrinioEquityQuoteQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioEquityQuoteQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
results: List[Dict] = []
def get_data(symbol):
base_url = "https://api-v2.intrinio.com"
url = f"{base_url}/securities/{symbol}/prices/realtime?source={query.source}&api_key={api_key}"
data = get_data_one(url, **kwargs)
data["symbol"] = symbol
results.append(data)
base_url = "https://api-v2.intrinio.com"
with ThreadPoolExecutor() as executor:
executor.map(get_data, [s.strip() for s in query.symbol.split(",")])
async def callback(response: ClientResponse, _: Any) -> dict:
"""Return the response."""
if response.status != 200:
return {}
return results
response_data = await response.json()
response_data["symbol"] = response.url.parts[-2]
return response_data
urls = [
f"{base_url}/securities/{s.strip()}/prices/realtime?source={query.source}&api_key={api_key}"
for s in query.symbol.split(",")
]
return await async_requests(urls, callback, **kwargs)
@staticmethod
def transform_data(

6
openbb_platform/providers/intrinio/openbb_intrinio/models/financial_attributes.py
View File

@@ -48,8 +48,8 @@ class IntrinioFinancialAttributesFetcher(
return IntrinioFinancialAttributesQueryParams(**transformed_params)
@staticmethod
def extract_data(
query: IntrinioFinancialAttributesQueryParams,
async def extract_data(
query: IntrinioFinancialAttributesQueryParams, # pylint: disable=unused-argument
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
@@ -64,7 +64,7 @@ class IntrinioFinancialAttributesFetcher(
url = f"{base_url}/historical_data/{query.symbol}/{query.tag}?{query_str}&api_key={api_key}"
# data = get_data_one(url).get("historical_data", [])
data = get_data_many(url, "historical_data")
data = await get_data_many(url, "historical_data")
return data

41
openbb_platform/providers/intrinio/openbb_intrinio/models/fred_series.py
View File

@@ -1,5 +1,6 @@
"""Intrinio FRED Series Model."""
import asyncio
from datetime import datetime
from typing import Any, Dict, List, Optional
@@ -9,8 +10,12 @@ from openbb_core.provider.standard_models.fred_series import (
SeriesData,
SeriesQueryParams,
)
from openbb_core.provider.utils.helpers import get_querystring
from openbb_intrinio.utils.helpers import async_get_data_one
from openbb_core.provider.utils.helpers import (
ClientResponse,
ClientSession,
async_requests,
get_querystring,
)
from pydantic import Field
@@ -79,31 +84,27 @@ class IntrinioFredSeriesFetcher(
f"?{query_str}&api_key={api_key}"
)
data = await async_get_data_one(url, query.limit, query.sleep, **kwargs)
async def callback(response: ClientResponse, session: ClientSession) -> dict:
"""Return the response."""
init_response = await response.json()
if query.all_pages:
all_data: list = data.get("historical_data", [])
next_page = data.get("next_page", None)
all_data: list = init_response.get("historical_data", [])
while next_page:
query_str = get_querystring(
query.model_dump(), ["symbol", "next_page", "all_pages", "sleep"]
)
if query.all_pages:
next_page = init_response.get("next_page", None)
while next_page:
if query.limit > 100:
await asyncio.sleep(query.sleep)
url = (
f"{base_url}/indices/economic/${query.symbol.replace('$', '')}/historical_data/level"
f"?{query_str}&next_page={next_page}&api_key={api_key}"
)
url = response.url.update_query(next_page=next_page).human_repr()
response_data = await session.get_json(url)
data = await async_get_data_one(url, query.limit, query.sleep, **kwargs)
all_data.extend(data.get("historical_data", []))
next_page = data.get("next_page", None)
all_data.extend(response_data.get("historical_data", []))
next_page = response_data.get("next_page", None)
return all_data
return data.get("historical_data", [])
return await async_requests([url], callback, **kwargs)
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/historical_attributes.py
View File

@@ -51,7 +51,7 @@ class IntrinioHistoricalAttributesFetcher(
return IntrinioHistoricalAttributesQueryParams(**transformed_params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioHistoricalAttributesQueryParams, # pylint: disable=unused-argument
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -63,7 +63,7 @@ class IntrinioHistoricalAttributesFetcher(
query_str = get_querystring(query.model_dump(by_alias=True), ["symbol", "tag"])
url = f"{base_url}/historical_data/{query.symbol}/{query.tag}?{query_str}&api_key={api_key}"
return get_data_many(url, "historical_data")
return await get_data_many(url, "historical_data")
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/historical_dividends.py
View File

@@ -69,7 +69,7 @@ class IntrinioHistoricalDividendsFetcher(
return IntrinioHistoricalDividendsQueryParams(**transformed_params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioHistoricalDividendsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -81,7 +81,7 @@ class IntrinioHistoricalDividendsFetcher(
query_str = get_querystring(query.model_dump(), ["symbol"])
url = f"{base_url}/{query.symbol}/prices/adjustments?{query_str}&api_key={api_key}"
return get_data_many(url, "stock_price_adjustments", **kwargs)
return await get_data_many(url, "stock_price_adjustments", **kwargs)
@staticmethod
def transform_data(

61
openbb_platform/providers/intrinio/openbb_intrinio/models/income_statement.py
View File

@@ -1,8 +1,6 @@
"""Intrinio Income Statement Model."""
from concurrent.futures import ThreadPoolExecutor
from itertools import repeat
from typing import Any, Dict, List, Optional
from openbb_core.provider.abstract.fetcher import Fetcher
@@ -10,6 +8,11 @@ from openbb_core.provider.standard_models.income_statement import (
IncomeStatementData,
IncomeStatementQueryParams,
)
from openbb_core.provider.utils.helpers import (
ClientResponse,
ClientSession,
async_requests,
)
from openbb_intrinio.utils.helpers import get_data_one
from pydantic import alias_generators
@@ -53,7 +56,7 @@ class IntrinioIncomeStatementFetcher(
return IntrinioIncomeStatementQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioIncomeStatementQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -61,12 +64,10 @@ class IntrinioIncomeStatementFetcher(
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
statement_code = "income_statement"
period_type = "FY" if query.period == "annual" else "QTR"
data_tags = ["ebit", "ebitda", "ebitdamargin"]
fundamentals_data: Dict = {}
data: List[Dict] = []
base_url = "https://api-v2.intrinio.com"
fundamentals_url_params = f"statement_code={statement_code}&type={period_type}"
fundamentals_url = (
@@ -74,47 +75,45 @@ class IntrinioIncomeStatementFetcher(
f"{fundamentals_url_params}&api_key={api_key}"
)
fundamentals_data = get_data_one(fundamentals_url, **kwargs).get(
"fundamentals", []
)
fundamentals_response = await get_data_one(fundamentals_url, **kwargs)
fundamentals_data = fundamentals_response.get("fundamentals", [])
fiscal_periods = [
f"{item['fiscal_year']}-{item['fiscal_period']}"
for item in fundamentals_data
]
fiscal_periods = fiscal_periods[: query.limit]
def get_financial_statement_data(period: str, data: List[Dict]) -> None:
statement_data: Dict = {}
calculations_data: List = []
async def callback(response: ClientResponse, session: ClientSession) -> Dict:
"""Return the response."""
intrinio_id = response.url.parts[-2].replace(statement_code, "calculations")
intrinio_id = f"{query.symbol}-{statement_code}-{period}"
statement_url = f"{base_url}/fundamentals/{intrinio_id}/standardized_financials?api_key={api_key}"
statement_data = get_data_one(statement_url, **kwargs)
statement_data = await response.json()
intrinio_id = f"{query.symbol}-calculations-{period}"
calculations_url = f"{base_url}/fundamentals/{intrinio_id}/standardized_financials?api_key={api_key}"
calculations_data = get_data_one(calculations_url, **kwargs).get(
"standardized_financials", []
)
calculations_response = await session.get(calculations_url)
calculations_data = await calculations_response.json()
calculations_data = [
item
for item in calculations_data
for item in calculations_data.get("standardized_financials", [])
if item["data_tag"]["tag"] in data_tags
]
data.append(
{
"date": statement_data["fundamental"]["end_date"],
"period": statement_data["fundamental"]["fiscal_period"],
"financials": statement_data["standardized_financials"]
+ calculations_data,
}
)
return {
"date": statement_data["fundamental"]["end_date"],
"period": statement_data["fundamental"]["fiscal_period"],
"financials": statement_data["standardized_financials"]
+ calculations_data,
}
with ThreadPoolExecutor() as executor:
executor.map(get_financial_statement_data, fiscal_periods, repeat(data))
intrinio_id = f"{query.symbol}-{statement_code}"
urls = [
f"{base_url}/fundamentals/{intrinio_id}-{period}/standardized_financials?api_key={api_key}"
for period in fiscal_periods
]
return data
return await async_requests(urls, callback, **kwargs)
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/insider_trading.py
View File

@@ -121,7 +121,7 @@ class IntrinioInsiderTradingFetcher(
return IntrinioInsiderTradingQueryParams(**transformed_params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioInsiderTradingQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -133,7 +133,7 @@ class IntrinioInsiderTradingFetcher(
query_str = get_querystring(query.model_dump(by_alias=True), ["symbol"])
url = f"{base_url}/{query.symbol}/insider_transaction_filings?{query_str}&api_key={api_key}"
return get_data_many(url, "transaction_filings", **kwargs)
return await get_data_many(url, "transaction_filings", **kwargs)
@staticmethod
def transform_data(

23
openbb_platform/providers/intrinio/openbb_intrinio/models/institutional_ownership.py
View File

@@ -1,6 +1,6 @@
"""Intrinio Institutional Ownership Model."""
from concurrent.futures import ThreadPoolExecutor
import asyncio
from typing import Any, Dict, List, Optional
from openbb_core.provider.abstract.fetcher import Fetcher
@@ -75,7 +75,7 @@ class IntrinioInstitutionalOwnershipFetcher(
return IntrinioInstitutionalOwnershipQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioInstitutionalOwnershipQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -90,19 +90,28 @@ class IntrinioInstitutionalOwnershipFetcher(
f"{base_url}/securities/{query.symbol}/institutional_ownership?"
f"{query_str}&api_key={api_key}"
)
data = get_data_many(url, "ownership", **kwargs)
def get_owner_name(item: Dict) -> Dict:
async def get_owner_name(item: Dict) -> Dict:
cik = item["owner_cik"]
cik_url = f"{base_url}/owners/{cik}?api_key={api_key}"
cik_data = get_data_one(cik_url, **kwargs)
cik_data = await get_data_one(cik_url, **kwargs)
owner_name = cik_data["owner_name"]
item["symbol"] = query.symbol
item["owner_name"] = owner_name
return item
with ThreadPoolExecutor() as executor:
data = list(executor.map(get_owner_name, data))
results = await asyncio.gather(
*[
get_owner_name(item)
for item in await get_data_many(url, "ownership", **kwargs)
],
return_exceptions=True,
)
for item in results:
if isinstance(item, Exception):
continue
data.append(item)
return data

24
openbb_platform/providers/intrinio/openbb_intrinio/models/key_metrics.py
View File

@@ -7,7 +7,10 @@ from openbb_core.provider.standard_models.key_metrics import (
KeyMetricsData,
KeyMetricsQueryParams,
)
from openbb_intrinio.utils.helpers import get_data_one
from openbb_core.provider.utils.helpers import (
ClientResponse,
async_requests,
)
from pydantic import Field
@@ -52,14 +55,13 @@ class IntrinioKeyMetricsFetcher(
return IntrinioKeyMetricsQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioKeyMetricsQueryParams, # pylint: disable=unused-argument
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> Dict:
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
data: Dict = {}
tags = [
"beta",
"volume",
@@ -70,10 +72,18 @@ class IntrinioKeyMetricsFetcher(
"pricetoearnings",
]
data["symbol"] = query.symbol
for tag in tags:
url = f"https://api-v2.intrinio.com/companies/{query.symbol}/data_point/{tag}?api_key={api_key}"
data[tag] = get_data_one(url).get("value")
urls = [
f"https://api-v2.intrinio.com/companies/{query.symbol}/data_point/{tag}?api_key={api_key}"
for tag in tags
]
async def callback(response: ClientResponse, _: Any) -> Dict:
"""Return the response."""
return {response.url.parts[-1]: await response.json()}
data: Dict = {"symbol": query.symbol}
for result in await async_requests(urls, callback, **kwargs):
data.update(result)
return data

4
openbb_platform/providers/intrinio/openbb_intrinio/models/latest_attributes.py
View File

@@ -36,7 +36,7 @@ class IntrinioLatestAttributesFetcher(
return IntrinioLatestAttributesQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioLatestAttributesQueryParams, # pylint: disable=unused-argument
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -44,7 +44,7 @@ class IntrinioLatestAttributesFetcher(
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
url = f"https://api-v2.intrinio.com/companies/{query.symbol}/data_point/{query.tag}?api_key={api_key}"
return get_data(url)
return await get_data(url)
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/market_indices.py
View File

@@ -67,7 +67,7 @@ class IntrinioMarketIndicesFetcher(
return IntrinioMarketIndicesQueryParams(**transformed_params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioMarketIndicesQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -79,7 +79,7 @@ class IntrinioMarketIndicesFetcher(
query_str = get_querystring(query.model_dump(by_alias=True), ["symbol", "tag"])
url = f"{base_url}/{query.symbol}/historical_data/{query.tag}?{query_str}&api_key={api_key}"
return get_data_many(url, "historical_data", **kwargs)
return await get_data_many(url, "historical_data", **kwargs)
@staticmethod
def transform_data(

46
openbb_platform/providers/intrinio/openbb_intrinio/models/options_chains.py
View File

@@ -1,12 +1,10 @@
"""Intrinio Options Chains Model."""
from concurrent.futures import ThreadPoolExecutor
from datetime import (
date as dateType,
datetime,
timedelta,
)
from itertools import repeat
from typing import Any, Dict, List, Optional
from dateutil import parser
@@ -15,6 +13,10 @@ from openbb_core.provider.standard_models.options_chains import (
OptionsChainsData,
OptionsChainsQueryParams,
)
from openbb_core.provider.utils.helpers import (
ClientResponse,
async_requests,
)
from openbb_intrinio.utils.helpers import get_data_many, get_weekday
from pydantic import Field, field_validator
@@ -70,7 +72,7 @@ class IntrinioOptionsChainsFetcher(
return IntrinioOptionsChainsQueryParams(**transform_params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioOptionsChainsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -78,39 +80,33 @@ class IntrinioOptionsChainsFetcher(
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
data: List = []
base_url = "https://api-v2.intrinio.com/options"
def get_options_chains(
expiration: str, data: List[IntrinioOptionsChainsData]
) -> None:
"""Return the data for the given expiration."""
url = (
f"{base_url}/chain/{query.symbol}/{expiration}/eod?"
f"date={query.date}&api_key={api_key}"
)
response = get_data_many(url, "chain", **kwargs)
data.extend(response)
def get_data(date: str) -> None:
"""Fetch data for a given date using ThreadPoolExecutor."""
async def get_urls(date: str) -> List[str]:
"""Return the urls for the given date."""
url = (
f"{base_url}/expirations/{query.symbol}/eod?"
f"after={date}&api_key={api_key}"
)
expirations = get_data_many(url, "expirations", **kwargs)
expiration = await get_data_many(url, "expirations", **kwargs)
return [
f"{base_url}/chain/{query.symbol}/{expiration}/eod?date={query.date}&api_key={api_key}"
for expiration in expiration
]
with ThreadPoolExecutor() as executor:
executor.map(get_options_chains, expirations, repeat(data))
async def callback(response: ClientResponse, _: Any) -> list:
"""Return the response."""
response_data = await response.json()
return response_data.get("chain", [])
date = get_weekday(query.date)
get_data(date)
results = await async_requests(await get_urls(date), callback, **kwargs)
if not data:
date = get_weekday(query.date - timedelta(days=1))
get_data(date)
if not results:
urls = await get_urls(get_weekday(date - timedelta(days=1)))
results = await async_requests(urls, callback, **kwargs)
return data
return results
@staticmethod
def transform_data(

4
openbb_platform/providers/intrinio/openbb_intrinio/models/options_unusual.py
View File

@@ -73,7 +73,7 @@ class IntrinioOptionsUnusualFetcher(
return IntrinioOptionsUnusualQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioOptionsUnusualQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -89,7 +89,7 @@ class IntrinioOptionsUnusualFetcher(
if query.symbol
else base_url + f"?source={query.source}&api_key={api_key}"
)
response = get_data_one(url, **kwargs)
response = await get_data_one(url, **kwargs)
if "trades" in response:
data = sorted(

8
openbb_platform/providers/intrinio/openbb_intrinio/models/search_attributes.py
View File

@@ -43,23 +43,23 @@ class IntrinioSearchAttributesFetcher(
return IntrinioSearchAttributesQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioSearchAttributesQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
data: List[Dict] = []
base_url = "https://api-v2.intrinio.com"
query_str = get_querystring(query.model_dump(by_alias=True), [])
url = f"{base_url}/data_tags/search?{query_str}&api_key={api_key}"
data = get_data_one(url).get("tags", [])
data = await get_data_one(url, **kwargs)
# Intrinio doesn't return the correct number of results when using the limit parameter
# Temporary fix until they fix it
data = data[: query.limit]
data = data.get("tags", [])[: query.limit]
return data

26
openbb_platform/providers/intrinio/openbb_intrinio/models/share_statistics.py
View File

@@ -8,7 +8,10 @@ from openbb_core.provider.standard_models.share_statistics import (
ShareStatisticsData,
ShareStatisticsQueryParams,
)
from openbb_intrinio.utils.helpers import get_data_one
from openbb_core.provider.utils.helpers import (
ClientResponse,
async_requests,
)
from pydantic import Field
@@ -53,26 +56,33 @@ class IntrinioShareStatisticsFetcher(
return IntrinioShareStatisticsQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioShareStatisticsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the Intrinio endpoint."""
api_key = credentials.get("intrinio_api_key") if credentials else ""
data: Dict = {}
data = {"symbol": query.symbol, "date": datetime.now().date()}
tags = [
"public_float",
"weightedavebasicdilutedsharesos",
"adjweightedavebasicdilutedsharesos",
]
for tag in tags:
url = f"https://api-v2.intrinio.com/companies/{query.symbol}/data_point/{tag}/number?api_key={api_key}"
data[tag] = get_data_one(url, **kwargs).get("value")
data["symbol"] = query.symbol
data["date"] = datetime.now().date()
urls = [
f"https://api-v2.intrinio.com/companies/{query.symbol}/data_point/{tag}/number?api_key={api_key}"
for tag in tags
]
async def callback(response: ClientResponse, _: Any) -> Dict:
"""Return the response."""
return {response.url.parts[-2]: await response.json()}
for result in await async_requests(urls, callback, **kwargs):
data.update(result)
return [data]
@staticmethod

4
openbb_platform/providers/intrinio/openbb_intrinio/models/world_news.py
View File

@@ -50,7 +50,7 @@ class IntrinioWorldNewsFetcher(
return IntrinioWorldNewsQueryParams(**params)
@staticmethod
def extract_data(
async def extract_data(
query: IntrinioWorldNewsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
@@ -61,7 +61,7 @@ class IntrinioWorldNewsFetcher(
base_url = "https://api-v2.intrinio.com"
url = f"{base_url}/companies/news?page_size={query.limit}&api_key={api_key}"
return get_data_many(url, "news", **kwargs)
return await get_data_many(url, "news", **kwargs)
@staticmethod
def transform_data(

62
openbb_platform/providers/intrinio/openbb_intrinio/utils/helpers.py
View File

@@ -7,10 +7,8 @@ from datetime import (
timedelta,
)
from io import StringIO
from typing import Any, List, Optional, TypeVar, Union
from typing import Any, Dict, List, Optional, TypeVar, Union
import requests
from openbb_core.provider import helpers
from openbb_core.provider.utils.errors import EmptyDataError
from openbb_core.provider.utils.helpers import (
ClientResponse,
@@ -18,7 +16,6 @@ from openbb_core.provider.utils.helpers import (
async_request,
)
from pydantic import BaseModel
from requests.exceptions import SSLError
T = TypeVar("T", bound=BaseModel)
@@ -59,30 +56,34 @@ def request(url: str) -> BasicResponse:
return BasicResponse(response)
def get_data(url: str, **kwargs: Any) -> Union[list, dict]:
"""Get data from Intrinio endpoint."""
try:
r: Union[requests.Response, BasicResponse] = helpers.make_request(url, **kwargs)
except SSLError:
r = request(url)
async def response_callback(
response: ClientResponse, _: ClientSession
) -> Union[dict, List[dict]]:
"""Callback for async_request."""
data = await response.json()
data = r.json()
if r.status_code != 200:
error = data.get("error")
message = data.get("message")
value = error or message
raise RuntimeError(f"Error in Intrinio request -> {value}")
if isinstance(data, dict) and response.status != 200:
if message := data.get("error", None) or data.get("message", None):
raise RuntimeError(f"Error in Intrinio request -> {message}")
if error := data.get("Error Message", None):
raise RuntimeError(f"Intrinio Error Message -> {error}")
if isinstance(data, (str, float)):
data = {"value": data}
if len(data) == 0:
if isinstance(data, list) and len(data) == 0:
raise EmptyDataError()
return data
def get_data_many(
async def get_data(url: str, **kwargs: Any) -> Union[list, dict]:
"""Get data from FMP endpoint."""
return await async_request(url, response_callback=response_callback, **kwargs)
async def get_data_many(
url: str, sub_dict: Optional[str] = None, **kwargs: Any
) -> List[dict]:
"""Get data from Intrinio endpoint and convert to list of schemas.
@@ -99,7 +100,7 @@ def get_data_many(
List[dict]
Dictionary of data.
"""
data = get_data(url, **kwargs)
data = await get_data(url, **kwargs)
if sub_dict and isinstance(data, dict):
data = data.get(sub_dict, [])
if isinstance(data, dict):
@@ -107,9 +108,9 @@ def get_data_many(
return data
def get_data_one(url: str, **kwargs: Any) -> dict:
async def get_data_one(url: str, **kwargs: Any) -> Dict[str, Any]:
"""Get data from Intrinio endpoint and convert to schema."""
data = get_data(url, **kwargs)
data = await get_data(url, **kwargs)
if isinstance(data, list):
if len(data) == 0:
raise ValueError("Expected dict, got empty list")
@@ -129,29 +130,14 @@ def get_weekday(date: dateType) -> str:
return date.strftime("%Y-%m-%d")
async def response_callback(response: ClientResponse, _: ClientSession) -> dict:
"""Return the response."""
data: dict = await response.json()
if message := data.get("error", None) or data.get("message", None):
raise RuntimeError(f"Error in Intrinio request -> {message}")
if error := data.get("Error Message", None):
raise RuntimeError(f"Intrinio Error Message -> {error}")
return data
async def async_get_data_one(
url: str, limit: int = 1, sleep: float = 1, **kwargs: Any
) -> dict:
) -> Union[list, dict]:
if limit > 100:
await asyncio.sleep(sleep)
try:
data: dict = await async_request(
url, response_callback=response_callback, **kwargs
)
data = await get_data(url, **kwargs)
except Exception as e:
if "limit" not in str(e):
raise e

3257
openbb_platform/providers/intrinio/tests/record/http/test_intrinio_fetchers/test_intrinio_options_chains_fetcher.yaml vendored
View File

File diff suppressed because it is too large Load Diff