quant-web/app.py

"""QuantTrade Streamlit dashboard."""

from __future__ import annotations

import os

import pandas as pd
import plotly.graph_objects as go
import streamlit as st
from plotly.subplots import make_subplots

from auth import get_current_user, logout
from backtest import load_ohlcv, run_ma_crossover
from strategy_db import delete_strategy, init_db, list_strategies, load_strategy, save_strategy
from telemetry import capture_exception, init_telemetry

init_telemetry("quant-streamlit")
init_db()

st.set_page_config(
    page_title="QuantTrade",
    page_icon="📈",
    layout="wide",
)

DEFAULT_TICKERS = os.environ.get(
    "CORE_TICKERS",
    "SPY,QQQ,AAPL,MSFT,GOOGL,AMZN,NVDA,META,IWM,TLT",
).split(",")


def render_equity_chart(result) -> None:
    fig = make_subplots(
        rows=2,
        cols=1,
        shared_xaxes=True,
        vertical_spacing=0.08,
        row_heights=[0.65, 0.35],
        subplot_titles=(f"{result.ticker} Price", "Strategy Equity"),
    )

    fig.add_trace(
        go.Scatter(x=result.price.index, y=result.price.values, name="Close", line=dict(color="#60a5fa")),
        row=1,
        col=1,
    )
    fig.add_trace(
        go.Scatter(
            x=result.equity_curve.index,
            y=result.equity_curve.values,
            name="Equity",
            line=dict(color="#34d399"),
        ),
        row=2,
        col=1,
    )
    fig.update_layout(height=640, template="plotly_dark", margin=dict(l=20, r=20, t=40, b=20))
    st.plotly_chart(fig, use_container_width=True)


def main() -> None:
    user = get_current_user()
    if not user:
        return

    st.title("QuantTrade")
    st.caption("VectorBT backtests on local Parquet market data")

    with st.sidebar:
        st.subheader("Account")
        st.write(f"Signed in as **{user}**")
        if st.button("Logout", use_container_width=True):
            logout()
            st.rerun()

        st.divider()
        st.subheader("Strategy")
        ticker = st.selectbox(
            "Ticker",
            options=[t.strip().upper() for t in DEFAULT_TICKERS if t.strip()],
            index=0,
        )
        fast_window = st.slider("Fast MA", min_value=5, max_value=100, value=20, step=1)
        slow_window = st.slider("Slow MA", min_value=20, max_value=250, value=50, step=1)
        init_cash = st.number_input("Initial cash", min_value=1000.0, value=10_000.0, step=1000.0)
        fees = st.number_input("Fees (fraction)", min_value=0.0, max_value=0.05, value=0.001, step=0.0005)

        run_clicked = st.button("Run Backtest", type="primary", use_container_width=True)

        st.divider()
        st.subheader("Saved Strategies")
        saved = list_strategies(user)
        saved_names = [s.name for s in saved]
        selected_name = st.selectbox("Load strategy", options=["—"] + saved_names)

        strategy_name = st.text_input("Strategy name", placeholder="My SPY crossover")
        col_save, col_delete = st.columns(2)
        with col_save:
            save_clicked = st.button("Save Strategy", use_container_width=True)
        with col_delete:
            delete_clicked = st.button("Delete", use_container_width=True)

    params = {
        "fast_window": fast_window,
        "slow_window": slow_window,
        "init_cash": init_cash,
        "fees": fees,
    }

    if save_clicked:
        if not strategy_name.strip():
            st.sidebar.error("Enter a strategy name before saving.")
        else:
            save_strategy(user, strategy_name.strip(), ticker, params)
            st.sidebar.success(f"Saved '{strategy_name.strip()}'.")
            st.rerun()

    if delete_clicked and selected_name != "—":
        delete_strategy(user, selected_name)
        st.sidebar.success(f"Deleted '{selected_name}'.")
        st.rerun()

    active_ticker = ticker
    active_params = dict(params)

    if selected_name != "—":
        loaded = load_strategy(user, selected_name)
        if loaded:
            active_ticker = loaded.ticker
            active_params.update(loaded.params)
            st.info(f"Loaded strategy **{loaded.name}** ({loaded.ticker}). Adjust sliders or run.")

    if run_clicked or selected_name != "—":
        try:
            load_ohlcv(active_ticker)
            result = run_ma_crossover(
                ticker=active_ticker,
                fast_window=int(active_params["fast_window"]),
                slow_window=int(active_params["slow_window"]),
                init_cash=float(active_params.get("init_cash", init_cash)),
                fees=float(active_params.get("fees", fees)),
            )

            c1, c2, c3, c4 = st.columns(4)
            c1.metric("Sharpe Ratio", f"{result.sharpe_ratio:.2f}")
            c2.metric("Max Drawdown", f"{result.max_drawdown:.1%}")
            c3.metric("Total Return", f"{result.total_return:.1%}")
            c4.metric("Bars", f"{len(result.price):,}")

            render_equity_chart(result)

            with st.expander("Raw stats"):
                st.write(
                    pd.DataFrame(
                        {
                            "Metric": ["Ticker", "Fast MA", "Slow MA", "Sharpe", "Max DD", "Return"],
                            "Value": [
                                result.ticker,
                                result.fast_window,
                                result.slow_window,
                                result.sharpe_ratio,
                                result.max_drawdown,
                                result.total_return,
                            ],
                        }
                    )
                )
        except FileNotFoundError:
            st.warning(
                f"No Parquet data for **{active_ticker}** yet. "
                "Wait for the harvester seed job or check container logs."
            )
        except ValueError as exc:
            st.error(str(exc))
        except Exception as exc:
            capture_exception(exc)
            st.error("Backtest failed. The error was reported to Bugsink.")
            st.exception(exc)
    else:
        st.info("Configure parameters in the sidebar and click **Run Backtest**.")


if __name__ == "__main__":
    main()