webapp.py

"""Gradio web UI for the Gemma-powered stock analyser.

Gradio is imported lazily inside :func:`build_ui` so the request-handler
functions can be unit-tested on machines without Gradio installed.

Run on Google Colab:
    !python webapp.py
    # → produces a public ``*.gradio.live`` link valid for 72 h.

Run locally (no GPU required if the model is already cached, but Gemma
inference will fall back to CPU and be very slow):
    python webapp.py
"""
from __future__ import annotations

import pandas as pd

from src.backtest import backtest
from src.display import (
    render_backtest_equity_chart,
    render_candlestick,
)
from src.pipeline import analyze


# ─────────────────────────────────────────────────── analyze tab ─────
def _run_analyze(symbol: str, horizon: str, use_debate: bool):
    if not symbol:
        return ("⚠️  Please enter a symbol.", "", "", None, None)
    try:
        result = analyze(symbol.strip().upper(), horizon=horizon, use_debate=use_debate)
    except Exception as err:
        return (f"❌ {type(err).__name__}: {err}", "", "", None, None)

    a, pf = result.advice, result.price_features
    stance = a.get("stance", "?")
    conf = float(a.get("confidence", 0.0) or 0.0)
    badge = {"BUY": "🟢", "SELL": "🔴", "HOLD": "🟡"}.get(stance, "⚪")

    def _fmt(value, fmt=".2f"):
        if value is None or (isinstance(value, float) and pd.isna(value)):
            return "n/a"
        return format(value, fmt)

    header = (
        f"# {badge} {result.symbol} — **{stance}**  "
        f"_(confidence {conf:.0%})_\n\n"
        f"Price ${_fmt(pf['close'])} · 1d {_fmt(pf['pct_1d'], '+.2f')}% · 5d "
        f"{_fmt(pf['pct_5d'], '+.2f')}% · RSI14 {_fmt(pf['rsi14'], '.1f')}\n\n"
        f"**Aggregate sentiment:** {result.aggregate_polarity:+.3f} "
        f"(deterministic, time-decay + impact weighted)\n\n"
        f"### Reasoning\n{a.get('reasoning', '(none)')}\n\n"
        f"**Key risks:** "
        + ", ".join(a.get("key_risks", []) or ["(none)"])
        + "\n\n**Catalysts:** "
        + ", ".join(a.get("catalysts", []) or ["(none)"])
    )

    bull = a.get("bull_case", "(synthesis mode — debate disabled)")
    bear = a.get("bear_case", "(synthesis mode — debate disabled)")

    news_rows = [
        [
            n.get("title", "")[:80],
            f"{float(n.get('polarity', 0.0) or 0.0):+.2f}",
            n.get("impact", "?"),
            n.get("horizon", "?"),
            (n.get("rationale", "") or "")[:120],
        ]
        for n in result.news_scores
    ]
    news_df = pd.DataFrame(
        news_rows, columns=["Title", "Polarity", "Impact", "Horizon", "Rationale"]
    )

    chart = render_candlestick(result.stock_df, result.symbol)
    return header, bull, bear, news_df, chart


# ─────────────────────────────────────────────────── backtest tab ────
def _run_backtest(symbol: str, days: int):
    if not symbol:
        return ("⚠️  Please enter a symbol.", None, None)
    try:
        bt = backtest(symbol.strip().upper(), backtest_days=int(days))
    except Exception as err:
        return (f"❌ {type(err).__name__}: {err}", None, None)

    m = bt.metrics
    summary = (
        f"## 📊 Backtest — {bt.symbol}\n"
        f"| Metric | Value |\n|---|---|\n"
        f"| Days replayed | {int(m['n_days'])} |\n"
        f"| Days traded | {int(m['n_traded'])} |\n"
        f"| Hit rate | {m['hit_rate']:.1%} |\n"
        f"| Avg daily return | {m['avg_return_per_day']*100:+.2f}% |\n"
        f"| Cumulative return | {m['cumulative_return']*100:+.2f}% |\n"
        f"| Annualised Sharpe | {m['annualized_sharpe']:+.2f} |\n"
        f"| Max drawdown | {m['max_drawdown']*100:+.2f}% |\n"
    )

    rows = [
        [
            d.date.strftime("%Y-%m-%d"),
            d.n_news,
            f"{d.aggregate_polarity:+.2f}",
            d.predicted_stance,
            f"{d.next_day_return_pct:+.2f}%",
            "✅" if d.hit else "❌",
        ]
        for d in bt.days
    ]
    days_df = pd.DataFrame(
        rows,
        columns=["Date", "#News", "Polarity", "Predicted", "Next-day %", "Hit"],
    )
    chart = render_backtest_equity_chart(bt)
    return summary, chart, days_df


# ─────────────────────────────────────────────────── UI layout ───────
def build_ui():
    """Build the Gradio Blocks UI. Gradio is imported lazily here so unit
    tests can import the helper functions without Gradio installed.
    """
    import gradio as gr  # lazy

    with gr.Blocks(
        title="Stock Analysis · Gemma 4",
        theme=gr.themes.Soft(primary_hue="indigo"),
    ) as ui:
        gr.Markdown(
            "# 📊 Stock Analysis — Gemma 4 (4-bit, multi-agent)\n"
            "Bull vs Bear debate + per-news sentiment + Plotly charts + backtesting."
        )

        # ─── Analyze tab ───
        with gr.Tab("🔍 Analyze"):
            with gr.Row():
                symbol_in = gr.Textbox(label="Symbol", value="AAPL", scale=2)
                horizon_in = gr.Dropdown(
                    ["1d", "1w", "1m"], value="1w", label="Horizon", scale=1
                )
                debate_in = gr.Checkbox(
                    value=True, label="Bull vs Bear debate", scale=1
                )
                run_btn = gr.Button("Analyze", variant="primary", scale=1)

            header_out = gr.Markdown()
            with gr.Row():
                bull_out = gr.Textbox(label="🐂 Bull case", lines=12, max_lines=20)
                bear_out = gr.Textbox(label="🐻 Bear case", lines=12, max_lines=20)
            chart_out = gr.Plot(label="Price chart")
            news_out = gr.Dataframe(
                label="Per-news sentiment", wrap=True, interactive=False
            )

            run_btn.click(
                _run_analyze,
                inputs=[symbol_in, horizon_in, debate_in],
                outputs=[header_out, bull_out, bear_out, news_out, chart_out],
            )

        # ─── Backtest tab ───
        with gr.Tab("⏮️  Backtest"):
            with gr.Row():
                bt_symbol = gr.Textbox(label="Symbol", value="AAPL", scale=2)
                bt_days = gr.Slider(7, 22, value=21, step=1, label="Days", scale=1)
                bt_btn = gr.Button("Run backtest", variant="primary", scale=1)

            bt_summary = gr.Markdown()
            bt_chart = gr.Plot(label="Equity curve + per-day return")
            bt_table = gr.Dataframe(
                label="Per-day decisions", wrap=True, interactive=False
            )

            bt_btn.click(
                _run_backtest,
                inputs=[bt_symbol, bt_days],
                outputs=[bt_summary, bt_chart, bt_table],
            )

        gr.Markdown(
            "<sub>Models: Gemma 4 (4-bit NF4) · Data: yfinance + News API · "
            "Backtest capped at 22 trading days (NewsAPI free-tier limit).</sub>"
        )

    return ui


if __name__ == "__main__":
    build_ui().queue(max_size=8).launch(share=True, show_error=True)