summaryrefslogtreecommitdiffstats
path: root/openbb_platform/extensions/derivatives/openbb_derivatives/derivatives_views.py
blob: 7400f2cf5d37bbee0472b15f924fe8cbaa7f31d7 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
"""Views for the Derivatives Extension."""

from typing import TYPE_CHECKING, Any, Dict, Tuple

if TYPE_CHECKING:
    from openbb_charting.core.openbb_figure import OpenBBFigure


class DerivativesViews:
    """Derivatives Views."""

    @staticmethod
    def derivatives_futures_historical(  # noqa: PLR0912
        **kwargs,
    ) -> Tuple["OpenBBFigure", Dict[str, Any]]:
        """Get Derivatives Futures Historical Chart."""
        # pylint: disable=import-outside-toplevel
        from openbb_charting.charts.price_historical import price_historical

        return price_historical(**kwargs)

    @staticmethod
    def derivatives_futures_curve(  # noqa: PLR0912
        **kwargs,
    ) -> Tuple["OpenBBFigure", Dict[str, Any]]:
        """Futures curve chart. All parameters are optional, and are kwargs.
        Parameters can be directly accessed from the function end point by
        entering as a nested dictionary to the 'chart_params' key.

        From the API, `chart_params` must be passed as a JSON in the request body with `extra_params`.

        If using the chart post-request, the parameters are passed directly
        as `key=value` pairs in the `charting.to_chart` or `charting.show` methods.

        Parameters
        ----------
        data : Optional[Union[List[Data], DataFrame]]
            Data for the chart. Required fields are: 'expiration' and 'price'.
            Multiple dates will be plotted on the same chart.
            If not supplied, the original OBBject.results will be used.
            If a DataFrame is supplied, flat data is expected, without a set index.
        title: Optional[str]
            Title for the chart. If not supplied, a default title will be used.
        colors: Optional[List[str]]
            List of colors to use for the chart. If not supplied, the default colorway will be used.
            Colors should be in hex format, or named Plotly colors. Invalid colors will raise a Plotly error.
        layout_kwargs: Optional[Dict[str, Any]]
            Additional layout parameters for the chart, passed directly to `figure.update_layout` before output.
            See Plotly documentation for available options.

        Returns
        -------
        Tuple[OpenBBFigure, Dict[str, Any]]
            Tuple with the OpenBBFigure object, and the JSON-serialized content.
            If using the API, only the JSON content will be returned.

        Examples
        --------
        ```python
        from openbb import obb
        data = obb.derivatives.futures.curve(symbol="vx", provider="cboe", date=["2020-03-31", "2024-06-28"], chart=True)
        data.show()
        ```

        Redraw the chart, from the same data, with a custom colorway and title:

        ```python
        data.charting.to_chart(colors=["green", "red"], title="VIX Futures Curve - 2020 vs. 2024")
        ```
        """
        # pylint: disable=import-outside-toplevel
        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.model.abstract.error import OpenBBError
        from openbb_core.provider.abstract.data import Data
        from pandas import DataFrame, to_datetime

        data = kwargs.get("data", None)
        symbol = kwargs.get("standard_params", {}).get("symbol", "")
        df: DataFrame = DataFrame()
        if data:
            if isinstance(data, DataFrame) and not data.empty:  # noqa: SIM108
                df = data
            elif isinstance(data, (list, Data)):
                df = DataFrame([d.model_dump(exclude_none=True, exclude_unset=True) for d in data])  # type: ignore
            else:
                pass
        else:
            df = DataFrame(
                [
                    d.model_dump(exclude_none=True, exclude_unset=True)  # type: ignore
                    for d in kwargs["obbject_item"]
                ]
                if isinstance(kwargs.get("obbject_item"), list)
                else kwargs["obbject_item"].model_dump(exclude_none=True, exclude_unset=True)  # type: ignore
            )

        if df.empty:
            raise OpenBBError("Error: No data to plot.")

        if "expiration" not in df.columns:
            raise OpenBBError("Expiration field not found in the data.")

        if "price" not in df.columns:
            raise ValueError("Price field not found in the data.")

        provider = kwargs.get("provider", "")

        df["expiration"] = (
            to_datetime(df["expiration"], errors="ignore")
            .dt
            .strftime("%b-%Y")
        )

        if (
            provider == "cboe"
            and "date" in df.columns
            and len(df["date"].unique()) > 1
            and "symbol" in df.columns
        ):
            df["expiration"] = df.symbol

        # Use a complete list of expirations to categorize the x-axis across all dates.
        expirations = df["expiration"].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):
            """Create a scatter for each date in the data."""
            for date in dates:
                color = colors[color_count % len(colors)]
                plot_df = (
                    df[df["date"].astype(str) == date].copy()
                    if "date" in df.columns
                    else df.copy()
                )
                plot_df = plot_df.drop(
                    columns=["date"] if "date" in plot_df.columns else []
                ).rename(columns={"expiration": "Expiration", "price": "Price"})
                figure.add_scatter(
                    x=plot_df["Expiration"],
                    y=plot_df["Price"],
                    mode="lines+markers",
                    name=date,
                    line=dict(width=3, color=color),
                    marker=dict(size=10, color=color),
                    hovertemplate=(
                        "Expiration: %{x}<br>Price: $%{y}<extra></extra>"
                        if len(dates) == 1
                        else "%{fullData.name}<br>Expiration: %{x}<br>Price: $%{y}<extra></extra>"
                    ),
                )
                color_count += 1
            return figure, color_count

        dates = (
            df.date.astype(str).unique().tolist()
            if "date" in df.columns
            else ["Current"]
        )
        figure, color_count = create_fig(figure, df, dates, color_count)

        # Set the title for the chart
        title: str = ""
        if provider == "cboe":
            vx_eod_symbols = ["vx", "vix", "vx_eod", "^vix"]
            title = (
                "VIX EOD Futures Curve"
                if symbol.lower() in vx_eod_symbols
                else "VIX Mid-Morning TWAP Futures Curve"
            )
            if len(dates) == 1 and dates[0] != "Current":
                title = f"{title} for {dates[0]}"
        elif provider == "yfinance":
            title = f"{symbol.upper()} Futures Curve"

        # Use the supplied title, if any.
        title = kwargs.get("title", title)

        # 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="",
                ticklen=0,
                showgrid=False,
                type="category",
                categoryorder="array",
                categoryarray=expirations,
            ),
            yaxis=dict(
                title="Price ($)",
                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=10,
                t=10,
            ),
        )

        layout_kwargs = kwargs.get("layout_kwargs", {})
        if layout_kwargs:
            figure.update_layout(layout_kwargs)

        content = figure.show(external=True).to_plotly_json()

        return figure, content