diff --git a/openbb_platform/core/openbb_core/app/command_runner.py b/openbb_platform/core/openbb_core/app/command_runner.py
index a8c368839f8..4d9985a2172 100644
--- a/openbb_platform/core/openbb_core/app/command_runner.py
+++ b/openbb_platform/core/openbb_core/app/command_runner.py
@@ -316,15 +316,18 @@ class StaticCommandRunner:
elif isinstance(extra_params, dict) and "chart_params" in extra_params:
chart_params = kwargs["extra_params"].get("chart_params", {})
- if "chart_params" in kwargs:
+ if "chart_params" in kwargs and kwargs["chart_params"] is not None:
chart_params.update(kwargs.pop("chart_params", {}))
- if "kwargs" in kwargs:
+ if (
+ "kwargs" in kwargs
+ and "chart_params" in kwargs["kwargs"]
+ and kwargs["kwargs"].get("chart_params") is not None
+ ):
chart_params.update(kwargs.pop("kwargs", {}).get("chart_params", {}))
if chart_params:
kwargs.update(chart_params)
-
obbject.charting.show(render=False, **kwargs)
# pylint: disable=R0913, R0914
@@ -374,7 +377,6 @@ class StaticCommandRunner:
# pylint: disable=protected-access
obbject._route = route
obbject._standard_params = kwargs.get("standard_params", None)
-
if chart and obbject.results:
cls._chart(obbject, **kwargs)
diff --git a/openbb_platform/extensions/technical/integration/test_technical_api.py b/openbb_platform/extensions/technical/integration/test_technical_api.py
index 1ed903f85c4..481fce9f356 100644
--- a/openbb_platform/extensions/technical/integration/test_technical_api.py
+++ b/openbb_platform/extensions/technical/integration/test_technical_api.py
@@ -959,3 +959,42 @@ def test_technical_ema(params, data_type):
result = requests.post(url, headers=get_headers(), timeout=10, data=body)
assert isinstance(result, requests.Response)
assert result.status_code == 200
+
+
+@parametrize(
+ "params",
+ [
+ (
+ {
+ "data": "",
+ "study": "price",
+ "benchmark": "SPY",
+ "long_period": 252,
+ "short_period": 21,
+ "window": 21,
+ "trading_periods": 252,
+ "chart_params": {"show_tails": False},
+ }
+ ),
+ ],
+)
+@pytest.mark.integration
+def test_technical_relative_rotation(params):
+ params = {p: v for p, v in params.items() if v}
+ data_params = dict(
+ symbol="AAPL,MSFT,GOOGL,AMZN,SPY",
+ provider="yfinance",
+ start_date="2022-01-01",
+ end_date="2024-01-01",
+ )
+ data_query_str = get_querystring(data_params, [])
+ data_url = f"http://0.0.0.0:8000/api/v1/equity/price/historical?{data_query_str}"
+ data_result = requests.get(data_url, headers=get_headers(), timeout=10).json()[
+ "results"
+ ]
+ body = json.dumps({"data": data_result})
+ query_str = get_querystring(params, ["data"])
+ url = f"http://0.0.0.0:8000/api/v1/technical/relative_rotation?{query_str}"
+ result = requests.post(url, headers=get_headers(), timeout=10, data=body)
+ assert isinstance(result, requests.Response)
+ assert result.status_code == 200
diff --git a/openbb_platform/extensions/technical/integration/test_technical_python.py b/openbb_platform/extensions/technical/integration/test_technical_python.py
index afc0d60b909..d866c3623c8 100644
--- a/openbb_platform/extensions/technical/integration/test_technical_python.py
+++ b/openbb_platform/extensions/technical/integration/test_technical_python.py
@@ -926,3 +926,45 @@ def test_technical_ema(params, data_type, obb):
assert result
assert isinstance(result, OBBject)
assert len(result.results) > 0
+
+
+@parametrize(
+ "params",
+ [
+ (
+ {
+ "data": "",
+ "study": "price",
+ "benchmark": "SPY",
+ "long_period": 252,
+ "short_period": 21,
+ "window": 21,
+ "trading_periods": 252,
+ "chart_params": {"show_tails": False},
+ }
+ ),
+ ],
+)
+@pytest.mark.integration
+def test_technical_relative_rotation(params, obb):
+ params["data"] = obb.equity.price.historical(
+ "AAPL,MSFT,GOOGL,AMZN,SPY",
+ provider="yfinance",
+ start_date="2022-01-01",
+ end_date="2024-01-01",
+ ).results
+ result = obb.technical.relative_rotation(
+ data=params["data"],
+ benchmark=params["benchmark"],
+ study=params["study"],
+ long_period=params["long_period"],
+ short_period=params["short_period"],
+ window=params["window"],
+ trading_periods=params["trading_periods"],
+ )
+ assert result
+ assert isinstance(result, OBBject)
+ assert hasattr(result.results, "rs_ratios")
+ assert len(result.results.rs_ratios) > 0 # type: ignore
+ assert hasattr(result.results, "rs_momentum")
+ assert len(result.results.rs_momentum) > 0 # type: ignore
diff --git a/openbb_platform/extensions/technical/openbb_technical/helpers.py b/openbb_platform/extensions/technical/openbb_technical/helpers.py
index d4eca5d6201..7c7fec91dc7 100644
--- a/openbb_platform/extensions/technical/openbb_technical/helpers.py
+++ b/openbb_platform/extensions/technical/openbb_technical/helpers.py
@@ -1,13 +1,11 @@
"""Technical Analysis Helpers."""
-import warnings
from typing import Any, List, Literal, Optional, Tuple, Union
+from warnings import warn
import numpy as np
import pandas as pd
-_warn = warnings.warn
-
def validate_data(data: list, length: Union[int, List[int]]) -> None:
"""Validate data."""
@@ -51,11 +49,11 @@ def parkinson(
Dataframe with results.
"""
if window < 1:
- _warn("Error: Window must be at least 1, defaulting to 30.")
+ warn("Error: Window must be at least 1, defaulting to 30.")
window = 30
if trading_periods and is_crypto:
- _warn("is_crypto is overridden by trading_periods.")
+ warn("is_crypto is overridden by trading_periods.")
if not trading_periods:
trading_periods = 365 if is_crypto else 252
@@ -106,11 +104,11 @@ def standard_deviation(
Dataframe with results.
"""
if window < 2:
- _warn("Error: Window must be at least 2, defaulting to 30.")
+ warn("Error: Window must be at least 2, defaulting to 30.")
window = 30
if trading_periods and is_crypto:
- _warn("is_crypto is overridden by trading_periods.")
+ warn("is_crypto is overridden by trading_periods.")
if not trading_periods:
trading_periods = 365 if is_crypto else 252
@@ -159,11 +157,11 @@ def garman_klass(
Dataframe with results.
"""
if window < 1:
- _warn("Error: Window must be at least 1, defaulting to 30.")
+ warn("Error: Window must be at least 1, defaulting to 30.")
window = 30
if trading_periods and is_crypto:
- _warn("is_crypto is overridden by trading_periods.")
+ warn("is_crypto is overridden by trading_periods.")
if not trading_periods:
trading_periods = 365 if is_crypto else 252
@@ -220,11 +218,11 @@ def hodges_tompkins(
>>> df = obb.technical.hodges_tompkins(data, is_crypto = True)
"""
if window < 2:
- _warn("Error: Window must be at least 2, defaulting to 30.")
+ warn("Error: Window must be at least 2, defaulting to 30.")
window = 30
if trading_periods and is_crypto:
- _warn("is_crypto is overridden by trading_periods.")
+ warn("is_crypto is overridden by trading_periods.")
if not trading_periods:
trading_periods = 365 if is_crypto else 252
@@ -280,11 +278,11 @@ def rogers_satchell(
Pandas Series with results.
"""
if window < 1:
- _warn("Error: Window must be at least 1, defaulting to 30.")
+ warn("Error: Window must be at least 1, defaulting to 30.")
window = 30
if trading_periods and is_crypto:
- _warn("is_crypto is overridden by trading_periods.")
+ warn("is_crypto is overridden by trading_periods.")
if not trading_periods:
trading_periods = 365 if is_crypto else 252
@@ -337,11 +335,11 @@ def yang_zhang(
Dataframe with results.
"""
if window < 2:
- _warn("Error: Window must be at least 2, defaulting to 30.")
+ warn("Error: Window must be at least 2, defaulting to 30.")
window = 30
if trading_periods and is_crypto:
- _warn("is_crypto is overridden by trading_periods.")
+ warn("is_crypto is overridden by trading_periods.")
if not trading_periods:
trading_periods = 365 if is_crypto else 252
@@ -547,12 +545,12 @@ def calculate_fib_levels(
if start_date and end_date:
if start_date not in data.index:
date0 = data.index[data.index.get_indexer([end_date], method="nearest")[0]]
- _warn(f"Start date not in data. Using nearest: {date0}")
+ warn(f"Start date not in data. Using nearest: {date0}")
else:
date0 = start_date
if end_date not in data.index:
date1 = data.index[data.index.get_indexer([end_date], method="nearest")[0]]
- _warn(f"End date not in data. Using nearest: {date1}")
+ warn(f"End date not in data. Using nearest: {date1}")
else:
date1 = end_date
diff --git a/openbb_platform/extensions/technical/openbb_technical/relative_rotation.py b/openbb_platform/extensions/technical/openbb_technical/relative_rotation.py
new file mode 100644
index 00000000000..5bd3f7a295f
--- /dev/null
+++ b/openbb_platform/extensions/technical/openbb_technical/relative_rotation.py
@@ -0,0 +1,554 @@
+"""Relative Rotation Model."""
+
+# pylint: disable=too-many-arguments, too-many-instance-attributes, protected-access
+# pylint: disable=too-many-locals, too-few-public-methods, unused-argument
+
+import contextlib
+from typing import Any, Dict, List, Literal, Optional, Tuple, Union
+
+import numpy as np
+from openbb_core.app.model.obbject import OBBject
+from openbb_core.app.utils import basemodel_to_df, convert_to_basemodel, df_to_basemodel
+from openbb_core.provider.abstract.data import Data
+from openbb_core.provider.abstract.fetcher import Fetcher
+from openbb_core.provider.abstract.query_params import QueryParams
+from pandas import DataFrame, Series, to_datetime
+from pydantic import Field, field_validator
+
+
+def absolute_maximum_scale(data: Series) -> Series:
+ """Absolute Maximum Scale Normaliztion Method."""
+ return data / data.abs().max()
+
+
+def min_max_scaling(data: Series) -> Series:
+ """Min/Max ScalingNormalization Method."""
+ return (data - data.min()) / (data.max() - data.min())
+
+
+def z_score_standardization(data: Series) -> Series:
+ """Z-Score Standardization Method."""
+ return (data - data.mean()) / data.std()
+
+
+def normalize(data: DataFrame, method: Literal["z", "m", "a"] = "z") -> DataFrame:
+ """
+ Normalize a Pandas DataFrame based on method.
+
+ Parameters
+ ----------
+ data: DataFrame
+ Pandas DataFrame with any number of columns to be normalized.
+ method: Literal["z", "m", "a"]
+ Normalization method.
+ z: Z-Score Standardization
+ m: Min/Max Scaling
+ a: Absolute Maximum Scale
+
+ Returns
+ -------
+ DataFrame
+ Normalized DataFrame.
+ """
+ methods = {
+ "z": z_score_standardization,
+ "m": min_max_scaling,
+ "a": absolute_maximum_scale,
+ }
+
+ df = data.copy()
+
+ for col in df.columns:
+ df.loc[:, col] = methods[f"{method}"](df.loc[:, col])
+
+ return df
+
+
+def standard_deviation(
+ data: DataFrame,
+ window: int = 21,
+ trading_periods: int = 252,
+) -> DataFrame:
+ """
+ Measures how widely returns are dispersed from the average return.
+
+ It is the most common (and biased) estimator of volatility.
+
+ Parameters
+ ----------
+ data : pd.DataFrame
+ Dataframe of OHLC prices.
+ window : int [default: 21]
+ Length of window to calculate over.
+ trading_periods : Optional[int] [default: 252]
+ Number of trading periods in a year.
+
+ Returns
+ -------
+ pd.DataFrame : results
+ Dataframe with results.
+ """
+ data = data.copy()
+ results = DataFrame()
+ if window < 2:
+ window = 21
+
+ for col in data.columns.tolist():
+ log_return = (data[col] / data[col].shift(1)).apply(np.log)
+
+ result = log_return.rolling(window=window, center=False).std() * np.sqrt(
+ trading_periods
+ )
+ results[col] = result
+
+ return results.dropna()
+
+
+def calculate_momentum(
+ data: Series, long_period: int = 252, short_period: int = 21
+) -> Series:
+ """
+ Momentum is calculated as the log trailing 12-month return minus trailing one-month return.
+
+ Higher values indicate larger, positive momentum exposure.
+
+ Momentum = ln(1 + r12) - ln(1 + r1)
+
+ Parameters
+ ----------
+ data: Series
+ Time series data to calculate the momentum for.
+ long_period: Optional[int]
+ Long period to base the calculation on. Default is one standard trading year.
+ short_period: Optional[int]
+ Short period to subtract from the long period. Default is one trading month.
+
+ Returns
+ -------
+ Series
+ Pandas Series with the calculated momentum.
+ """
+ df = data.copy()
+ epsilon = 1e-10
+ momentum_long = np.log(1 + df.pct_change(long_period) + epsilon)
+ momentum_short = np.log(1 + df.pct_change(short_period) + epsilon)
+ data = momentum_long - momentum_short # type: ignore
+
+ return data
+
+
+def get_momentum(
+ data: DataFrame, long_period: int = 252, short_period: int = 21
+) -> DataFrame:
+ """
+ Calculate the Relative-Strength Momentum Indicator.
+
+ Takes the Relative Strength Ratio as the input.
+
+ Parameters
+ ----------
+ data: DataFrame
+ Indexed time series data formatted with each column representing a ticker.
+ long_period: Optional[int]
+ Long period to base the calculation on. Default is one standard trading year.
+ short_period: Optional[int]
+ Short period to subtract from the long period. Default is one trading month.
+
+ Returns
+ -------
+ DataFrame
+ Pandas DataFrame with the calculated historical momentum factor exposure score.
+ """
+ df = data.copy()
+ rs_momentum = DataFrame()
+ for ticker in df.columns.to_list():
+ rs_momentum.loc[:, ticker] = calculate_momentum(
+ df.loc[:, ticker], long_period, short_period
+ ) # type: ignore
+
+ return rs_momentum
+
+
+def calculate_relative_strength_ratio(
+ symbols_data: DataFrame,
+ benchmark_data: DataFrame,
+) -> DataFrame:
+ """Calculate the Relative Strength Ratio for each ticker (column) in a DataFrame against the benchmark.
+
+ Symbols data and benchmark data should have the same index,
+ and each column should represent a ticker.
+
+ Parameters
+ ----------
+ symbols_data: DataFrame
+ Pandas DataFrame with the symbols data to compare against the benchmark.
+ benchmark_data: DataFrame
+ Pandas DataFrame with the benchmark data.
+
+ Returns
+ -------
+ DataFrame
+ Pandas DataFrame with the calculated relative strength
+ ratio for each ticker joined with the benchmark values.
+ """
+ return (
+ symbols_data.div(benchmark_data.iloc[:, 0], axis=0)
+ .multiply(100)
+ .join(benchmark_data.iloc[:, 0])
+ .dropna()
+ )
+
+
+def process_data(
+ symbols_data: DataFrame,
+ benchmark_data: DataFrame,
+ long_period: int = 252,
+ short_period: int = 21,
+ normalize_method: Literal["z", "m", "a"] = "z",
+) -> Tuple[DataFrame, DataFrame]:
+ """Process the raw data into normalized indicator values.
+
+ Parameters
+ ----------
+ symbols_data: DataFrame
+ Indexed time series data formatted with each column representing a ticker.
+ benchmark_data: DataFrame
+ Indexed time series data of the benchmark symbol.
+ long_period: Optional[int]
+ Long period to base the calculation on. Default is one standard trading year.
+ short_period: Optional[int]
+ Short period to subtract from the long period. Default is one trading month.
+ normalize_method: Literal["z", "m", "a"]
+
+ Returns
+ -------
+ Tuple[DataFrame, DataFrame]
+ Tuple of Pandas DataFrames with the normalized ratio and momentum indicator values.
+ """
+ ratio_data = calculate_relative_strength_ratio(symbols_data, benchmark_data)
+ momentum_data = get_momentum(ratio_data, long_period, short_period)
+ normalized_ratio = normalize(ratio_data, normalize_method)
+ normalized_momentum = normalize(momentum_data, normalize_method)
+
+ return normalized_ratio, normalized_momentum
+
+
+class RelativeRotation:
+ """Relative Rotation Class."""
+
+ def __init__(
+ self,
+ data: Union[List[Data], DataFrame],
+ benchmark: str,
+ study: Optional[Literal["price", "volume", "volatility"]] = "price",
+ long_period: Optional[int] = 252,
+ short_period: Optional[int] = 21,
+ window: Optional[int] = 21,
+ trading_periods: Optional[int] = 252,
+ ):
+ """Initialize the class."""
+ benchmark = benchmark.upper()
+ df = DataFrame()
+
+ target_col = "volume" if study == "volume" else "close"
+
+ if isinstance(data, OBBject):
+ data = data.results # type: ignore
+
+ if isinstance(data, List) and (
+ all(isinstance(d, Data) for d in data)
+ or all(isinstance(d, dict) for d in data)
+ ):
+ with contextlib.suppress(Exception):
+ df = basemodel_to_df(convert_to_basemodel(data), index="date")
+
+ if isinstance(data, DataFrame) and not df.empty:
+ df = data.copy()
+ if "date" in df.columns:
+ df.set_index("date", inplace=True)
+
+ if df.empty:
+ raise ValueError(
+ "Data must be a list of Data objects or a DataFrame with a 'date' column."
+ )
+
+ if "symbol" in df.columns:
+ df = df.pivot(columns="symbol", values=target_col)
+
+ if benchmark not in df.columns:
+ raise RuntimeError("The benchmark symbol was not found in the data.")
+
+ benchmark_data = df.pop(benchmark).to_frame()
+ symbols_data = df
+
+ if len(symbols_data) <= 252 and study in ["price", "volume"]: # type: ignore
+ raise ValueError(
+ "Supplied data must be daily intervals and have more than one year of back data to calculate"
+ " the most recent day in the time series."
+ )
+
+ if study == "volatility" and len(symbols_data) <= 504: # type: ignore
+ raise ValueError(
+ "Supplied data must be daily intervals and have more than two years of back data to calculate"
+ " the most recent day in the time series as a volatility study."
+ )
+ self.symbols = df.columns.to_list()
+ self.benchmark = benchmark
+ self.study = study
+ self.long_period = long_period
+ self.short_period = short_period
+ self.window = window
+ self.trading_periods = trading_periods
+ self.symbols_data = symbols_data # type: ignore
+ self.benchmark_data = benchmark_data # type: ignore
+ self._process_data() # type: ignore
+ self.symbols_data = df_to_basemodel(self.symbols_data.reset_index()) # type: ignore
+ self.benchmark_data = df_to_basemodel(self.benchmark_data.reset_index()) # type: ignore
+
+ def _process_data(self):
+ """Process the data."""
+ if self.study == "volatility":
+ self.symbols_data = standard_deviation(
+ self.symbols_data, # type: ignore
+ window=self.window, # type: ignore
+ trading_periods=self.trading_periods, # type: ignore
+ )
+ self.benchmark_data = standard_deviation(
+ self.benchmark_data, # type: ignore
+ window=self.window, # type: ignore
+ trading_periods=self.trading_periods, # type: ignore
+ )
+ ratios, momentum = process_data(
+ self.symbols_data, # type: ignore
+ self.benchmark_data, # type: ignore
+ long_period=self.long_period, # type: ignore
+ short_period=self.short_period, # type: ignore
+ )
+ # Re-index rs_ratios using the new index
+ index_after_dropping_nans = momentum.dropna().index
+ ratios = ratios.reindex(index_after_dropping_nans)
+ self.rs_ratios = df_to_basemodel(ratios.reset_index())
+ self.rs_momentum = df_to_basemodel(momentum.dropna().reset_index())
+ self.end_date = to_datetime(ratios.index[-1]).strftime("%Y-%m-%d")
+ self.start_date = to_datetime(ratios.index[0]).strftime("%Y-%m-%d")
+ return self
+
+
+def _get_type_name(t):
+ """Get the type name of a type hint."""
+ if hasattr(t, "__origin__"):
+ if hasattr(t.__origin__, "__name__"):
+ return f"{t.__origin__.__name__}[{', '.join([_get_type_name(arg) for arg in t.__args__])}]"
+ if hasattr(t.__origin__, "_name"):
+ return f"{t.__origin__._name}[{', '.join([_get_type_name(arg) for arg in t.__args__])}]"
+ if isinstance(t, str):
+ return t
+ if hasattr(t, "__name__"):
+ return t.__name__
+ if hasattr(t, "_name"):
+ return t._name
+ return str(t)
+
+
+class RelativeRotationQueryParams(QueryParams):
+ """Relative Rotation Query Parameters."""
+
+ data: List[Data] = Field(
+ description="The data to be used for the relative rotation calculations."
+ + " This should be the multi-symbol output from the"
+ + " 'equity.price.historical' endpoint, or similar, at a daily interval."
+ + " Or a pivot table with the 'date' column as the index, the symbols as the columns,"
+ + " and the 'study' as the values."
+ + " It is recommended to use the 'equity.price.historical' endpoint to get the data,"
+ + " and feed the results as-is."
+ )
+ benchmark: str = Field(description="The symbol to be used as the benchmark.")
+ study: Literal["price", "volume", "volatility"] = Field(
+ default="price",
+ description="The data point for the calculations."
+ + " If 'price', the closing price will be used."
+ + " If 'volatility', the standard deviation of the closing price will be used."
+ + " If 'data' is supplied as a pivot table,"
+ + " the 'study' will assume the values are the closing price and 'volume' will be ignored.",
+ )
+ long_period: Optional[int] = Field(
+ default=252,
+ description="The length of the long period for momentum calculation, by default is 252."
+ + " Adjust this value, to 365, when supplying assets such as crypto.",
+ )
+ short_period: Optional[int] = Field(
+ default=21,
+ description="The length of the short period for momentum calculation, by default is 21."
+ + " Adjust this value, to 30, when supplying assets such as crypto.",
+ )
+ window: Optional[int] = Field(
+ default=21,
+ description="The length of window for the standard deviation calculation, by default is 21."
+ + " Adjust this value, to 30, when supplying assets such as crypto.",
+ )
+ trading_periods: Optional[int] = Field(
+ default=252,
+ description="The number of trading periods per year,"
+ + " for the standard deviation calculation, by default is 252."
+ + " Adjust this value, to 365, when supplying assets such as crypto.",
+ )
+ chart_params: Optional[Dict[str, Any]] = Field(
+ default=None,
+ description="Additional parameters to pass when `chart=True` and the `openbb-charting` extension is installed."
+ + " Parameters can be passed again to redraw the chart using the charting.to_chart() method of the response."
+ + "\n"
+ + "\n ChartParams"
+ + "\n -----------"
+ + "\n date: Optional[str]"
+ + "\n A target end date within the data, by default is the last date in the data."
+ + "\n show_tails: bool"
+ + "\n Show the tails on the chart, by default is True."
+ + "\n tail_periods: Optional[int]"
+ + "\n Number of periods to show in the tails, by default is 16."
+ + "\n tail_interval: Literal['day', 'week', 'month']"
+ + "\n Interval to show the tails, by default is 'week'."
+ + "\n title: Optional[str]"
+ + "\n Title of the chart.",
+ )
+
+ @field_validator("benchmark", mode="before", check_fields=False)
+ @classmethod
+ def to_upper(cls, v):
+ """Convert the benchmark symbol to uppercase."""
+ return v.upper()
+
+ @field_validator("data", mode="before", check_fields=False)
+ @classmethod
+ def convert_data(cls, v):
+ """Validate the data format."""
+ if isinstance(v, OBBject):
+ return v.results
+ if isinstance(v, Data):
+ return v
+ if isinstance(v, (list, dict)):
+ return convert_to_basemodel(v)
+ if isinstance(v, DataFrame):
+ return df_to_basemodel(v.reset_index())
+ return v
+
+ def __init__(self, **data):
+ """Initialize the class."""
+ super().__init__(**data)
+ fields = self.__class__.model_fields
+ doc_str = (
+ "\n"
+ + self.__class__.__name__
+ + "\n\n"
+ + " Parameters\n"
+ + " ----------\n"
+ + "\n".join(
+ [
+ f" {k} : {_get_type_name(v.annotation)}\n {v.description}"
+ for k, v in fields.items()
+ ]
+ )
+ + "\n"
+ )
+ self.__doc__ = doc_str
+
+
+class RelativeRotationData(Data):
+ """Relative Rotation Data Model."""
+
+ symbols: List[str] = Field(
+ description="The symbols that are being compared against the benchmark."
+ )
+ benchmark: str = Field(description="The benchmark symbol, as entered by the user.")
+ study: Literal["price", "volume", "volatility"] = Field(
+ description="The data point for the study, as entered by the user."
+ )
+ long_period: int = Field(
+ description="The length of the long period for momentum calculation,"
+ + " as entered by the user."
+ )
+ short_period: int = Field(
+ description="The length of the short period for momentum calculation,"
+ + " as entered by the user."
+ )
+ window: int = Field(
+ description="The length of window for the standard deviation calculation,"
+ + " as entered by the user.",
+ )
+ trading_periods: int = Field(
+ description="The number of trading periods per year,"
+ + " for the standard deviation calculation, as entered by the user."
+ )
+ start_date: str = Field(
+ description="The start date of the data after adjusting"
+ + " the length of the data for the calculations."
+ )
+ end_date: str = Field(description="The end date of the data.")
+ symbols_data: List[Data] = Field(
+ description="The data representing the selected 'study' for each symbol."
+ )
+ benchmark_data: List[Data] = Field(
+ description="The data representing the selected 'study' for the benchmark."
+ )
+ rs_ratios: List[Data] = Field(
+ description="The normalized relative strength ratios data."
+ )
+ rs_momentum: List[Data] = Field(
+ description="The normalized relative strength momentum data."
+ )
+
+ def __init__(self, **data):
+ """Initialize the class."""
+ super().__init__(**data)
+ fields = self.__class__.model_fields
+ doc_str = (
+ "\n"
+ + self.__class__.__name__
+ + "\n\n"
+ + " Attributes\n"
+ + " ----------\n"
+ + "\n".join(
+ [
+ f" {k} : {_get_type_name(v.annotation)}\n {v.description}"
+ for k, v in fields.items()
+ ]
+ )
+ + "\n"
+ )
+ self.__doc__ = doc_str
+
+
+class RelativeRotationFetcher(
+ Fetcher[RelativeRotationQueryParams, RelativeRotationData]
+):
+ """Relative Rotation Fetcher."""
+
+ @staticmethod
+ def transform_query(params: Dict[str, Any]) -> RelativeRotationQueryParams:
+ """Transform the query parameters."""
+ return RelativeRotationQueryParams.model_validate(**params)
+
+ @staticmethod
+ def extract_data(
+ query: RelativeRotationQueryParams,
+ credentials: Optional[Dict[str, str]],
+ **kwargs: Any,
+ ) -> Dict:
+ """Extract the data."""
+ return RelativeRotation(
+ query.data,
+ query.benchmark,
+ study=query.study,
+ long_period=query.long_period,
+ short_period=query.short_period,
+ window=query.window,
+ trading_periods=query.trading_periods,
+ ).__dict__
+
+ @staticmethod
+ def transform_data(
+ query: RelativeRotationQueryParams,
+ data: Dict,
+ **kwargs: Any,
+ ) -> RelativeRotationData:
+ """Transform the data."""
+ return RelativeRotationData.model_validate(data)
diff --git a/openbb_platform/extensions/technical/openbb_technical/technical_router.py b/openbb_platform/extensions/technical/openbb_technical/technical_router.py
index f9b5b3ac3be..47aa24d2700 100644
--- a/openbb_platform/extensions/technical/openbb_technical/technical_router.py
+++ b/openbb_platform/extensions/technical/openbb_technical/technical_router.py
@@ -1,7 +1,7 @@
"""Technical Analysis Router."""
# pylint: disable=too-many-lines
-from typing import List, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional
import pandas as pd
import pandas_ta as ta
@@ -17,17 +17,159 @@ from openbb_core.app.utils import (
from openbb_core.provider.abstract.data import Data
from pydantic import NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt
-from .helpers import (
+from openbb_technical.helpers import (
calculate_cones,
calculate_fib_levels,
clenow_momentum,
validate_data,
)
+from openbb_technical.relative_rotation import (
+ RelativeRotationData,
+ RelativeRotationFetcher,
+ RelativeRotationQueryParams,
+)
# TODO: Split this into multiple files
router = Router(prefix="", description="Technical Analysis tools.")
+@router.command(
+ methods=["POST"],
+ examples=[
+ PythonEx(
+ description="Calculate the Relative Strength Ratio and Relative Strength Momentum"
+ + " for a group of symbols against a benchmark.",
+ code=[
+ "stock_data = obb.equity.price.historical("
+ + "symbol='AAPL,MSFT,GOOGL,META,AMZN,TSLA,SPY', start_date='2022-01-01', provider='yfinance')",
+ "rr_data = obb.technical.relative_rotation(data=stock_data.results, benchmark='SPY')",
+ "rs_ratios = rr_data.results.rs_ratios",
+ "rs_momentum = rr_data.results.rs_momentum",
+ ],
+ ),
+ PythonEx(
+ description="When the assets are not traded 252 days per year,"
+ + "adjust the momentum and volatility periods accordingly.",
+ code=[
+ "crypto_data = obb.crypto.price.historical("
+ + " symbol='BTCUSD,ETHUSD,SOLUSD', start_date='2021-01-01', provider='yfinance')",
+ "rr_data = obb.technical.relative_rotation(data=crypto_data.results, benchmark='BTCUSD',"
+ + " long_period=365, short_period=30, window=30, trading_periods=365)",
+ ],
+ ),
+ APIEx(
+ description="Note that the mock data displayed here is insufficient."
+ + " It must contain multiple symbols, with the benchmark, and be daily data at least 1 year in length.",
+ parameters={"benchmark": "SPY", "data": APIEx.mock_data("timeseries")},
+ ),
+ ],
+)
+async def relative_rotation(
+ data: List[Data],
+ benchmark: str,
+ study: Literal["price", "volume", "volatility"] = "price",
+ long_period: Optional[int] = 252,
+ short_period: Optional[int] = 21,
+ window: Optional[int] = 21,
+ trading_periods: Optional[int] = 252,
+ chart_params: Optional[Dict[str, Any]] = None,
+) -> OBBject[RelativeRotationData]:
+ """Calculate the Relative Strength Ratio and Relative Strength Momentum for a group of symbols against a benchmark.
+
+ Parameters
+ ----------
+ data : list[Data]
+ The data to be used for the relative rotation calculations.
+ This should be the multi-symbol output from the 'equity.price.historical' endpoint, or similar.
+ Or a pivot table with the 'date' column as the index, the symbols as the columns, and the 'study' as the values.
+ It is recommended to use the 'equity.price.historical' endpoint to get the data, and feed the results as-is.
+ benchmark : str
+ The symbol to be used as the benchmark.
+ study : Literal[price, volume, volatility]
+ The data point for the calculations. If 'price', the closing price will be used.
+ If 'volatility', the standard deviation of the closing price will be used.
+ If 'data' is supplied as a pivot table,
+ the 'study' will assume the values are the closing price and 'volume' will be ignored.
+ long_period : int, optional
+ The length of the long period for momentum calculation, by default 252.
+ Adjust this value when supplying a time series with an interval that is not daily.
+ For example, if the data is monthly, the long period should be 12.
+ short_period : int, optional
+ The length of the short period for momentum calculation, by default 21.
+ Adjust this value when supplying a time series with an interval that is not daily.
+ window : int, optional
+ The length of window for the standard deviation calculation, by default 21.
+ Adjust this value when supplying a time series with an interval that is not daily.
+ trading_periods : int, optional
+ The number of trading periods per year, for the standard deviation calculation, by default 252.
+ Adjust this value when supplying a time series with an interval that is not daily.
+ chart_params : dict[str, Any], optional
+ Additional parameters to pass when `chart=True` and the `openbb-charting` extension is installed.
+ Parameters can be passed again to redraw the chart using the charting.to_chart() method of the response.
+
+ ChartParams
+ -----------
+ date : str, optional
+ A target end date within the data to use for the chart, by default is the last date in the data.
+ show_tails : bool
+ Show the tails on the chart, by default True.
+ tail_periods : int
+ Number of periods to show in the tails, by default 16.
+ tail_interval : Literal[day, week, month]
+ Interval to show the tails, by default 'week'.
+ title : str, optional
+ Title of the chart.
+
+ Returns
+ -------
+ OBBject[RelativeRotationData]
+ results : RelativeRotationData
+ symbols : list[str]:
+ The symbols that are being compared against the benchmark.
+ benchmark : str
+ The benchmark symbol.
+ study : Literal[price, volume, volatility]
+ The data point for the selected.
+ long_period : int
+ The length of the long period for momentum calculation, as entered by the user.
+ short_period : int
+ The length of the short period for momentum calculation, as entered by the user.
+ window : int
+ The length of window for the standard deviation calculation.
+ trading_periods : int
+ The number of trading periods per year, for the standard deviation calculation.
+ start_date : str
+ The start date of the data after adjusting the length of the data for the calculations.
+ end_date : str
+ The end date of the data.
+ symbols_data : list[Data]
+ The data representing the selected 'study' for each symbol.
+ benchmark_data : list[Data]
+ The data representing the selected 'study' for the benchmark.
+ rs_ratios : list[Data]
+ The normalized relative strength ratios data.
+ rs_momentum : list[Data]
+ The normalized relative strength momentum data.
+ """
+
+ params = RelativeRotationQueryParams(
+ data=data,
+ benchmark=benchmark,
+ study=study,
+ long_period=long_period,
+ short_period=short_period,
+ window=window,
+ trading_periods=trading_periods,
+ chart_params=chart_params,
+ )
+
+ return OBBject(
+ results=RelativeRotationFetcher.transform_data(
+ params, RelativeRotationFetcher.extract_data(params, {})
+ )
+ )
+
+
@router.command(
methods=["POST"],
examples=[
diff --git a/openbb_platform/obbject_extensions/charting/integration/test_charting_api.py b/openbb_platform/obbject_extensions/charting/integration/test_charting_api.py
index ab23e816822..0c6fb8f4c39 100644
--- a/openbb_platform/obbject_extensions/charting/integration/test_charting_api.py
+++ b/openbb_platform/obbject_extensions/charting/integration/test_charting_api.py
@@ -572,6 +572,52 @@ def test_charting_economy_fred_series(params, headers):
assert list(chart.keys()) == ["content", "format"]
+@parametrize(
+ "params",
+ [
+ (
+ {
+ "data": "",
+ "study": "price",
+ "benchmark": "SPY",
+ "long_period": 252,
+ "short_period": 21,
+ "window": 21,
+ "trading_periods": 252,
+ "normalize_method": "z",
+ }
+ ),
+ ],
+)
+@pytest.mark.integration
+def test_charting_technical_relative_rotation(params):
+ params = {p: v for p, v in params.items() if v}
+ data_params = dict(
+ symbol="AAPL,MSFT,GOOGL,AMZN,SPY",
+ provider="yfinance",
+ start_date="2022-01-01",
+ end_date="2024-01-01",
+ )
+ data_query_str = get_querystring(data_params, [])
+ data_url = f"http://0.0.0.0:8000/api/v1/equity/price/historical?{data_query_str}"
+ data_result = requests.get(data_url, headers=get_headers(), timeout=10).json()[
+ "results"
+ ]
+ body = json.dumps({"data": data_result})
+ query_str = get_querystring(params, ["data"])
+ url = f"http://0.0.0.0:8000/api/v1/technical/relative_rotation?{query_str}"
+ result = requests.post(url, headers=get_headers(), timeout=10, data=body)
+ assert isinstance(result, requests.Response)
+ assert result.status_code == 200
+
+ chart = result.json()["chart"]
+ fig = chart.pop("fig", {})
+
+ assert chart
+ assert not fig
+ assert list(chart.keys()) == ["content", "format"]
+
+
@parametrize(
"params",
[
diff --git a/openbb_platform/obbject_extensions/charting/integration/test_charting_python.py b/openbb_platform/obbject_extensions/charting/integration/test_charting_python.py
index 9349f826e95..3e8bdd0de7d 100644
--- a/openbb_platform/obbject_extensions/charting/integration/test_charting_python.py
+++ b/openbb_platform/obbject_extensions/charting/integration/test_charting_python.py
@@ -475,6 +475,50 @@ def test_charting_economy_fred_series(params, obb):
assert isinstance(result.chart.fig, OpenBBFigure)
+@parametrize(
+ "params",
+ [
+ (
+ {
+ "data": "",
+ "study": "price",
+ "benchmark": "SPY",
+ "long_period": 252,
+ "short_period": 21,
+ "window": 21,
+ "trading_periods": 252,
+ "normalize_method": "z",
+ "chart": True,
+ }
+ ),
+ ],
+)
+@pytest.mark.integration
+def test_charting_technical_relative_rotation(params, obb):
+ params["data"] = obb.equity.price.historical(
+ "AAPL,MSFT,GOOGL,AMZN,SPY",
+ provider="yfinance",
+ start_date="2022-01-01",
+ end_date="2024-01-01",
+ ).results
+ result = obb.technical.relative_rotation(
+ data=params["data"],
+ benchmark=params["benchmark"],
+ study=params["study"],
+ long_period=params["long_period"],
+ short_period=params["short_period"],
+ window=params["window"],
+ trading_periods=params["trading_periods"],
+ normalize_method=params["normalize_method"],
+ chart=params["chart"],
+ )
+ assert result
+ assert isinstance(result, OBBject)
+ assert len(result.results.rs_ratios) > 0 # type: ignore
+ assert result.chart.content # type: ignore
+ assert isinstance(result.chart.fig, OpenBBFigure) # type: ignore
+
+
@parametrize(
"params",
[
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py b/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py
index b13bf0d5a05..fc271fbeaa1 100644
--- a/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py
+++ b/openbb_platform/obbject_extensions/charting/openbb_charting/charting_router.py
@@ -15,6 +15,7 @@ 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,
@@ -445,6 +446,7 @@ def equity_price_historical( # noqa: PLR0912
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)
@@ -471,8 +473,9 @@ def equity_price_historical( # noqa: PLR0912
xaxis=dict(
ticklen=0,
showgrid=True,
- showline=True,
gridcolor="rgba(128,128,128,0.3)",
+ showline=True,
+ mirror=True,
),
margin=dict(l=20, r=20, b=20, t=20),
dragmode="pan",
@@ -1219,3 +1222,73 @@ def economy_fred_series( # noqa: PLR0912
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
+ )
+
+ 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)
+ 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
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/query_params.py b/openbb_platform/obbject_extensions/charting/openbb_charting/query_params.py
index 003d13b35ba..db8274042c0 100644
--- a/openbb_platform/obbject_extensions/charting/openbb_charting/query_params.py
+++ b/openbb_platform/obbject_extensions/charting/openbb_charting/query_params.py
@@ -15,11 +15,17 @@ MAMODES = Literal["ema", "sma", "wma", "hna", "zlma", "rma"]
def _get_type_name(t):
"""Get the type name of a type hint."""
-
- if hasattr(t, "__origin__") and hasattr(t.__origin__, "__name__"):
- return f"{t.__origin__.__name__}[{', '.join([_get_type_name(arg) for arg in t.__args__])}]"
+ if hasattr(t, "__origin__"):
+ if hasattr(t.__origin__, "__name__"):
+ return f"{t.__origin__.__name__}[{', '.join([_get_type_name(arg) for arg in t.__args__])}]"
+ if hasattr(t.__origin__, "_name"):
+ return f"{t.__origin__._name}[{', '.join([_get_type_name(arg) for arg in t.__args__])}]" # pylint: disable=W0212
+ if isinstance(t, str):
+ return t
if hasattr(t, "__name__"):
return t.__name__
+ if hasattr(t, "_name"):
+ return t._name # pylint: disable=W0212
return str(t)
@@ -344,6 +350,31 @@ class TechnicalRSIChartQueryParams(ChartQueryParams):
)
+class TechnicalRelativeRotationChartQueryParams(ChartQueryParams):
+ """Technical Relative Rotation Chart Query Params."""
+
+ date: Optional[str] = Field(
+ default=None,
+ description="A target end date within the data to use for the chart, by default is the last date in the data.",
+ )
+ show_tails: bool = Field(
+ default=True,
+ description="Show the tails on the chart, by default True.",
+ )
+ tail_periods: Optional[int] = Field(
+ default=16,
+ description="Number of periods to show in the tails, by default 16.",
+ )
+ tail_interval: Literal["day", "week", "month"] = Field(
+ default="week",
+ description="The interval to show the tails, by default 'week'.",
+ )
+ title: Optional[str] = Field(
+ default=None,
+ description="Title of the chart.",
+ )
+
+
class ChartParams:
"""Chart Query Params."""
@@ -356,16 +387,17 @@ class ChartParams:
etf_holdings = EtfHoldingsChartQueryParams
etf_price_performance = EquityPricePerformanceChartQueryParams
index_price_historical = EquityPriceHistoricalChartQueryParams
- technical_cones = TechnicalConesChartQueryParams
- technical_sma = TechnicalSMAChartQueryParams
- technical_ema = TechnicalEMAChartQueryParams
- technical_hma = TechnicalHMAChartQueryParams
- technical_wma = TechnicalWMAChartQueryParams
- technical_zlma = TechnicalZLMAChartQueryParams
technical_adx = TechnicalADXChartQueryParams
technical_aroon = TechnicalArooonChartQueryParams
+ technical_cones = TechnicalConesChartQueryParams
+ technical_ema = TechnicalEMAChartQueryParams
+ technical_hma = TechnicalHMAChartQueryParams
technical_macd = TechnicalMACDChartQueryParams
+ technical_relative_rotation = TechnicalRelativeRotationChartQueryParams
technical_rsi = TechnicalRSIChartQueryParams
+ technical_sma = TechnicalSMAChartQueryParams
+ technical_wma = TechnicalWMAChartQueryParams
+ technical_zlma = TechnicalZLMAChartQueryParams
class IndicatorsQueryParams(BaseQueryParams):
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/dark.pltstyle.json b/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/dark.pltstyle.json
index 9504aacce04..d5d75ac6826 100644
--- a/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/dark.pltstyle.json
+++ b/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/dark.pltstyle.json
@@ -134,7 +134,7 @@
}
},
- "gridcolor": "rgba(128, 128, 128, 0.25)",
+ "gridcolor": "rgba(128, 128, 128, 0.33)",
"linecolor": "#F5EFF3",
"mirror": true,
"ticks": "outside"
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/light.pltstyle.json b/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/light.pltstyle.json
index d634f13e9ed..4f3198c4989 100644
--- a/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/light.pltstyle.json
+++ b/openbb_platform/obbject_extensions/charting/openbb_charting/styles/default/light.pltstyle.json
@@ -109,7 +109,10 @@
"zeroline": false,
"tick0": 1,
"title": {
- "standoff": 20
+ "standoff": 20,
+ "font": {
+ "size": 16
+ }
},
"linecolor": "#A9A9A9",
"mirror": true,
@@ -125,9 +128,12 @@
"side": "right",
"tick0": 0.5,
"title": {
- "standoff": 20
+ "standoff": 20,
+ "font": {
+ "size": 16
+ }
},
- "gridcolor": "rgba(128, 128, 128, 0.25)",
+ "gridcolor": "rgba(128, 128, 128, 0.33)",
"linecolor": "#A9A9A9",
"mirror": true,
"ticks": "outside"
diff --git a/openbb_platform/obbject_extensions/charting/openbb_charting/utils/relative_rotation.py b/openbb_platform/obbject_extensions/charting/openbb_charting/utils/relative_rotation.py
new file mode 100644
index 00000000000..b543d5d2633
--- /dev/null
+++ b/openbb_platform/obbject_extensions/charting/openbb_charting/utils/relative_rotation.py
@@ -0,0 +1,582 @@
+"""Relative Rotation Chart Helpers."""
+
+from datetime import date as dateType
+from typing import Literal, Optional
+from warnings import warn
+
+from pandas import DataFrame, to_datetime
+from plotly import graph_objects as go
+from plotly.graph_objects import Figure
+
+color_sequence = [
+ "burlywood",
+ "orange",
+ "grey",
+ "magenta",
+ "cyan",
+ "yellowgreen",
+ "#1f77b4",
+ "#aec7e8",
+ "#ff7f0e",
+ "#ffbb78",
+ "#d62728",
+ "#ff9896",
+ "#9467bd",
+ "#c5b0d5",
+ "#8c564b",
+ "#c49c94",
+ "#e377c2",
+ "#f7b6d2",
+ "#7f7f7f",
+ "#c7c7c7",
+ "#bcbd22",
+ "#dbdb8d",
+ "#17becf",
+ "#9edae5",
+ "#7e7e7e",
+ "#1b9e77",
+ "#d95f02",
+ "#7570b3",
+ "#e7298a",
+ "#66a61e",
+ "#e6ab02",
+ "#a6761d",
+ "#666666",
+ "#f0027f",
+ "#bf5b17",
+ "#d9f202",
+ "#8dd3c7",
+ "#ffffb3",
+ "#bebada",
+ "#fb8072",
+ "#80b1d3",
+ "#fdb462",
+ "#b3de69",
+ "#fccde5",
+ "#d9d9d9",
+ "#bc80bd",
+ "#ccebc5",
+ "#ffed6f",
+ "#6a3d9a",
+ "#b15928",
+ "#b2df8a",
+ "#33a02c",
+ "#fb9a99",
+ "#e31a1c",
+ "#fdbf6f",
+ "#ff7f00",
+ "#cab2d6",
+ "#6a3d9a",
+ "#ffff99",
+ "#b15928",
+]
+
+
+def create_rrg_with_tails(
+ ratios_data: DataFrame,
+ momentum_data: DataFrame,
+ study: str,
+ benchmark_symbol: str,
+ tail_periods: int,
+ tail_interval: Literal["day", "week", "month"],
+) -> Figure:
+ """Create The Relative Rotation Graph With Tails.
+
+ Parameters
+ ----------
+ ratios_data : DataFrame
+ The DataFrame containing the RS-Ratio values.
+ momentum_data : DataFrame
+ The DataFrame containing the RS-Momentum values.
+ study : str
+ The study that was selected when loading the raw data.
+ If custom data is supplied, this will override the study for the chart titles.
+ benchmark_symbol : str
+ The symbol of the benchmark.
+ tail_periods : int
+ The number of periods to display in the tails.
+ tail_interval : Literal["day", "week", "month"]
+
+ Returns
+ -------
+ Figure
+ Plotly GraphObjects Figure.
+ """
+ color = 0
+ symbols = ratios_data.columns.to_list()
+
+ tail_dict = {"week": "W", "month": "M"}
+
+ ratios_data.index = to_datetime(ratios_data.index)
+ momentum_data.index = to_datetime(momentum_data.index)
+ if tail_interval != "day":
+ ratios_data = ratios_data.resample(tail_dict[tail_interval]).last()
+ momentum_data = momentum_data.resample(tail_dict[tail_interval]).last()
+ ratios_data = ratios_data.iloc[-tail_periods:]
+ momentum_data = momentum_data.iloc[-tail_periods:]
+ _tail_periods = len(ratios_data)
+ tail_title = (
+ f"The Previous {_tail_periods} {tail_interval.capitalize()}s "
+ f"Ending {ratios_data.index[-1].strftime('%Y-%m-%d')}"
+ )
+ x_min = ratios_data.min().min()
+ x_max = ratios_data.max().max()
+ y_min = momentum_data.min().min()
+ y_max = momentum_data.max().max()
+ # Create an empty list to store the scatter traces
+ traces = []
+
+ for symbol in symbols:
+ # Select a single row from each dataframe
+ x_data = ratios_data[symbol]
+ y_data = momentum_data[symbol]
+ name = symbol.upper().replace("^", "").replace(":US", "")
+ # Create a trace for the line
+ line_trace = go.Scattergl(
+ x=x_data[:-1], # All but the last data point
+ y=y_data[:-1], # All but the last data point
+ mode="lines+markers",
+ line=dict(
+ color=color_sequence[color % len(color_sequence)], width=2, dash="dash"
+ ),
+ marker=dict(size=6, color=color_sequence[color % len(color_sequence)]),
+ opacity=0.3,
+ showlegend=False,
+ name=name,
+ text=name,
+ hovertemplate="%{fullData.name}"
+ + "
RS-Ratio: %{x:.4f}"
+ + "RS-Momentum: %{y:.4f}"
+ + "",
+ hoverlabel=dict(font_size=10),
+ )
+ special_name = "-" in name or len(name) > 5
+ marker_size = 38 if special_name else 30
+ # Create a trace for the last data point
+ marker_trace = go.Scatter(
+ x=[x_data.iloc[-1]], # Only the last data point
+ y=[y_data.iloc[-1]], # Only the last data point
+ mode="markers+text",
+ name=name,
+ text=[name],
+ textposition="middle center",
+ textfont=(
+ dict(size=10, color="black")
+ if len(name) < 4
+ else dict(size=8, color="black")
+ ),
+ marker=dict(
+ size=marker_size,
+ color=color_sequence[color % len(color_sequence)],
+ line=dict(color="black", width=1),
+ ),
+ showlegend=False,
+ hovertemplate="%{fullData.name}"
+ + "
RS-Ratio: %{x:.4f}"
+ + "RS-Momentum: %{y:.4f}"
+ + "",
+ hoverlabel=dict(font_size=10),
+ )
+ traces.extend([line_trace, marker_trace])
+ color += 1
+ padding = 0.1
+ y_range = [y_min - padding * abs(y_min) - 0.3, y_max + padding * abs(y_max) + 0.3]
+ x_range = [x_min - padding * abs(x_min) - 0.3, x_max + padding * abs(x_max) + 0.3]
+
+ layout = go.Layout(
+ title={
+ "text": (
+ f"Relative Rotation Against {benchmark_symbol.replace('^', '')} {study.capitalize()} For {tail_title}"
+ ),
+ "x": 0.5,
+ "xanchor": "center",
+ "font": dict(color="white", size=18),
+ },
+ xaxis=dict(
+ title="RS-Ratio",
+ zerolinecolor="black",
+ range=x_range,
+ showspikes=False,
+ ),
+ yaxis=dict(
+ title="
RS-Momentum",
+ zerolinecolor="black",
+ range=y_range,
+ showspikes=False,
+ side="left",
+ title_standoff=5,
+ ),
+ shapes=[
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=0,
+ y0=0,
+ x1=x_range[1],
+ y1=y_range[1],
+ fillcolor="lightgreen",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=x_range[0],
+ y0=0,
+ x1=0,
+ y1=y_range[1],
+ fillcolor="lightblue",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=x_range[0],
+ y0=y_range[0],
+ x1=0,
+ y1=0,
+ fillcolor="lightpink",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=0,
+ y0=y_range[0],
+ x1=x_range[1],
+ y1=0,
+ fillcolor="lightyellow",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=x_range[0],
+ y0=y_range[0],
+ x1=x_range[1],
+ y1=y_range[1],
+ line=dict(
+ color="Black",
+ width=1,
+ ),
+ fillcolor="rgba(0,0,0,0)",
+ layer="above",
+ ),
+ ],
+ annotations=[
+ go.layout.Annotation(
+ x=1,
+ xref="paper",
+ y=1,
+ yref="paper",
+ text="Leading",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="darkgreen",
+ ),
+ ),
+ go.layout.Annotation(
+ x=1,
+ xref="paper",
+ y=0,
+ yref="paper",
+ text="Weakening",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="goldenrod",
+ ),
+ ),
+ go.layout.Annotation(
+ x=0,
+ xref="paper",
+ y=0,
+ yref="paper",
+ text="Lagging",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="red",
+ ),
+ ),
+ go.layout.Annotation(
+ x=0,
+ xref="paper",
+ yref="paper",
+ y=1,
+ text="Improving",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="blue",
+ ),
+ ),
+ ],
+ autosize=True,
+ margin=dict(
+ l=30,
+ r=50,
+ b=50,
+ t=50,
+ pad=0,
+ ),
+ dragmode="pan",
+ hovermode="closest",
+ )
+
+ fig = go.Figure(data=traces, layout=layout)
+
+ return fig
+
+
+def create_rrg_without_tails(
+ ratios_data: DataFrame,
+ momentum_data: DataFrame,
+ benchmark_symbol: str,
+ study: str,
+ date: Optional[dateType] = None,
+) -> Figure:
+ """Create the Plotly Figure Object without Tails.
+
+ Parameters
+ ----------
+ ratios_data : DataFrame
+ The DataFrame containing the RS-Ratio values.
+ momentum_data : DataFrame
+ The DataFrame containing the RS-Momentum values.
+ benchmark_symbol : str
+ The symbol of the benchmark.
+ study: str
+ The study that was selected when loading the raw data.
+ If custom data is supplied, this will override the study for the chart titles.
+ date : Optional[dateType], optional
+ A specific date within the data to target for display, by default None.
+
+ Returns
+ -------
+ Figure
+ Plotly GraphObjects Figure.
+ """
+
+ if date is not None and date not in ratios_data.index.astype(str):
+ warn(f"Date {str(date)} not found in data, using the last available date.")
+ date = ratios_data.index[-1]
+ if date is None:
+ date = ratios_data.index[-1]
+
+ # Select a single row from each dataframe
+ row_x = ratios_data.loc[to_datetime(date).date()] # type: ignore
+ row_y = momentum_data.loc[to_datetime(date).date()] # type: ignore
+
+ x_max = row_x.max() + 0.5
+ x_min = row_x.min() - 0.5
+ y_max = row_y.max() + 0.5
+ y_min = row_y.min() - 0.5
+
+ # Create an empty list to store the scatter traces
+ traces = []
+
+ # Loop through each column in the row_x dataframe
+ for i, (column_name, value_x) in enumerate(row_x.items()):
+ # Retrieve the corresponding value from the row_y dataframe
+ value_y = row_y[column_name] # type: ignore
+ marker_name = column_name.upper().replace("^", "").replace(":US", "") # type: ignore
+ special_name = "-" in marker_name or len(marker_name) > 5
+ marker_size = 38 if special_name else 30
+ # Create a scatter trace for each column
+ trace = go.Scatter(
+ x=[value_x],
+ y=[value_y],
+ mode="markers+text",
+ text=[marker_name],
+ textposition="middle center",
+ textfont=dict(size=10 if len(marker_name) < 4 else 8, color="black"),
+ marker=dict(
+ size=marker_size,
+ color=color_sequence[i % len(color_sequence)],
+ line=dict(color="black", width=1),
+ ),
+ name=column_name,
+ showlegend=False,
+ hovertemplate="%{fullData.name}"
+ + "
RS-Ratio: %{x:.4f}"
+ + "RS-Momentum: %{y:.4f}"
+ + "",
+ )
+ # Add the trace to the list
+ traces.append(trace)
+
+ padding = 0.1
+ y_range = [y_min - padding * abs(y_min) - 0.3, y_max + padding * abs(y_max)]
+ x_range = [x_min - padding * abs(x_min), x_max + padding * abs(x_max)]
+
+ layout = go.Layout(
+ title={
+ "text": (
+ f"RS-Ratio vs RS-Momentum of {study.capitalize()} "
+ f"Against {benchmark_symbol.replace('^', '')} - {to_datetime(row_x.name).strftime('%Y-%m-%d')}" # type: ignore
+ ),
+ "x": 0.5,
+ "xanchor": "center",
+ "font": dict(size=20),
+ },
+ xaxis=dict(
+ title="RS-Ratio",
+ zerolinecolor="black",
+ range=x_range,
+ showspikes=False,
+ ),
+ yaxis=dict(
+ title="
RS-Momentum",
+ zerolinecolor="black",
+ range=y_range,
+ side="left",
+ title_standoff=5,
+ showspikes=False,
+ ),
+ shapes=[
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=0,
+ y0=0,
+ x1=x_range[1],
+ y1=y_range[1],
+ fillcolor="lightgreen",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=x_range[0],
+ y0=0,
+ x1=0,
+ y1=y_range[1],
+ fillcolor="lightblue",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=x_range[0],
+ y0=y_range[0],
+ x1=0,
+ y1=0,
+ fillcolor="lightpink",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=0,
+ y0=y_range[0],
+ x1=x_range[1],
+ y1=0,
+ fillcolor="lightyellow",
+ opacity=0.3,
+ layer="below",
+ line_width=0,
+ ),
+ go.layout.Shape(
+ type="rect",
+ xref="x",
+ yref="y",
+ x0=x_range[0],
+ y0=y_range[0],
+ x1=x_range[1],
+ y1=y_range[1],
+ line=dict(
+ color="Black",
+ width=1,
+ ),
+ fillcolor="rgba(0,0,0,0)",
+ layer="above",
+ ),
+ ],
+ annotations=[
+ go.layout.Annotation(
+ x=1,
+ xref="paper",
+ y=1,
+ yref="paper",
+ text="Leading",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="darkgreen",
+ ),
+ ),
+ go.layout.Annotation(
+ x=1,
+ xref="paper",
+ y=0,
+ yref="paper",
+ text="Weakening",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="goldenrod",
+ ),
+ ),
+ go.layout.Annotation(
+ x=0,
+ xref="paper",
+ y=0,
+ yref="paper",
+ text="Lagging",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="red",
+ ),
+ ),
+ go.layout.Annotation(
+ x=0,
+ xref="paper",
+ yref="paper",
+ y=1,
+ text="Improving",
+ showarrow=False,
+ font=dict(
+ size=18,
+ color="blue",
+ ),
+ ),
+ ],
+ autosize=True,
+ margin=dict(
+ l=30,
+ r=50,
+ b=50,
+ t=50,
+ pad=0,
+ ),
+ dragmode="pan",
+ )
+
+ fig = go.Figure(data=traces, layout=layout)
+
+ return fig