1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
|
"""Trading Economics Economic Calendar Model."""
from datetime import datetime
from typing import Any, Dict, List, Optional, Union
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.market_historical import (
MarketHistoricalData,
MarketHistoricalQueryParams,
)
from openbb_core.provider.utils.helpers import (
ClientResponse,
amake_requests,
get_querystring,
)
from pandas import to_datetime
from pydantic import field_validator
class TEMarketHistoricalQueryParams(MarketHistoricalQueryParams):
"""Trading Economics Market Historical Query.
Source: https://docs.tradingeconomics.com/markets/historical/
"""
__alias_dict__ = {"start_date": "d1", "end_date": "d2"}
__json_schema_extra__ = {"symbol": ["multiple_items_allowed"]}
class TEMarketHistorical(MarketHistoricalData):
"""Trading Economics Market Historical Data."""
__alias_dict__ = {
"symbol": "Symbol",
"date": "Date",
"open": "Open",
"high": "High",
"low": "Low",
"close": "Close",
}
@field_validator("date", mode="before")
@classmethod
def validate_date(cls, v: str) -> datetime:
"""Return formatted datetime."""
return to_datetime(v, utc=True, dayfirst=True)
class TEMarketHistoricalFetcher(
Fetcher[
TEMarketHistoricalQueryParams,
List[TEMarketHistorical],
]
):
"""Transform the query, extract and transform the data from the Trading Economics endpoints."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> TEMarketHistoricalQueryParams:
"""Transform the query params."""
return TEMarketHistoricalQueryParams(**params)
@staticmethod
async def aextract_data(
query: TEMarketHistoricalQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> Union[dict, List[dict]]:
"""Return the raw data from the TE endpoint."""
api_key = credentials.get("tradingeconomics_api_key") if credentials else ""
symbols = query.symbol.split(",")
base_url = "https://api.tradingeconomics.com/markets/historical"
query_str = get_querystring(query.model_dump(), ["symbol"])
urls = []
for symbol in symbols:
urls.append(f"{base_url}/{symbol}?{query_str}&c={api_key}")
async def callback(response: ClientResponse, _: Any) -> Union[dict, List[dict]]:
"""Return the response."""
if response.status != 200:
raise RuntimeError(f"Error in TE request -> {await response.text()}")
return await response.json()
return await amake_requests(urls, response_callback=callback, **kwargs)
# pylint: disable=unused-argument
@staticmethod
def transform_data(
query: TEMarketHistoricalQueryParams, data: List[Dict], **kwargs: Any
) -> List[TEMarketHistorical]:
"""Return the transformed data."""
return [TEMarketHistorical.model_validate(d) for d in data]
|
