Pratik Bhadane
346 lines
11 KiB
Python
346 lines
11 KiB
Python
"""
|
|
ferro_ta.dashboard — Interactive dashboards and exploration helpers.
|
|
===================================================================
|
|
|
|
Optional helpers for interactive exploration in Jupyter notebooks (via
|
|
ipywidgets) and a Streamlit template. All widgets are optional: if ipywidgets
|
|
or streamlit are not installed, a clear ``ImportError`` is raised with install
|
|
instructions.
|
|
|
|
Functions
|
|
---------
|
|
indicator_widget(close, indicator_fn, param_name, param_range)
|
|
Create an ipywidgets slider that updates an indicator plot in real time.
|
|
|
|
backtest_widget(close, strategy_fn, param_name, param_range)
|
|
Create an ipywidgets slider that re-runs a backtest and shows equity curve.
|
|
|
|
streamlit_app()
|
|
Launch a minimal Streamlit dashboard (call from a ``streamlit run`` script).
|
|
|
|
Notes
|
|
-----
|
|
To install optional dependencies::
|
|
|
|
pip install ferro-ta[dashboard] # installs ipywidgets
|
|
pip install streamlit # for Streamlit app
|
|
|
|
Only the Python layer is in this module — all heavy computation delegated to
|
|
existing ferro-ta indicator and backtest functions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Sequence
|
|
from typing import Any, Union
|
|
|
|
import numpy as np
|
|
from numpy.typing import ArrayLike, NDArray
|
|
|
|
__all__ = [
|
|
"indicator_widget",
|
|
"backtest_widget",
|
|
"streamlit_app",
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jupyter / ipywidgets helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def indicator_widget(
|
|
close: ArrayLike,
|
|
indicator_fn: Callable[..., Any],
|
|
param_name: str,
|
|
param_range: Sequence[int],
|
|
title: str = "Indicator",
|
|
) -> Any:
|
|
"""Create an interactive Jupyter widget with a parameter slider.
|
|
|
|
Renders a ``matplotlib`` chart with the close price overlaid by the
|
|
indicator output. Dragging the slider updates the chart in real time.
|
|
|
|
Parameters
|
|
----------
|
|
close : array-like — close price series
|
|
indicator_fn : callable — indicator function, e.g. ``ferro_ta.SMA``.
|
|
Signature: ``fn(close, **{param_name: value}) -> ndarray``.
|
|
param_name : str — name of the integer parameter to vary (e.g. ``'timeperiod'``).
|
|
param_range : sequence of int — values to iterate over (e.g. ``range(5, 51)``).
|
|
title : str — chart title.
|
|
|
|
Returns
|
|
-------
|
|
ipywidgets ``Output`` widget — display it in a Jupyter cell.
|
|
|
|
Requires
|
|
--------
|
|
``ipywidgets``, ``matplotlib``
|
|
|
|
Examples
|
|
--------
|
|
>>> from ferro_ta import SMA
|
|
>>> from ferro_ta.tools.dashboard import indicator_widget
|
|
>>> w = indicator_widget(close, SMA, 'timeperiod', range(5, 51))
|
|
>>> display(w) # in a Jupyter cell
|
|
"""
|
|
try:
|
|
import ipywidgets as widgets
|
|
import matplotlib.pyplot as plt
|
|
except ImportError as exc:
|
|
raise ImportError(
|
|
"indicator_widget requires ipywidgets and matplotlib.\n"
|
|
"Install with: pip install ipywidgets matplotlib"
|
|
) from exc
|
|
|
|
c = np.asarray(close, dtype=np.float64)
|
|
param_values = list(param_range)
|
|
|
|
out = widgets.Output()
|
|
|
|
def update(change: Any) -> None:
|
|
value = change["new"]
|
|
with out:
|
|
out.clear_output(wait=True)
|
|
fig, ax = plt.subplots(figsize=(12, 4))
|
|
ax.plot(c, label="Close", alpha=0.5)
|
|
ind_out = indicator_fn(c, **{param_name: value})
|
|
if isinstance(ind_out, tuple):
|
|
for arr in ind_out:
|
|
ax.plot(np.asarray(arr, dtype=np.float64), alpha=0.8)
|
|
else:
|
|
ax.plot(
|
|
np.asarray(ind_out, dtype=np.float64),
|
|
label=f"{indicator_fn.__name__}({param_name}={value})",
|
|
)
|
|
ax.set_title(f"{title} — {param_name}={value}")
|
|
ax.legend()
|
|
plt.tight_layout()
|
|
plt.show()
|
|
|
|
slider = widgets.IntSlider(
|
|
value=param_values[len(param_values) // 2],
|
|
min=min(param_values),
|
|
max=max(param_values),
|
|
step=1,
|
|
description=param_name,
|
|
continuous_update=False,
|
|
)
|
|
slider.observe(update, names="value")
|
|
update({"new": slider.value})
|
|
|
|
return widgets.VBox([slider, out])
|
|
|
|
|
|
def backtest_widget(
|
|
close: ArrayLike,
|
|
strategy: Union[str, Callable[..., Any]] = "rsi_30_70",
|
|
param_name: str = "timeperiod",
|
|
param_range: Sequence[int] = range(5, 30),
|
|
title: str = "Backtest",
|
|
) -> Any:
|
|
"""Create an interactive Jupyter widget that re-runs a backtest on slider change.
|
|
|
|
Parameters
|
|
----------
|
|
close : array-like — close prices
|
|
strategy : str or callable — backtest strategy (see ``ferro_ta.backtest.backtest``).
|
|
param_name : str — strategy parameter name to vary.
|
|
param_range: sequence of int — parameter values to iterate.
|
|
title : str — chart title.
|
|
|
|
Returns
|
|
-------
|
|
ipywidgets ``VBox`` widget.
|
|
|
|
Requires
|
|
--------
|
|
``ipywidgets``, ``matplotlib``
|
|
"""
|
|
try:
|
|
import ipywidgets as widgets
|
|
import matplotlib.pyplot as plt
|
|
except ImportError as exc:
|
|
raise ImportError(
|
|
"backtest_widget requires ipywidgets and matplotlib.\n"
|
|
"Install with: pip install ipywidgets matplotlib"
|
|
) from exc
|
|
|
|
from ferro_ta.analysis.backtest import backtest
|
|
|
|
c = np.asarray(close, dtype=np.float64)
|
|
param_values = list(param_range)
|
|
out = widgets.Output()
|
|
|
|
def update(change: Any) -> None:
|
|
value = change["new"]
|
|
with out:
|
|
out.clear_output(wait=True)
|
|
result = backtest(c, strategy=strategy, **{param_name: value})
|
|
fig, axes = plt.subplots(2, 1, figsize=(12, 6), sharex=True)
|
|
axes[0].plot(c, label="Close", alpha=0.7)
|
|
axes[0].set_title(f"{title} — {param_name}={value}")
|
|
axes[0].legend()
|
|
axes[1].plot(result.equity, label="Equity", color="green")
|
|
axes[1].axhline(1.0, color="gray", linestyle="--", alpha=0.5)
|
|
axes[1].set_title(
|
|
f"Equity (trades={result.n_trades}, final={result.final_equity:.3f})"
|
|
)
|
|
axes[1].legend()
|
|
plt.tight_layout()
|
|
plt.show()
|
|
|
|
slider = widgets.IntSlider(
|
|
value=param_values[len(param_values) // 2],
|
|
min=min(param_values),
|
|
max=max(param_values),
|
|
step=1,
|
|
description=param_name,
|
|
continuous_update=False,
|
|
)
|
|
slider.observe(update, names="value")
|
|
update({"new": slider.value})
|
|
return widgets.VBox([slider, out])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Streamlit app template
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def streamlit_app() -> None:
|
|
"""Run a minimal Streamlit TA dashboard.
|
|
|
|
Call this function from a Python script and run with::
|
|
|
|
streamlit run your_script.py
|
|
|
|
The dashboard provides:
|
|
- A file uploader for OHLCV CSV data (or uses synthetic data as fallback).
|
|
- An indicator selector (SMA, EMA, RSI, MACD, Bollinger Bands).
|
|
- A parameter slider.
|
|
- A price + indicator chart.
|
|
- A backtest panel (RSI strategy) with equity curve.
|
|
|
|
Requires
|
|
--------
|
|
``streamlit``, ``matplotlib`` or ``plotly`` (optional)
|
|
|
|
Examples
|
|
--------
|
|
Create a file ``ta_dashboard.py``::
|
|
|
|
from ferro_ta.tools.dashboard import streamlit_app
|
|
streamlit_app()
|
|
|
|
Then run::
|
|
|
|
streamlit run ta_dashboard.py
|
|
"""
|
|
try:
|
|
import streamlit as st
|
|
except ImportError as exc:
|
|
raise ImportError(
|
|
"streamlit_app requires streamlit.\nInstall with: pip install streamlit"
|
|
) from exc
|
|
|
|
import ferro_ta as ft
|
|
from ferro_ta.analysis.backtest import backtest
|
|
|
|
st.title("ferro-ta Interactive Dashboard")
|
|
|
|
# ---- Data ----
|
|
st.sidebar.header("Data")
|
|
uploaded = st.sidebar.file_uploader("Upload OHLCV CSV", type=["csv"])
|
|
|
|
if uploaded is not None:
|
|
try:
|
|
import pandas as pd
|
|
|
|
df = pd.read_csv(uploaded)
|
|
cols = {c.lower(): c for c in df.columns}
|
|
close = df[cols["close"]].values.astype(np.float64)
|
|
except (ImportError, KeyError, ValueError) as e:
|
|
st.error(f"Could not read CSV: {e}")
|
|
close = _synthetic_close()
|
|
else:
|
|
st.info(
|
|
"Using synthetic data. Upload a CSV with a 'close' column to use real data."
|
|
)
|
|
close = _synthetic_close()
|
|
|
|
n = len(close)
|
|
st.sidebar.write(f"Bars loaded: {n}")
|
|
|
|
# ---- Indicator ----
|
|
st.sidebar.header("Indicator")
|
|
indicator_name = st.sidebar.selectbox(
|
|
"Indicator", ["SMA", "EMA", "RSI", "MACD", "BBANDS"]
|
|
)
|
|
timeperiod = st.sidebar.slider("Period", min_value=2, max_value=200, value=20)
|
|
|
|
st.subheader(f"Price + {indicator_name}({timeperiod})")
|
|
|
|
try:
|
|
import matplotlib.pyplot as plt
|
|
|
|
fig, ax = plt.subplots(figsize=(12, 4))
|
|
ax.plot(close, label="Close", alpha=0.5)
|
|
|
|
if indicator_name == "SMA":
|
|
ax.plot(np.asarray(ft.SMA(close, timeperiod=timeperiod)), label="SMA")
|
|
elif indicator_name == "EMA":
|
|
ax.plot(np.asarray(ft.EMA(close, timeperiod=timeperiod)), label="EMA")
|
|
elif indicator_name == "RSI":
|
|
fig2, ax2 = plt.subplots(figsize=(12, 2))
|
|
ax2.plot(
|
|
np.asarray(ft.RSI(close, timeperiod=timeperiod)),
|
|
label="RSI",
|
|
color="orange",
|
|
)
|
|
ax2.axhline(30, color="green", linestyle="--", alpha=0.5)
|
|
ax2.axhline(70, color="red", linestyle="--", alpha=0.5)
|
|
ax2.set_title("RSI")
|
|
st.pyplot(fig2)
|
|
elif indicator_name == "MACD":
|
|
macd, signal, hist = ft.MACD(close)
|
|
ax.plot(np.asarray(macd), label="MACD")
|
|
ax.plot(np.asarray(signal), label="Signal")
|
|
elif indicator_name == "BBANDS":
|
|
upper, middle, lower = ft.BBANDS(close, timeperiod=timeperiod)
|
|
ax.plot(np.asarray(upper), label="Upper", linestyle="--")
|
|
ax.plot(np.asarray(middle), label="Middle")
|
|
ax.plot(np.asarray(lower), label="Lower", linestyle="--")
|
|
|
|
ax.legend()
|
|
st.pyplot(fig)
|
|
except (ImportError, ValueError, RuntimeError) as e:
|
|
st.error(f"Error computing indicator: {e}")
|
|
|
|
# ---- Backtest panel ----
|
|
st.subheader("Backtest (RSI 30/70 strategy)")
|
|
if st.button("Run Backtest"):
|
|
result = backtest(close, strategy="rsi_30_70", timeperiod=timeperiod)
|
|
try:
|
|
import matplotlib.pyplot as plt
|
|
|
|
fig3, ax3 = plt.subplots(figsize=(12, 3))
|
|
ax3.plot(result.equity, color="green", label="Equity")
|
|
ax3.axhline(1.0, color="gray", linestyle="--")
|
|
ax3.set_title(
|
|
f"Equity trades={result.n_trades} final={result.final_equity:.4f}"
|
|
)
|
|
ax3.legend()
|
|
st.pyplot(fig3)
|
|
except ImportError:
|
|
st.write(
|
|
f"Final equity: {result.final_equity:.4f} trades: {result.n_trades}"
|
|
)
|
|
|
|
|
|
def _synthetic_close(n: int = 500) -> NDArray:
|
|
"""Generate a synthetic close price series for the dashboard demo."""
|
|
rng = np.random.default_rng(42)
|
|
return np.cumprod(1 + rng.normal(0, 0.01, n)) * 100.0
|