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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
"""AlphaVantage Historical EPS Model."""
# pylint: disable=unused-argument
from datetime import date as dateType
from typing import Any, Dict, List, Literal, Optional, Union
from warnings import warn
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.historical_eps import (
HistoricalEpsData,
HistoricalEpsQueryParams,
)
from openbb_core.provider.utils.descriptions import QUERY_DESCRIPTIONS
from openbb_core.provider.utils.errors import EmptyDataError
from openbb_core.provider.utils.helpers import (
ClientResponse,
ClientSession,
amake_requests,
)
from pydantic import Field, field_validator
class AlphaVantageHistoricalEpsQueryParams(HistoricalEpsQueryParams):
"""
AlphaVantage Historical EPS Query Params.
Source: https://www.alphavantage.co/documentation/#earnings
"""
__json_schema_extra__ = {"symbol": {"multiple_items_allowed": True}}
period: Literal["annual", "quarter"] = Field(
default="quarter", description=QUERY_DESCRIPTIONS.get("period", "")
)
limit: Optional[int] = Field(
default=None, description=QUERY_DESCRIPTIONS.get("limit", "")
)
class AlphaVantageHistoricalEpsData(HistoricalEpsData):
"""AlphaVantage Historical EPS Data."""
__alias_dict__ = {
"date": "fiscalDateEnding",
"eps_actual": "reportedEPS",
"eps_estimated": "estimatedEPS",
"surprise_percent": "surprisePercentage",
"reported_date": "reportedDate",
}
surprise: Optional[float] = Field(
default=None,
description="Surprise in EPS (Actual - Estimated).",
)
surprise_percent: Optional[Union[float, str]] = Field(
default=None,
description="EPS surprise as a normalized percent.",
json_schema_extra={"x-unit_measurement": "percent", "x-frontend_multiply": 100},
)
reported_date: Optional[dateType] = Field(
default=None,
description="Date of the earnings report.",
)
@field_validator(
"eps_estimated",
"eps_actual",
"surprise",
mode="before",
check_fields=False,
)
@classmethod
def validate_null(cls, v):
"""Clean None returned as a string."""
return None if str(v).strip() == "None" or str(v) == "0" else v
@field_validator("surprise_percent", mode="before", check_fields=False)
@classmethod
def normalize_percent(cls, v):
"""Normalize percent values."""
if isinstance(v, str) and v == "None" or str(v) == "0":
return None
return float(v) / 100
class AVHistoricalEpsFetcher(
Fetcher[AlphaVantageHistoricalEpsQueryParams, List[AlphaVantageHistoricalEpsData]]
):
"""AlphaVantage Historical EPS Fetcher."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> AlphaVantageHistoricalEpsQueryParams:
"""Transform the query params."""
return AlphaVantageHistoricalEpsQueryParams(**params)
@staticmethod
async def aextract_data(
query: AlphaVantageHistoricalEpsQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the AlphaVantage endpoint."""
api_key = credentials.get("alpha_vantage_api_key") if credentials else ""
BASE_URL = "https://www.alphavantage.co/query?function=EARNINGS&"
# We are allowing multiple symbols to be passed in the query, so we need to handle that.
symbols = query.symbol.split(",")
urls = [f"{BASE_URL}symbol={symbol}&apikey={api_key}" for symbol in symbols]
results: List = []
messages: List = []
# We need to make a custom callback function for this async request.
async def response_callback(response: ClientResponse, _: ClientSession):
"""Response callback function."""
symbol = response.url.query.get("symbol", None)
data = await response.json()
target = (
"annualEarnings" if query.period == "annual" else "quarterlyEarnings"
)
message = data.get("Information", "")
if message:
messages.append(message)
warn(f"Symbol Error for {symbol}: {message}")
result: List = []
# If data is returned, append it to the results list.
if data:
result = [
{
"symbol": symbol,
**d,
}
for d in data.get(target, []) # type: ignore
]
if query.limit is not None:
results.extend(result[: query.limit])
else:
results.extend(result)
# If no data is returned, raise a warning and move on to the next symbol.
if not data:
warn(f"Symbol Error: No data found for {symbol}")
await amake_requests(urls, response_callback, **kwargs) # type: ignore
if not results:
raise EmptyDataError(f"No data was returned -> \n{messages[-1]}")
return results
@staticmethod
def transform_data(
query: AlphaVantageHistoricalEpsQueryParams,
data: List[Dict],
**kwargs: Any,
) -> List[AlphaVantageHistoricalEpsData]:
"""Transform the raw data into the standard model."""
if not data:
raise EmptyDataError("No data found.")
return [AlphaVantageHistoricalEpsData.model_validate(d) for d in data]
|
