src/ui/chart.py

"""The forecast chart, as the design's own SVG.

Every number in here comes from the design file rather than from taste: the
1200x352 viewBox, the 14/252 plot band, the volume strip that starts at y=266
and draws upward from 322, the 1.4px wicks. Reproducing them exactly is what
makes the port faithful; rounding them to something tidier is what makes a port
look approximately right in a way nobody can name.

The chart is rendered server-side as markup rather than drawn by a plotting
library, because the design specifies the shapes directly and a library would
fight it the whole way.
"""

from __future__ import annotations

import numpy as np
import pandas as pd

from ..adapters import OHLCV_COLUMNS
from bit_ui.markup import DASH, a, e  # noqa: F401

# --- the design's geometry, verbatim --------------------------------------
VIEW_W, VIEW_H = 1200, 352
PLOT_TOP, PLOT_BOTTOM = 14.0, 252.0
VOL_BASELINE = 266.0     # the rule under the price plot
VOL_FLOOR = 322.0        # volume bars grow upward from here
VOL_SCALE = 46.0
NOW_TOP, NOW_BOTTOM = 8.0, 330.0
GRID_LINES = 4           # produces 5 labels, hi..lo
HISTORY_BARS = 56        # how much history the design shows

# Series colours, in selection order. The design's own palette.
SERIES_COLORS = (
    "var(--accent-amber-strong)",
    "var(--mute-teal)",
    "var(--mute-violet)",
)
# Matchup fan styles cycle solid / hatch / open, as the design does.
FAN_DASH = ("none", "6 4", "2 2 6 2")


def series_color(index: int) -> str:
    return SERIES_COLORS[index % len(SERIES_COLORS)]


class Scale:
    """Maps bar index and price onto the design's coordinate system."""

    def __init__(self, n_history: int, horizon: int, lo: float, hi: float):
        self.n_history = n_history
        self.horizon = horizon
        self.slots = max(1, n_history + horizon)
        self.slot_w = VIEW_W / self.slots
        self.bar_w = max(3.0, self.slot_w * 0.6)
        pad = (hi - lo) * 0.06 or max(abs(hi), 1.0) * 0.01
        self.lo, self.hi = lo - pad, hi + pad

    def x(self, i: float) -> float:
        return self.slot_w * (i + 0.5)

    def y(self, v: float) -> float:
        span = self.hi - self.lo
        if span <= 0:
            return PLOT_BOTTOM
        return PLOT_BOTTOM - ((v - self.lo) / span) * (PLOT_BOTTOM - PLOT_TOP)


def _fmt_price(v: float, decimals: int) -> str:
    return f"{v:,.{decimals}f}"


def decimals_for(asset: str) -> int:
    """BTC prints whole dollars in the design; everything else two places."""
    return 0 if asset.startswith("BTC") else 2


def build(history: pd.DataFrame, runs: list, asset: str, horizon: int,
          show_ghosts: bool = True, ghost_limit: int = 7) -> dict:
    """Render the chart. Returns the SVG plus the overlay values around it.

    `runs` is a list of `ForecastRun`. Ghost paths are drawn only for runs
    whose *declared capability* is `ohlcv_paths` -- never because a result
    happens to carry a `paths` array. That is the capability gate, and it lives
    here so no renderer downstream has to know a model's name.
    """
    hist = history.tail(HISTORY_BARS).reset_index(drop=True)
    n = len(hist)
    if n == 0:
        return {"svg": "", "grid": [], "time_labels": [], "now_left": "0%",
                "now_time": "", "forecast_width": "0"}

    lo = float(hist["low"].min())
    hi = float(hist["high"].max())
    for run in runs:
        q = run.result.quantiles
        lo = min(lo, float(q.min()))
        hi = max(hi, float(q.max()))
    scale = Scale(n, horizon, lo, hi)

    parts: list[str] = [
        f'<svg viewBox="0 0 {VIEW_W} {VIEW_H}" preserveAspectRatio="none" '
        f'role="img" aria-label="Price history and forecast for {a(asset)}">',
        '<defs><pattern id="faHatch" width="7" height="7" '
        'patternUnits="userSpaceOnUse" patternTransform="rotate(45)">'
        '<line x1="0" y1="0" x2="0" y2="7" stroke="var(--text-tertiary)" '
        'stroke-width="2.4" opacity="0.5"></line></pattern></defs>',
    ]

    # -- grid ------------------------------------------------------------
    grid = []
    dec = decimals_for(asset)
    for i in range(GRID_LINES + 1):
        v = scale.hi - (scale.hi - scale.lo) * (i / GRID_LINES)
        y = scale.y(v)
        parts.append(f'<line x1="0" y1="{y:.1f}" x2="{VIEW_W}" y2="{y:.1f}" '
                     f'stroke="var(--border-subtle)" stroke-width="1" opacity="0.55"></line>')
        grid.append({"top": f"{y / VIEW_H * 100:.2f}%", "label": _fmt_price(v, dec)})

    # -- the forecast region ---------------------------------------------
    now_x = scale.x(n - 0.5)
    parts.append(f'<rect x="{now_x:.1f}" y="8" width="{VIEW_W - now_x:.1f}" '
                 f'height="252" fill="var(--bg-canvas)" opacity="0.5"></rect>')

    # -- ghost paths, behind the fans ------------------------------------
    if show_ghosts:
        for idx, run in enumerate(runs):
            if run.capabilities.get("output") != "ohlcv_paths":
                continue          # the capability gate
            paths = run.result.paths
            if paths is None:
                continue
            color = series_color(idx)
            parts.append(_ghost_cells(paths, scale, n, horizon, color, ghost_limit))

    # -- fans -------------------------------------------------------------
    for idx, run in enumerate(runs):
        parts.append(_fan(run, scale, n, horizon, idx, len(runs)))

    # -- candles ----------------------------------------------------------
    parts.append(_candles(hist, scale))

    # -- the now line and the volume strip --------------------------------
    parts.append(f'<line x1="{now_x:.1f}" y1="{NOW_TOP}" x2="{now_x:.1f}" '
                 f'y2="{NOW_BOTTOM}" stroke="var(--accent-amber)" '
                 f'stroke-width="1.4" stroke-dasharray="5 4"></line>')
    parts.append(f'<line x1="0" y1="{VOL_BASELINE}" x2="{VIEW_W}" y2="{VOL_BASELINE}" '
                 f'stroke="var(--border-subtle)" stroke-width="1"></line>')
    parts.append(_volumes(hist, runs, scale, n, horizon))

    parts.append("</svg>")

    ts = pd.to_datetime(hist["ts"], utc=True)
    return {
        "svg": "".join(parts),
        "grid": grid,
        "time_labels": _time_labels(ts, scale, n, horizon),
        "now_left": f"{now_x / VIEW_W * 100:.2f}%",
        "now_time": f"· {ts.iloc[-1].strftime('%H:%M')} UTC",
        "forecast_width": f"{VIEW_W - now_x:.1f}",
    }


# --------------------------------------------------------------------------
# Pieces
# --------------------------------------------------------------------------


def _candles(hist: pd.DataFrame, scale: Scale) -> str:
    out = []
    bw = scale.bar_w
    o = hist["open"].to_numpy(dtype="float64")
    h = hist["high"].to_numpy(dtype="float64")
    lw = hist["low"].to_numpy(dtype="float64")
    c = hist["close"].to_numpy(dtype="float64")
    for i in range(len(hist)):
        up = c[i] >= o[i]
        color = "var(--fin-up)" if up else "var(--fin-down)"
        x = scale.x(i)
        top, bot = scale.y(max(o[i], c[i])), scale.y(min(o[i], c[i]))
        wick_top, wick_bot = scale.y(h[i]), scale.y(lw[i])
        out.append(
            f'<rect x="{x - 0.7:.1f}" y="{wick_top:.1f}" width="1.4" '
            f'height="{max(1.0, wick_bot - wick_top):.1f}" fill="{color}"></rect>'
            f'<rect x="{x - bw / 2:.1f}" y="{top:.1f}" width="{bw:.1f}" '
            f'height="{max(1.4, bot - top):.1f}" fill="{color}"></rect>')
    return "".join(out)


def _fan(run, scale: Scale, n: int, horizon: int, idx: int, total: int) -> str:
    """The q10/q50/q90 envelope, starting from the last real close."""
    q = run.result
    lo_i, hi_i = q.level_index(0.1), q.level_index(0.9)
    mid_i = q.level_index(0.5)
    color = series_color(idx)

    steps = min(horizon, q.horizon)
    upper, lower, mid = [], [], []
    for t in range(steps):
        x = scale.x(n + t)
        upper.append(f"{x:.1f} {scale.y(q.quantiles[t, hi_i]):.1f}")
        lower.append(f"{x:.1f} {scale.y(q.quantiles[t, lo_i]):.1f}")
        mid.append(f"{x:.1f} {scale.y(q.quantiles[t, mid_i]):.1f}")
    if not mid:
        return ""

    last_close = float(run.context["close"].iloc[-1])
    start = f"{scale.x(n - 1):.1f} {scale.y(last_close):.1f}"

    matchup = total > 1
    style = idx % 3 if matchup else 0
    dash = FAN_DASH[style] if matchup else "none"
    fill = "url(#faHatch)" if style == 1 and matchup else color
    fill_op = "0.05" if style == 2 and matchup else ("0.9" if style == 1 and matchup
                                                     else ("0.14" if matchup else "0.18"))

    band = "M" + start + " L" + " L".join(upper) + " L" + " L".join(reversed(lower)) + " Z"
    return (
        f'<path d="{band}" fill="{fill}" opacity="{fill_op}"></path>'
        f'<path d="M{start} L{" L".join(upper)}" fill="none" stroke="{color}" '
        f'stroke-width="1.2" stroke-dasharray="2 3" opacity="0.85"></path>'
        f'<path d="M{start} L{" L".join(lower)}" fill="none" stroke="{color}" '
        f'stroke-width="1.2" stroke-dasharray="2 3" opacity="0.85"></path>'
        f'<path d="M{start} L{" L".join(mid)}" fill="none" stroke="{color}" '
        f'stroke-width="2" stroke-dasharray="{dash}"></path>')


def _ghost_cells(paths: np.ndarray, scale: Scale, n: int, horizon: int,
                 color: str, limit: int) -> str:
    """Sampled OHLCV paths, drawn as faint candle bodies.

    Only `limit` of them are drawn. Sending every sampled path would multiply
    the payload by the sample count for no readable gain -- past a handful the
    ghosts stop being individually legible and become a smear.
    """
    o_i, c_i = OHLCV_COLUMNS.index("open"), OHLCV_COLUMNS.index("close")
    out = []
    bw = scale.bar_w
    take = min(limit, paths.shape[0])
    # Evenly spaced through the samples rather than the first N, so the ghosts
    # represent the spread instead of whichever paths happened to be drawn first.
    picks = np.linspace(0, paths.shape[0] - 1, take).round().astype(int)
    steps = min(horizon, paths.shape[1])
    for s in picks:
        for t in range(steps):
            o = float(paths[s, t, o_i])
            c = float(paths[s, t, c_i])
            top, bot = scale.y(max(o, c)), scale.y(min(o, c))
            out.append(
                f'<rect x="{scale.x(n + t) - bw * 0.35:.1f}" y="{top:.1f}" '
                f'width="{bw * 0.7:.1f}" height="{max(1.2, bot - top):.1f}" '
                f'fill="{color}" opacity="0.17"></rect>')
    return "".join(out)


def _volumes(hist: pd.DataFrame, runs: list, scale: Scale, n: int,
             horizon: int) -> str:
    vol = hist["volume"].to_numpy(dtype="float64")
    peak = float(vol.max()) or 1.0
    o = hist["open"].to_numpy(dtype="float64")
    c = hist["close"].to_numpy(dtype="float64")
    bw = scale.bar_w
    out = []
    for i in range(n):
        h = (vol[i] / peak) * VOL_SCALE
        color = "var(--fin-up)" if c[i] >= o[i] else "var(--fin-down)"
        out.append(f'<rect x="{scale.x(i) - bw / 2:.1f}" y="{VOL_FLOOR - h:.1f}" '
                   f'width="{bw:.1f}" height="{h:.1f}" fill="{color}" opacity="0.55"></rect>')

    # Forecast-side volume, only from a model that actually forecasts volume.
    for idx, run in enumerate(runs[:1]):
        if run.capabilities.get("output") != "ohlcv_paths" or run.result.paths is None:
            continue
        v_i = OHLCV_COLUMNS.index("volume")
        mean_vol = run.result.paths[:, :, v_i].mean(axis=0)
        color = series_color(idx)
        for t in range(min(horizon, len(mean_vol))):
            h = (float(mean_vol[t]) / peak) * VOL_SCALE
            h = max(0.0, min(h, VOL_SCALE * 1.5))
            out.append(f'<rect x="{scale.x(n + t) - bw / 2:.1f}" '
                       f'y="{VOL_FLOOR - h:.1f}" width="{bw:.1f}" height="{h:.1f}" '
                       f'fill="{color}" opacity="0.22"></rect>')
    return "".join(out)


def _time_labels(ts: pd.Series, scale: Scale, n: int, horizon: int) -> list[dict]:
    if len(ts) < 2:
        return []
    step = ts.diff().dropna().mode()
    step = step.iloc[0] if len(step) else pd.Timedelta("1h")
    last = ts.iloc[-1]
    out = []
    for i in range(6):
        idx = round((scale.slots - 1) * i / 5)
        when = last + step * (idx - (n - 1))
        x = scale.x(idx)
        out.append({
            "left": f"{min(97.0, max(3.0, x / VIEW_W * 100)):.2f}%",
            "label": when.strftime("%d %H:%M"),
        })
    return out