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
|
"""Yahoo Finance Futures Historical Price Model."""
# pylint: disable=unused-argument
# ruff: noqa: SIM105
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional
from dateutil.relativedelta import relativedelta
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.futures_historical import (
FuturesHistoricalData,
FuturesHistoricalQueryParams,
)
from openbb_core.provider.utils.descriptions import QUERY_DESCRIPTIONS
from openbb_core.provider.utils.errors import EmptyDataError
from openbb_yfinance.utils.helpers import get_futures_data, yf_download
from openbb_yfinance.utils.references import INTERVALS, MONTHS, PERIODS
from pandas import Timestamp, to_datetime
from pydantic import Field, field_validator
class YFinanceFuturesHistoricalQueryParams(FuturesHistoricalQueryParams):
"""Yahoo Finance Futures historical Price Query.
Source: https://finance.yahoo.com/crypto/
"""
interval: Optional[INTERVALS] = Field(default="1d", description="Data granularity.")
period: Optional[PERIODS] = Field(
default=None, description=QUERY_DESCRIPTIONS.get("period", "")
)
class YFinanceFuturesHistoricalData(FuturesHistoricalData):
"""Yahoo Finance Futures Historical Price Data."""
@field_validator("date", mode="before", check_fields=False)
@classmethod
def date_validate(cls, v):
"""Return datetime object from string."""
if isinstance(v, Timestamp):
return v.to_pydatetime()
return v
class YFinanceFuturesHistoricalFetcher(
Fetcher[
YFinanceFuturesHistoricalQueryParams,
List[YFinanceFuturesHistoricalData],
]
):
"""Transform the query, extract and transform the data from the Yahoo Finance endpoints."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> YFinanceFuturesHistoricalQueryParams:
"""Transform the query. Setting the start and end dates for a 1 year period."""
transformed_params = params.copy()
symbols = params["symbol"].split(",")
new_symbols = []
futures_data = get_futures_data()
for symbol in symbols:
if params.get("expiration"):
expiry_date = datetime.strptime(
transformed_params["expiration"], "%Y-%m"
)
if "." not in symbol:
exchange = futures_data[futures_data["Ticker"] == symbol][
"Exchange"
].values[0]
new_symbol = (
f"{symbol}{MONTHS[expiry_date.month]}{str(expiry_date.year)[-2:]}.{exchange}"
if "." not in symbol
else symbol
)
new_symbols.append(new_symbol)
else:
new_symbols.append(symbol)
formatted_symbols = []
for s in new_symbols:
if "." not in s.upper() and "=F" not in s.upper():
formatted_symbols.append(f"{s.upper()}=F")
else:
formatted_symbols.append(s.upper())
transformed_params["symbol"] = ",".join(formatted_symbols)
now = datetime.now()
if params.get("start_date") is None:
transformed_params["start_date"] = (now - relativedelta(years=1)).strftime(
"%Y-%m-%d"
)
if params.get("end_date") is None:
transformed_params["end_date"] = now.strftime("%Y-%m-%d")
return YFinanceFuturesHistoricalQueryParams(**transformed_params)
@staticmethod
def extract_data(
query: YFinanceFuturesHistoricalQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Return the raw data from the Yahoo Finance endpoint."""
data = yf_download(
query.symbol,
start=query.start_date,
end=query.end_date,
interval=query.interval, # type: ignore
prepost=True,
auto_adjust=False,
actions=False,
)
if data.empty:
raise EmptyDataError()
days = (
1
if query.interval in ["1m", "2m", "5m", "15m", "30m", "60m", "1h", "90m"]
else 0
)
if "date" in data.columns:
data.set_index("date", inplace=True)
data.index = to_datetime(data.index)
if query.start_date:
data = data[
(data.index >= to_datetime(query.start_date))
& (data.index <= to_datetime(query.end_date + timedelta(days=days)))
]
data.reset_index(inplace=True)
data.rename(columns={"index": "date"}, inplace=True)
return data.to_dict("records")
@staticmethod
def transform_data(
query: YFinanceFuturesHistoricalQueryParams,
data: List[Dict],
**kwargs: Any,
) -> List[YFinanceFuturesHistoricalData]:
"""Transform the data to the standard format."""
return [YFinanceFuturesHistoricalData.model_validate(d) for d in data]
|
