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
|
"""FMP Price Performance Model."""
# pylint: disable=unused-argument
from typing import Any, Dict, List, Optional
from warnings import warn
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.recent_performance import (
RecentPerformanceData,
RecentPerformanceQueryParams,
)
from openbb_fmp.utils.helpers import create_url, get_data_many
from pydantic import Field, model_validator
class FMPPricePerformanceQueryParams(RecentPerformanceQueryParams):
"""FMP Price Performance Query.
Source: https://site.financialmodelingprep.com/developer/docs/stock-split-calendar-api/
"""
__json_schema_extra__ = {"symbol": ["multiple_items_allowed"]}
class FMPPricePerformanceData(RecentPerformanceData):
"""FMP Price Performance Data."""
symbol: str = Field(description="The ticker symbol.")
__alias_dict__ = {
"one_day": "1D",
"one_week": "5D",
"one_month": "1M",
"three_month": "3M",
"six_month": "6M",
"one_year": "1Y",
"three_year": "3Y",
"five_year": "5Y",
"ten_year": "10Y",
}
@model_validator(mode="before")
@classmethod
def replace_zero(cls, values): # pylint: disable=no-self-argument
"""Replace zero with None and convert percents to normalized values."""
if isinstance(values, dict):
for k, v in values.items():
if k != "symbol":
values[k] = None if v == 0 else float(v) / 100
return values
class FMPPricePerformanceFetcher(
Fetcher[
FMPPricePerformanceQueryParams,
List[FMPPricePerformanceData],
]
):
"""Transform the query, extract and transform the data from the FMP endpoints."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> FMPPricePerformanceQueryParams:
"""Transform the query params."""
return FMPPricePerformanceQueryParams(**params)
@staticmethod
async def aextract_data(
query: FMPPricePerformanceQueryParams,
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 ""
url = create_url(
version=3,
endpoint=f"stock-price-change/{query.symbol}",
api_key=api_key,
exclude=["symbol"],
)
return await get_data_many(url, **kwargs)
@staticmethod
def transform_data(
query: FMPPricePerformanceQueryParams,
data: List[Dict],
**kwargs: Any,
) -> List[FMPPricePerformanceData]:
"""Return the transformed data."""
symbols = query.symbol.split(",")
if len(data) != len(symbols):
data_symbols = [d["symbol"] for d in data]
missing_symbols = [
symbol for symbol in symbols if symbol not in data_symbols
]
warn(f"Missing data for symbols: {missing_symbols}")
return [FMPPricePerformanceData.model_validate(i) for i in data]
|
