Programming

1. Charts

"""
Figures for the web app.

Deliberately thin: it reuses the exact figure functions from
`lending_analytics.report`. If a chart looked one way in the HTML deliverable
and another way on the dashboard, someone would eventually notice and stop
trusting both. One definition, two renderers.
"""
from __future__ import annotations

import numpy as np
import pandas as pd
import plotly.io as pio

from lending_analytics import config as C
from lending_analytics import report as R
from lending_analytics.i18n import period_label, t

from . import bridge

BUTTONS_TO_KILL = [
    "select2d", "lasso2d", "autoScale2d", "hoverClosestCartesian",
    "hoverCompareCartesian", "toggleSpikelines", "zoomIn2d", "zoomOut2d",
]

PLOTLY_CONFIG = {
    "displaylogo": False,
    "responsive": True,
    "modeBarButtonsToRemove": BUTTONS_TO_KILL,
    "toImageButtonOptions": {"format": "png", "scale": 2, "filename": "portfolio"},
}

_TEMPLATE_READY = False


def ensure_template(spec) -> None:
    global _TEMPLATE_READY
    if not _TEMPLATE_READY:
        pio.templates["kb"] = R.make_template(spec.brand)
        pio.templates.default = "kb"
        _TEMPLATE_READY = True


def to_html(fig, *, include_plotlyjs: bool = False) -> str:
    """The embedding call — plotly.js is loaded once by the base template."""
    return fig.to_html(full_html=False, include_plotlyjs=include_plotlyjs,
                       config=PLOTLY_CONFIG, default_height="100%")


# --------------------------------------------------------------------------
# Registry: slug -> (label key, builder)
# --------------------------------------------------------------------------
def _b(fn, *marts, **kw):
    def build(m, spec):
        return fn(*[m[name] for name in marts], spec, **kw)
    return build


CHARTS: dict[str, tuple[str, callable]] = {
    "portfolio":     ("exposure",        _b(R.fig_portfolio, "portfolio_monthly")),
    "leading":       ("watch_ratio",     _b(R.fig_leading_indicator, "portfolio_monthly")),
    "origination":   ("amount",          _b(R.fig_origination, "origination_monthly")),
    "burden":        ("loans_per_client", _b(R.fig_burden, "client_burden")),
    "branch_rank":   ("filial",          _b(R.fig_branch_rank, "portfolio_monthly")),
    "segments":      ("passport",        _b(R.fig_segment_matrix, "portfolio_monthly")),
    "vintage_curve": ("sec_vintage",     _b(R.fig_vintage_curves, "vintages")),
    "vintage_heat":  ("sec_vintage",     _b(R.fig_vintage_heat, "vintage_marks")),
    "transitions":   ("sec_migration",   _b(R.fig_transition_matrix, "transitions")),
    "cure":          ("cured",           _b(R.fig_cure, "cure_rates")),
    "pd":            ("pd",              _b(R.fig_pd, "pd_observed")),
    "bridge":        ("sec_bridge",      _b(R.fig_bridge, "npl_bridge")),
    "bridge_trend":  ("sec_bridge",      _b(R.fig_bridge_trend, "npl_bridge")),
    "insurance":     ("sec_insurance",   _b(R.fig_insurance, "portfolio_monthly")),
}

# What each dashboard tab shows.
SECTIONS: dict[str, tuple[str, tuple[str, ...]]] = {
    "overview":    ("sec_overview",    ("portfolio", "leading")),
    "origination": ("sec_origination", ("origination", "burden")),
    "quality":     ("sec_quality",     ("branch_rank", "segments")),
    "vintage":     ("sec_vintage",     ("vintage_curve", "vintage_heat")),
    "migration":   ("sec_migration",   ("transitions", "cure", "pd")),
    "bridge":      ("sec_bridge",      ("bridge", "bridge_trend")),
    "insurance":   ("sec_insurance",   ("insurance",)),
}


def build_chart(slug: str, marts: dict, spec) -> str:
    ensure_template(spec)
    _, builder = CHARTS[slug]
    return to_html(builder(marts, spec))


def build_section(section: str, marts: dict, spec) -> list[dict]:
    _, slugs = SECTIONS[section]
    out = []
    for slug in slugs:
        try:
            out.append({"slug": slug, "title": t(CHARTS[slug][0], spec.lang),
                        "html": build_chart(slug, marts, spec), "error": None})
        except Exception as exc:  # a broken chart must not take the page down
            out.append({"slug": slug, "title": t(CHARTS[slug][0], spec.lang),
                        "html": "", "error": f"{type(exc).__name__}: {exc}"})
    return out


# --------------------------------------------------------------------------
# KPI strip
# --------------------------------------------------------------------------
def kpis(marts: dict, spec, period: int | None = None) -> list[dict]:
    pf = marts["portfolio_monthly"]
    d = pf[pf["dim_name"] == "__all__"].sort_values("period")
    if period:
        d = d[d["period"] <= period]
    if len(d) < 2:
        return []
    cur, prev = d.iloc[-1], d.iloc[-2]
    unit = t(f"unit_{spec.units}", spec.lang)
    div = C.MLRD if spec.units == "mlrd" else C.MLN

    def item(key, value, delta=None, good_down=True, suffix=""):
        cls = ""
        txt = ""
        if delta is not None and not pd.isna(delta):
            cls = "up" if (delta > 0) == good_down else "down"
            txt = f"{delta:+.2f} {'п.п.' if spec.lang == 'ru' else 'pp'}"
        return {"label": t(key, spec.lang) + suffix, "value": value,
                "delta": txt, "cls": cls}

    return [
        item("exposure", f"{cur['exposure'] / div:,.0f}".replace(",", " "),
             None, suffix=f", {unit}"),
        item("npl_ratio", f"{cur['npl_ratio']:.2%}",
             (cur["npl_ratio"] - prev["npl_ratio"]) * 100),
        item("watch_ratio", f"{cur['watch_ratio']:.2%}",
             (cur["watch_ratio"] - prev["watch_ratio"]) * 100),
        item("coverage_ratio", f"{cur['coverage_ratio']:.0%}",
             (cur["coverage_ratio"] - prev["coverage_ratio"]) * 100, good_down=False),
        item("n_clients", f"{int(cur['n_clients']):,}".replace(",", " ")),
        item("loans_per_client", f"{cur['loans_per_client']:.2f}"),
    ]


def branch_table(marts: dict, spec, period: int | None = None) -> list[dict]:
    """The table people screenshot and paste into chat. Worth getting right."""
    pf = marts["portfolio_monthly"]
    d = pf[pf["dim_name"] == "filial"]
    if d.empty:
        return []
    per = period or int(d["period"].max())
    periods = sorted(d["period"].unique())
    i = periods.index(per) if per in periods else len(periods) - 1
    prev = periods[max(i - 12, 0)]

    cur = d[d["period"] == per].set_index("dim_value")
    old = d[d["period"] == prev].set_index("dim_value")
    div = C.MLRD if spec.units == "mlrd" else C.MLN

    rows = []
    for name, r in cur.sort_values("npl_ratio", ascending=False).iterrows():
        delta = (r["npl_ratio"] - old["npl_ratio"].get(name, np.nan)) * 100
        rows.append({
            "name": name,
            "exposure": f"{r['exposure'] / div:,.1f}".replace(",", " "),
            "npl": f"{r['npl_ratio']:.2%}",
            "watch": f"{r['watch_ratio']:.2%}",
            "coverage": f"{r['coverage_ratio']:.0%}" if pd.notna(r["coverage_ratio"]) else "—",
            "n_loans": f"{int(r['n_loans']):,}".replace(",", " "),
            "delta": f"{delta:+.2f}" if pd.notna(delta) else "—",
            "delta_cls": "up" if pd.notna(delta) and delta > 0 else "down",
        })
    return rows


def narrative(marts: dict, spec) -> dict[str, list[str]]:
    try:
        return R.commentary(marts, spec)
    except Exception:
        return {}


def period_choices(spec) -> list[dict]:
    return [{"value": p, "label": period_label(p, spec.lang)}
            for p in reversed(bridge.available_periods())]
Helpful? Dislike 0 Log in to react