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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
"""FMP Major Indices end of day fetcher."""
from datetime import datetime
from typing import Any, Dict, List, Literal, Optional
from dateutil.relativedelta import relativedelta
from openbb_fmp.utils.helpers import get_data_many
from openbb_provider.abstract.fetcher import Fetcher
from openbb_provider.standard_models.major_indices_historical import (
MajorIndicesHistoricalData,
MajorIndicesHistoricalQueryParams,
)
from openbb_provider.utils.helpers import get_querystring
from pydantic import Field, NonNegativeInt, validator
class FMPMajorIndicesHistoricalQueryParams(MajorIndicesHistoricalQueryParams):
"""FMP Major Indices end of day Query.
Source: https://site.financialmodelingprep.com/developer/docs/historical-index-price-api/
"""
timeseries: Optional[NonNegativeInt] = Field(
default=None, description="Number of days to look back."
)
interval: Literal[
"1min", "5min", "15min", "30min", "1hour", "4hour", "1day"
] = Field(default="1day", description="Data granularity.")
class FMPMajorIndicesHistoricalData(MajorIndicesHistoricalData):
"""FMP Major Indices end of day Data."""
adj_close: Optional[float] = Field(
description="Adjusted Close Price of the symbol.",
default=None,
)
unadjusted_volume: Optional[float] = Field(
description="Unadjusted volume of the symbol.",
default=None,
)
change: Optional[float] = Field(
description="Change in the price of the symbol from the previous day.",
default=None,
)
change_percent: Optional[float] = Field(
description=r"Change % in the price of the symbol.",
default=None,
)
label: Optional[str] = Field(
description="Human readable format of the date.", default=None
)
change_over_time: Optional[float] = Field(
description=r"Change % in the price of the symbol over a period of time.",
default=None,
)
@validator("date", pre=True, check_fields=False)
def date_validate(cls, v) -> datetime: # pylint: disable=E0213
"""Return the date as a datetime object."""
try:
return datetime.strptime(v, "%Y-%m-%d %H:%M:%S")
except Exception:
return datetime.strptime(v, "%Y-%m-%d")
class FMPMajorIndicesHistoricalFetcher(
Fetcher[
FMPMajorIndicesHistoricalQueryParams,
List[FMPMajorIndicesHistoricalData],
]
):
"""Transform the query, extract and transform the data from the FMP endpoints."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> FMPMajorIndicesHistoricalQueryParams:
"""Transform the query params."""
transformed_params = params
now = datetime.now().date()
if params.get("start_date") is None:
transformed_params["start_date"] = now - relativedelta(years=1)
if params.get("end_date") is None:
transformed_params["end_date"] = now
return FMPMajorIndicesHistoricalQueryParams(**transformed_params)
@staticmethod
def extract_data(
query: FMPMajorIndicesHistoricalQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the FMP endpoint."""
api_key = credentials.get("fmp_api_key") if credentials else ""
base_url = "https://financialmodelingprep.com/api/v3"
query_str = (
get_querystring(query.dict(), ["symbol"])
.replace("start_date", "from")
.replace("end_date", "to")
)
url_params = f"{query.symbol}?{query_str}&apikey={api_key}"
url = f"{base_url}/historical-chart/{query.interval}/{url_params}"
if query.interval == "1day":
url = f"{base_url}/historical-price-full/index/{url_params}"
return get_data_many(url, "historical", **kwargs)
@staticmethod
def transform_data(data: List[Dict]) -> List[FMPMajorIndicesHistoricalData]:
"""Return the transformed data."""
return [FMPMajorIndicesHistoricalData(**d) for d in data]
|
