diff options
Diffstat (limited to 'openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py')
| -rw-r--r-- | openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py | 268 |
1 files changed, 266 insertions, 2 deletions
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py index f13ba058623..44b92a2c6c5 100644 --- a/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py @@ -1,5 +1,6 @@ """Charting router.""" +import json from typing import Any, Dict, Tuple import pandas as pd @@ -9,6 +10,10 @@ from openbb_core.app.utils import basemodel_to_df from openbb_charting.core.chart_style import ChartStyle from openbb_charting.core.openbb_figure import OpenBBFigure from openbb_charting.core.plotly_ta.ta_class import PlotlyTA +from openbb_charting.query_params import ( + FredSeriesChartQueryParams, + TechnicalConesChartQueryParams, +) CHART_FORMAT = ChartFormat.plotly @@ -179,7 +184,9 @@ def technical_ema(**kwargs) -> Tuple["OpenBBFigure", Dict[str, Any]]: return _ta_ma(ma_type, **kwargs) -def technical_cones(**kwargs) -> Tuple["OpenBBFigure", Dict[str, Any]]: +def technical_cones( + **kwargs: TechnicalConesChartQueryParams, +) -> Tuple["OpenBBFigure", Dict[str, Any]]: """Volatility Cones Chart.""" data = kwargs.get("data") @@ -187,7 +194,7 @@ def technical_cones(**kwargs) -> Tuple["OpenBBFigure", Dict[str, Any]]: if isinstance(data, pd.DataFrame) and not data.empty and "window" in data.columns: df_ta = data.set_index("window") else: - df_ta = basemodel_to_df(kwargs["obbject_item"], index="window") + df_ta = basemodel_to_df(kwargs["obbject_item"], index="window") # type: ignore df_ta.columns = [col.title().replace("_", " ") for col in df_ta.columns] @@ -276,3 +283,260 @@ def technical_cones(**kwargs) -> Tuple["OpenBBFigure", Dict[str, Any]]: content = fig.to_plotly_json() return fig, content + + +def economy_fred_series( + **kwargs: FredSeriesChartQueryParams, +) -> Tuple["OpenBBFigure", Dict[str, Any]]: + """FRED Series Chart.""" + + ytitle_dict = { + "chg": "Change", + "ch1": "Change From Year Ago", + "pch": "Percent Change", + "pc1": "Percent Change From Year Ago", + "pca": "Compounded Annual Rate Of Change", + "cch": "Continuously Compounded Rate Of Change", + "cca": "Continuously Compounded Annual Rate Of Change", + "log": "Natural Log", + } + + colors = [ + "#1f77b4", + "#7f7f7f", + "#ff7f0e", + "#2ca02c", + "#d62728", + "#9467bd", + "#8c564b", + "#e377c2", + "#7f7f7f", + "#bcbd22", + "#17becf", + ] + + provider = kwargs.get("provider") + + if provider != "fred": + raise RuntimeError( + f"This charting method does not support {provider}. Supported providers: fred." + ) + + columns = basemodel_to_df(kwargs["obbject_item"], index=None).columns.to_list() # type: ignore + + allow_unsafe = kwargs.get("allow_unsafe", False) + dropnan = kwargs.get("dropna", True) + normalize = kwargs.get("normalize", False) + + data_cols = [] + data = kwargs.get("data") + + if isinstance(data, pd.DataFrame) and not data.empty: + data_cols = data.columns.to_list() + df_ta = data + + else: + df_ta = basemodel_to_df(kwargs["obbject_item"], index="date") # type: ignore + + # Check for unsupported external data injection. + if allow_unsafe is False and data_cols: + for data_col in data_cols: + if data_col not in columns: + raise RuntimeError( + f"Column '{data_col}' was not found in the original data." + + " External data injection is not supported unless `allow_unsafe = True`." + ) + + # Align the data so each column has the same index and length. + if dropnan: + df_ta = df_ta.dropna(how="any") + + if df_ta.empty or len(df_ta) < 2: + raise ValueError( + "No data is left after dropping NaN values. Try setting `dropnan = False`," + + " or use the `frequency` parameter on request ." + ) + + columns = df_ta.columns.to_list() + + def z_score_standardization(data: pd.Series) -> pd.Series: + """Z-Score Standardization Method.""" + return (data - data.mean()) / data.std() + + if normalize: + df_ta = df_ta.apply(z_score_standardization) + + # Extract the metadata from the warnings. + warnings = kwargs.get("warnings") + metadata = json.loads(warnings[0].message) if warnings else {} # type: ignore + + # Check if the request was transformed by the FRED API. + params = kwargs["extra_params"] if kwargs.get("extra_params") else {} + has_params = hasattr(params, "transform") and params.transform is not None # type: ignore + + # Get a unique list of all units of measurement in the DataFrame. + y_units = list({metadata.get(col).get("units") for col in columns if col in metadata}) # type: ignore + + if len(y_units) > 2 and has_params is False and allow_unsafe is True: + raise RuntimeError( + "This method supports up to 2 y-axis units." + + " Please use the 'transform' parameter, in the data request," + + " to compare all series on the same scale, or set `normalize = True`." + + " Override this error by setting `allow_unsafe = True`." + ) + + y1_units = y_units[0] + + y1title = y1_units + + y2title = y_units[1] if len(y_units) > 1 else None + + xtitle = "" + + # If the request was transformed, the y-axis will be shared under these conditions. + if has_params and any( + i in params.transform for i in ["pc1", "pch", "pca", "cch", "cca", "log"] # type: ignore + ): + y1title = "Log" if params.transform == "Log" else "Percent" # type: ignore + y2title = None + + # Set the title for the chart. + if kwargs.get("title"): + title = kwargs.get("title") + else: + if metadata.get(columns[0]): + title = metadata.get(columns[0]).get("title") if len(columns) == 1 else "FRED Series" # type: ignore + else: + title = "FRED Series" + transform_title = ytitle_dict.get(params.transform) if has_params is True else "" # type: ignore + title = f"{title} - {transform_title}" if transform_title else title + + # Define this to use as a check. + y3title = "" + + # Create the figure object with subplots. + fig = OpenBBFigure().create_subplots( + rows=1, cols=1, shared_xaxes=True, shared_yaxes=False + ) + fig.update_layout(ChartStyle().plotly_template.get("layout", {})) + text_color = "black" if ChartStyle().plt_style == "light" else "white" + + # For each series in the DataFrame, add a scatter plot. + for i, col in enumerate(df_ta.columns): + + # Check if the y-axis should be shared for this series. + on_y1 = ( + ( + metadata.get(col).get("units") == y1_units + or y2title is None # type: ignore + ) + if metadata.get(col) + else False + ) + if normalize: + on_y1 = True + yaxes = "y2" if not on_y1 else "y1" + on_y3 = not metadata.get(col) and normalize is False + if on_y3: + yaxes = "y3" + y3title = df_ta[col].name + fig.add_scatter( + x=df_ta.index, + y=df_ta[col], + name=df_ta[col].name, + mode="lines", + hovertemplate=f"{df_ta[col].name}: %{{y}}<extra></extra>", + line=dict(width=1, color=colors[i % len(colors)]), + yaxis=yaxes, + ) + + # Set the y-axis titles, if supplied. + if kwargs.get("y1title"): + y1title = kwargs.get("y1title") + if kwargs.get("y2title") and y2title is not None: + y2title = kwargs.get("y2title") + # Set the x-axis title, if suppiled. + if kwargs.get("xtitle"): + xtitle = kwargs.get("xtitle") + # If the data was normalized, set the title to reflect this. + if normalize: + y1title = None + y2title = None + y3title = None + title = f"{title} - Normalized" + + # Now update the layout of the complete figure. + fig.update_layout( + title=dict(text=title, x=0.5, font=dict(size=16)), + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + font=dict(color=text_color), + legend=dict( + orientation="h", + yanchor="bottom", + xanchor="right", + y=1.02, + x=1, + bgcolor="rgba(0,0,0,0)", + ), + yaxis=( + dict( + ticklen=0, + side="right", + title=dict(text=y1title, standoff=30, font=dict(size=18)), + tickfont=dict(size=14), + anchor="x", + ) + if y1title + else None + ), + yaxis2=( + dict( + overlaying="y", + side="left", + ticklen=0, + showgrid=False, + title=dict( + text=y2title if y2title else None, standoff=10, font=dict(size=18) + ), + tickfont=dict(size=14), + anchor="x", + ) + if y2title + else None + ), + yaxis3=( + dict( + overlaying="y", + side="left", + ticklen=0, + position=0, + showgrid=False, + showticklabels=True, + title=( + dict(text=y3title, standoff=10, font=dict(size=16)) + if y3title + else None + ), + tickfont=dict(size=12, color="rgba(128,128,128,0.75)"), + anchor="free", + ) + if y3title + else None + ), + xaxis=dict( + ticklen=0, + showgrid=False, + title=( + dict(text=xtitle, standoff=30, font=dict(size=18)) if xtitle else None + ), + domain=[0.095, 0.95] if y3title else None, + ), + margin=dict(r=25, l=25) if normalize is False else None, + autosize=True, + dragmode="pan", + ) + + content = fig.to_plotly_json() + + return fig, content |