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
|
"""YFinance Price Target Consensus Model"""
# pylint: disable=unused-argument
import asyncio
import warnings
from typing import Any, Dict, List, Optional
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.price_target_consensus import (
PriceTargetConsensusData,
PriceTargetConsensusQueryParams,
)
from pydantic import Field
from yfinance import Ticker
_warn = warnings.warn
class YFinancePriceTargetConsensusQueryParams(PriceTargetConsensusQueryParams):
"""YFinance Price Target Consensus query params."""
class YFinancePriceTargetConsensusData(PriceTargetConsensusData):
"""YFinance Price Target Consensus Data."""
__alias_dict__ = {
"target_high": "targetHighPrice",
"target_low": "targetLowPrice",
"target_consensus": "targetMeanPrice",
"target_median": "targetMedianPrice",
"recommendation": "recommendationKey",
"recommendation_mean": "recommendationMean",
"number_of_analysts": "numberOfAnalystOpinions",
"current_price": "currentPrice",
}
recommendation: Optional[str] = Field(
default=None,
description="Recommendation - buy, sell, etc.",
)
recommendation_mean: Optional[float] = Field(
default=None,
description="Mean recommendation score where 1 is strong buy and 5 is strong sell.",
)
current_price: Optional[float] = Field(
default=None,
description="Current price of the stock.",
)
class YFinancePriceTargetConsensusFetcher(
Fetcher[
YFinancePriceTargetConsensusQueryParams, List[YFinancePriceTargetConsensusData]
]
):
"""YFinance Price Target Consensus fetcher."""
@staticmethod
def transform_query(
params: Dict[str, Any]
) -> YFinancePriceTargetConsensusQueryParams:
"""Transform the query."""
return YFinancePriceTargetConsensusQueryParams(**params)
@staticmethod
async def aextract_data(
query: YFinancePriceTargetConsensusQueryParams,
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Extract the raw data from YFinance"""
symbols = query.symbol.split(",")
results = []
fields = [
"symbol",
"currentPrice",
"targetHighPrice",
"targetLowPrice",
"targetMeanPrice",
"targetMedianPrice",
"recommendationMean",
"recommendationKey",
"numberOfAnalystOpinions",
]
async def get_one(symbol):
"""Get the data for one ticker symbol."""
result = {}
ticker = {}
try:
ticker = Ticker(symbol).get_info()
except Exception as e:
_warn(f"Error getting data for {symbol}: {e}")
if ticker:
for field in fields:
if field in ticker:
result[field] = ticker.get(field, None)
if result:
results.append(result)
tasks = [get_one(symbol) for symbol in symbols]
await asyncio.gather(*tasks)
return results
@staticmethod
def transform_data(
query: YFinancePriceTargetConsensusQueryParams,
data: List[Dict],
**kwargs: Any,
) -> List[YFinancePriceTargetConsensusData]:
"""Transform the data."""
return [YFinancePriceTargetConsensusData.model_validate(d) for d in data]
|
