diff --git a/openbb_platform/core/openbb_core/app/model/obbject.py b/openbb_platform/core/openbb_core/app/model/obbject.py index 75078ff2919..d9600746092 100644 --- a/openbb_platform/core/openbb_core/app/model/obbject.py +++ b/openbb_platform/core/openbb_core/app/model/obbject.py @@ -72,7 +72,7 @@ class OBBject(Tagged, Generic[T]): _standard_params: Optional[Dict[str, Any]] = PrivateAttr( default_factory=dict, ) - _standard_params: Optional[Dict[str, Any]] = PrivateAttr( + _extra_params: Optional[Dict[str, Any]] = PrivateAttr( default_factory=dict, ) diff --git a/openbb_platform/extensions/crypto/openbb_crypto/crypto_views.py b/openbb_platform/extensions/crypto/openbb_crypto/crypto_views.py new file mode 100644 index 00000000000..15e2f92b34c --- /dev/null +++ b/openbb_platform/extensions/crypto/openbb_crypto/crypto_views.py @@ -0,0 +1,17 @@ +"""Views for the crypto Extension.""" + +from typing import Any, Dict, Tuple + +from openbb_charting.charts.price_historical import price_historical +from openbb_charting.core.openbb_figure import OpenBBFigure + + +class CryptoViews: + """Crypto Views.""" + + @staticmethod + def crypto_price_historical( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Crypto Price Historical Chart.""" + return price_historical(**kwargs) diff --git a/openbb_platform/extensions/crypto/pyproject.toml b/openbb_platform/extensions/crypto/pyproject.toml index e3c74377c1f..818fa9e853d 100644 --- a/openbb_platform/extensions/crypto/pyproject.toml +++ b/openbb_platform/extensions/crypto/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] crypto = "openbb_crypto.crypto_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +crypto = "openbb_crypto.crypto_views:CryptoViews" diff --git a/openbb_platform/extensions/currency/openbb_currency/currency_views.py b/openbb_platform/extensions/currency/openbb_currency/currency_views.py new file mode 100644 index 00000000000..dbc70f7b43c --- /dev/null +++ b/openbb_platform/extensions/currency/openbb_currency/currency_views.py @@ -0,0 +1,17 @@ +"""Views for the Currency Extension.""" + +from typing import Any, Dict, Tuple + +from openbb_charting.charts.price_historical import price_historical +from openbb_charting.core.openbb_figure import OpenBBFigure + + +class CurrencyViews: + """Currency Views.""" + + @staticmethod + def currency_price_historical( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Currency Price Historical Chart.""" + return price_historical(**kwargs) diff --git a/openbb_platform/extensions/currency/pyproject.toml b/openbb_platform/extensions/currency/pyproject.toml index 5dcef853d34..e8117d8f3cd 100644 --- a/openbb_platform/extensions/currency/pyproject.toml +++ b/openbb_platform/extensions/currency/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] currency = "openbb_currency.currency_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +currency = "openbb_currency.currency_views:CurrencyViews" diff --git a/openbb_platform/extensions/economy/openbb_economy/economy_views.py b/openbb_platform/extensions/economy/openbb_economy/economy_views.py new file mode 100644 index 00000000000..c0bc2396aa5 --- /dev/null +++ b/openbb_platform/extensions/economy/openbb_economy/economy_views.py @@ -0,0 +1,307 @@ +"""Views for the Economy Extension.""" + +from typing import Any, Dict, Optional, Tuple +from warnings import warn + +import pandas as pd +from openbb_charting.charts.generic_charts import bar_chart +from openbb_charting.charts.helpers import ( + z_score_standardization, +) +from openbb_charting.core.openbb_figure import OpenBBFigure +from openbb_charting.styles.colors import LARGE_CYCLER +from openbb_core.app.utils import basemodel_to_df + + +class EconomyViews: + """economy Views.""" + + @staticmethod + def economy_fred_series( # noqa: PLR0912 + **kwargs, + ) -> 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", + } + + 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() + + metadata = kwargs["extra"].get("results_metadata", {}) # 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 has_params is True and not y_units: + y_units = [ytitle_dict.get(params.transform)] # type: ignore + + if normalize or ( + kwargs.get("bar") is True + and len(y_units) > 1 + and ( + has_params is False + or not any( + i in params.transform for i in ["pc1", "pch", "pca", "cch", "cca", "log"] # type: ignore + ) + ) + ): + normalize = True + df_ta = df_ta.apply(z_score_standardization) + + if len(y_units) > 2 and has_params is False and allow_unsafe is False: + 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] if y_units else None + y1title = y1_units + y2title = y_units[1] if len(y_units) > 1 else None + xtitle = str(kwargs.get("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. + title: str = "" + if isinstance(kwargs, dict) and title in kwargs: + title = kwargs["title"] # type: ignore + else: + if metadata.get(columns[0]): # type: ignore + 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: Optional[str] = "" + + if kwargs.get("plot_bar") is True or len(df_ta.index) < 100: + margin = dict(l=10, r=5, b=75 if xtitle else 30) + try: + if normalize: + y1title = None + title = f"{title} - Normalized" if title else "Normalized" + bar_mode = kwargs.get("barmode", "group") + fig = bar_chart( + df_ta.reset_index(), + "date", + df_ta.columns.to_list(), + title=title, + xtitle=xtitle, + ytitle=y1title, + barmode=bar_mode, # type: ignore + layout_kwargs=dict(margin=margin), # type: ignore + ) + if kwargs.get("layout_kwargs"): + fig.update_layout(kwargs.get("layout_kwargs")) + + if kwargs.get("title"): + fig.set_title(str(kwargs.get("title"))) # type: ignore + + content = fig.to_plotly_json() + + return fig, content # type: ignore + except Exception as _: + warn("Bar chart failed. Attempting line chart.") + + # Create the figure object with subplots. + fig = OpenBBFigure().create_subplots( + rows=1, cols=1, shared_xaxes=True, shared_yaxes=False + ) + + # 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 # type: ignore + or y2title is None # type: ignore + or kwargs.get("same_axis") is True + ) + if metadata.get(col) # type: ignore + 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 # type: ignore + if on_y3: + yaxes = "y3" + y3title = df_ta[col].name # type: ignore + 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}}", + line=dict(width=2, color=LARGE_CYCLER[i % len(LARGE_CYCLER)]), + yaxis="y1" if kwargs.get("same_axis") is True else 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 isinstance(kwargs, dict) and "xtitle" in kwargs: + xtitle = kwargs["xtitle"] + # If the data was normalized, set the title to reflect this. + if normalize: + y1title = None + y2title = None + y3title = None + title = f"{title} - Normalized" if title else "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)", + legend=dict( + orientation="h", + yanchor="bottom", + xanchor="right", + y=1.02, + x=0.95, + bgcolor="rgba(0,0,0,0)", + font=dict(size=12), + ), + yaxis=( + dict( + ticklen=0, + side="right", + showline=True, + mirror=True, + title=dict(text=y1title, standoff=30, font=dict(size=16)), + tickfont=dict(size=14), + anchor="x", + gridcolor="rgba(128,128,128,0.3)", + ) + 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=16), + ), + 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.9)"), + anchor="free", + ) + if y3title + else None + ), + xaxis=dict( + ticklen=0, + showgrid=True, + showline=True, + mirror=True, + title=( + dict(text=xtitle, standoff=30, font=dict(size=16)) + if xtitle + else None + ), + gridcolor="rgba(128,128,128,0.3)", + domain=[0.095, 0.95] if y3title else None, + ), + margin=( + dict(r=25, l=25, b=75 if xtitle else 30) if normalize is False else None + ), + autosize=True, + dragmode="pan", + ) + if kwargs.get("layout_kwargs"): + fig.update_layout(kwargs.get("layout_kwargs")) + if kwargs.get("title"): + fig.set_title(str(kwargs.get("title"))) + content = fig.to_plotly_json() + + return fig, content diff --git a/openbb_platform/extensions/economy/pyproject.toml b/openbb_platform/extensions/economy/pyproject.toml index 4b2deb7faeb..64ca2811998 100644 --- a/openbb_platform/extensions/economy/pyproject.toml +++ b/openbb_platform/extensions/economy/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] economy = "openbb_economy.economy_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +economy = "openbb_economy.economy_views:EconomyViews" diff --git a/openbb_platform/extensions/equity/openbb_equity/equity_views.py b/openbb_platform/extensions/equity/openbb_equity/equity_views.py new file mode 100644 index 00000000000..659abb3738c --- /dev/null +++ b/openbb_platform/extensions/equity/openbb_equity/equity_views.py @@ -0,0 +1,25 @@ +"""Views for the Equity Extension.""" + +from typing import Any, Dict, Tuple + +from openbb_charting.charts.price_historical import price_historical +from openbb_charting.charts.price_performance import price_performance +from openbb_charting.core.openbb_figure import OpenBBFigure + + +class EquityViews: + """Equity Views.""" + + @staticmethod + def equity_price_historical( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Equity Price Historical Chart.""" + return price_historical(**kwargs) + + @staticmethod + def equity_price_performance( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Equity Price Performance Chart.""" + return price_performance(**kwargs) diff --git a/openbb_platform/extensions/equity/pyproject.toml b/openbb_platform/extensions/equity/pyproject.toml index 6858807d5d6..a7c87a674d0 100644 --- a/openbb_platform/extensions/equity/pyproject.toml +++ b/openbb_platform/extensions/equity/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] equity = "openbb_equity.equity_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +equity = "openbb_equity.equity_views:EquityViews" diff --git a/openbb_platform/extensions/etf/openbb_etf/etf_views.py b/openbb_platform/extensions/etf/openbb_etf/etf_views.py new file mode 100644 index 00000000000..71e2c9e3c23 --- /dev/null +++ b/openbb_platform/extensions/etf/openbb_etf/etf_views.py @@ -0,0 +1,84 @@ +"""Views for the ETF Extension.""" + +from typing import Any, Dict, Tuple, Union + +import pandas as pd +from openbb_charting.charts.generic_charts import bar_chart +from openbb_charting.charts.price_historical import price_historical +from openbb_charting.charts.price_performance import price_performance +from openbb_charting.core.openbb_figure import OpenBBFigure +from openbb_core.app.utils import basemodel_to_df +from plotly.graph_objs import Figure + + +class EtfViews: + """Etf Views.""" + + @staticmethod + def etf_historical( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Etf Price Historical Chart.""" + return price_historical(**kwargs) + + @staticmethod + def etf_price_performance( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Etf Price Performance Chart.""" + return price_performance(**kwargs) + + @staticmethod + def etf_holdings(**kwargs) -> Tuple[Union[OpenBBFigure, Figure], Dict[str, Any]]: + """Equity Compare Groups Chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + elif "data" in kwargs and isinstance(kwargs["data"], list): + data = basemodel_to_df(kwargs["data"], index=None) # type: ignore + else: + data = basemodel_to_df(kwargs["obbject_item"], index=None) # type: ignore + + if "weight" not in data.columns: + raise ValueError("No 'weight' column found in the data.") + + orientation = kwargs.get("orientation", "h") + limit = kwargs.get("limit", 20) + symbol = kwargs["standard_params"].get("symbol") # type: ignore + title = kwargs.get("title", f"Top {limit} {symbol} Holdings") + layout_kwargs = kwargs.get("layout_kwargs", {}) + + data = data.sort_values("weight", ascending=False) + limit = min(limit, len(data)) # type: ignore + target = data.head(limit)[["symbol", "weight"]].set_index("symbol") + target = target.multiply(100) + axis_title = "Weight (%)" + + fig = bar_chart( + target.reset_index(), + "symbol", + ["weight"], + title=title, # type: ignore + xtitle=axis_title if orientation == "h" else None, + ytitle=axis_title if orientation == "v" else None, + orientation=orientation, # type: ignore + ) + + fig.update_layout( + hovermode="x" if orientation == "v" else "y", + margin=dict(r=0, l=50) if orientation == "h" else None, + ) + + fig.update_traces( + hovertemplate=( + "%{y:.3f}%" + if orientation == "v" + else "%{x:.3f}%" + ) + ) + + if layout_kwargs: + fig.update_layout(**layout_kwargs) # type: ignore + + content = fig.show(external=True).to_plotly_json() # type: ignore + + return fig, content diff --git a/openbb_platform/extensions/etf/pyproject.toml b/openbb_platform/extensions/etf/pyproject.toml index a036b093c36..dc7ff104ee2 100644 --- a/openbb_platform/extensions/etf/pyproject.toml +++ b/openbb_platform/extensions/etf/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] etf = "openbb_etf.etf_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +etf = "openbb_etf.etf_views:EtfViews" diff --git a/openbb_platform/extensions/fixedincome/openbb_fixedincome/fixedincome_views.py b/openbb_platform/extensions/fixedincome/openbb_fixedincome/fixedincome_views.py new file mode 100644 index 00000000000..4c76bcc0b46 --- /dev/null +++ b/openbb_platform/extensions/fixedincome/openbb_fixedincome/fixedincome_views.py @@ -0,0 +1,183 @@ +"""Views for the Fixed Income Extension.""" + +from typing import Any, Dict, Optional, Tuple + +import pandas as pd +from openbb_charting.charts.helpers import ( + duration_sorter, +) +from openbb_charting.core.chart_style import ChartStyle +from openbb_charting.core.openbb_figure import OpenBBFigure +from openbb_charting.styles.colors import LARGE_CYCLER +from openbb_core.app.utils import basemodel_to_df +from openbb_core.provider.abstract.data import Data + + +class FixedIncomeViews: + """FixedIncome Views.""" + + @staticmethod + def fixedincome_government_yield_curve( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Government Yield Curve Chart.""" + data = kwargs.get("data", None) + df: pd.DataFrame = pd.DataFrame() + if data: + if isinstance(data, pd.DataFrame) and not data.empty: # noqa: SIM108 + df = data + elif isinstance(data, (list, Data)): + df = basemodel_to_df(data, index=None) # type: ignore + else: + pass + else: + df = pd.DataFrame([d.model_dump() for d in kwargs["obbject_item"]]) # type: ignore + + if df.empty: + raise ValueError("Error: No data to plot.") + + if "maturity" not in df.columns: + raise ValueError("Error: Maturity column not found in the data.") + + if "rate" not in df.columns: + raise ValueError("Error: Rate column not found in the data.") + + if "date" not in df.columns: + raise ValueError("Error: Date column not found in the data.") + + provider = kwargs.get("provider") + df["date"] = df["date"].astype(str) + maturities = duration_sorter(df["maturity"].unique().tolist()) + + # Use the supplied colors, if any. + colors = kwargs.get("colors", []) + if not colors: + colors = LARGE_CYCLER + color_count = 0 + + figure = OpenBBFigure().create_subplots(shared_xaxes=True) + figure.update_layout(ChartStyle().plotly_template.get("layout", {})) + + def create_fig(figure, df, dates, color_count, country: Optional[str] = None): + """Create a scatter for each date in the data.""" + for date in dates: + color = colors[color_count % len(colors)] + plot_df = df[df["date"] == date].copy() + plot_df["rate"] = plot_df["rate"].apply(lambda x: x * 100) + plot_df = ( + plot_df.drop(columns=["date"]) + .set_index("maturity") + .filter(items=maturities, axis=0) + .reset_index() + .rename(columns={"maturity": "Maturity", "rate": "Yield"}) + ) + plot_df["Maturity"] = [ + ( + d.split("_")[1] + " " + d.split("_")[0].title() + if d != "long_term" + else "Long Term" + ) + for d in plot_df["Maturity"] + ] + figure.add_scatter( + x=plot_df["Maturity"], + y=plot_df["Yield"], + # fill=fill, + mode="lines+markers", + name=f"{country} - {date}" if country else date, + line=dict(width=3, color=color), + marker=dict(size=10, color=color), + hovertemplate=( + "Maturity: %{x}
Yield: %{y}%" + if len(dates) == 1 + else "%{fullData.name}
Maturity: %{x}
Yield: %{y}%" + ), + ) + color_count += 1 + return figure, color_count + + dates = df.date.unique().tolist() + figure, color_count = create_fig(figure, df, dates, color_count) + extra_params = kwargs.get("extra_params") + extra_params = ( + extra_params if isinstance(extra_params, dict) else extra_params.__dict__ + ) + # Set the title for the chart + country: str = "" + if provider in ("federal_reserve", "fmp"): + country = "United States" + elif provider == "ecb": + curve_type = ( + extra_params.get("yield_curve_type", "").replace("_", " ").title() + ) + grade = extra_params.get("rating", "").replace("_", " ") + grade = grade.upper() if grade == "aaa" else "All Ratings" + country = f"Euro Area ({grade}) {curve_type}" + elif provider == "fred": + curve_type = extra_params.get("yield_curve_type", "") + curve_type = ( + "Real Rates" + if curve_type == "real" + else curve_type.replace("_", " ").title() + ) + country = f"United States {curve_type}" + elif provider == "econdb": + country = extra_params.get("country", "") + country = country.replace("_", " ").title() if country else "United States" + country = country + " " if country else "" + title = kwargs.get("title", "") + if not title: + title = f"{country}Yield Curve" + if len(dates) == 1: + title = f"{country} Yield Curve - {dates[0]}" + + # Update the layout of the figure. + figure.update_layout( + title=dict(text=title, x=0.5, font=dict(size=20)), + plot_bgcolor="rgba(255,255,255,0)", + xaxis=dict( + title="Maturity", + ticklen=0, + showgrid=False, + type="category", + categoryorder="array", + categoryarray=( + [ + ( + d.split("_")[1] + " " + d.split("_")[0].title() + if d != "long_term" + else "Long Term" + ) + for d in maturities + ] + ), + ), + yaxis=dict( + title="Yield (%)", + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + ), + legend=dict( + orientation="v", + yanchor="top", + xanchor="right", + y=0.95, + x=0, + xref="paper", + font=dict(size=12), + bgcolor="rgba(0,0,0,0)", + ), + margin=dict( + b=25, + t=10, + ), + ) + + layout_kwargs = kwargs.get("layout_kwargs", {}) + if layout_kwargs: + figure.update_layout(layout_kwargs) + + content = figure.show(external=True) + + return figure, content diff --git a/openbb_platform/extensions/fixedincome/pyproject.toml b/openbb_platform/extensions/fixedincome/pyproject.toml index c19cbd70b10..233b0b80d1a 100644 --- a/openbb_platform/extensions/fixedincome/pyproject.toml +++ b/openbb_platform/extensions/fixedincome/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] fixedincome = "openbb_fixedincome.fixedincome_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +fixedincome = "openbb_fixedincome.fixedincome_views:FixedIncomeViews" diff --git a/openbb_platform/extensions/index/openbb_index/index_views.py b/openbb_platform/extensions/index/openbb_index/index_views.py new file mode 100644 index 00000000000..1185444e417 --- /dev/null +++ b/openbb_platform/extensions/index/openbb_index/index_views.py @@ -0,0 +1,17 @@ +"""Views for the index Extension.""" + +from typing import Any, Dict, Tuple + +from openbb_charting.charts.price_historical import price_historical +from openbb_charting.core.openbb_figure import OpenBBFigure + + +class IndexViews: + """Index Views.""" + + @staticmethod + def index_price_historical( # noqa: PLR0912 + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Index Price Historical Chart.""" + return price_historical(**kwargs) diff --git a/openbb_platform/extensions/index/pyproject.toml b/openbb_platform/extensions/index/pyproject.toml index d36fa96ca18..96401793960 100644 --- a/openbb_platform/extensions/index/pyproject.toml +++ b/openbb_platform/extensions/index/pyproject.toml @@ -17,3 +17,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] index = "openbb_index.index_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +index = "openbb_index.index_views:IndexViews" diff --git a/openbb_platform/extensions/technical/openbb_technical/technical_views.py b/openbb_platform/extensions/technical/openbb_technical/technical_views.py new file mode 100644 index 00000000000..b59c46aef29 --- /dev/null +++ b/openbb_platform/extensions/technical/openbb_technical/technical_views.py @@ -0,0 +1,511 @@ +"""Views for the technical Extension.""" + +from typing import Any, Dict, Tuple + +import pandas as pd +from openbb_charting.charts import relative_rotation +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.core.to_chart import to_chart +from openbb_charting.styles.colors import LARGE_CYCLER +from openbb_core.app.utils import basemodel_to_df + + +class TechnicalViews: + """Technical Views.""" + + @staticmethod + def technical_sma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Plot simple moving average chart.""" + if "ma_type" not in kwargs: + kwargs["ma_type"] = "sma" + return _ta_ma(**kwargs) + + @staticmethod + def technical_ema(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Exponential moving average chart.""" + if "ma_type" not in kwargs: + kwargs["ma_type"] = "ema" + return _ta_ma(**kwargs) + + @staticmethod + def technical_hma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Hull moving average chart.""" + if "ma_type" not in kwargs: + kwargs["ma_type"] = "hma" + return _ta_ma(**kwargs) + + @staticmethod + def technical_wma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Weighted moving average chart.""" + if "ma_type" not in kwargs: + kwargs["ma_type"] = "wma" + return _ta_ma(**kwargs) + + @staticmethod + def technical_zlma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Zero lag moving average chart.""" + if "ma_type" not in kwargs: + kwargs["ma_type"] = "zlma" + return _ta_ma(**kwargs) + + @staticmethod + def technical_aroon(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Technical Aroon Chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + else: + data = basemodel_to_df( + kwargs["obbject_item"], index=kwargs.get("index", "date") + ) + + if "date" in data.columns: + data = data.set_index("date") + + if "symbol" in data.columns and len(data.symbol.unique()) > 1: + raise ValueError( + "Please provide data with only one symbol and columns for OHLC." + ) + + symbol = kwargs.get("symbol", "") + + volume = kwargs.get("volume") is True + title = f"Aroon Indicator & Oscillator {symbol}" + + length = kwargs.get("length", 25) + scalar = kwargs.get("scalar", 100) + symbol = kwargs.get("symbol", "") + + ta = PlotlyTA() + fig = ta.plot( # type: ignore + data, + dict(aroon=dict(length=length, scalar=scalar)), + title, + False, + volume=volume, + ) + + content = fig.show(external=True).to_plotly_json() + + return fig, content + + @staticmethod + def technical_macd(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Plot moving average convergence divergence chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + else: + data = basemodel_to_df( + kwargs["obbject_item"], index=kwargs.get("index", "date") + ) + + if "date" in data.columns: + data = data.set_index("date") + + if "symbol" in data.columns and len(data.symbol.unique()) > 1: + raise ValueError( + "Please provide data with only one symbol and columns for OHLC." + ) + + fast = kwargs.get("fast", 12) + slow = kwargs.get("slow", 26) + signal = kwargs.get("signal", 9) + symbol = kwargs.get("symbol", "") + + title = f"{symbol.upper()} MACD" + volume = kwargs.get("volume") is True + + ta = PlotlyTA() + fig = ta.plot( # type: ignore + data, + dict(macd=dict(fast=fast, slow=slow, signal=signal)), + title, + False, + volume=volume, + ) + content = fig.show(external=True).to_plotly_json() + + return fig, content + + @staticmethod + def technical_adx(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Average directional movement index chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + else: + data = basemodel_to_df( + kwargs["obbject_item"], index=kwargs.get("index", "date") + ) + + if "date" in data.columns: + data = data.set_index("date") + + if "symbol" in data.columns and len(data.symbol.unique()) > 1: + raise ValueError( + "Please provide data with only one symbol and columns for OHLC." + ) + + length = kwargs.get("length", 14) + scalar = kwargs.get("scalar", 100.0) + drift = kwargs.get("drift", 1) + symbol = kwargs.get("symbol", "") + + ta = PlotlyTA() + fig = ta.plot( # type: ignore + data, + dict(adx=dict(length=length, scalar=scalar, drift=drift)), + f"Average Directional Movement Index (ADX) {symbol}", + False, + volume=False, + ) + content = fig.show(external=True).to_plotly_json() + + return fig, content + + @staticmethod + def technical_rsi(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Relative strength index chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + else: + data = basemodel_to_df( + kwargs["obbject_item"], index=kwargs.get("index", "date") + ) + + if "date" in data.columns: + data = data.set_index("date") + + if "symbol" in data.columns and len(data.symbol.unique()) > 1: + raise ValueError( + "Please provide data with only one symbol and columns for OHLC." + ) + + window = kwargs.get("window", 14) + scalar = kwargs.get("scalar", 100.0) + drift = kwargs.get("drift", 1) + symbol = kwargs.get("symbol", "") + + ta = PlotlyTA() + fig = ta.plot( # type: ignore + data, + dict(rsi=dict(length=window, scalar=scalar, drift=drift)), + f"{symbol.upper()} RSI {window}", + False, + volume=False, + ) + content = fig.show(external=True).to_plotly_json() + + return fig, content + + @staticmethod + def technical_cones(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Volatility Cones Chart.""" + data = kwargs.get("data") + + 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") # type: ignore + + df_ta.columns = [col.title().replace("_", " ") for col in df_ta.columns] + + # Check if the data is formatted as expected. + if not all( + col in df_ta.columns for col in ["Realized", "Min", "Median", "Max"] + ): + raise ValueError("Data supplied does not match the expected format.") + + model = ( + str(kwargs.get("model")) + .replace("std", "Standard Deviation") + .replace("_", "-") + .title() + if kwargs.get("model") + else "Standard Deviation" + ) + + symbol = str(kwargs.get("symbol")) + " - " if kwargs.get("symbol") else "" + + title = ( + str(kwargs.get("title")) + if kwargs.get("title") + else f"{symbol}Realized Volatility Cones - {model} Model" + ) + + colors = [ + "green", + "red", + "burlywood", + "grey", + "orange", + "blue", + ] + + fig = OpenBBFigure() + + fig.update_layout(ChartStyle().plotly_template.get("layout", {})) + + text_color = "black" if ChartStyle().plt_style == "light" else "white" + + for i, col in enumerate(df_ta.columns): + fig.add_scatter( + x=df_ta.index, + y=df_ta[col], + name=col, + mode="lines+markers", + hovertemplate=f"{col}: %{{y}}", + marker=dict( + color=colors[i], + size=11, + ), + ) + + fig.set_title(title) + + fig.update_layout( + 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, + showgrid=True, + showline=True, + mirror=True, + zeroline=False, + gridcolor="rgba(128,128,128,0.3)", + ), + xaxis=dict( + type="category", + tickmode="array", + ticklen=0, + tickvals=df_ta.index, + ticktext=df_ta.index, + title_text="Period", + showgrid=False, + showline=True, + mirror=True, + zeroline=False, + ), + margin=dict(l=20, r=20, b=20), + dragmode="pan", + ) + + content = fig.to_plotly_json() + + return fig, content + + @staticmethod + def technical_relative_rotation( + **kwargs: Any, + ) -> Tuple["OpenBBFigure", Dict[str, Any]]: + """Relative Rotation Chart.""" + ratios_df = basemodel_to_df(kwargs["obbject_item"].rs_ratios, index="date") # type: ignore + momentum_df = basemodel_to_df(kwargs["obbject_item"].rs_momentum, index="date") # type: ignore + benchmark_symbol = kwargs["obbject_item"].benchmark # type: ignore + study = kwargs.get("study", None) + study = str(kwargs["obbject_item"].study) if study is None else str(study) + show_tails = kwargs.get("show_tails") + show_tails = True if show_tails is None else show_tails + tail_periods = int(kwargs.get("tail_periods")) if "tail_periods" in kwargs else 16 # type: ignore + tail_interval = str(kwargs.get("tail_interval")) if "tail_interval" in kwargs else "week" # type: ignore + date = kwargs.get("date") if "date" in kwargs else None # type: ignore + show_tails = False if date is not None else show_tails + if ratios_df.empty or momentum_df.empty: + raise RuntimeError("Error: No data to plot.") + + if show_tails is True: + fig = relative_rotation.create_rrg_with_tails( + ratios_df, momentum_df, study, benchmark_symbol, tail_periods, tail_interval # type: ignore + ) + + if show_tails is False: + fig = relative_rotation.create_rrg_without_tails( + ratios_df, momentum_df, benchmark_symbol, study, date # type: ignore + ) + + figure = OpenBBFigure(fig) # pylint: disable=E0606 + font_color = "black" if ChartStyle().plt_style == "light" else "white" + figure.update_layout( + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(255,255,255,1)", + font=dict(color=font_color), + yaxis=dict( + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + side="left", + showline=True, + zeroline=True, + mirror=True, + ticklen=0, + tickfont=dict(size=14), + titlefont=dict(size=16), + ), + xaxis=dict( + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + showline=True, + zeroline=True, + mirror=True, + ticklen=0, + tickfont=dict(size=14), + titlefont=dict(size=16), + hoverformat="", + ), + hoverlabel=dict( + font_size=12, + ), + hovermode="x", + hoverdistance=50, + ) + if kwargs.get("title") is not None: + figure.set_title(str(kwargs.get("title"))) + content = figure.to_plotly_json() + + return figure, content + + +def _ta_ma(**kwargs): + """Plot moving average helper.""" + index = ( + kwargs.get("index") + if "index" in kwargs and kwargs.get("index") is not None + else "date" + ) + data = kwargs.get("data") + ma_type = ( + kwargs["ma_type"] + if "ma_type" in kwargs and kwargs.get("ma_type") is not None + else "sma" + ) + ma_types = ma_type.split(",") if isinstance(ma_type, str) else ma_type + + if isinstance(data, pd.DataFrame) and not data.empty: + data = data.set_index(index) if index in data.columns else data + + if data is None: + data = basemodel_to_df(kwargs["obbject_item"], index=index) + + if isinstance(data, list): + data = basemodel_to_df(data, index=index) + + window = ( + kwargs.get("length", []) + if "length" in kwargs and kwargs.get("length") is not None + else [50] + ) + offset = kwargs.get("offset", 0) + target = ( + kwargs.get("target") + if "target" in kwargs and kwargs.get("target") is not None + else "close" + ) + + if target not in data.columns and "close" in data.columns: + target = "close" + + if target not in data.columns and "close" not in data.columns: + raise ValueError(f"Column '{target}', or 'close', not found in the data.") + + df = data.copy() + if target in data.columns: + df = df[[target]] + df.columns = ["close"] + title = ( + kwargs.get("title") + if "title" in kwargs and kwargs.get("title") is not None + else f"{ma_type.upper()}" + ) + + fig = OpenBBFigure() + fig = fig.create_subplots( + 1, + 1, + shared_xaxes=True, + vertical_spacing=0.06, + horizontal_spacing=0.01, + row_width=[1], + specs=[[{"secondary_y": True}]], + ) + fig.update_layout(ChartStyle().plotly_template.get("layout", {})) + + ma_df = pd.DataFrame() + window = [window] if isinstance(window, int) else window + for w in window: + for ma_type in ma_types: + ma_df[f"{ma_type.upper()} {w}"] = getattr(df.ta, ma_type)( + length=w, offset=offset + ) + + if kwargs.get("dropnan") is True: + ma_df = ma_df.dropna() + data = data.iloc[-len(ma_df) :] + + if ( + "candles" in kwargs + and kwargs.get("candles") is True + and kwargs.get("target") is None + ): + volume = kwargs.get("volume") is True + fig, _ = to_chart(data, candles=True, volume=volume) + + else: + ma_df[f"{target}".title()] = data[target] + + for i, col in enumerate(ma_df.columns): + name = col.replace("_", " ") + fig.add_scatter( + x=ma_df.index, + y=ma_df[col], + name=name, + mode="lines", + hovertemplate=f"{name}: %{{y}}", + line=dict(width=1, color=LARGE_CYCLER[i]), + showlegend=True, + ) + + 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)", + showlegend=True, + legend=dict( + orientation="h", + yanchor="bottom", + xanchor="right", + y=1.02, + x=0.95, + bgcolor="rgba(0,0,0,0)", + ), + xaxis=dict( + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + zeroline=True, + mirror=True, + ), + yaxis=dict( + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + zeroline=True, + mirror=True, + autorange=True, + ), + ) + + content = fig.show(external=True).to_plotly_json() + + return fig, content diff --git a/openbb_platform/extensions/technical/pyproject.toml b/openbb_platform/extensions/technical/pyproject.toml index 08b8354d242..f1770b619ac 100644 --- a/openbb_platform/extensions/technical/pyproject.toml +++ b/openbb_platform/extensions/technical/pyproject.toml @@ -21,3 +21,6 @@ build-backend = "poetry.core.masonry.api" [tool.poetry.plugins."openbb_core_extension"] technical = "openbb_technical.technical_router:router" + +[tool.poetry.plugins."openbb_charting_extension"] +technical = "openbb_technical.technical_views:TechnicalViews" diff --git a/openbb_platform/obbject_extensions/charting/README.md b/openbb_platform/obbject_extensions/charting/README.md index 039001608c0..8b87d86d226 100644 --- a/openbb_platform/obbject_extensions/charting/README.md +++ b/openbb_platform/obbject_extensions/charting/README.md @@ -9,7 +9,7 @@ The library includes: - prebuilt charts for a set of commands that are built-in OpenBB extensions >[!NOTE] -> The charting library is an [`OBBject` extension](https://docs.openbb.co/platform/development/developer-guidelines/obbject_extensions) which means you'll have the functionality it exposes on every command result. +> The charting library is an `OBBject` extension which means you'll have the functionality it exposes on every command result. ## Installation @@ -120,15 +120,44 @@ Example usage: ## Add a visualization to an existing Platform command -One should first ensure that the already implemented endpoint is available in the [charting router](/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py). +To add a visualization to an existing command, you'll need to add a `poetry` plugin to your `pyproject.toml` file. The syntax should be the following: -To do so, you can run: - `python openbb_platform/obbject_extensions/charting/openbb_charting/builder.py` - which will read all the available endpoints and add them to the charting router. +```toml +[tool.poetry.plugins."openbb_charting_extension"] +my_extension = "openbb_my_extension.my_extension_views:MyExtensionViews" +``` -Afterwards, you'll need to add the visualization to the [charting router]. The convention to match the endpoint with the respective charting function is the following: +Where the `openbb_charting_extension` is **mandatory**, otherwise the charting extension won't be able to find the visualization. + +And the suggested structure for the `my_extension_views` module is the following: + +```python +"""Views for MyExtension.""" + +from typing import Any, Dict, Tuple + +from openbb_charting.charts.price_historical import price_historical +from openbb_charting.core.openbb_figure import OpenBBFigure + + +class MyExtensionViews: + """MyExtension Views.""" + + @staticmethod + def my_extension_price_historical( + **kwargs, + ) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """MyExtension Price Historical Chart.""" + return price_historical(**kwargs) +``` + +> Note that `my_extension_views` lives under the `openbb_my_extension` package. + +Afterwards, you'll need to add the visualization to your new `MyExtensionViews` class. The convention to match the endpoint with the respective charting function is the following: - `/equity/price/historical` -> `equity_price_historical` - `/technical/ema` -> `technical_ema` +- `/my_extension/price_historical` -> `my_extension_price_historical` When you spot the charting function on the charting router file, you can add the visualization to it. diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/__init__.py b/openbb_platform/obbject_extensions/charting/openbb_charting/__init__.py index d20908557d4..fb00de240b0 100644 --- a/openbb_platform/obbject_extensions/charting/openbb_charting/__init__.py +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/__init__.py @@ -1,561 +1,14 @@ """OpenBB OBBject extension for charting.""" -# pylint: disable=too-many-arguments,unused-argument - import warnings -from typing import ( - Any, - Callable, - Dict, - List, - Literal, - Optional, - Tuple, - Union, -) -from warnings import warn -import numpy as np -import pandas as pd -from openbb_core.app.model.charts.chart import Chart from openbb_core.app.model.extension import Extension -from openbb_core.app.model.obbject import OBBject -from openbb_core.app.utils import basemodel_to_df, convert_to_basemodel -from openbb_core.provider.abstract.data import Data -from plotly.graph_objs import Figure -from openbb_charting import charting_router -from openbb_charting.core.backend import Backend, create_backend, get_backend -from openbb_charting.core.openbb_figure import OpenBBFigure -from openbb_charting.query_params import ChartParams, IndicatorsParams -from openbb_charting.utils.generic_charts import bar_chart, line_chart -from openbb_charting.utils.helpers import get_charting_functions +from openbb_charting.charting import Charting as _Charting warnings.filterwarnings( "ignore", category=UserWarning, module="openbb_core.app.model.extension", lineno=47 ) ext = Extension(name="charting", description="Create custom charts from OBBject data.") - - -@ext.obbject_accessor -class Charting: - """Charting extension. - - Methods - ------- - show - Display chart and save it to the OBBject. - to_chart - Redraw the chart and save it to the OBBject, with an optional entry point for Data. - functions - Return a list of Platform commands with charting functions. - get_params - Return the charting parameters for the function the OBBject was created from. - indicators - Return the list of the available technical indicators to use with the `to_chart` method and OHLC+V data. - table - Display an interactive table. - create_line_chart - Create a line chart from external data. - create_bar_chart - Create a bar chart, on a single x-axis with one or more values for the y-axis, from external data. - toggle_chart_style - Toggle the chart style, of an existing chart, between light and dark mode. - """ - - def __init__(self, obbject): - """Initialize Charting extension.""" - # pylint: disable=import-outside-toplevel - from openbb_core.app.model.charts.charting_settings import ChartingSettings - - self._obbject: OBBject = obbject - self._charting_settings = ChartingSettings( - user_settings=self._obbject._user_settings, # type: ignore - system_settings=self._obbject._system_settings, # type: ignore - ) - self._backend: Backend = self._handle_backend() - - @classmethod - def indicators(cls): - """Return an instance of the IndicatorsParams class, containing all available indicators and their parameters. - - Without assigning to a variable, it will print the the information to the console. - """ - return IndicatorsParams() - - @classmethod - def functions(cls): - """Return a list of the available functions.""" - return get_charting_functions() - - def _handle_backend(self) -> Backend: - """Create and start the backend.""" - create_backend(self._charting_settings) - backend = get_backend() - backend.start(debug=self._charting_settings.debug_mode) - return backend - - @staticmethod - def _get_chart_function(route: str) -> Callable: - """Given a route, it returns the chart function. The module must contain the given route.""" - if route is None: - raise ValueError("OBBject was initialized with no function route.") - adjusted_route = route.replace("/", "_")[1:] - return getattr(charting_router, adjusted_route) - - def get_params(self) -> ChartParams: - """Return the ChartQueryParams class for the function the OBBject was created from. - - Without assigning to a variable, it will print the docstring to the console. - """ - if self._obbject._route is None: # pylint: disable=protected-access - raise ValueError("OBBject was initialized with no function route.") - charting_function = ( - self._obbject._route # pylint: disable=protected-access - ).replace("/", "_")[1:] - if hasattr(ChartParams, charting_function): - return getattr(ChartParams, charting_function)() - raise ValueError( - f"Error: No chart parameters are defined for the route: {charting_function}" - ) - - def _prepare_data_as_df( - self, data: Optional[Union[pd.DataFrame, pd.Series]] - ) -> Tuple[pd.DataFrame, bool]: - """Convert supplied data to a DataFrame.""" - has_data = ( - isinstance(data, (Data, pd.DataFrame, pd.Series)) and not data.empty # type: ignore - ) or (bool(data)) - index = ( - data.index.name - if has_data and isinstance(data, (pd.DataFrame, pd.Series)) - else None - ) - data_as_df: pd.DataFrame = ( - basemodel_to_df(convert_to_basemodel(data), index=index) - if has_data - else self._obbject.to_dataframe(index=index) - ) - if "date" in data_as_df.columns: - data_as_df = data_as_df.set_index("date") - if "provider" in data_as_df.columns: - data_as_df.drop(columns="provider", inplace=True) - return data_as_df, has_data - - # pylint: disable=too-many-locals - def create_line_chart( - self, - data: Union[ - list, - dict, - pd.DataFrame, - List[pd.DataFrame], - pd.Series, - List[pd.Series], - np.ndarray, - Data, - ], - index: Optional[str] = None, - target: Optional[str] = None, - title: Optional[str] = None, - x: Optional[str] = None, - xtitle: Optional[str] = None, - y: Optional[Union[str, List[str]]] = None, - ytitle: Optional[str] = None, - y2: Optional[Union[str, List[str]]] = None, - y2title: Optional[str] = None, - layout_kwargs: Optional[dict] = None, - scatter_kwargs: Optional[dict] = None, - normalize: bool = False, - returns: bool = False, - same_axis: bool = False, - render: bool = True, - **kwargs, - ) -> Union[OpenBBFigure, Figure, None]: - """Create a line chart from external data and render a chart or return the OpenBBFigure. - - Parameters - ---------- - data : Union[Data, pd.DataFrame, pd.Series] - Data to be plotted (OHLCV data). - index : Optional[str], optional - Index column, by default None - target : Optional[str], optional - Target column to be plotted, by default None - title : Optional[str], optional - Chart title, by default None - x : Optional[str], optional - X-axis column, by default None - xtitle : Optional[str], optional - X-axis title, by default None - y : Optional[Union[str, List[str]]], optional - Y-axis column(s), by default None - If None are supplied, the layout is optimized for the contents of data. - Where many units/scales are present, - it will attempt to divide based on the range of values. - ytitle : Optional[str], optional - Y-axis title, by default None - y2 : Optional[Union[str, List[str]]], optional - Y2-axis column(s), by default None - y2title : Optional[str], optional - Y2-axis title, by default None - layout_kwargs : Optional[dict], optional - Additional Plotly Layout parameters for `fig.update_layout`, by default None - scatter_kwargs : Optional[dict], optional - Additional Plotly parameters applied on creation of each scatter plot, by default None - normalize : bool, optional - Normalize the data with Z-Score Standardization, by default False - returns : bool, optional - Convert the data to cumulative returns, by default False - same_axis: bool, optional - If True, forces all data onto the same Y-axis, by default False - render: bool, optional - If True, the chart will be rendered, by default True - **kwargs: Dict[str, Any] - Extra parameters to be passed to `figure.show()` - """ - fig = line_chart( - data=data, - index=index, - target=target, - title=title, - x=x, - xtitle=xtitle, - y=y, - ytitle=ytitle, - y2=y2, - y2title=y2title, - layout_kwargs=layout_kwargs, - scatter_kwargs=scatter_kwargs, - normalize=normalize, - returns=returns, - same_axis=same_axis, - **kwargs, - ) - fig = self._set_chart_style(fig) - if render: - return fig.show(**kwargs) - - return fig - - def create_bar_chart( - self, - data: Union[ - list, - dict, - pd.DataFrame, - List[pd.DataFrame], - pd.Series, - List[pd.Series], - np.ndarray, - Data, - ], - x: str, - y: Union[str, List[str]], - barmode: Literal["group", "stack", "relative", "overlay"] = "group", - xtype: Literal[ - "category", "multicategory", "date", "log", "linear" - ] = "category", - title: Optional[str] = None, - xtitle: Optional[str] = None, - ytitle: Optional[str] = None, - orientation: Literal["h", "v"] = "v", - colors: Optional[List[str]] = None, - layout_kwargs: Optional[Dict[str, Any]] = None, - bar_kwargs: Optional[Dict[str, Any]] = None, - render: bool = True, - **kwargs, - ) -> Union[OpenBBFigure, Figure, None]: - """Create a bar chart on a single x-axis with one or more values for the y-axis. - - Parameters - ---------- - data : Union[list, dict, pd.DataFrame, List[pd.DataFrame], pd.Series, List[pd.Series], np.ndarray, Data] - Data to plot. - x : str - The x-axis column name. - y : Union[str, List[str]] - The y-axis column name(s). - barmode : Literal["group", "stack", "relative", "overlay"], optional - The bar mode, by default "group". - xtype : Literal["category", "multicategory", "date", "log", "linear"], optional - The x-axis type, by default "category". - title : str, optional - The title of the chart, by default None. - xtitle : str, optional - The x-axis title, by default None. - ytitle : str, optional - The y-axis title, by default None. - colors: List[str], optional - Manually set the colors to cycle through for each column in 'y', by default None. - bar_kwargs : Dict[str, Any], optional - Additional keyword arguments to apply with figure.add_bar(), by default None. - layout_kwargs : Dict[str, Any], optional - Additional keyword arguments to apply with figure.update_layout(), by default None. - Returns - ------- - OpenBBFigure - The OpenBBFigure object. - """ - fig = bar_chart( - data=data, - x=x, - y=y, - barmode=barmode, - xtype=xtype, - title=title, - xtitle=xtitle, - ytitle=ytitle, - orientation=orientation, - colors=colors, - bar_kwargs=bar_kwargs, - layout_kwargs=layout_kwargs, - ) - fig = self._set_chart_style(fig) - if render: - return fig.show(**kwargs) - - return fig - - # pylint: disable=inconsistent-return-statements - def show(self, render: bool = True, **kwargs): - """Display chart and save it to the OBBject.""" - try: - charting_function = self._get_chart_function( - self._obbject._route # pylint: disable=protected-access - ) - kwargs["obbject_item"] = self._obbject # pylint: disable=protected-access - kwargs["charting_settings"] = ( - self._charting_settings - ) # pylint: disable=protected-access - kwargs["standard_params"] = ( - self._obbject._standard_params - ) # pylint: disable=protected-access - kwargs["extra_params"] = ( - self._obbject._extra_params - ) # pylint: disable=protected-access - kwargs["provider"] = ( - self._obbject.provider - ) # pylint: disable=protected-access - kwargs["extra"] = self._obbject.extra # pylint: disable=protected-access - fig, content = charting_function(**kwargs) - fig = self._set_chart_style(fig) - content = fig.show(external=True, **kwargs).to_plotly_json() - self._obbject.chart = Chart( - fig=fig, content=content, format=charting_router.CHART_FORMAT - ) - if render: - fig.show(**kwargs) - except Exception: # pylint: disable=W0718 - try: - fig = self.create_line_chart(data=self._obbject.results, render=False, **kwargs) # type: ignore - fig = self._set_chart_style(fig) - content = fig.show(external=True, **kwargs).to_plotly_json() # type: ignore - self._obbject.chart = Chart( - fig=fig, content=content, format=charting_router.CHART_FORMAT - ) - if render: - return fig.show(**kwargs) # type: ignore - except Exception as e: - raise RuntimeError( - "Failed to automatically create a generic chart with the data provided." - ) from e - - # pylint: disable=too-many-locals,inconsistent-return-statements - def to_chart( - self, - data: Optional[ - Union[ - list, - dict, - pd.DataFrame, - List[pd.DataFrame], - pd.Series, - List[pd.Series], - np.ndarray, - Data, - ] - ] = None, - target: Optional[str] = None, - index: Optional[str] = None, - indicators: Optional[Dict[str, Dict[str, Any]]] = None, - symbol: str = "", - candles: bool = True, - volume: bool = True, - volume_ticks_x: int = 7, - render: bool = True, - **kwargs, - ): - """Create an OpenBBFigure with user customizations (if any) and save it to the OBBject. - - This function is used to populate, or re-populate, the OBBject with a chart using the data within - the OBBject or external data supplied via the `data` parameter. - This function modifies the original OBBject by overwriting the existing chart. - - Parameters - ---------- - data : Union[Data, pd.DataFrame, pd.Series] - Data to be plotted. - indicators : Dict[str, Dict[str, Any]], optional - Indicators to be plotted, by default None - symbol : str, optional - Symbol to be plotted. This is used for labels and titles, by default "" - candles : bool, optional - If True, candles will be plotted, by default True - volume : bool, optional - If True, volume will be plotted, by default True - volume_ticks_x : int, optional - Volume ticks, by default 7 - render : bool, optional - If True, the chart will be rendered, by default True - kwargs: Dict[str, Any] - Extra parameters to be passed to the chart constructor. - - Examples - -------- - Plotting a time series with TA indicators - - >>> from openbb import obb - >>> res = obb.equity.price.historical("AAPL") - >>> indicators = dict( - >>> sma=dict(length=[20,30,50]), - >>> adx=dict(length=14), - >>> rsi=dict(length=14), - >>> macd=dict(fast=12, slow=26, signal=9), - >>> bbands=dict(length=20, std=2), - >>> stoch=dict(length=14), - >>> ema=dict(length=[20,30,50]), - >>> ) - >>> res.charting.to_chart(**{"indicators": indicators}) - - Get all the available indicators - - >>> res = obb.equity.price.historical("AAPL") - >>> indicators = res.charting.indicators() - >>> indicators? - """ - data_as_df, has_data = self._prepare_data_as_df(data) # type: ignore - if target is not None: - data_as_df = data_as_df[[target]] - kwargs["candles"] = candles - kwargs["volume"] = volume - kwargs["volume_ticks_x"] = volume_ticks_x - kwargs["indicators"] = indicators if indicators else {} - kwargs["symbol"] = symbol - kwargs["target"] = target - kwargs["index"] = index - kwargs["obbject_item"] = self._obbject # pylint: disable=protected-access - kwargs["charting_settings"] = ( - self._charting_settings - ) # pylint: disable=protected-access - kwargs["standard_params"] = ( - self._obbject._standard_params - ) # pylint: disable=protected-access - kwargs["extra_params"] = ( - self._obbject._extra_params - ) # pylint: disable=protected-access - kwargs["provider"] = self._obbject.provider # pylint: disable=protected-access - kwargs["extra"] = self._obbject.extra # pylint: disable=protected-access - try: - if has_data: - self.show(data=data_as_df, render=render, **kwargs) - else: - self.show(**kwargs, render=render) - except Exception: # pylint: disable=W0718 - try: - fig = self.create_line_chart(data=data_as_df, render=False, **kwargs) - fig = self._set_chart_style(fig) - content = fig.show(external=True, **kwargs).to_plotly_json() # type: ignore - self._obbject.chart = Chart( - fig=fig, content=content, format=charting_router.CHART_FORMAT - ) - if render: - return fig.show(**kwargs) # type: ignore - except Exception as e: # pylint: disable=W0718 - raise RuntimeError( - "Failed to automatically create a generic chart with the data provided." - ) from e - - def _set_chart_style(self, figure: Figure): - """Set the user preference for light or dark mode.""" - style = self._charting_settings.chart_style - font_color = "black" if style == "light" else "white" - paper_bgcolor = "white" if style == "light" else "black" - figure = figure.update_layout( - dict( # pylint: disable=R1735 - font_color=font_color, paper_bgcolor=paper_bgcolor - ) - ) - return figure - - def toggle_chart_style(self): - """Toggle the chart style between light and dark mode.""" - if not hasattr(self._obbject.chart, "fig"): - raise ValueError( - "Error: No chart has been created. Please create a chart first." - ) - current = self._charting_settings.chart_style - new = "light" if current == "dark" else "dark" - self._charting_settings.chart_style = new - figure = self._obbject.chart.fig # type: ignore[union-attr] - updated_figure = self._set_chart_style(figure) - self._obbject.chart.fig = updated_figure # type: ignore[union-attr] - self._obbject.chart.content = updated_figure.show( # type: ignore[union-attr] - external=True - ).to_plotly_json() - - def table( - self, - data: Optional[Union[pd.DataFrame, pd.Series]] = None, - title: str = "", - ): - """Display an interactive table. - - Parameters - ---------- - data : Optional[Union[pd.DataFrame, pd.Series]], optional - Data to be plotted, by default None. - If no data is provided the OBBject results will be used. - title : str, optional - Title of the table, by default "". - """ - data_as_df, _ = self._prepare_data_as_df(data) - if isinstance(data_as_df.index, pd.RangeIndex): - data_as_df.reset_index(inplace=True, drop=True) - else: - data_as_df.reset_index(inplace=True) - if self._backend.isatty: - try: - self._backend.send_table( - df_table=data_as_df, - title=title - or self._obbject._route, # pylint: disable=protected-access - theme=self._charting_settings.table_style, - ) - except Exception as e: # pylint: disable=W0718 - warn(f"Failed to show figure with backend. {e}") - - else: - from plotly import ( # pylint:disable=import-outside-toplevel - optional_imports, - ) - - ipython_display = optional_imports.get_module("IPython.display") - if ipython_display: - ipython_display.display(ipython_display.HTML(data_as_df.to_html())) - else: - warn("IPython.display is not available.") - - def url( - self, - url: str, - title: str = "", - width: Optional[int] = None, - height: Optional[int] = None, - ): - """Return the URL of the chart.""" - try: - self._backend.send_url(url=url, title=title, width=width, height=height) - except Exception as e: # pylint: disable=W0718 - warn(f"Failed to show figure with backend. {e}") +Charting = ext.obbject_accessor(_Charting) diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/builder.py b/openbb_platform/obbject_extensions/charting/openbb_charting/builder.py deleted file mode 100644 index 8fb4302cd7a..00000000000 --- a/openbb_platform/obbject_extensions/charting/openbb_charting/builder.py +++ /dev/null @@ -1,53 +0,0 @@ -"""Script used to create charting functions in charting_router.py.""" - -from inspect import getmembers, isfunction -from typing import List - -from openbb_core.app.router import RouterLoader - -from openbb_charting import charting_router - -# refers to the root of the repo, not the root of the extension -# this script should be run from the root of the repo -CHARTING_ROUTER = ( - "openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py" -) - - -def get_routes() -> List[str]: - """Get all routes from the api router.""" - router = RouterLoader.from_extensions() - return [route.path for route in router.api_router.routes] # type: ignore - - -def create_charting_functions_w_routes(routes_to_add: List[str]): - """Create new charting functions in charting_router.py.""" - with open(CHARTING_ROUTER, "a") as f: - for route in routes_to_add: - original_route = route.replace("_", "/") - f.write( - f""" -def {route}(**kwargs) -> Tuple["OpenBBFigure", Dict[str, Any]]: - raise NotImplementedError("Command `{original_route}` does not have a charting function") -""" - ) - - -def get_charting_functions() -> List[str]: - """Get all functions from charting_router.py.""" - functions = getmembers(charting_router, isfunction) - if functions: - return [function[0] for function in functions] - return [] - - -if __name__ == "__main__": - routes = get_routes() - # format routes to be valid function names - routes = [route.replace("/", "_")[1:] for route in routes] - # get functions already in charting_router.py - known_functions = get_charting_functions() - # remove known functions from routes - routes = [route for route in routes if route not in known_functions] - # create new functions in charting_router.py - create_charting_functions_w_routes(routes) diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/charting.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charting.py new file mode 100644 index 00000000000..662b656142a --- /dev/null +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charting.py @@ -0,0 +1,563 @@ +"""Charting Class implementation.""" + +# pylint: disable=too-many-arguments,unused-argument + +from typing import ( + Any, + Callable, + ClassVar, + Dict, + List, + Literal, + Optional, + Tuple, + Type, + Union, +) +from warnings import warn + +import numpy as np +import pandas as pd +from importlib_metadata import entry_points +from openbb_core.app.model.charts.chart import Chart +from openbb_core.app.model.obbject import OBBject +from openbb_core.app.utils import basemodel_to_df, convert_to_basemodel +from openbb_core.provider.abstract.data import Data +from plotly.graph_objs import Figure + +from openbb_charting.charts.generic_charts import bar_chart, line_chart +from openbb_charting.charts.helpers import get_charting_functions +from openbb_charting.core.backend import Backend, create_backend, get_backend +from openbb_charting.core.openbb_figure import OpenBBFigure +from openbb_charting.query_params import ChartParams, IndicatorsParams + + +class Charting: + """Charting extension. + + Methods + ------- + show + Display chart and save it to the OBBject. + to_chart + Redraw the chart and save it to the OBBject, with an optional entry point for Data. + functions + Return a list of Platform commands with charting functions. + get_params + Return the charting parameters for the function the OBBject was created from. + indicators + Return the list of the available technical indicators to use with the `to_chart` method and OHLC+V data. + table + Display an interactive table. + create_line_chart + Create a line chart from external data. + create_bar_chart + Create a bar chart, on a single x-axis with one or more values for the y-axis, from external data. + toggle_chart_style + Toggle the chart style, of an existing chart, between light and dark mode. + """ + + _extension_views: ClassVar[List[Type]] = [ + entry_point.load() + for entry_point in entry_points(group="openbb_charting_extension") + ] + + def __init__(self, obbject): + """Initialize Charting extension.""" + # pylint: disable=import-outside-toplevel + from openbb_core.app.model.charts.charting_settings import ChartingSettings + + self._obbject: OBBject = obbject + self._charting_settings = ChartingSettings( + user_settings=self._obbject._user_settings, # type: ignore + system_settings=self._obbject._system_settings, # type: ignore + ) + self._backend: Backend = self._handle_backend() + self._functions: Dict[str, Callable] = self._get_functions() + + @classmethod + def indicators(cls): + """Return an instance of the IndicatorsParams class, containing all available indicators and their parameters. + + Without assigning to a variable, it will print the the information to the console. + """ + return IndicatorsParams() + + @classmethod + def functions(cls) -> List[str]: + """Return a list of the available functions.""" + functions: List[str] = [] + for view in cls._extension_views: + functions.extend(get_charting_functions(view)) + + return functions + + def _get_functions(self) -> Dict[str, Callable]: + """Return a dict with the available functions.""" + functions: Dict[str, Callable] = {} + for view in self._extension_views: + functions.update(get_charting_functions(view, with_objects=True)) + + return functions + + def _handle_backend(self) -> Backend: + """Create and start the backend.""" + create_backend(self._charting_settings) + backend = get_backend() + backend.start(debug=self._charting_settings.debug_mode) + return backend + + def _get_chart_function(self, route: str) -> Callable: + """Given a route, it returns the chart function. The module must contain the given route.""" + if route is None: + raise ValueError("OBBject was initialized with no function route.") + adjusted_route = route.replace("/", "_")[1:] + if adjusted_route not in self._functions: + raise ValueError( + f"Could not find the route `{adjusted_route}` in the charting functions." + ) + return self._functions[adjusted_route] + + def get_params(self) -> ChartParams: + """Return the ChartQueryParams class for the function the OBBject was created from. + + Without assigning to a variable, it will print the docstring to the console. + """ + if self._obbject._route is None: # pylint: disable=protected-access + raise ValueError("OBBject was initialized with no function route.") + charting_function = ( + self._obbject._route # pylint: disable=protected-access + ).replace("/", "_")[1:] + if hasattr(ChartParams, charting_function): + return getattr(ChartParams, charting_function)() + raise ValueError( + f"Error: No chart parameters are defined for the route: {charting_function}" + ) + + def _prepare_data_as_df( + self, data: Optional[Union[pd.DataFrame, pd.Series]] + ) -> Tuple[pd.DataFrame, bool]: + """Convert supplied data to a DataFrame.""" + has_data = ( + isinstance(data, (Data, pd.DataFrame, pd.Series)) and not data.empty # type: ignore + ) or (bool(data)) + index = ( + data.index.name + if has_data and isinstance(data, (pd.DataFrame, pd.Series)) + else None + ) + data_as_df: pd.DataFrame = ( + basemodel_to_df(convert_to_basemodel(data), index=index) + if has_data + else self._obbject.to_dataframe(index=index) + ) + if "date" in data_as_df.columns: + data_as_df = data_as_df.set_index("date") + if "provider" in data_as_df.columns: + data_as_df.drop(columns="provider", inplace=True) + return data_as_df, has_data + + # pylint: disable=too-many-locals + def create_line_chart( + self, + data: Union[ + list, + dict, + pd.DataFrame, + List[pd.DataFrame], + pd.Series, + List[pd.Series], + np.ndarray, + Data, + ], + index: Optional[str] = None, + target: Optional[str] = None, + title: Optional[str] = None, + x: Optional[str] = None, + xtitle: Optional[str] = None, + y: Optional[Union[str, List[str]]] = None, + ytitle: Optional[str] = None, + y2: Optional[Union[str, List[str]]] = None, + y2title: Optional[str] = None, + layout_kwargs: Optional[dict] = None, + scatter_kwargs: Optional[dict] = None, + normalize: bool = False, + returns: bool = False, + same_axis: bool = False, + render: bool = True, + **kwargs, + ) -> Union[OpenBBFigure, Figure, None]: + """Create a line chart from external data and render a chart or return the OpenBBFigure. + + Parameters + ---------- + data : Union[Data, pd.DataFrame, pd.Series] + Data to be plotted (OHLCV data). + index : Optional[str], optional + Index column, by default None + target : Optional[str], optional + Target column to be plotted, by default None + title : Optional[str], optional + Chart title, by default None + x : Optional[str], optional + X-axis column, by default None + xtitle : Optional[str], optional + X-axis title, by default None + y : Optional[Union[str, List[str]]], optional + Y-axis column(s), by default None + If None are supplied, the layout is optimized for the contents of data. + Where many units/scales are present, + it will attempt to divide based on the range of values. + ytitle : Optional[str], optional + Y-axis title, by default None + y2 : Optional[Union[str, List[str]]], optional + Y2-axis column(s), by default None + y2title : Optional[str], optional + Y2-axis title, by default None + layout_kwargs : Optional[dict], optional + Additional Plotly Layout parameters for `fig.update_layout`, by default None + scatter_kwargs : Optional[dict], optional + Additional Plotly parameters applied on creation of each scatter plot, by default None + normalize : bool, optional + Normalize the data with Z-Score Standardization, by default False + returns : bool, optional + Convert the data to cumulative returns, by default False + same_axis: bool, optional + If True, forces all data onto the same Y-axis, by default False + render: bool, optional + If True, the chart will be rendered, by default True + **kwargs: Dict[str, Any] + Extra parameters to be passed to `figure.show()` + """ + fig = line_chart( + data=data, + index=index, + target=target, + title=title, + x=x, + xtitle=xtitle, + y=y, + ytitle=ytitle, + y2=y2, + y2title=y2title, + layout_kwargs=layout_kwargs, + scatter_kwargs=scatter_kwargs, + normalize=normalize, + returns=returns, + same_axis=same_axis, + **kwargs, + ) + fig = self._set_chart_style(fig) + if render: + return fig.show(**kwargs) + + return fig + + def create_bar_chart( + self, + data: Union[ + list, + dict, + pd.DataFrame, + List[pd.DataFrame], + pd.Series, + List[pd.Series], + np.ndarray, + Data, + ], + x: str, + y: Union[str, List[str]], + barmode: Literal["group", "stack", "relative", "overlay"] = "group", + xtype: Literal[ + "category", "multicategory", "date", "log", "linear" + ] = "category", + title: Optional[str] = None, + xtitle: Optional[str] = None, + ytitle: Optional[str] = None, + orientation: Literal["h", "v"] = "v", + colors: Optional[List[str]] = None, + layout_kwargs: Optional[Dict[str, Any]] = None, + bar_kwargs: Optional[Dict[str, Any]] = None, + render: bool = True, + **kwargs, + ) -> Union[OpenBBFigure, Figure, None]: + """Create a bar chart on a single x-axis with one or more values for the y-axis. + + Parameters + ---------- + data : Union[list, dict, pd.DataFrame, List[pd.DataFrame], pd.Series, List[pd.Series], np.ndarray, Data] + Data to plot. + x : str + The x-axis column name. + y : Union[str, List[str]] + The y-axis column name(s). + barmode : Literal["group", "stack", "relative", "overlay"], optional + The bar mode, by default "group". + xtype : Literal["category", "multicategory", "date", "log", "linear"], optional + The x-axis type, by default "category". + title : str, optional + The title of the chart, by default None. + xtitle : str, optional + The x-axis title, by default None. + ytitle : str, optional + The y-axis title, by default None. + colors: List[str], optional + Manually set the colors to cycle through for each column in 'y', by default None. + bar_kwargs : Dict[str, Any], optional + Additional keyword arguments to apply with figure.add_bar(), by default None. + layout_kwargs : Dict[str, Any], optional + Additional keyword arguments to apply with figure.update_layout(), by default None. + Returns + ------- + OpenBBFigure + The OpenBBFigure object. + """ + fig = bar_chart( + data=data, + x=x, + y=y, + barmode=barmode, + xtype=xtype, + title=title, + xtitle=xtitle, + ytitle=ytitle, + orientation=orientation, + colors=colors, + bar_kwargs=bar_kwargs, + layout_kwargs=layout_kwargs, + ) + fig = self._set_chart_style(fig) + if render: + return fig.show(**kwargs) + + return fig + + # pylint: disable=inconsistent-return-statements + def show(self, render: bool = True, **kwargs): + """Display chart and save it to the OBBject.""" + try: + charting_function = self._get_chart_function( + self._obbject._route # pylint: disable=protected-access + ) + kwargs["obbject_item"] = self._obbject + kwargs["charting_settings"] = self._charting_settings # + kwargs["standard_params"] = ( + self._obbject._standard_params # pylint: disable=protected-access + ) + kwargs["extra_params"] = ( + self._obbject._extra_params # pylint: disable=protected-access + ) + kwargs["provider"] = self._obbject.provider + kwargs["extra"] = self._obbject.extra + fig, content = charting_function(**kwargs) + fig = self._set_chart_style(fig) + content = fig.show(external=True, **kwargs).to_plotly_json() + self._obbject.chart = Chart(fig=fig, content=content) + if render: + fig.show(**kwargs) + except Exception: # pylint: disable=W0718 + try: + fig = self.create_line_chart(data=self._obbject.results, render=False, **kwargs) # type: ignore + fig = self._set_chart_style(fig) + content = fig.show(external=True, **kwargs).to_plotly_json() # type: ignore + self._obbject.chart = Chart(fig=fig, content=content) + if render: + return fig.show(**kwargs) # type: ignore + except Exception as e: + raise RuntimeError( + "Failed to automatically create a generic chart with the data provided." + ) from e + + # pylint: disable=too-many-locals,inconsistent-return-statements + def to_chart( + self, + data: Optional[ + Union[ + list, + dict, + pd.DataFrame, + List[pd.DataFrame], + pd.Series, + List[pd.Series], + np.ndarray, + Data, + ] + ] = None, + target: Optional[str] = None, + index: Optional[str] = None, + indicators: Optional[Dict[str, Dict[str, Any]]] = None, + symbol: str = "", + candles: bool = True, + volume: bool = True, + volume_ticks_x: int = 7, + render: bool = True, + **kwargs, + ): + """Create an OpenBBFigure with user customizations (if any) and save it to the OBBject. + + This function is used to populate, or re-populate, the OBBject with a chart using the data within + the OBBject or external data supplied via the `data` parameter. + This function modifies the original OBBject by overwriting the existing chart. + + Parameters + ---------- + data : Union[Data, pd.DataFrame, pd.Series] + Data to be plotted. + indicators : Dict[str, Dict[str, Any]], optional + Indicators to be plotted, by default None + symbol : str, optional + Symbol to be plotted. This is used for labels and titles, by default "" + candles : bool, optional + If True, candles will be plotted, by default True + volume : bool, optional + If True, volume will be plotted, by default True + volume_ticks_x : int, optional + Volume ticks, by default 7 + render : bool, optional + If True, the chart will be rendered, by default True + kwargs: Dict[str, Any] + Extra parameters to be passed to the chart constructor. + + Examples + -------- + Plotting a time series with TA indicators + + >>> from openbb import obb + >>> res = obb.equity.price.historical("AAPL") + >>> indicators = dict( + >>> sma=dict(length=[20,30,50]), + >>> adx=dict(length=14), + >>> rsi=dict(length=14), + >>> macd=dict(fast=12, slow=26, signal=9), + >>> bbands=dict(length=20, std=2), + >>> stoch=dict(length=14), + >>> ema=dict(length=[20,30,50]), + >>> ) + >>> res.charting.to_chart(**{"indicators": indicators}) + + Get all the available indicators + + >>> res = obb.equity.price.historical("AAPL") + >>> indicators = res.charting.indicators() + >>> indicators? + """ + data_as_df, has_data = self._prepare_data_as_df(data) # type: ignore + if target is not None: + data_as_df = data_as_df[[target]] + kwargs["candles"] = candles + kwargs["volume"] = volume + kwargs["volume_ticks_x"] = volume_ticks_x + kwargs["indicators"] = indicators if indicators else {} + kwargs["symbol"] = symbol + kwargs["target"] = target + kwargs["index"] = index + kwargs["obbject_item"] = self._obbject + kwargs["charting_settings"] = self._charting_settings + kwargs["standard_params"] = ( + self._obbject._standard_params # pylint: disable=protected-access + ) + kwargs["extra_params"] = ( + self._obbject._extra_params # pylint: disable=protected-access + ) + kwargs["provider"] = self._obbject.provider # pylint: disable=protected-access + kwargs["extra"] = self._obbject.extra # pylint: disable=protected-access + try: + if has_data: + self.show(data=data_as_df, render=render, **kwargs) + else: + self.show(**kwargs, render=render) + except Exception: # pylint: disable=W0718 + try: + fig = self.create_line_chart(data=data_as_df, render=False, **kwargs) + fig = self._set_chart_style(fig) + content = fig.show(external=True, **kwargs).to_plotly_json() # type: ignore + self._obbject.chart = Chart(fig=fig, content=content) + if render: + return fig.show(**kwargs) # type: ignore + except Exception as e: # pylint: disable=W0718 + raise RuntimeError( + "Failed to automatically create a generic chart with the data provided." + ) from e + + def _set_chart_style(self, figure: Figure): + """Set the user preference for light or dark mode.""" + style = self._charting_settings.chart_style + font_color = "black" if style == "light" else "white" + paper_bgcolor = "white" if style == "light" else "black" + figure = figure.update_layout( + dict( # pylint: disable=R1735 + font_color=font_color, paper_bgcolor=paper_bgcolor + ) + ) + return figure + + def toggle_chart_style(self): + """Toggle the chart style between light and dark mode.""" + if not hasattr(self._obbject.chart, "fig"): + raise ValueError( + "Error: No chart has been created. Please create a chart first." + ) + current = self._charting_settings.chart_style + new = "light" if current == "dark" else "dark" + self._charting_settings.chart_style = new + figure = self._obbject.chart.fig # type: ignore[union-attr] + updated_figure = self._set_chart_style(figure) + self._obbject.chart.fig = updated_figure # type: ignore[union-attr] + self._obbject.chart.content = updated_figure.show( # type: ignore[union-attr] + external=True + ).to_plotly_json() + + def table( + self, + data: Optional[Union[pd.DataFrame, pd.Series]] = None, + title: str = "", + ): + """Display an interactive table. + + Parameters + ---------- + data : Optional[Union[pd.DataFrame, pd.Series]], optional + Data to be plotted, by default None. + If no data is provided the OBBject results will be used. + title : str, optional + Title of the table, by default "". + """ + data_as_df, _ = self._prepare_data_as_df(data) + if isinstance(data_as_df.index, pd.RangeIndex): + data_as_df.reset_index(inplace=True, drop=True) + else: + data_as_df.reset_index(inplace=True) + if self._backend.isatty: + try: + self._backend.send_table( + df_table=data_as_df, + title=title + or self._obbject._route, # pylint: disable=protected-access + theme=self._charting_settings.table_style, + ) + except Exception as e: # pylint: disable=W0718 + warn(f"Failed to show figure with backend. {e}") + + else: + from plotly import ( # pylint:disable=import-outside-toplevel + optional_imports, + ) + + ipython_display = optional_imports.get_module("IPython.display") + if ipython_display: + ipython_display.display(ipython_display.HTML(data_as_df.to_html())) + else: + warn("IPython.display is not available.") + + def url( + self, + url: str, + title: str = "", + width: Optional[int] = None, + height: Optional[int] = None, + ): + """Return the URL of the chart.""" + try: + self._backend.send_url(url=url, title=title, width=width, height=height) + except Exception as e: # pylint: disable=W0718 + warn(f"Failed to show figure with backend. {e}") diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py deleted file mode 100644 index 26b31d6289c..00000000000 --- a/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py +++ /dev/null @@ -1,1445 +0,0 @@ -"""Charting router.""" - -# pylint: disable=too-many-arguments, too-many-locals, too-many-branches, too-many-statements, unused-argument, too-many-lines - -from typing import Any, Dict, Optional, Tuple, Union -from warnings import warn - -import pandas as pd -from openbb_core.app.model.charts.chart import ChartFormat -from openbb_core.app.utils import basemodel_to_df -from openbb_core.provider.abstract.data import Data -from plotly.graph_objs import Figure - -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.core.to_chart import to_chart -from openbb_charting.styles.colors import LARGE_CYCLER -from openbb_charting.utils import relative_rotation -from openbb_charting.utils.generic_charts import bar_chart -from openbb_charting.utils.helpers import ( - calculate_returns, - duration_sorter, - heikin_ashi, - should_share_axis, - z_score_standardization, -) - -CHART_FORMAT = ChartFormat.plotly - - -def equity_price_performance( - **kwargs, -) -> Tuple[Union[OpenBBFigure, Figure], Dict[str, Any]]: # noqa: PLR0912 - """Equity Price Performance Chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - elif "data" in kwargs and isinstance(kwargs["data"], list): - data = basemodel_to_df(kwargs["data"], index=kwargs.get("index", "symbol")) # type: ignore - else: - data = basemodel_to_df( - kwargs["obbject_item"], index=kwargs.get("index", "symbol") # type: ignore - ) - - cols = [ - "one_day", - "one_week", - "one_month", - "three_month", - "six_month", - "ytd", - "one_year", - "two_year", - "three_year", - "four_year", - "five_year", - ] - - df = pd.DataFrame() - chart_df = pd.DataFrame() - - if "symbol" in data.columns: - data = data.set_index("symbol") - chart_cols = [] - - if len(data) == 0: - raise ValueError("No data was found in the DataFrame.") - - data = data.drop_duplicates(keep="first") - - for col in cols: - if col in data.columns and data[col].notnull().any(): - df[col.replace("_", " ").title() if col != "ytd" else col.upper()] = data[ - col - ].apply(lambda x: round(x * 100, 4) if x is not None else None) - - if df.empty: - raise ValueError(f"No columns matching, {cols}, were found in the data.") - - chart_df = df.T - chart_cols = chart_df.columns.to_list() - - if "limit" in kwargs and isinstance(kwargs.get("limit"), int): - limit = kwargs.pop("limit", 10) - chart_df = chart_df.head(limit) # type: ignore - - layout_kwargs: Dict[str, Any] = kwargs.get("layout_kwargs", {}) - - title = ( - f"{kwargs.pop('title')}" if "title" in kwargs else "Equity Price Performance" - ) - orientation = ( - kwargs.pop("orientation") - if "orientation" in kwargs and kwargs.get("orientation") is not None - else "v" - ) - - ytitle = "Performance (%)" - xtitle = None - - if orientation == "h": - xtitle = ytitle # type: ignore - ytitle = None # type: ignore - - fig = bar_chart( - chart_df.reset_index(), - x="index", - y=chart_cols, - title=title, - xtitle=xtitle, - ytitle=ytitle, - orientation=orientation, # type: ignore - ) - fig.update_traces( - hovertemplate=( - "%{fullData.name}:%{y:.2f}%" - if orientation == "v" - else "%{fullData.name}:%{x:.2f}%" - ) - ) - - fig.update_layout(**layout_kwargs) - content = fig.show(external=True).to_plotly_json() # type: ignore - - return fig, content - - -def etf_price_performance( - **kwargs, -) -> Tuple[Union[OpenBBFigure, Figure], Dict[str, Any]]: - """ETF Historical Chart.""" - fig, content = equity_price_performance(**kwargs) - if "title" in kwargs and kwargs.get("title") is not None: - fig.set_title(kwargs.get("title")) # type: ignore - else: - fig.set_title("ETF Price Performance") # type: ignore - - content = fig.show(external=True).to_plotly_json() # type: ignore - return fig, content - - -def etf_holdings(**kwargs) -> Tuple[Union[OpenBBFigure, Figure], Dict[str, Any]]: - """Equity Compare Groups Chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - elif "data" in kwargs and isinstance(kwargs["data"], list): - data = basemodel_to_df(kwargs["data"], index=None) # type: ignore - else: - data = basemodel_to_df(kwargs["obbject_item"], index=None) # type: ignore - - if "weight" not in data.columns: - raise ValueError("No 'weight' column found in the data.") - - orientation = kwargs.get("orientation", "h") - limit = kwargs.get("limit", 20) - symbol = kwargs["standard_params"].get("symbol") # type: ignore - title = kwargs.get("title", f"Top {limit} {symbol} Holdings") - layout_kwargs = kwargs.get("layout_kwargs", {}) - - data = data.sort_values("weight", ascending=False) - limit = min(limit, len(data)) # type: ignore - target = data.head(limit)[["symbol", "weight"]].set_index("symbol") - target = target.multiply(100) - axis_title = "Weight (%)" - - fig = bar_chart( - target.reset_index(), - "symbol", - ["weight"], - title=title, # type: ignore - xtitle=axis_title if orientation == "h" else None, - ytitle=axis_title if orientation == "v" else None, - orientation=orientation, # type: ignore - ) - - fig.update_layout( - hovermode="x" if orientation == "v" else "y", - margin=dict(r=0, l=50) if orientation == "h" else None, - ) - - fig.update_traces( - hovertemplate=( - "%{y:.3f}%" - if orientation == "v" - else "%{x:.3f}%" - ) - ) - - if layout_kwargs: - fig.update_layout(**layout_kwargs) # type: ignore - - content = fig.show(external=True).to_plotly_json() # type: ignore - - return fig, content - - -def equity_price_historical( # noqa: PLR0912 - **kwargs, -) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Equity Price Historical Chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - elif "data" in kwargs and isinstance(kwargs["data"], list): - data = basemodel_to_df(kwargs["data"], index=kwargs.get("index", "date")) # type: ignore - else: - data = basemodel_to_df( - kwargs["obbject_item"], index=kwargs.get("index", "date") # type: ignore - ) - - if "date" in data.columns: - data = data.set_index("date") - - target = str(kwargs.get("target")) - normalize = kwargs.get("normalize") is True - returns = kwargs.get("returns") is True - same_axis = kwargs.get("same_axis") is True - text_color = "black" if ChartStyle().plt_style == "light" else "white" - title = f"{kwargs.get('title')}" if "title" in kwargs else "Historical Prices" - y1title = "" - y2title = "" - candles = True - multi_symbol = ( - bool(kwargs.get("multi_symbol") is True) - or ( - "symbol" in data.columns - and target in data.columns - and len(data.symbol.unique()) > 1 - ) - or ("target" in kwargs and kwargs.get("target") is not None) - or "symbol" in data.columns - or ( - "symbol" not in data.columns - and bool(data.columns.isin(["open", "high", "low", "close"]).all()) - ) - ) - target = "close" if target is None or target == "None" or target == "" else target - - if multi_symbol is True: - if "symbol" not in data.columns and target in data.columns: - data = data[[target]] - y1title = target.title() - if "symbol" in data.columns and target in data.columns: - data = data.pivot(columns="symbol", values=target) - y1title = target - title = f"Historical {target.title()}" - - indicators = kwargs.get("indicators", {}) - candles = bool(~data.columns.isin(["open", "high", "low", "close"]).all()) - candles = candles if kwargs.get("candles", True) else False - volume = kwargs.get("volume", True) if "volume" in data.columns else False - - if normalize is True: - if "symbol" not in data.columns and target in data.columns: - data = data[[target]] - multi_symbol = True - candles = False - volume = False - - if returns is True: - if "symbol" not in data.columns and target in data.columns: - data = data[[target]] - multi_symbol = True - candles = False - volume = False - if ( # pylint: disable = R0916 - multi_symbol is False - and normalize is False - and returns is False - and candles is True - ) or (indicators and multi_symbol is False): - if ( - "heikin_ashi" in kwargs - and kwargs["heikin_ashi"] is True - and candles is True - ): - data = heikin_ashi(data) - title = f"{title} - Heikin Ashi" - _volume = False - if "atr" in indicators: # type: ignore - _volume = volume - volume = False - ta = PlotlyTA() - fig = ta.plot( # type: ignore - data, - indicators=indicators if indicators else {}, # type: ignore - symbol=target if candles is False else "", - candles=candles, - volume=volume, # type: ignore - ) - if _volume is True and "atr" in indicators: # type: ignore - fig.add_inchart_volume(data) - fig.update_layout( - paper_bgcolor="rgba(0,0,0,0)", - plot_bgcolor="rgba(0,0,0,0)", - font=dict(color=text_color), - showlegend=True, - legend=dict( - orientation="v", - yanchor="top", - xanchor="right", - y=0.95, - x=-0.01, - xref="paper", - font=dict(size=12), - bgcolor="rgba(0,0,0,0)", - ), - xaxis=dict( - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - zeroline=True, - mirror=True, - showline=True, - ), - xaxis2=dict( - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - zeroline=True, - mirror=True, - showline=True, - ), - yaxis=dict( - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - zeroline=True, - mirror=True, - showline=True, - tickfont=dict(size=14), - ), - yaxis2=dict( - ticklen=0, - gridcolor="rgba(128,128,128,0.3)", - ), - yaxis3=dict( - ticklen=0, - gridcolor="rgba(128,128,128,0.3)", - ), - dragmode="pan", - hovermode="x", - ) - - if kwargs.get("title"): - title = kwargs["title"] - fig.update_layout(title=dict(text=title, x=0.5)) - - content = fig.to_plotly_json() - - return fig, content - - if multi_symbol is True or candles is False: - - if "symbol" not in data.columns and target in data.columns: - data = data[[target]] - - if "symbol" in data.columns: - data = data.pivot(columns="symbol", values=target) - - title: str = kwargs.get("title", "Historical Prices") # type: ignore - - y1title = data.iloc[:, 0].name - y2title = "" - - if len(data.columns) > 2 or normalize is True or returns is True: - if returns is True or (len(data.columns) > 2 and normalize is False): - data = data.apply(calculate_returns) - title = f"{title} - Cumulative Returns" - y1title = "Percent" - if normalize is True: - if returns is True: - title = f"{title.replace(' - Cumulative Returns', '')} - Normalized Cumulative Returns" - else: - title = title + " - Normalized" - data = data.apply(z_score_standardization) - y1title = None # type: ignore - y2title = None # type: ignore - - fig = OpenBBFigure() - - for i, col in enumerate(data.columns): - - hovertemplate = f"{data[col].name}: %{{y}}" - yaxis = "y1" - if y1title and y1title != "Percent": - yaxis = ( - ( - "y1" - if should_share_axis(data, col, y1title) # type: ignore - or col == y1title - or normalize is True - or returns is True - else "y2" - ) - if same_axis is False - else "y1" - ) - - if yaxis == "y2": - y2title = data[col].name - - fig.add_scatter( - x=data.index, - y=data[col], - name=data[col].name, - mode="lines", - hovertemplate=hovertemplate, - line=dict(width=2, color=LARGE_CYCLER[i % len(LARGE_CYCLER)]), - yaxis=yaxis, - ) - - if normalize is True or returns is True: - y1title = "Percent" if returns is True else None # type: ignore - y2title = None # type: ignore - - if same_axis is True: - y1title = None # type: ignore - y2title = None # type: ignore - - fig.update_layout( - paper_bgcolor="rgba(0,0,0,0)", - plot_bgcolor="rgba(0,0,0,0)", - legend=( - dict( - orientation="v", - yanchor="top", - xanchor="right", - y=0.95, - x=-0.01, - bgcolor="rgba(0,0,0,0)", - ) - if len(data.columns) > 2 - else dict( - orientation="h", - yanchor="bottom", - xanchor="right", - y=1.02, - x=0.98, - bgcolor="rgba(0,0,0,0)", - ) - ), - yaxis1=( - dict( - side="right", - ticklen=0, - showgrid=True, - showline=True, - mirror=True, - gridcolor="rgba(128,128,128,0.3)", - title=dict( - text=y1title if y1title else None, standoff=20, font=dict(size=20) - ), - tickfont=dict(size=14), - anchor="x", - ) - ), - yaxis2=( - dict( - overlaying="y", - side="left", - ticklen=0, - showgrid=False, - title=dict( - text=y2title if y2title else None, standoff=10, font=dict(size=20) - ), - tickfont=dict(size=14), - anchor="x", - ) - if y2title - else None - ), - xaxis=dict( - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - showline=True, - mirror=True, - ), - margin=dict(l=20, r=20, b=20, t=20), - dragmode="pan", - hovermode="x", - ) - if kwargs.get("title"): - title = kwargs["title"] - fig.update_layout(title=dict(text=title, x=0.5)) - - content = fig.show(external=True).to_plotly_json() - - return fig, content - - -def etf_historical(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """ETF Historical Chart.""" - return equity_price_historical(**kwargs) - - -def index_price_historical(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Index Price Historical Chart.""" - return equity_price_historical(**kwargs) - - -def currency_price_historical(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Currency Price Historical Chart.""" - return equity_price_historical(**kwargs) - - -def crypto_price_historical(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Crypto Price Historical Chart.""" - return equity_price_historical(**kwargs) - - -def _ta_ma(**kwargs): - """Plot moving average helper.""" - index = ( - kwargs.get("index") - if "index" in kwargs and kwargs.get("index") is not None - else "date" - ) - data = kwargs.get("data") - ma_type = ( - kwargs["ma_type"] - if "ma_type" in kwargs and kwargs.get("ma_type") is not None - else "sma" - ) - ma_types = ma_type.split(",") if isinstance(ma_type, str) else ma_type - - if isinstance(data, pd.DataFrame) and not data.empty: - data = data.set_index(index) if index in data.columns else data - - if data is None: - data = basemodel_to_df(kwargs["obbject_item"], index=index) - - if isinstance(data, list): - data = basemodel_to_df(data, index=index) - - window = ( - kwargs.get("length", []) - if "length" in kwargs and kwargs.get("length") is not None - else [50] - ) - offset = kwargs.get("offset", 0) - target = ( - kwargs.get("target") - if "target" in kwargs and kwargs.get("target") is not None - else "close" - ) - - if target not in data.columns and "close" in data.columns: - target = "close" - - if target not in data.columns and "close" not in data.columns: - raise ValueError(f"Column '{target}', or 'close', not found in the data.") - - df = data.copy() - if target in data.columns: - df = df[[target]] - df.columns = ["close"] - title = ( - kwargs.get("title") - if "title" in kwargs and kwargs.get("title") is not None - else f"{ma_type.upper()}" - ) - - fig = OpenBBFigure() - fig = fig.create_subplots( - 1, - 1, - shared_xaxes=True, - vertical_spacing=0.06, - horizontal_spacing=0.01, - row_width=[1], - specs=[[{"secondary_y": True}]], - ) - fig.update_layout(ChartStyle().plotly_template.get("layout", {})) - - ma_df = pd.DataFrame() - window = [window] if isinstance(window, int) else window - for w in window: - for ma_type in ma_types: - ma_df[f"{ma_type.upper()} {w}"] = getattr(df.ta, ma_type)( - length=w, offset=offset - ) - - if kwargs.get("dropnan") is True: - ma_df = ma_df.dropna() - data = data.iloc[-len(ma_df) :] - - if ( - "candles" in kwargs - and kwargs.get("candles") is True - and kwargs.get("target") is None - ): - volume = kwargs.get("volume") is True - fig, _ = to_chart(data, candles=True, volume=volume) - - else: - ma_df[f"{target}".title()] = data[target] - - for i, col in enumerate(ma_df.columns): - name = col.replace("_", " ") - fig.add_scatter( - x=ma_df.index, - y=ma_df[col], - name=name, - mode="lines", - hovertemplate=f"{name}: %{{y}}", - line=dict(width=1, color=LARGE_CYCLER[i]), - showlegend=True, - ) - - 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)", - showlegend=True, - legend=dict( - orientation="h", - yanchor="bottom", - xanchor="right", - y=1.02, - x=0.95, - bgcolor="rgba(0,0,0,0)", - ), - xaxis=dict( - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - zeroline=True, - mirror=True, - ), - yaxis=dict( - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - zeroline=True, - mirror=True, - autorange=True, - ), - ) - - content = fig.show(external=True).to_plotly_json() - - return fig, content - - -def technical_sma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Plot simple moving average chart.""" - if "ma_type" not in kwargs: - kwargs["ma_type"] = "sma" - return _ta_ma(**kwargs) - - -def technical_ema(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Exponential moving average chart.""" - if "ma_type" not in kwargs: - kwargs["ma_type"] = "ema" - return _ta_ma(**kwargs) - - -def technical_hma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Hull moving average chart.""" - if "ma_type" not in kwargs: - kwargs["ma_type"] = "hma" - return _ta_ma(**kwargs) - - -def technical_wma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Weighted moving average chart.""" - if "ma_type" not in kwargs: - kwargs["ma_type"] = "wma" - return _ta_ma(**kwargs) - - -def technical_zlma(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Zero lag moving average chart.""" - if "ma_type" not in kwargs: - kwargs["ma_type"] = "zlma" - return _ta_ma(**kwargs) - - -def technical_aroon(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Technical Aroon Chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - else: - data = basemodel_to_df( - kwargs["obbject_item"], index=kwargs.get("index", "date") - ) - - if "date" in data.columns: - data = data.set_index("date") - - if "symbol" in data.columns and len(data.symbol.unique()) > 1: - raise ValueError( - "Please provide data with only one symbol and columns for OHLC." - ) - - symbol = kwargs.get("symbol", "") - - volume = kwargs.get("volume") is True - title = f"Aroon Indicator & Oscillator {symbol}" - - length = kwargs.get("length", 25) - scalar = kwargs.get("scalar", 100) - symbol = kwargs.get("symbol", "") - - ta = PlotlyTA() - fig = ta.plot( # type: ignore - data, - dict(aroon=dict(length=length, scalar=scalar)), - title, - False, - volume=volume, - ) - - content = fig.show(external=True).to_plotly_json() - - return fig, content - - -def technical_macd(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Plot moving average convergence divergence chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - else: - data = basemodel_to_df( - kwargs["obbject_item"], index=kwargs.get("index", "date") - ) - - if "date" in data.columns: - data = data.set_index("date") - - if "symbol" in data.columns and len(data.symbol.unique()) > 1: - raise ValueError( - "Please provide data with only one symbol and columns for OHLC." - ) - - fast = kwargs.get("fast", 12) - slow = kwargs.get("slow", 26) - signal = kwargs.get("signal", 9) - symbol = kwargs.get("symbol", "") - - title = f"{symbol.upper()} MACD" - volume = kwargs.get("volume") is True - - ta = PlotlyTA() - fig = ta.plot( # type: ignore - data, - dict(macd=dict(fast=fast, slow=slow, signal=signal)), - title, - False, - volume=volume, - ) - content = fig.show(external=True).to_plotly_json() - - return fig, content - - -def technical_adx(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Average directional movement index chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - else: - data = basemodel_to_df( - kwargs["obbject_item"], index=kwargs.get("index", "date") - ) - - if "date" in data.columns: - data = data.set_index("date") - - if "symbol" in data.columns and len(data.symbol.unique()) > 1: - raise ValueError( - "Please provide data with only one symbol and columns for OHLC." - ) - - length = kwargs.get("length", 14) - scalar = kwargs.get("scalar", 100.0) - drift = kwargs.get("drift", 1) - symbol = kwargs.get("symbol", "") - - ta = PlotlyTA() - fig = ta.plot( # type: ignore - data, - dict(adx=dict(length=length, scalar=scalar, drift=drift)), - f"Average Directional Movement Index (ADX) {symbol}", - False, - volume=False, - ) - content = fig.show(external=True).to_plotly_json() - - return fig, content - - -def technical_rsi(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Relative strength index chart.""" - if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): - data = kwargs["data"] - else: - data = basemodel_to_df( - kwargs["obbject_item"], index=kwargs.get("index", "date") - ) - - if "date" in data.columns: - data = data.set_index("date") - - if "symbol" in data.columns and len(data.symbol.unique()) > 1: - raise ValueError( - "Please provide data with only one symbol and columns for OHLC." - ) - - window = kwargs.get("window", 14) - scalar = kwargs.get("scalar", 100.0) - drift = kwargs.get("drift", 1) - symbol = kwargs.get("symbol", "") - - ta = PlotlyTA() - fig = ta.plot( # type: ignore - data, - dict(rsi=dict(length=window, scalar=scalar, drift=drift)), - f"{symbol.upper()} RSI {window}", - False, - volume=False, - ) - content = fig.show(external=True).to_plotly_json() - - return fig, content - - -def technical_cones(**kwargs) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Volatility Cones Chart.""" - data = kwargs.get("data") - - 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") # type: ignore - - df_ta.columns = [col.title().replace("_", " ") for col in df_ta.columns] - - # Check if the data is formatted as expected. - if not all(col in df_ta.columns for col in ["Realized", "Min", "Median", "Max"]): - raise ValueError("Data supplied does not match the expected format.") - - model = ( - str(kwargs.get("model")) - .replace("std", "Standard Deviation") - .replace("_", "-") - .title() - if kwargs.get("model") - else "Standard Deviation" - ) - - symbol = str(kwargs.get("symbol")) + " - " if kwargs.get("symbol") else "" - - title = ( - str(kwargs.get("title")) - if kwargs.get("title") - else f"{symbol}Realized Volatility Cones - {model} Model" - ) - - colors = [ - "green", - "red", - "burlywood", - "grey", - "orange", - "blue", - ] - - fig = OpenBBFigure() - - fig.update_layout(ChartStyle().plotly_template.get("layout", {})) - - text_color = "black" if ChartStyle().plt_style == "light" else "white" - - for i, col in enumerate(df_ta.columns): - fig.add_scatter( - x=df_ta.index, - y=df_ta[col], - name=col, - mode="lines+markers", - hovertemplate=f"{col}: %{{y}}", - marker=dict( - color=colors[i], - size=11, - ), - ) - - fig.set_title(title) - - fig.update_layout( - 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, - showgrid=True, - showline=True, - mirror=True, - zeroline=False, - gridcolor="rgba(128,128,128,0.3)", - ), - xaxis=dict( - type="category", - tickmode="array", - ticklen=0, - tickvals=df_ta.index, - ticktext=df_ta.index, - title_text="Period", - showgrid=False, - showline=True, - mirror=True, - zeroline=False, - ), - margin=dict(l=20, r=20, b=20), - dragmode="pan", - ) - - content = fig.to_plotly_json() - - return fig, content - - -def economy_fred_series( # noqa: PLR0912 - **kwargs, -) -> 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", - } - - 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() - - metadata = kwargs["extra"].get("results_metadata", {}) # 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 has_params is True and not y_units: - y_units = [ytitle_dict.get(params.transform)] # type: ignore - - if normalize or ( - kwargs.get("bar") is True - and len(y_units) > 1 - and ( - has_params is False - or not any( - i in params.transform for i in ["pc1", "pch", "pca", "cch", "cca", "log"] # type: ignore - ) - ) - ): - normalize = True - df_ta = df_ta.apply(z_score_standardization) - - if len(y_units) > 2 and has_params is False and allow_unsafe is False: - 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] if y_units else None - y1title = y1_units - y2title = y_units[1] if len(y_units) > 1 else None - xtitle = str(kwargs.get("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. - title: str = "" - if isinstance(kwargs, dict) and title in kwargs: - title = kwargs["title"] # type: ignore - else: - if metadata.get(columns[0]): # type: ignore - 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: Optional[str] = "" - - if kwargs.get("plot_bar") is True or len(df_ta.index) < 100: - margin = dict(l=10, r=5, b=75 if xtitle else 30) - try: - if normalize: - y1title = None - title = f"{title} - Normalized" if title else "Normalized" - bar_mode = kwargs.get("barmode", "group") - fig = bar_chart( - df_ta.reset_index(), - "date", - df_ta.columns.to_list(), - title=title, - xtitle=xtitle, - ytitle=y1title, - barmode=bar_mode, # type: ignore - layout_kwargs=dict(margin=margin), # type: ignore - ) - if kwargs.get("layout_kwargs"): - fig.update_layout(kwargs.get("layout_kwargs")) - - if kwargs.get("title"): - fig.set_title(str(kwargs.get("title"))) # type: ignore - - content = fig.to_plotly_json() - - return fig, content # type: ignore - except Exception as _: - warn("Bar chart failed. Attempting line chart.") - - # Create the figure object with subplots. - fig = OpenBBFigure().create_subplots( - rows=1, cols=1, shared_xaxes=True, shared_yaxes=False - ) - - # 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 # type: ignore - or y2title is None # type: ignore - or kwargs.get("same_axis") is True - ) - if metadata.get(col) # type: ignore - 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 # type: ignore - if on_y3: - yaxes = "y3" - y3title = df_ta[col].name # type: ignore - 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}}", - line=dict(width=2, color=LARGE_CYCLER[i % len(LARGE_CYCLER)]), - yaxis="y1" if kwargs.get("same_axis") is True else 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 isinstance(kwargs, dict) and "xtitle" in kwargs: - xtitle = kwargs["xtitle"] - # If the data was normalized, set the title to reflect this. - if normalize: - y1title = None - y2title = None - y3title = None - title = f"{title} - Normalized" if title else "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)", - legend=dict( - orientation="h", - yanchor="bottom", - xanchor="right", - y=1.02, - x=0.95, - bgcolor="rgba(0,0,0,0)", - font=dict(size=12), - ), - yaxis=( - dict( - ticklen=0, - side="right", - showline=True, - mirror=True, - title=dict(text=y1title, standoff=30, font=dict(size=16)), - tickfont=dict(size=14), - anchor="x", - gridcolor="rgba(128,128,128,0.3)", - ) - 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=16) - ), - 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.9)"), - anchor="free", - ) - if y3title - else None - ), - xaxis=dict( - ticklen=0, - showgrid=True, - showline=True, - mirror=True, - title=( - dict(text=xtitle, standoff=30, font=dict(size=16)) if xtitle else None - ), - gridcolor="rgba(128,128,128,0.3)", - domain=[0.095, 0.95] if y3title else None, - ), - margin=dict(r=25, l=25, b=75 if xtitle else 30) if normalize is False else None, - autosize=True, - dragmode="pan", - ) - if kwargs.get("layout_kwargs"): - fig.update_layout(kwargs.get("layout_kwargs")) - if kwargs.get("title"): - fig.set_title(str(kwargs.get("title"))) - content = fig.to_plotly_json() - - return fig, content - - -def technical_relative_rotation( - **kwargs: Any, -) -> Tuple["OpenBBFigure", Dict[str, Any]]: - """Relative Rotation Chart.""" - ratios_df = basemodel_to_df(kwargs["obbject_item"].rs_ratios, index="date") # type: ignore - momentum_df = basemodel_to_df(kwargs["obbject_item"].rs_momentum, index="date") # type: ignore - benchmark_symbol = kwargs["obbject_item"].benchmark # type: ignore - study = kwargs.get("study", None) - study = str(kwargs["obbject_item"].study) if study is None else str(study) - show_tails = kwargs.get("show_tails") - show_tails = True if show_tails is None else show_tails - tail_periods = int(kwargs.get("tail_periods")) if "tail_periods" in kwargs else 16 # type: ignore - tail_interval = str(kwargs.get("tail_interval")) if "tail_interval" in kwargs else "week" # type: ignore - date = kwargs.get("date") if "date" in kwargs else None # type: ignore - show_tails = False if date is not None else show_tails - if ratios_df.empty or momentum_df.empty: - raise RuntimeError("Error: No data to plot.") - - if show_tails is True: - fig = relative_rotation.create_rrg_with_tails( - ratios_df, momentum_df, study, benchmark_symbol, tail_periods, tail_interval # type: ignore - ) - else: - fig = relative_rotation.create_rrg_without_tails( - ratios_df, momentum_df, benchmark_symbol, study, date # type: ignore - ) - - figure = OpenBBFigure(fig) - font_color = "black" if ChartStyle().plt_style == "light" else "white" - figure.update_layout( - paper_bgcolor="rgba(0,0,0,0)", - plot_bgcolor="rgba(255,255,255,1)", - font=dict(color=font_color), - yaxis=dict( - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - side="left", - showline=True, - zeroline=True, - mirror=True, - ticklen=0, - tickfont=dict(size=14), - titlefont=dict(size=16), - ), - xaxis=dict( - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - showline=True, - zeroline=True, - mirror=True, - ticklen=0, - tickfont=dict(size=14), - titlefont=dict(size=16), - hoverformat="", - ), - hoverlabel=dict( - font_size=12, - ), - hovermode="x", - hoverdistance=50, - ) - if kwargs.get("title") is not None: - figure.set_title(str(kwargs.get("title"))) - content = figure.to_plotly_json() - - return figure, content - - -def fixedincome_government_yield_curve( # noqa: PLR0912 - **kwargs, -) -> Tuple[OpenBBFigure, Dict[str, Any]]: - """Government Yield Curve Chart.""" - data = kwargs.get("data", None) - df: pd.DataFrame = pd.DataFrame() - if data: - if isinstance(data, pd.DataFrame) and not data.empty: # noqa: SIM108 - df = data - elif isinstance(data, (list, Data)): - df = basemodel_to_df(data, index=None) # type: ignore - else: - pass - else: - df = pd.DataFrame([d.model_dump() for d in kwargs["obbject_item"]]) # type: ignore - - if df.empty: - raise ValueError("Error: No data to plot.") - - if "maturity" not in df.columns: - raise ValueError("Error: Maturity column not found in the data.") - - if "rate" not in df.columns: - raise ValueError("Error: Rate column not found in the data.") - - if "date" not in df.columns: - raise ValueError("Error: Date column not found in the data.") - - provider = kwargs.get("provider") - df["date"] = df["date"].astype(str) - maturities = duration_sorter(df["maturity"].unique().tolist()) - - # Use the supplied colors, if any. - colors = kwargs.get("colors", []) - if not colors: - colors = LARGE_CYCLER - color_count = 0 - - figure = OpenBBFigure().create_subplots(shared_xaxes=True) - figure.update_layout(ChartStyle().plotly_template.get("layout", {})) - - def create_fig(figure, df, dates, color_count, country: Optional[str] = None): - """Create a scatter for each date in the data.""" - for date in dates: - color = colors[color_count % len(colors)] - plot_df = df[df["date"] == date].copy() - plot_df["rate"] = plot_df["rate"].apply(lambda x: x * 100) - plot_df = ( - plot_df.drop(columns=["date"]) - .set_index("maturity") - .filter(items=maturities, axis=0) - .reset_index() - .rename(columns={"maturity": "Maturity", "rate": "Yield"}) - ) - plot_df["Maturity"] = [ - ( - d.split("_")[1] + " " + d.split("_")[0].title() - if d != "long_term" - else "Long Term" - ) - for d in plot_df["Maturity"] - ] - figure.add_scatter( - x=plot_df["Maturity"], - y=plot_df["Yield"], - # fill=fill, - mode="lines+markers", - name=f"{country} - {date}" if country else date, - line=dict(width=3, color=color), - marker=dict(size=10, color=color), - hovertemplate=( - "Maturity: %{x}
Yield: %{y}%" - if len(dates) == 1 - else "%{fullData.name}
Maturity: %{x}
Yield: %{y}%" - ), - ) - color_count += 1 - return figure, color_count - - dates = df.date.unique().tolist() - figure, color_count = create_fig(figure, df, dates, color_count) - extra_params = kwargs.get("extra_params") - extra_params = ( - extra_params if isinstance(extra_params, dict) else extra_params.__dict__ - ) - # Set the title for the chart - country: str = "" - if provider in ("federal_reserve", "fmp"): - country = "United States" - elif provider == "ecb": - curve_type = extra_params.get("yield_curve_type", "").replace("_", " ").title() - grade = extra_params.get("rating", "").replace("_", " ") - grade = grade.upper() if grade == "aaa" else "All Ratings" - country = f"Euro Area ({grade}) {curve_type}" - elif provider == "fred": - curve_type = extra_params.get("yield_curve_type", "") - curve_type = ( - "Real Rates" - if curve_type == "real" - else curve_type.replace("_", " ").title() - ) - country = f"United States {curve_type}" - elif provider == "econdb": - country = extra_params.get("country", "") - country = country.replace("_", " ").title() if country else "United States" - country = country + " " if country else "" - title = kwargs.get("title", "") - if not title: - title = f"{country}Yield Curve" - if len(dates) == 1: - title = f"{country} Yield Curve - {dates[0]}" - - # Update the layout of the figure. - figure.update_layout( - title=dict(text=title, x=0.5, font=dict(size=20)), - plot_bgcolor="rgba(255,255,255,0)", - xaxis=dict( - title="Maturity", - ticklen=0, - showgrid=False, - type="category", - categoryorder="array", - categoryarray=( - [ - ( - d.split("_")[1] + " " + d.split("_")[0].title() - if d != "long_term" - else "Long Term" - ) - for d in maturities - ] - ), - ), - yaxis=dict( - title="Yield (%)", - ticklen=0, - showgrid=True, - gridcolor="rgba(128,128,128,0.3)", - ), - legend=dict( - orientation="v", - yanchor="top", - xanchor="right", - y=0.95, - x=0, - xref="paper", - font=dict(size=12), - bgcolor="rgba(0,0,0,0)", - ), - margin=dict( - b=25, - t=10, - ), - ) - - layout_kwargs = kwargs.get("layout_kwargs", {}) - if layout_kwargs: - figure.update_layout(layout_kwargs) - - content = figure.show(external=True) - - return figure, content diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/__init__.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/__init__.py similarity index 100% rename from openbb_platform/obbject_extensions/charting/openbb_charting/utils/__init__.py rename to openbb_platform/obbject_extensions/charting/openbb_charting/charts/__init__.py diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/generic_charts.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/generic_charts.py similarity index 99% rename from openbb_platform/obbject_extensions/charting/openbb_charting/utils/generic_charts.py rename to openbb_platform/obbject_extensions/charting/openbb_charting/charts/generic_charts.py index 9254408b4c8..066865a8d80 100644 --- a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/generic_charts.py +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/generic_charts.py @@ -10,14 +10,14 @@ from openbb_core.app.utils import basemodel_to_df, convert_to_basemodel from openbb_core.provider.abstract.data import Data from plotly.graph_objs import Figure -from openbb_charting.core.chart_style import ChartStyle -from openbb_charting.core.openbb_figure import OpenBBFigure -from openbb_charting.styles.colors import LARGE_CYCLER -from openbb_charting.utils.helpers import ( +from openbb_charting.charts.helpers import ( calculate_returns, should_share_axis, z_score_standardization, ) +from openbb_charting.core.chart_style import ChartStyle +from openbb_charting.core.openbb_figure import OpenBBFigure +from openbb_charting.styles.colors import LARGE_CYCLER def line_chart( # noqa: PLR0912 @@ -48,7 +48,6 @@ def line_chart( # noqa: PLR0912 **kwargs, ) -> Union[OpenBBFigure, Figure]: """Create a line chart.""" - if data is None: raise ValueError("Error: Data is a required field.") @@ -381,7 +380,6 @@ def bar_chart( # noqa: PLR0912 OpenBBFigure The OpenBBFigure object. """ - try: figure = OpenBBFigure() except Exception as _: @@ -523,7 +521,6 @@ def bar_increasing_decreasing( # pylint: disable=W0102 OpenBBFigure The OpenBBFigure object. """ - try: figure = OpenBBFigure() except Exception as _: diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/helpers.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/helpers.py similarity index 83% rename from openbb_platform/obbject_extensions/charting/openbb_charting/utils/helpers.py rename to openbb_platform/obbject_extensions/charting/openbb_charting/charts/helpers.py index eff4ca13a18..e6a818b0109 100644 --- a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/helpers.py +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/helpers.py @@ -1,27 +1,30 @@ """Helper functions for charting.""" -import importlib from inspect import getmembers, getsource, isfunction -from typing import List +from typing import Callable, Dict, List, Type, Union import pandas as pd from pandas_ta import candles -def get_charting_functions() -> List[str]: +def get_charting_functions( + view: Type, with_objects: bool = False +) -> Union[List[str], Dict[str, Callable]]: """Discover charting functions.""" + implemented_functions: Union[List[str], Dict[str, Callable]] = ( + [] if not with_objects else {} + ) - charting_router = importlib.import_module("openbb_charting.charting_router") - - implemented_functions = [] - - for name, obj in getmembers(charting_router, isfunction): + for name, obj in getmembers(view, isfunction): if ( - obj.__module__ == charting_router.__name__ + obj.__module__ == view.__module__ and not name.startswith("_") and "NotImplementedError" not in getsource(obj) ): - implemented_functions.append(name) + if with_objects: + implemented_functions[name] = obj + else: + implemented_functions.append(name) return implemented_functions @@ -68,7 +71,6 @@ def heikin_ashi(data: pd.DataFrame) -> pd.DataFrame: pd.DataFrame DataFrame copy with Heikin Ashi candle calculations. """ - df = data.copy() check_columns = ["open", "high", "low", "close"] diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_historical.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_historical.py new file mode 100644 index 00000000000..0187b222086 --- /dev/null +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_historical.py @@ -0,0 +1,314 @@ +"""Price historical charting utility.""" + +from typing import Any, Dict, Tuple + +import pandas as pd +from openbb_core.app.utils import basemodel_to_df + +from openbb_charting.charts.helpers import ( + calculate_returns, + heikin_ashi, + should_share_axis, + z_score_standardization, +) +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.styles.colors import LARGE_CYCLER + +# pylint: disable=too-many-branches + + +def price_historical( # noqa: PLR0912 + **kwargs, +) -> Tuple[OpenBBFigure, Dict[str, Any]]: + """Equity Price Historical Chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + elif "data" in kwargs and isinstance(kwargs["data"], list): + data = basemodel_to_df(kwargs["data"], index=kwargs.get("index", "date")) # type: ignore + else: + data = basemodel_to_df( + kwargs["obbject_item"], index=kwargs.get("index", "date") # type: ignore + ) + + if "date" in data.columns: + data = data.set_index("date") + + target = str(kwargs.get("target")) + normalize = kwargs.get("normalize") is True + returns = kwargs.get("returns") is True + same_axis = kwargs.get("same_axis") is True + text_color = "black" if ChartStyle().plt_style == "light" else "white" + title = f"{kwargs.get('title')}" if "title" in kwargs else "Historical Prices" + y1title = "" + y2title = "" + candles = True + multi_symbol = ( + bool(kwargs.get("multi_symbol") is True) + or ( + "symbol" in data.columns + and target in data.columns + and len(data.symbol.unique()) > 1 + ) + or ("target" in kwargs and kwargs.get("target") is not None) + or "symbol" in data.columns + or ( + "symbol" not in data.columns + and bool(data.columns.isin(["open", "high", "low", "close"]).all()) + ) + ) + target = "close" if target is None or target == "None" or target == "" else target + + if multi_symbol is True: + if "symbol" not in data.columns and target in data.columns: + data = data[[target]] + y1title = target.title() + if "symbol" in data.columns and target in data.columns: + data = data.pivot(columns="symbol", values=target) + y1title = target + title = f"Historical {target.title()}" + + indicators = kwargs.get("indicators", {}) + candles = bool(~data.columns.isin(["open", "high", "low", "close"]).all()) + candles = candles if kwargs.get("candles", True) else False + volume = kwargs.get("volume", True) if "volume" in data.columns else False + + if normalize is True: + if "symbol" not in data.columns and target in data.columns: + data = data[[target]] + multi_symbol = True + candles = False + volume = False + + if returns is True: + if "symbol" not in data.columns and target in data.columns: + data = data[[target]] + multi_symbol = True + candles = False + volume = False + if ( # pylint: disable = R0916 + multi_symbol is False + and normalize is False + and returns is False + and candles is True + ) or (indicators and multi_symbol is False): + if ( + "heikin_ashi" in kwargs + and kwargs["heikin_ashi"] is True + and candles is True + ): + data = heikin_ashi(data) + title = f"{title} - Heikin Ashi" + _volume = False + if "atr" in indicators: # type: ignore + _volume = volume + volume = False + ta = PlotlyTA() + fig = ta.plot( # type: ignore + data, + indicators=indicators if indicators else {}, # type: ignore + symbol=target if candles is False else "", + candles=candles, + volume=volume, # type: ignore + ) + if _volume is True and "atr" in indicators: # type: ignore + fig.add_inchart_volume(data) + fig.update_layout( + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + font=dict(color=text_color), + showlegend=True, + legend=dict( + orientation="v", + yanchor="top", + xanchor="right", + y=0.95, + x=-0.01, + xref="paper", + font=dict(size=12), + bgcolor="rgba(0,0,0,0)", + ), + xaxis=dict( + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + zeroline=True, + mirror=True, + showline=True, + ), + xaxis2=dict( + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + zeroline=True, + mirror=True, + showline=True, + ), + yaxis=dict( + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + zeroline=True, + mirror=True, + showline=True, + tickfont=dict(size=14), + ), + yaxis2=dict( + ticklen=0, + gridcolor="rgba(128,128,128,0.3)", + ), + yaxis3=dict( + ticklen=0, + gridcolor="rgba(128,128,128,0.3)", + ), + dragmode="pan", + hovermode="x", + ) + + if kwargs.get("title"): + title = kwargs["title"] + fig.update_layout(title=dict(text=title, x=0.5)) + + content = fig.to_plotly_json() + + return fig, content + + if multi_symbol is True or candles is False: + + if "symbol" not in data.columns and target in data.columns: + data = data[[target]] + + if "symbol" in data.columns: + data = data.pivot(columns="symbol", values=target) + + title: str = kwargs.get("title", "Historical Prices") # type: ignore + + y1title = data.iloc[:, 0].name + y2title = "" + + if len(data.columns) > 2 or normalize is True or returns is True: + if returns is True or (len(data.columns) > 2 and normalize is False): + data = data.apply(calculate_returns) + title = f"{title} - Cumulative Returns" + y1title = "Percent" + if normalize is True: + if returns is True: + title = f"{title.replace(' - Cumulative Returns', '')} - Normalized Cumulative Returns" + else: + title = title + " - Normalized" + data = data.apply(z_score_standardization) + y1title = None # type: ignore + y2title = None # type: ignore + + fig = OpenBBFigure() + + for i, col in enumerate(data.columns): + + hovertemplate = f"{data[col].name}: %{{y}}" + yaxis = "y1" + if y1title and y1title != "Percent": + yaxis = ( + ( + "y1" + if should_share_axis(data, col, y1title) # type: ignore + or col == y1title + or normalize is True + or returns is True + else "y2" + ) + if same_axis is False + else "y1" + ) + + if yaxis == "y2": + y2title = data[col].name + + fig.add_scatter( + x=data.index, + y=data[col], + name=data[col].name, + mode="lines", + hovertemplate=hovertemplate, + line=dict(width=2, color=LARGE_CYCLER[i % len(LARGE_CYCLER)]), + yaxis=yaxis, + ) + + if normalize is True or returns is True: + y1title = "Percent" if returns is True else None # type: ignore + y2title = None # type: ignore + + if same_axis is True: + y1title = None # type: ignore + y2title = None # type: ignore + + fig.update_layout( + paper_bgcolor="rgba(0,0,0,0)", + plot_bgcolor="rgba(0,0,0,0)", + legend=( + dict( + orientation="v", + yanchor="top", + xanchor="right", + y=0.95, + x=-0.01, + bgcolor="rgba(0,0,0,0)", + ) + if len(data.columns) > 2 + else dict( + orientation="h", + yanchor="bottom", + xanchor="right", + y=1.02, + x=0.98, + bgcolor="rgba(0,0,0,0)", + ) + ), + yaxis1=( + dict( + side="right", + ticklen=0, + showgrid=True, + showline=True, + mirror=True, + gridcolor="rgba(128,128,128,0.3)", + title=dict( + text=y1title if y1title else None, standoff=20, font=dict(size=20) + ), + tickfont=dict(size=14), + anchor="x", + ) + ), + yaxis2=( + dict( + overlaying="y", + side="left", + ticklen=0, + showgrid=False, + title=dict( + text=y2title if y2title else None, standoff=10, font=dict(size=20) + ), + tickfont=dict(size=14), + anchor="x", + ) + if y2title + else None + ), + xaxis=dict( + ticklen=0, + showgrid=True, + gridcolor="rgba(128,128,128,0.3)", + showline=True, + mirror=True, + ), + margin=dict(l=20, r=20, b=20, t=20), + dragmode="pan", + hovermode="x", + ) + if kwargs.get("title"): + title = kwargs["title"] + fig.update_layout(title=dict(text=title, x=0.5)) + + content = fig.show(external=True).to_plotly_json() + + return fig, content diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_performance.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_performance.py new file mode 100644 index 00000000000..3590a30a7e2 --- /dev/null +++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_performance.py @@ -0,0 +1,106 @@ +"""Price performance charting implementation.""" + +from typing import Any, Dict, Tuple, Union + +import pandas as pd +from openbb_core.app.utils import basemodel_to_df +from plotly.graph_objs import Figure + +from openbb_charting.charts.generic_charts import bar_chart +from openbb_charting.core.openbb_figure import OpenBBFigure + + +def price_performance( + **kwargs, +) -> Tuple[Union[OpenBBFigure, Figure], Dict[str, Any]]: # noqa: PLR0912 + """Equity Price Performance Chart.""" + if "data" in kwargs and isinstance(kwargs["data"], pd.DataFrame): + data = kwargs["data"] + elif "data" in kwargs and isinstance(kwargs["data"], list): + data = basemodel_to_df(kwargs["data"], index=kwargs.get("index", "symbol")) # type: ignore + else: + data = basemodel_to_df( + kwargs["obbject_item"], index=kwargs.get("index", "symbol") # type: ignore + ) + + cols = [ + "one_day", + "one_week", + "one_month", + "three_month", + "six_month", + "ytd", + "one_year", + "two_year", + "three_year", + "four_year", + "five_year", + ] + + df = pd.DataFrame() + chart_df = pd.DataFrame() + + if "symbol" in data.columns: + data = data.set_index("symbol") + chart_cols = [] + + if len(data) == 0: + raise ValueError("No data was found in the DataFrame.") + + data = data.drop_duplicates(keep="first") + + for col in cols: + if col in data.columns and data[col].notnull().any(): + df[col.replace("_", " ").title() if col != "ytd" else col.upper()] = data[ + col + ].apply(lambda x: round(x * 100, 4) if x is not None else None) + + if df.empty: + raise ValueError(f"No columns matching, {cols}, were found in the data.") + + chart_df = df.T + chart_cols = chart_df.columns.to_list() + + if "limit" in kwargs and isinstance(kwargs.get("limit"), int): + limit = kwargs.pop("limit", 10) + chart_df = chart_df.head(limit) # type: ignore + + layout_kwargs: Dict[str, Any] = kwargs.get("layout_kwargs", {}) + + title = ( + f"{kwargs.pop('title')}" if "title" in kwargs else "Equity Price Performance" + ) + orientation = ( + kwargs.pop("orientation") + if "orientation" in kwargs and kwargs.get("orientation") is not None + else "v" + ) + + ytitle = "Performance (%)" + xtitle = None + + if orientation == "h": + xtitle = ytitle # type: ignore + ytitle = None # type: ignore + + fig = bar_chart( + chart_df.reset_index(), + x="index", + y=chart_cols, + title=title, + xtitle=xtitle, + ytitle=ytitle, + orientation=orientation, # type: ignore + ) + fig.update_traces( + hovertemplate=( + "%{fullData.name}:%{y:.2f}%" + if orientation == "v" + else "%{fullData.name}:%{x:.2f}%" + ) + ) + + fig.update_layout(**layout_kwargs) + content = fig.show(external=True).to_plotly_json() # type: ignore + + return fig, content diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/relative_rotation.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charts/relative_rotation.py similarity index 100% rename from openbb_platform/obbject_extensions/charting/openbb_charting/utils/relative_rotation.py rename to openbb_platform/obbject_extensions/charting/openbb_charting/charts/relative_rotation.py diff --git a/openbb_platform/obbject_extensions/charting/tests/test_charting.py b/openbb_platform/obbject_extensions/charting/tests/test_charting.py index 9e148c72808..6a6c3c4a7b1 100644 --- a/openbb_platform/obbject_extensions/charting/tests/test_charting.py +++ b/openbb_platform/obbject_extensions/charting/tests/test_charting.py @@ -122,16 +122,16 @@ def test_handle_backend(mock_create_backend, mock_get_backend, obbject): mock_get_backend.assert_called_once() -@patch("openbb_charting.charting_router") -def test_get_chart_function(mock_charting_router): +def test_get_chart_function(obbject): """Test _get_chart_function method.""" # Arrange mock_function = MagicMock() - mock_charting_router.some_function = mock_function + charting = Charting(obbject) + charting._functions = {"some_function": mock_function} route = "/some/function" # Act - result = Charting._get_chart_function(route) + result = charting._get_chart_function(route) # Assert assert result == mock_function