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()))
    # The realised path has to fit on the chart, or a forecast that missed
    # badly would render as a fan with nothing beside it -- which reads as a
    # good forecast rather than a bad one.
    realized = _realized_of(runs)
    if realized is not None and len(realized):
        lo = min(lo, float(realized["low"].min()))
        hi = max(hi, float(realized["high"].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))

    # -- what actually happened, drawn through the fan ---------------------
    if realized is not None and len(realized):
        parts.append(_realized(realized, runs[0], scale, n))

    # -- 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)
    cached = bool(runs and getattr(runs[0], "from_cache", False))
    return {
        "svg": "".join(parts),
        "grid": grid,
        "time_labels": _time_labels(ts, scale, n, horizon),
        "now_left": f"{now_x / VIEW_W * 100:.2f}%",
        # The line marks where the forecast begins. On a live forecast that is
        # now; on an archived one it is when it was issued, and calling it
        # "NOW" there would be a plain lie about a three-day-old chart.
        "boundary_label": "ISSUED" if cached else "NOW",
        "now_time": f"· {ts.iloc[-1].strftime('%H:%M')} UTC",
        "forecast_width": f"{VIEW_W - now_x:.1f}",
    }


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


def _realized_of(runs: list):
    """The realised bars of the primary run, if it has any."""
    if not runs:
        return None
    return getattr(runs[0], "realized", None)


def _realized(realized: pd.DataFrame, run, scale: Scale, n: int) -> str:
    """Candles that printed after the forecast was issued.

    Positioned by matching each bar's timestamp to the forecast step it was
    predicting, rather than by counting forward: a market that closed for a
    weekend mid-horizon would otherwise slide every later bar off its step.

    Drawn narrower than the history candles and outlined, so the frozen fan
    stays readable underneath rather than being buried by the answer.
    """
    targets = {pd.Timestamp(t): i for i, t in enumerate(run.target_ts)}
    ts = pd.to_datetime(realized["ts"], utc=True)
    o = realized["open"].to_numpy(dtype="float64")
    h = realized["high"].to_numpy(dtype="float64")
    lw = realized["low"].to_numpy(dtype="float64")
    c = realized["close"].to_numpy(dtype="float64")

    bw = scale.bar_w * 0.62
    out = []
    for i in range(len(realized)):
        step = targets.get(pd.Timestamp(ts.iloc[i]))
        if step is None:
            continue
        x = scale.x(n + step)
        up = c[i] >= o[i]
        color = "var(--fin-up)" if up else "var(--fin-down)"
        top, bot = scale.y(max(o[i], c[i])), scale.y(min(o[i], c[i]))
        out.append(
            f'<rect x="{x - 0.6:.1f}" y="{scale.y(h[i]):.1f}" width="1.2" '
            f'height="{max(1.0, scale.y(lw[i]) - scale.y(h[i])):.1f}" '
            f'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>')

    # The realised close, as a line, so the path is legible even where the
    # candles are thin.
    points = []
    for i in range(len(realized)):
        step = targets.get(pd.Timestamp(ts.iloc[i]))
        if step is not None:
            points.append(f"{scale.x(n + step):.1f} {scale.y(c[i]):.1f}")
    if len(points) > 1:
        start = f"{scale.x(n - 1):.1f} {scale.y(float(run.context['close'].iloc[-1])):.1f}"
        out.append(f'<path d="M{start} L{" L".join(points)}" fill="none" '
                   f'stroke="var(--text-primary)" stroke-width="1.6"></path>')
    return "".join(out)


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