jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-11-06 09:22:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

[Feature] Charting Modularity (#6477)

Browse Source 
* add accessors to the charting extension

* equity views

* get charting functions from the extensions/accessors

* use the private variable instead

* etf views

* index views

* crypto and currency views

* technical views

* economy views

* fixedincome views

* right description for price historical

* unused

* adjust test and deprecate the charting router

* import views only with charting installed

* adding views throught dependencies instead

* removing unnecessary infra and adding entry point to charting extension

* fixed income toml

* removing accessor

* typo

* typo

* adding extra params to obbject

* remove things from init

* rename utils to charts instead

* updated readme

---------

Co-authored-by: Danglewood <85772166+deeleeramone@users.noreply.github.com>
Co-authored-by: Igor Radovanovic <74266147+IgorWounds@users.noreply.github.com>
This commit is contained in:
Henrique Joaquim
2024-06-18 14:45:10 +01:00
committed by GitHub
parent 6445b974d7
commit 5c79635da2
29 changed files with 2226 additions and 2075 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
openbb_platform/core/openbb_core/app/model/obbject.py
View File

@@ -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,
)

17
openbb_platform/extensions/crypto/openbb_crypto/crypto_views.py Normal file
View File

@@ -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)

3
openbb_platform/extensions/crypto/pyproject.toml vendored
View File

@@ -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"

17
openbb_platform/extensions/currency/openbb_currency/currency_views.py Normal file
View File

@@ -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)

3
openbb_platform/extensions/currency/pyproject.toml vendored
View File

@@ -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"

307
openbb_platform/extensions/economy/openbb_economy/economy_views.py Normal file
View File

@@ -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}}<extra></extra>",
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

3
openbb_platform/extensions/economy/pyproject.toml vendored
View File

@@ -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"

25
openbb_platform/extensions/equity/openbb_equity/equity_views.py Normal file
View File

@@ -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)

3
openbb_platform/extensions/equity/pyproject.toml vendored
View File

@@ -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"

84
openbb_platform/extensions/etf/openbb_etf/etf_views.py Normal file
View File

@@ -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}%<extra></extra>"
if orientation == "v"
else "%{x:.3f}%<extra></extra>"
)
)
if layout_kwargs:
fig.update_layout(**layout_kwargs) # type: ignore
content = fig.show(external=True).to_plotly_json() # type: ignore
return fig, content

3
openbb_platform/extensions/etf/pyproject.toml vendored
View File

@@ -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"

183
openbb_platform/extensions/fixedincome/openbb_fixedincome/fixedincome_views.py Normal file
View File

@@ -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}<br>Yield: %{y}%<extra></extra>"
if len(dates) == 1
else "%{fullData.name}<br>Maturity: %{x}<br>Yield: %{y}%<extra></extra>"
),
)
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

3
openbb_platform/extensions/fixedincome/pyproject.toml vendored
View File

@@ -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"

17
openbb_platform/extensions/index/openbb_index/index_views.py Normal file
View File

@@ -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)

3
openbb_platform/extensions/index/pyproject.toml vendored
View File

@@ -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"

511
openbb_platform/extensions/technical/openbb_technical/technical_views.py Normal file
View File

@@ -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}}<extra></extra>",
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}}<extra></extra>",
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

3
openbb_platform/extensions/technical/pyproject.toml vendored
View File

@@ -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"

39
openbb_platform/obbject_extensions/charting/README.md vendored
View File

@@ -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.

551
openbb_platform/obbject_extensions/charting/openbb_charting/__init__.py
View File

@@ -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)

53
openbb_platform/obbject_extensions/charting/openbb_charting/builder.py
View File

@@ -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)

563
openbb_platform/obbject_extensions/charting/openbb_charting/charting.py Normal file
View File

@@ -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}")

1445
openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py
View File

File diff suppressed because it is too large Load Diff

0
openbb_platform/obbject_extensions/charting/openbb_charting/utils/__init__.py → openbb_platform/obbject_extensions/charting/openbb_charting/charts/__init__.py
View File

11
openbb_platform/obbject_extensions/charting/openbb_charting/utils/generic_charts.py → openbb_platform/obbject_extensions/charting/openbb_charting/charts/generic_charts.py
View File

@@ -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 _:

24
openbb_platform/obbject_extensions/charting/openbb_charting/utils/helpers.py → openbb_platform/obbject_extensions/charting/openbb_charting/charts/helpers.py
View File

@@ -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"]

314
openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_historical.py Normal file
View File

@@ -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}}<extra></extra>"
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

106
openbb_platform/obbject_extensions/charting/openbb_charting/charts/price_performance.py Normal file
View File

@@ -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}%<extra></extra>"
if orientation == "v"
else "%{fullData.name}:%{x:.2f}%<extra></extra>"
)
)
fig.update_layout(**layout_kwargs)
content = fig.show(external=True).to_plotly_json() # type: ignore
return fig, content

0
openbb_platform/obbject_extensions/charting/openbb_charting/utils/relative_rotation.py → openbb_platform/obbject_extensions/charting/openbb_charting/charts/relative_rotation.py
View File

8
openbb_platform/obbject_extensions/charting/tests/test_charting.py
View File

@@ -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