Programming
6. process
"""
Analytics engine for the portfolio report.
IMPORTANT — the one rule that matters when porting notebook cells here:
NOTHING may execute at import time.
Django imports this module when it loads views.py, which happens on every
`runserver`, every autoreload, and every `manage.py` command — including
`migrate` and `collectstatic`. If a notebook cell that loads 8M rows sits at
module level, every one of those becomes a two-minute hang.
So: cell 1 and cell 2 become module constants (cheap). Cells 3-13 become
function bodies. The only entry point is run().
Everything else is copied from the notebook unchanged.
"""
from __future__ import annotations
import hashlib
import html as _html
import math
import textwrap
import datetime as dt
from pathlib import Path
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import plotly.io as pio
# ==========================================================================
# CONFIG (notebook cell 1 — constants only, no I/O)
# ==========================================================================
NPL_CATEGORIES = (3, 4, 5)
WATCH_DPD = (31, 90)
MIN_COHORT = 30
MIN_BALANCE = 0 # static section keeps only exposure > this
VINTAGE_MARKS = ((30, 3), (30, 6), (90, 9), (90, 12))
DPD_BUCKETS = (
(-10**9, 0, "CUR"), (1, 30, "B01_30"), (31, 60, "B31_60"), (61, 90, "B61_90"),
(91, 180, "B91_180"), (181, 360, "B181_360"), (361, 10**9, "B360P"),
)
BUCKET_ORDER = [b[2] for b in DPD_BUCKETS] + ["EXIT"]
DEFAULT_BUCKETS = ("B91_180", "B181_360", "B360P")
MOB_BINS = [-1, 3, 6, 12, 24, 36, 10**9]
MOB_LABELS = ["0-3 мес.", "4-6 мес.", "7-12 мес.", "13-24 мес.", "25-36 мес.", "36+ мес."]
CURRENCY_NAMES = {0: "Сум", 840: "USD", 978: "EUR"}
UNIT, UNIT_NAME = 1e9, "млрд сум"
TOP_PASSPORTS, TOP_OKED, MAX_LABELS = 7, 10, 14
TITLE, BANK = "Обзор кредитного портфеля", "Kapitalbank"
BRAND = {
"ink": "#0B0B0C", "ink_soft": "#2A2A2E", "gold": "#F2C230",
"gold_deep": "#C9971A", "paper": "#FFFFFF", "rule": "#E4E1D8",
"muted": "#8A867C", "bad": "#B3341F", "good": "#2F6B4F", "cool": "#4A6E8A",
}
MONTHS_RU = ["янв","фев","мар","апр","май","июн","июл","авг","сен","окт","ноя","дек"]
LABEL = {
"CUR": "Без просрочки", "B01_30": "1-30 дн.", "B31_60": "31-60 дн.",
"B61_90": "61-90 дн.", "B91_180": "91-180 дн.", "B181_360": "181-360 дн.",
"B360P": "360+ дн.", "EXIT": "Выбыл", "non_auto": "Не авто",
"pervichka": "Авто новые", "vtorichka": "Авто с пробегом",
"norm": "Без суда", "official": "В суде", "neofficial": "Досудебная",
}
BUCKET_COLORS = {
"CUR": "#2F6B4F", "B01_30": "#8FA85C", "B31_60": "#F2C230", "B61_90": "#D99A1E",
"B91_180": "#C2591F", "B181_360": "#A33418", "B360P": "#6E1F0E", "EXIT": "#8A867C",
}
SEQ = [BRAND["gold"], BRAND["cool"], BRAND["good"], BRAND["gold_deep"], BRAND["bad"],
"#6E8FA8", "#A88C4A", "#5C6E4A", "#8A867C", "#3E5C74", "#C2A25A", "#7A5C3E"]
CANONICAL = [
"contragent","anketa","filial","tobo","valuta","dogovor_sana","vidacha_sana",
"department","department_number","passport","contragent_name",
"brutto_95_summa","brutto_summa","protsent_16309_summa","protsent_16377_summa",
"protsent_16379_summa","spisat_95413_summa","spisat_91501_summa","reserve_summa",
"vidacha_tekushiy_summa","pogashen_tekushiy_summa","maks_dni","kategoriya_origin",
"kategoriya_npl_90","straxovka_summa","straxovka_brutto_95_summa",
"kategoriya_straxovka","sud","avto","report_date","all_time_given_amount",
]
OPTIONAL = ["oked_name"]
RENAME = {
"vidacha_tm_summa": "vidacha_tekushiy_summa",
"pogashen_tm_summa": "pogashen_tekushiy_summa",
"vidacha_alltime_summa": "all_time_given_amount",
}
DATE_COLS = ("dogovor_sana", "vidacha_sana", "report_date")
AMOUNT_COLS = [c for c in CANONICAL if c.endswith("_summa") or c == "all_time_given_amount"]
PCFG = {"displaylogo": False, "responsive": True,
"modeBarButtonsToRemove": ["select2d","lasso2d","autoScale2d",
"hoverClosestCartesian","hoverCompareCartesian",
"toggleSpikelines","zoomIn2d","zoomOut2d"],
"toImageButtonOptions": {"format": "png", "scale": 2}}
_TEMPLATE_READY = False
def _ensure_template():
"""Registering the plotly template is cheap but still deferred out of import."""
global _TEMPLATE_READY
if _TEMPLATE_READY:
return
pio.templates["kb"] = go.layout.Template(layout=go.Layout(
font=dict(family="Cambria, Georgia, serif", size=13, color=BRAND["ink"]),
paper_bgcolor="white", plot_bgcolor="white", colorway=SEQ,
margin=dict(l=86, r=96, t=104, b=104),
hoverlabel=dict(bgcolor=BRAND["ink"], font_color="white", font_size=12),
xaxis=dict(showgrid=False, linecolor=BRAND["rule"], ticks="outside",
tickcolor=BRAND["rule"], tickfont=dict(size=11), automargin=True),
yaxis=dict(gridcolor=BRAND["rule"], zerolinecolor=BRAND["rule"],
tickfont=dict(size=11), automargin=True),
legend=dict(orientation="h", yanchor="bottom", y=1.03, x=0,
font=dict(size=11.5), bgcolor="rgba(0,0,0,0)"),
title=dict(font=dict(size=17), x=0, xanchor="left", y=.965, yanchor="top"),
))
pio.templates.default = "kb"
_TEMPLATE_READY = True
# ==========================================================================
# HELPERS (notebook cell 2)
# ==========================================================================
def plabel(p):
y, m = divmod(int(p), 100)
return f"{MONTHS_RU[m-1]} {str(y)[2:]}"
def plabel_full(p):
y, m = divmod(int(p), 100)
return f"{MONTHS_RU[m-1]} {y}"
lab = lambda k: LABEL.get(str(k), str(k))
money = lambda v: np.asarray(v, dtype="float64") / UNIT
def fnum(v, dec=1):
try:
return f"{float(v):,.{dec}f}".replace(",", " ")
except (TypeError, ValueError):
return ""
def thin(values, fmt=lambda v: fnum(v, 1), max_labels=None, keep_last=True):
n = len(values)
step = max(1, math.ceil(n / (max_labels or MAX_LABELS)))
return [fmt(v) if (i % step == 0 or (i == n - 1 and keep_last)) else ""
for i, v in enumerate(values)]
def headroom(line_max, factor=2.6):
top = float(line_max) * factor if line_max and np.isfinite(line_max) else 1.0
return dict(range=[0, top])
def unit_note(fig, text=None, extra_bottom=0):
fig.add_annotation(text=text or f"единица измерения: {UNIT_NAME}",
xref="paper", yref="paper", x=1, y=-0.16 - extra_bottom,
xanchor="right", yanchor="top", showarrow=False,
font=dict(size=10.5, color=BRAND["muted"]))
return fig
def kind_tag(fig, kind, legend_items=None):
txt = {"СТОК": "СТОК · снимок на отчётную дату, по месяцам не суммируется",
"ПОТОК": "ПОТОК · движение внутри месяца, суммируется",
"КОГОРТЫ": "КОГОРТЫ · разрез по месяцу выдачи",
"СРЕЗ": "СРЕЗ · одна отчётная дата"}.get(kind, kind)
n = legend_items if legend_items is not None else len(
[t for t in fig.data if t.showlegend is not False and t.name])
rows = max(1, math.ceil(n / 4))
tag_y = 1.045
fig.add_annotation(text=txt, xref="paper", yref="paper", x=0, y=tag_y,
xanchor="left", yanchor="bottom", showarrow=False,
font=dict(size=10.5, color=BRAND["gold_deep"]))
fig.update_layout(legend=dict(y=tag_y + 0.035, yanchor="bottom", x=0),
margin=dict(t=104 + 26 * rows))
return fig
# ==========================================================================
# LOAD (notebook cell 3)
# ==========================================================================
def normalise(df):
df = df.rename(columns=lambda c: str(c).strip().lower().replace(" ", "_")).rename(columns=RENAME)
missing = [c for c in CANONICAL if c not in df.columns]
if missing:
raise KeyError("в данных не хватает колонок: " + ", ".join(missing))
have_opt = [c for c in OPTIONAL if c in df.columns]
df = df[CANONICAL + have_opt].copy()
for c in DATE_COLS:
df[c] = pd.to_datetime(df[c], errors="coerce")
for c in AMOUNT_COLS:
df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0.0)
df["maks_dni"] = pd.to_numeric(df["maks_dni"], errors="coerce").fillna(0).astype("int32")
for c in ("kategoriya_origin", "kategoriya_npl_90", "kategoriya_straxovka"):
df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0).astype("int8")
df["valuta"] = pd.to_numeric(df["valuta"], errors="coerce").fillna(0).astype("int32")
for c in ("anketa", "contragent"):
df[c] = pd.to_numeric(df[c], errors="coerce").astype("int64")
for c in ["filial","tobo","department","department_number","passport",
"contragent_name","sud","avto"] + have_opt:
df[c] = df[c].astype("string").str.strip()
df["avto"] = df["avto"].fillna("0").replace(
{"0": "non_auto", "0.0": "non_auto", "": "non_auto",
"nan": "non_auto", "None": "non_auto", "<NA>": "non_auto"})
df.loc[~df["avto"].isin(["pervichka", "vtorichka"]), "avto"] = "non_auto"
df["sud"] = df["sud"].fillna("norm")
if have_opt:
df["oked_name"] = df["oked_name"].fillna("Не указано").replace({"": "Не указано"})
return df
def read_months(data_dir, period_from=None, period_to=None):
"""
Read only the monthly files the request actually needs.
The filename is the report_date, so the window can be applied before any
file is opened. Asking for 12 months reads 12 files, not 44.
"""
files = sorted(Path(data_dir).glob("*.parquet"))
if not files:
raise FileNotFoundError(f"нет parquet-файлов в {data_dir}")
def file_period(p):
rd = pd.Timestamp(p.stem) - pd.Timedelta(days=1)
return rd.year * 100 + rd.month
chosen = []
for f in files:
try:
per = file_period(f)
except ValueError: # filename isn't a date — read it anyway
chosen.append(f)
continue
if period_from and per < int(period_from):
continue
if period_to and per > int(period_to):
continue
chosen.append(f)
if not chosen:
raise ValueError(f"нет данных за период {period_from}–{period_to}")
return normalise(pd.concat([pd.read_parquet(f) for f in chosen], ignore_index=True))
def data_fingerprint(data_dir):
"""
Identity of the input set: name, size and mtime of every parquet file.
Changes the moment a file is added, replaced or re-saved — which is exactly
when a cached report stops being valid.
"""
parts = []
for f in sorted(Path(str(data_dir)).glob("*.parquet")):
st = f.stat()
parts.append(f"{f.name}:{int(st.st_mtime)}:{st.st_size}")
return hashlib.sha256("|".join(parts).encode()).hexdigest()[:20]
def run_fingerprint(data_dir, period_from, period_to, selected_period):
"""Same parameters + same files = same report. No need to rebuild it."""
key = f"{data_fingerprint(data_dir)}|{period_from}|{period_to}|{selected_period}"
return hashlib.sha256(key.encode()).hexdigest()[:32]
def available_periods(data_dir):
"""Cheap scan for the dropdown — filenames only, no file is opened."""
out = []
for f in sorted(Path(data_dir).glob("*.parquet")):
try:
rd = pd.Timestamp(f.stem) - pd.Timedelta(days=1)
except ValueError:
continue
out.append(rd.year * 100 + rd.month)
return sorted(set(out))
# ==========================================================================
# PANEL (notebook cell 4)
# ==========================================================================
def bucket_of(dpd):
edges = np.array([b[0] for b in DPD_BUCKETS[1:]])
labels = [b[2] for b in DPD_BUCKETS]
idx = np.searchsorted(edges, np.asarray(dpd), side="left")
return pd.Categorical([labels[i] for i in idx], categories=BUCKET_ORDER, ordered=True)
def build_panel(raw):
d = raw.copy()
a = d["report_date"] - pd.Timedelta(days=1)
d["period"] = (a.dt.year * 100 + a.dt.month).astype("int32")
v = d["vidacha_sana"]
d["vintage"] = (v.dt.year * 100 + v.dt.month).astype("Int32")
d["mob"] = ((a.dt.year - v.dt.year) * 12 + (a.dt.month - v.dt.month)).astype("Int32")
d["exposure"] = d["brutto_95_summa"]
d["reserve"] = d["reserve_summa"]
d["disbursed"] = d["vidacha_tekushiy_summa"]
d["repaid"] = d["pogashen_tekushiy_summa"]
d["written_amt"] = d["spisat_95413_summa"]
d["insurance_paid"] = d["straxovka_summa"]
d["interest"] = (d["protsent_16309_summa"] + d["protsent_16377_summa"]
+ d["protsent_16379_summa"])
d["dpd"] = d["maks_dni"].clip(lower=0)
d["bucket"] = bucket_of(d["dpd"])
d["mob_band"] = pd.cut(d["mob"].astype("float"), MOB_BINS, labels=MOB_LABELS)
d["is_npl"] = d["kategoriya_npl_90"].isin(NPL_CATEGORIES)
d["is_watch"] = (~d["is_npl"]) & d["dpd"].between(*WATCH_DPD)
d["is_new"] = d["disbursed"] > 0
d["in_court"] = d["sud"].eq("official")
d["written_off"] = d["spisat_95413_summa"] > 0
d["is_npl_gross"] = d["kategoriya_straxovka"].isin(NPL_CATEGORIES)
d["insurance_saved"] = d["is_npl_gross"] & ~d["is_npl"]
d["exposure_gross"] = d["straxovka_brutto_95_summa"].where(
d["straxovka_brutto_95_summa"] > 0, d["exposure"])
d["npl_exposure"] = np.where(d["is_npl"], d["exposure"], 0.0)
d["watch_exposure"] = np.where(d["is_watch"], d["exposure"], 0.0)
d["npl_gross_exp"] = np.where(d["is_npl_gross"], d["exposure_gross"], 0.0)
d["cur_name"] = d["valuta"].map(CURRENCY_NAMES).fillna(d["valuta"].astype(str))
has_oked = "oked_name" in d.columns
keep = ["anketa","contragent","period","vintage","mob","mob_band","dpd","bucket",
"is_npl","is_watch","is_new","is_npl_gross","insurance_saved","in_court",
"written_off","exposure","exposure_gross","npl_exposure","watch_exposure",
"npl_gross_exp","reserve","disbursed","repaid","written_amt",
"insurance_paid","interest","filial","department","passport","avto",
"valuta","cur_name","sud"] + (["oked_name"] if has_oked else [])
p = d[keep].sort_values(["anketa", "period"], kind="stable").reset_index(drop=True)
g = p.groupby("anketa", sort=False)
p["mi"] = (p["period"] // 100) * 12 + (p["period"] % 100)
p["peak_dpd"] = g["dpd"].cummax()
prev_mi, next_mi = g["mi"].shift(1), g["mi"].shift(-1)
back, fwd = prev_mi.eq(p["mi"] - 1), next_mi.eq(p["mi"] + 1)
p["prev_is_npl"] = g["is_npl"].shift(1).where(back)
p["prev_exposure"] = g["exposure"].shift(1).where(back)
p["next_bucket"] = g["bucket"].shift(-1).astype("object").where(fwd, "EXIT")
p["exited_after"] = ~fwd
return p
# ==========================================================================
# MARTS (notebook cells 6-7)
# ==========================================================================
def by_period(p, dim=None):
keys = ["period"] + ([dim] if dim else [])
g = p.groupby(keys, observed=True)
m = pd.DataFrame({
# --- STOCK: measured on the reporting date, never summed across months
"exposure": g["exposure"].sum(), "npl_exposure": g["npl_exposure"].sum(),
"watch_exposure": g["watch_exposure"].sum(),
"npl_gross_exp": g["npl_gross_exp"].sum(),
"exposure_gross": g["exposure_gross"].sum(), "reserve": g["reserve"].sum(),
"n_loans": g["anketa"].size(), "n_clients": g["contragent"].nunique(),
"court_exp": g.apply(lambda x: x.loc[x["in_court"], "exposure"].sum(),
include_groups=False),
# --- FLOW: happened inside the month, summing across months is correct
"disbursed": g["disbursed"].sum(), "repaid": g["repaid"].sum(),
"written_amt": g["written_amt"].sum(), "n_disbursed": g["is_new"].sum(),
}).reset_index()
nz = lambda s: s.replace(0, np.nan)
m["npl_ratio"] = m["npl_exposure"] / nz(m["exposure"])
m["watch_ratio"] = m["watch_exposure"] / nz(m["exposure"])
m["npl_ratio_gross"] = m["npl_gross_exp"] / nz(m["exposure_gross"])
m["coverage"] = m["reserve"] / nz(m["npl_exposure"])
m["loans_per_client"] = m["n_loans"] / nz(m["n_clients"])
m["court_share"] = m["court_exp"] / nz(m["exposure"])
m["avg_ticket"] = m["exposure"] / nz(m["n_loans"])
if dim:
m = m.rename(columns={dim: "dim_value"})
m["dim_value"] = m["dim_value"].astype("string")
return m
def build_marts(panel, snap, periods, sel, has_oked):
# ---------------- Direction A: dynamics ----------------
A = {"total": by_period(panel).sort_values("period")}
for d in ["department", "filial", "passport", "avto", "cur_name", "sud"] \
+ (["oked_name"] if has_oked else []):
A[d] = by_period(panel, d)
# delinquency mix as within-month shares (they sum to 100%, so stacking is honest)
g = panel.groupby(["period", "bucket"], observed=True)["exposure"].sum().reset_index()
g["share"] = g["exposure"] / g.groupby("period", observed=True)["exposure"].transform("sum")
A["bucket_mix"] = g
# stock/flow reconciliation: how last month's photograph became this month's
t = A["total"].copy().reset_index(drop=True)
t["prev_exposure"] = t["exposure"].shift(1)
t["delta"] = t["exposure"] - t["prev_exposure"]
t["other"] = t["delta"] - t["disbursed"] + t["repaid"] + t["written_amt"]
A["stock_flow"] = t.dropna(subset=["prev_exposure"])
# NPL bridge: closing = opening + inflow - cured - exited + residual
stock = panel.groupby("period", observed=True)["npl_exposure"].sum()
q = panel[panel["prev_is_npl"].notna()]
inflow = q[q["is_npl"] & ~q["prev_is_npl"].astype(bool)] \
.groupby("period", observed=True)["exposure"].sum()
cured = q[~q["is_npl"] & q["prev_is_npl"].astype(bool)] \
.groupby("period", observed=True)["prev_exposure"].sum()
nxt = lambda x: ((x // 100 + 1) * 100 + 1) if x % 100 == 12 else x + 1
exited = (panel[panel["is_npl"] & panel["exited_after"]]
.assign(np_=lambda x: x["period"].map(nxt))
.groupby("np_", observed=True)["exposure"].sum())
rows = []
for i, per in enumerate(periods[1:], 1):
op, cl = float(stock.get(periods[i-1], 0)), float(stock.get(per, 0))
inf, cu, ex = (float(inflow.get(per, 0)), float(cured.get(per, 0)),
float(exited.get(per, 0)))
rows.append({"period": per, "opening": op, "inflow": inf, "cured": -cu,
"exited": -ex, "residual": cl - (op + inf - cu - ex), "closing": cl})
A["bridge"] = pd.DataFrame(rows)
# migration between delinquency buckets, month over month
tr = panel[panel["next_bucket"].notna()].copy()
tr["from_b"] = tr["bucket"].astype("string")
tr["to_b"] = tr["next_bucket"].astype("string")
gg = tr.groupby(["period", "from_b", "to_b"], observed=True)["anketa"] \
.size().reset_index(name="n")
gg["rate"] = gg["n"] / gg.groupby(["period", "from_b"], observed=True)["n"].transform("sum")
A["transitions"] = gg
# vintages — cohorts are keyed on origination month, so they ignore the window
vv = panel[panel["vintage"].notna() & panel["mob"].between(0, 36)].copy()
THRS = sorted({t_ for t_, _ in VINTAGE_MARKS})
for thr in THRS:
vv[f"ever_{thr}"] = vv["peak_dpd"] > thr
gv = vv.groupby(["vintage", "mob"], observed=True)
vin = pd.DataFrame({"n_loans": gv["anketa"].size()})
for thr in THRS:
vin[f"gl{thr}"] = gv[f"ever_{thr}"].sum() / gv["anketa"].size()
vin = vin.reset_index()
vin["reliable"] = vin["n_loans"] >= MIN_COHORT
A["vintages"] = vin
A["vin_marks"] = pd.concat(
[vin[vin["mob"] == mob].assign(metric=f"GL{thr}+@{mob}MOB", rate=vin[f"gl{thr}"])
[["vintage", "metric", "rate", "n_loans", "reliable"]]
for thr, mob in VINTAGE_MARKS], ignore_index=True)
# top products per department, fixed by their standing at SEL so the
# composition of the stack does not jump from month to month
top_pass = {}
for dep in sorted(snap["department"].dropna().unique()):
s = (snap[snap["department"] == dep]
.groupby("passport", observed=True)["exposure"].sum()
.sort_values(ascending=False))
top_pass[str(dep)] = [str(x) for x in s.head(TOP_PASSPORTS).index]
pp = panel.copy()
pp["passport_grp"] = pp["passport"].astype("string")
for dep, keep in top_pass.items():
mask = (pp["department"] == dep) & (~pp["passport"].isin(keep))
pp.loc[mask, "passport_grp"] = "Прочие"
gp = pp.groupby(["period", "department", "passport_grp"], observed=True)
pdp = pd.DataFrame({"exposure": gp["exposure"].sum(),
"npl_exposure": gp["npl_exposure"].sum(),
"n_loans": gp["anketa"].size(),
"disbursed": gp["disbursed"].sum()}).reset_index()
pdp["npl_ratio"] = pdp["npl_exposure"] / pdp["exposure"].replace(0, np.nan)
A["passport_dep"] = pdp
# ---------------- Direction B: one reporting date ----------------
tot_exp = float(snap["exposure"].sum())
tot_npl = float(snap["npl_exposure"].sum())
def structure(dim):
gg = snap.groupby(dim, observed=True)
m = pd.DataFrame({
"exposure": gg["exposure"].sum(), "npl_exposure": gg["npl_exposure"].sum(),
"watch_exposure": gg["watch_exposure"].sum(), "reserve": gg["reserve"].sum(),
"n_loans": gg["anketa"].size(), "n_clients": gg["contragent"].nunique(),
}).reset_index().rename(columns={dim: "name"})
m["name"] = m["name"].astype("string")
nz = lambda x: x.replace(0, np.nan)
m["share"] = m["exposure"] / tot_exp if tot_exp else np.nan
m["npl_ratio"] = m["npl_exposure"] / nz(m["exposure"])
m["watch_ratio"] = m["watch_exposure"] / nz(m["exposure"])
m["coverage"] = m["reserve"] / nz(m["npl_exposure"])
m["npl_contrib"] = m["npl_exposure"] / tot_npl if tot_npl else np.nan
m["avg_ticket"] = m["exposure"] / nz(m["n_loans"])
return m.sort_values("exposure", ascending=False)
B = {"_tot_exp": tot_exp, "_tot_npl": tot_npl,
"_n_clients": int(snap["contragent"].nunique()), "_n_loans": int(len(snap))}
for d in ["department", "filial", "passport", "avto", "cur_name", "sud",
"bucket", "mob_band"] + (["oked_name"] if has_oked else []):
B[d] = structure(d)
# branch rows tagged with their department, for the composition chart
dep_of = snap.groupby("filial", observed=True)["department"].first()
bf = B["filial"].copy()
bf["department"] = bf["name"].map(dep_of).astype("string")
B["dep_filial"] = bf
# what the book is made of, by origination month
vg = snap.groupby("vintage", observed=True)
vm = pd.DataFrame({"exposure": vg["exposure"].sum(),
"npl_exposure": vg["npl_exposure"].sum(),
"n_loans": vg["anketa"].size()}).reset_index()
vm["share"] = vm["exposure"] / tot_exp if tot_exp else np.nan
vm["npl_ratio"] = vm["npl_exposure"] / vm["exposure"].replace(0, np.nan)
B["vintage_mix"] = vm
# branch x bucket, normalised inside each branch so sizes are comparable
fb = snap.groupby(["filial", "bucket"], observed=True)["exposure"].sum().reset_index()
fb["share"] = fb["exposure"] / fb.groupby("filial", observed=True)["exposure"] \
.transform("sum").replace(0, np.nan)
B["filial_bucket"] = fb
# concentration — this snapshot only
cl = snap.groupby("contragent", observed=True)["exposure"].sum() \
.sort_values(ascending=False)
tot_cl = float(cl.sum()) or 1.0
B["lorenz"] = pd.DataFrame({
"client_pct": (np.arange(1, len(cl) + 1) / max(len(cl), 1)),
"exposure_pct": (cl.cumsum() / tot_cl).to_numpy() if len(cl) else np.array([])})
topn = lambda n: float(cl.head(min(n, len(cl))).sum() / tot_cl) if len(cl) else 0.0
B["concentration"] = {"n_clients": int(len(cl)), "top10": topn(10), "top50": topn(50),
"top100": topn(100), "top1pct": topn(max(1, len(cl)//100)),
"top5pct": topn(max(1, len(cl)//20)),
"top10pct": topn(max(1, len(cl)//10))}
# client burden — the average behaviour of a client holding N loans
cnt = snap.groupby("contragent", observed=True).agg(
n_loans=("anketa", "size"), exposure=("exposure", "sum"),
npl=("npl_exposure", "sum")).reset_index()
cnt["burden"] = pd.cut(cnt["n_loans"], [0, 1, 2, 3, 4, 10**9],
labels=["1", "2", "3", "4", "5+"])
bg = cnt.groupby("burden", observed=True)
bd = pd.DataFrame({"n_clients": bg["contragent"].size(),
"exposure": bg["exposure"].sum(),
"avg_exposure": bg["exposure"].mean(),
"avg_loans": bg["n_loans"].mean(),
"npl": bg["npl"].sum()}).reset_index()
bd["client_share"] = bd["n_clients"] / max(bd["n_clients"].sum(), 1)
bd["exposure_share"] = bd["exposure"] / (bd["exposure"].sum() or 1)
bd["npl_ratio"] = bd["npl"] / bd["exposure"].replace(0, np.nan)
B["burden"] = bd
# passport x department at SEL
pdm = snap.groupby(["department", "passport"], observed=True)
bpd = pd.DataFrame({"exposure": pdm["exposure"].sum(),
"npl_exposure": pdm["npl_exposure"].sum(),
"n_loans": pdm["anketa"].size()}).reset_index()
bpd["npl_ratio"] = bpd["npl_exposure"] / bpd["exposure"].replace(0, np.nan)
bpd["share"] = bpd["exposure"] / tot_exp if tot_exp else np.nan
B["passport_dep"] = bpd
return A, B, top_pass
# ==========================================================================
# CHARTS
#
# Every builder has the same shape: fn(A, B, ctx) -> (figure, note).
# They take marts as arguments rather than reading globals, because two
# requests can be running with different periods at the same time.
# ==========================================================================
def _x(df):
return [plabel(p) for p in df["period"]]
def _piv(df, val, agg="sum"):
return df.pivot_table(index="period", columns="dim_value", values=val, aggfunc=agg)
def _safemax(s, default=1.0):
try:
v = float(np.nanmax(np.asarray(s, dtype="float64")))
return v if np.isfinite(v) and v > 0 else default
except (ValueError, TypeError):
return default
# ---------------------------------------------------------------- A: dynamics
def a01_portfolio(A, B, ctx):
d = A["total"]; x = _x(d)
amt, npl, wat = money(d.exposure), d.npl_ratio * 100, d.watch_ratio * 100
f = go.Figure()
f.add_bar(x=x, y=amt, name=f"Портфель, {UNIT_NAME}", marker_color=BRAND["gold"],
text=thin(amt), textposition="outside", cliponaxis=False,
textfont=dict(size=10),
hovertemplate="%{x}<br>%{y:,.1f} " + UNIT_NAME + "<extra></extra>")
f.add_scatter(x=x, y=npl, name="NPL 90+, %", yaxis="y2", mode="lines+markers+text",
line=dict(color=BRAND["bad"], width=3), marker=dict(size=5),
text=thin(npl, lambda v: f"{v:.2f}%"), textposition="bottom center",
textfont=dict(size=9.5, color=BRAND["bad"]))
f.add_scatter(x=x, y=wat, name="Зона риска 31-90, %", yaxis="y2", mode="lines",
line=dict(color=BRAND["ink"], width=1.8, dash="dot"))
f.update_layout(title="A01 · Портфель и качество по отчётным датам", height=520,
bargap=.28, yaxis=dict(title=UNIT_NAME),
yaxis2=dict(overlaying="y", side="right", title="%", showgrid=False,
**headroom(max(_safemax(npl), _safemax(wat)), 2.8)))
unit_note(f); kind_tag(f, "СТОК")
return f, ("Столбцы — остаток портфеля на каждую отчётную дату. Соседние месяцы "
"описывают одну и ту же книгу, снятую заново, поэтому складывать их "
"нельзя. Линии намеренно уведены вниз, чтобы не пересекать столбцы.")
def a02_stock_flow(A, B, ctx):
d = A["stock_flow"]; x = _x(d)
dl = money(d.delta)
f = go.Figure()
f.add_bar(x=x, y=money(d.disbursed), name="Выдачи (+)", marker_color=BRAND["good"])
f.add_bar(x=x, y=-money(d.repaid), name="Погашения (−)", marker_color=BRAND["cool"])
f.add_bar(x=x, y=-money(d.written_amt), name="Списания (−)", marker_color=BRAND["bad"])
f.add_bar(x=x, y=money(d.other), name="Прочее (проценты, корректировки)",
marker_color=BRAND["muted"])
f.add_scatter(x=x, y=dl, name="Итоговое изменение портфеля",
mode="lines+markers+text", line=dict(color=BRAND["ink"], width=2.6),
marker=dict(size=5), text=thin(dl), textposition="top center",
textfont=dict(size=9.5))
f.update_layout(title="A02 · Из чего складывается изменение портфеля за месяц",
barmode="relative", height=520, yaxis=dict(title=UNIT_NAME))
unit_note(f); kind_tag(f, "ПОТОК")
return f, ("Мост между двумя снимками портфеля. Чёрная линия равна фактическому "
"изменению остатка, столбцы объясняют, откуда оно взялось. Именно этот "
"график стоит показать любому, кто попытается сложить два месяца.")
def a03_bucket_mix(A, B, ctx):
piv = A["bucket_mix"].pivot_table(index="period", columns="bucket",
values="share", aggfunc="sum").fillna(0)
keep = [b for b in BUCKET_ORDER if b in piv.columns]
piv = piv[keep]
x = [plabel(p) for p in piv.index]
step = max(1, math.ceil(len(x) / MAX_LABELS))
f = go.Figure()
for b in keep:
v = piv[b] * 100
txt = [f"{z:.1f}" if (i % step == 0 or i == len(v) - 1) and z >= 4 else ""
for i, z in enumerate(v)]
f.add_bar(x=x, y=v, name=lab(b), marker_color=BUCKET_COLORS.get(b, BRAND["muted"]),
text=txt, textposition="inside", insidetextanchor="middle",
textfont=dict(size=9, color="white"),
hovertemplate="%{x}<br>" + lab(b) + ": %{y:.2f}%<extra></extra>")
bad = piv[[c for c in keep if c != "CUR"]].sum(axis=1) * 100
f.add_scatter(x=x, y=bad, name="Всего с просрочкой, %", mode="lines+text",
line=dict(color=BRAND["ink"], width=2),
text=thin(bad, lambda v: f"{v:.1f}%"), textposition="top center",
textfont=dict(size=9.5))
f.update_layout(title="A03 · Структура портфеля по просрочке, % от портфеля месяца",
barmode="stack", height=540,
yaxis=dict(title="% портфеля", range=[0, 108]))
unit_note(f, "доли внутри каждого месяца, сумма столбца = 100%")
kind_tag(f, "СТОК")
return f, ("Стек здесь корректен: внутри каждого месяца доли складываются в 100%. "
"Цифры внутри полос — проценты. Чёрная линия — суммарная доля кредитов "
"с любой просрочкой.")
def a04_npl_departments(A, B, ctx):
piv = _piv(A["department"], "npl_ratio", "mean") * 100
x = [plabel(p) for p in piv.index]
f = go.Figure()
for i, c in enumerate(piv.columns):
v = piv[c]
f.add_scatter(x=x, y=v, name=str(c), mode="lines+text",
line=dict(width=2.6, color=SEQ[i % len(SEQ)]),
text=thin(v, lambda z: f"{z:.1f}", max_labels=9),
textposition="top center",
textfont=dict(size=9, color=SEQ[i % len(SEQ)]))
tv = A["total"].set_index("period").reindex(piv.index)["npl_ratio"] * 100
f.add_scatter(x=x, y=tv, name="Итого по банку", mode="lines",
line=dict(color=BRAND["ink"], width=1.6, dash="dash"))
f.update_layout(title="A04 · Доля NPL 90+ по департаментам", height=480,
yaxis=dict(title="% портфеля департамента"))
unit_note(f, "проценты внутри своего департамента"); kind_tag(f, "СТОК")
return f, ("Линии, а не стек: доли разных департаментов не суммируются в общую "
"долю банка. Пунктир — банк целиком, для сравнения.")
def a05_branch_heatmap(A, B, ctx):
piv = _piv(A["filial"], "npl_ratio", "mean") * 100
order = piv.iloc[-1].sort_values(ascending=False).index
piv = piv[order]
z = piv.T.to_numpy()
step = max(1, math.ceil(z.shape[1] / 12))
txt = [[f"{z[r, c]:.1f}" if (c % step == 0 or c == z.shape[1] - 1)
and np.isfinite(z[r, c]) else "" for c in range(z.shape[1])]
for r in range(z.shape[0])]
f = go.Figure(go.Heatmap(
z=z, x=[plabel(p) for p in piv.index], y=list(order),
text=txt, texttemplate="%{text}", textfont=dict(size=8.5),
colorscale=[[0, "#FFFFFF"], [.3, BRAND["gold"]], [.65, BRAND["gold_deep"]],
[1, BRAND["bad"]]],
colorbar=dict(title="NPL 90+, %", thickness=14, len=.85),
hovertemplate="%{y}<br>%{x}: %{z:.2f}%<extra></extra>"))
f.update_layout(title="A05 · Доля NPL 90+ по филиалам и месяцам",
height=max(520, 30 * len(order)))
unit_note(f, "значения в % портфеля филиала"); kind_tag(f, "СТОК")
return f, ("Филиалы отсортированы по последнему месяцу. Горизонтальная красная "
"полоса — филиал с устойчивой проблемой. Вертикальная — событие, "
"затронувшее весь банк сразу.")
def a06_bridge(A, B, ctx):
d = A["bridge"]; x = _x(d)
delta = money(d.closing - d.opening)
f = go.Figure()
for k, c, n in (("inflow", BRAND["bad"], "Новые дефолты (+)"),
("cured", BRAND["good"], "Восстановлено (−)"),
("exited", BRAND["cool"], "Выбытие с баланса (−)"),
("residual", BRAND["muted"], "Погашение и прочее")):
f.add_bar(x=x, y=money(d[k]), name=n, marker_color=c)
f.add_scatter(x=x, y=delta, name="Итоговое изменение NPL",
mode="lines+markers+text", line=dict(color=BRAND["ink"], width=2.6),
marker=dict(size=5), text=thin(delta), textposition="top center",
textfont=dict(size=9.5))
f.update_layout(title="A06 · Почему изменился NPL: приток, восстановление, выбытие",
barmode="relative", height=520, yaxis=dict(title=UNIT_NAME))
unit_note(f); kind_tag(f, "ПОТОК")
return f, ("«NPL вырос на 0.4 п.п.» — это не ответ. Ответ — вот эти четыре потока. "
"Восстановление и выбытие принципиально разные вещи: второе означает, "
"что кредит ушёл с баланса, а не выздоровел.")
def a07_vintage_quality(A, B, ctx):
vm = A["vin_marks"]; vm = vm[vm.reliable]
m12 = vm[vm.metric == "GL90+@12MOB"].set_index("vintage")["rate"] * 100
m6 = vm[vm.metric == "GL30+@6MOB"].set_index("vintage")["rate"] * 100
idx = sorted(set(m12.index) | set(m6.index))
if not idx:
raise ValueError("нет когорт с достаточным числом кредитов")
x = [plabel(v) for v in idx]
v12, v6 = m12.reindex(idx), m6.reindex(idx)
f = go.Figure()
f.add_bar(x=x, y=v6, name="GL30+@6MOB — просрочка 30+ к 6-му месяцу",
marker_color=BRAND["gold"], text=thin(v6, lambda z: f"{z:.1f}"),
textposition="outside", cliponaxis=False, textfont=dict(size=9.5))
f.add_bar(x=x, y=v12, name="GL90+@12MOB — дефолт к 12-му месяцу",
marker_color=BRAND["bad"], text=thin(v12, lambda z: f"{z:.1f}"),
textposition="outside", cliponaxis=False, textfont=dict(size=9.5))
if v12.notna().sum():
avg = float(np.nanmean(v12))
f.add_hline(y=avg, line=dict(color=BRAND["ink"], width=1.2, dash="dot"),
annotation_text=f"средний уровень GL90+@12MOB: {avg:.1f}%",
annotation_position="top left", annotation_font_size=11)
f.update_layout(title="A07 · Качество выдач: как портится каждая когорта",
barmode="group", height=500,
xaxis=dict(title="месяц выдачи кредита"),
yaxis=dict(title="% кредитов когорты"))
unit_note(f, "% от числа кредитов, выданных в этом месяце"); kind_tag(f, "КОГОРТЫ")
return f, ("Один столбец — один месяц выдачи. Жёлтый: сколько кредитов этой выдачи "
"ушли в просрочку 30+ дней к шестому месяцу жизни. Красный: сколько "
"дошли до дефолта к двенадцатому. Правая часть пустая: молодые когорты "
"ещё не дожили до контрольной точки.")
def a08_vintage_heat(A, B, ctx):
vm = A["vin_marks"]; vm = vm[vm.reliable]
piv = vm.pivot_table(index="metric", columns="vintage", values="rate",
aggfunc="mean") * 100
order = [f"GL{a_}+@{b_}MOB" for a_, b_ in VINTAGE_MARKS]
piv = piv.reindex([o for o in order if o in piv.index])
z = piv.to_numpy()
step = max(1, math.ceil(max(z.shape[1], 1) / 16))
txt = [[f"{z[r, c]:.1f}" if (c % step == 0 or c == z.shape[1] - 1)
and np.isfinite(z[r, c]) else "" for c in range(z.shape[1])]
for r in range(z.shape[0])]
f = go.Figure(go.Heatmap(
z=z, x=[plabel(v) for v in piv.columns], y=list(piv.index),
text=txt, texttemplate="%{text}", textfont=dict(size=9),
colorscale=[[0, "#FFFFFF"], [.35, BRAND["gold"]], [.7, BRAND["gold_deep"]],
[1, BRAND["bad"]]],
colorbar=dict(title="%", thickness=14),
hovertemplate="%{y}<br>выдача %{x}: %{z:.2f}%<extra></extra>"))
f.update_layout(title="A08 · GLx+@yMOB по когортам выдач", height=360,
xaxis=dict(title="месяц выдачи"))
unit_note(f, f"когорты меньше {MIN_COHORT} кредитов скрыты"); kind_tag(f, "КОГОРТЫ")
return f, ("Читается так: GL90+@12MOB — доля кредитов, выданных в этом месяце, "
"которые хотя бы раз достигли 90 дней просрочки за первые 12 месяцев "
"жизни. Чем правее и краснее, тем хуже свежие выдачи.")
def a09_migration(A, B, ctx):
tr = A["transitions"]
per = A["total"]["period"]
to_default = (tr[(tr.from_b == "B61_90") & tr.to_b.isin(DEFAULT_BUCKETS)]
.groupby("period", observed=True)["rate"].sum().reindex(per) * 100)
cured = (tr[(tr.from_b == "B01_30") & (tr.to_b == "CUR")]
.groupby("period", observed=True)["rate"].sum().reindex(per) * 100)
x = [plabel(p) for p in per]
f = go.Figure()
f.add_bar(x=x, y=to_default, name="Из корзины 61-90 ушли в дефолт",
marker_color=BRAND["bad"], text=thin(to_default, lambda v: f"{v:.0f}"),
textposition="outside", cliponaxis=False, textfont=dict(size=9.5),
hovertemplate="%{x}<br>%{y:.1f} из 100 кредитов<extra></extra>")
f.add_scatter(x=x, y=cured, name="Из корзины 1-30 вернулись в норму",
mode="lines+text", line=dict(color=BRAND["good"], width=2.6),
text=thin(cured, lambda v: f"{v:.0f}"), textposition="bottom center",
textfont=dict(size=9.5, color=BRAND["good"]))
f.update_layout(title="A09 · Из каждых 100 кредитов корзины — сколько ушли "
"в дефолт за следующий месяц",
height=480, yaxis=dict(title="кредитов из 100"))
unit_note(f, "из каждых 100 кредитов, находившихся в корзине месяц назад")
kind_tag(f, "ПОТОК")
return f, ("Читается буквально: из 100 кредитов, стоявших в корзине 61-90 дней, "
"столько-то оказались в дефолте через месяц. Это самый ранний "
"измеримый сигнал — он опережает долю NPL примерно на квартал. "
"Зелёная линия — обратный процесс.")
def a10_currency(A, B, ctx):
piv = _piv(A["cur_name"], "exposure")
piv = (piv.div(piv.sum(axis=1), axis=0).fillna(0) * 100)
x = [plabel(p) for p in piv.index]
step = max(1, math.ceil(len(x) / MAX_LABELS))
f = go.Figure()
for i, c in enumerate(piv.columns):
v = piv[c]
txt = [f"{z:.1f}" if (j % step == 0 or j == len(v) - 1) and z >= 4 else ""
for j, z in enumerate(v)]
f.add_bar(x=x, y=v, name=str(c), marker_color=SEQ[i % len(SEQ)], text=txt,
textposition="inside", insidetextanchor="middle",
textfont=dict(size=9, color="white"),
hovertemplate="%{x}<br>" + str(c) + ": %{y:.2f}%<extra></extra>")
non_som = 100 - piv.get("Сум", pd.Series(0.0, index=piv.index))
f.add_scatter(x=x, y=non_som, name="Валютная часть, %", mode="lines+text",
line=dict(color=BRAND["ink"], width=2),
text=thin(non_som, lambda v: f"{v:.1f}%"), textposition="top center",
textfont=dict(size=9.5))
f.update_layout(title="A10 · Доля валют в портфеле", barmode="stack", height=460,
yaxis=dict(title="% портфеля", range=[0, 108]))
unit_note(f, "доли внутри месяца, сумма = 100%"); kind_tag(f, "СТОК")
return f, ("Суммы в источнике уже приведены к сумам, поэтому здесь доля, а не "
"пересчёт. Рост валютной части — это валютный риск заёмщика, который "
"возвращается к банку кредитным риском.")
def a11_origination(A, B, ctx):
piv = _piv(A["department"], "disbursed")
x = [plabel(p) for p in piv.index]
tot = money(piv.sum(axis=1))
f = go.Figure()
for i, c in enumerate(piv.columns):
f.add_bar(x=x, y=money(piv[c]), name=str(c), marker_color=SEQ[i % len(SEQ)],
hovertemplate="%{x}<br>" + str(c) + ": %{y:,.1f} " + UNIT_NAME +
"<extra></extra>")
f.add_scatter(x=x, y=tot, name="Всего выдано за месяц", mode="text",
text=thin(tot), textposition="top center",
textfont=dict(size=9.5, color=BRAND["ink"]), showlegend=False)
f.update_layout(title="A11 · Выдачи за месяц по департаментам", barmode="stack",
height=500, yaxis=dict(title=UNIT_NAME))
unit_note(f); kind_tag(f, "ПОТОК")
return f, ("Поток: столбцы за разные месяцы складываются законно, это реальные "
"деньги, выданные в каждом месяце. Подписи сверху — суммарная выдача.")
def a12_coverage(A, B, ctx):
d = A["total"]; x = _x(d)
cov, crt = d.coverage * 100, d.court_share * 100
f = go.Figure()
f.add_scatter(x=x, y=cov, name="Покрытие NPL резервами, %", mode="lines+text",
line=dict(color=BRAND["good"], width=3),
text=thin(cov, lambda v: f"{v:.0f}"), textposition="top center",
textfont=dict(size=9.5, color=BRAND["good"]))
f.add_scatter(x=x, y=crt, name="Доля портфеля в суде, %", yaxis="y2",
mode="lines+text", line=dict(color=BRAND["cool"], width=2.4, dash="dot"),
text=thin(crt, lambda v: f"{v:.1f}"), textposition="bottom center",
textfont=dict(size=9.5, color=BRAND["cool"]))
f.add_hline(y=100, line=dict(color=BRAND["muted"], width=1, dash="dash"),
annotation_text="полное покрытие", annotation_position="right",
annotation_font_size=10)
f.update_layout(title="A12 · Покрытие резервами и судебная работа", height=460,
yaxis=dict(title="покрытие, %"),
yaxis2=dict(overlaying="y", side="right", title="в суде, %",
showgrid=False, **headroom(_safemax(crt), 2.6)))
unit_note(f, "покрытие = резервы / NPL 90+"); kind_tag(f, "СТОК")
return f, ("Покрытие ниже 100% означает, что часть ожидаемых потерь ещё не "
"признана резервом. Растущая судебная доля при падающем покрытии — "
"сочетание, которое всегда требует объяснения.")
def a13_insurance(A, B, ctx):
d = A["total"]; x = _x(d)
amt = money(d.exposure)
npl, gross, watch = (d.npl_ratio * 100, d.npl_ratio_gross * 100, d.watch_ratio * 100)
f = go.Figure()
f.add_bar(x=x, y=amt, name=f"Портфель, {UNIT_NAME}", marker_color="#EFE7CE",
text=thin(amt), textposition="outside", cliponaxis=False,
textfont=dict(size=9.5, color=BRAND["muted"]))
f.add_scatter(x=x, y=gross, name="NPL 90+ без страховых выплат, %", yaxis="y2",
mode="lines", line=dict(color=BRAND["bad"], width=2.6, dash="dash"))
f.add_scatter(x=x, y=npl, name="NPL 90+ фактический, %", yaxis="y2",
mode="lines+text", line=dict(color=BRAND["ink"], width=3),
fill="tonexty", fillcolor="rgba(242,194,48,.30)",
text=thin(npl, lambda v: f"{v:.2f}"), textposition="bottom center",
textfont=dict(size=9.5))
f.add_scatter(x=x, y=watch, name="Зона риска 31-90, %", yaxis="y2", mode="lines",
line=dict(color=BRAND["cool"], width=1.6, dash="dot"))
f.update_layout(title="A13 · Что страховка удерживает от NPL", height=520, bargap=.28,
yaxis=dict(title=UNIT_NAME),
yaxis2=dict(overlaying="y", side="right", title="%", showgrid=False,
**headroom(max(_safemax(gross), _safemax(watch)), 2.8)))
unit_note(f); kind_tag(f, "СТОК")
return f, ("Золотая заливка между линиями — работа страховой программы: NPL, "
"который был бы на балансе, если бы страховая не платила. Поля "
"straxovka_brutto_95_summa и kategoriya_straxovka есть в выгрузке, "
"но их почти никто не считает.")
def _passport_chart(dep, cid):
"""One stacked chart per department, products fixed by their standing at SEL."""
def fn(A, B, ctx):
d = A["passport_dep"]
d = d[d.department == dep]
if d.empty:
raise ValueError(f"нет данных по департаменту «{dep}»")
piv = d.pivot_table(index="period", columns="passport_grp",
values="exposure", aggfunc="sum").fillna(0)
cols = [c for c in piv.columns if c != "Прочие"] + \
(["Прочие"] if "Прочие" in piv.columns else [])
piv = piv[cols]
x = [plabel(p) for p in piv.index]
tot = money(piv.sum(axis=1))
f = go.Figure()
for i, c in enumerate(cols):
colour = BRAND["muted"] if c == "Прочие" else SEQ[i % len(SEQ)]
f.add_bar(x=x, y=money(piv[c]), name=str(c), marker_color=colour,
hovertemplate="%{x}<br>" + str(c) + ": %{y:,.1f} " + UNIT_NAME +
"<extra></extra>")
f.add_scatter(x=x, y=tot, name="Итого", mode="text", text=thin(tot),
textposition="top center",
textfont=dict(size=9.5, color=BRAND["ink"]), showlegend=False)
nplr = (d.groupby("period", observed=True)
.apply(lambda g: g.npl_exposure.sum() / max(g.exposure.sum(), 1),
include_groups=False) * 100).reindex(piv.index)
f.add_scatter(x=x, y=nplr, name="NPL 90+ департамента, %", yaxis="y2",
mode="lines+text", line=dict(color=BRAND["bad"], width=2.6),
text=thin(nplr, lambda v: f"{v:.1f}"),
textposition="bottom center",
textfont=dict(size=9, color=BRAND["bad"]))
f.update_layout(title=f"{cid} · «{dep}»: портфель по продуктам "
f"(топ-{TOP_PASSPORTS} на {plabel_full(ctx['sel'])})",
barmode="stack", height=520, yaxis=dict(title=UNIT_NAME),
yaxis2=dict(overlaying="y", side="right", title="NPL, %",
showgrid=False, **headroom(_safemax(nplr), 2.8)))
unit_note(f); kind_tag(f, "СТОК")
return f, (f"Продукты отобраны по объёму на {plabel_full(ctx['sel'])} и "
f"зафиксированы на всю историю, поэтому состав полос не «прыгает» "
f"от месяца к месяцу. Всё, что не вошло в топ-{TOP_PASSPORTS}, "
f"собрано в серые «Прочие». Красная линия — качество департамента "
f"целиком.")
return fn
def a_oked_dynamics(A, B, ctx):
top = list(B["oked_name"].head(TOP_OKED)["name"])
d = A["oked_name"].copy()
d["grp"] = np.where(d.dim_value.isin(top), d.dim_value, "Прочие отрасли")
piv = d.pivot_table(index="period", columns="grp", values="exposure",
aggfunc="sum").fillna(0)
piv = piv.div(piv.sum(axis=1), axis=0).fillna(0) * 100
cols = [c for c in piv.columns if c != "Прочие отрасли"] + \
(["Прочие отрасли"] if "Прочие отрасли" in piv.columns else [])
piv = piv[cols]
x = [plabel(p) for p in piv.index]
step = max(1, math.ceil(len(x) / 10))
f = go.Figure()
for i, c in enumerate(cols):
v = piv[c]
colour = BRAND["muted"] if c == "Прочие отрасли" else SEQ[i % len(SEQ)]
txt = [f"{z:.0f}" if (j % step == 0 or j == len(v) - 1) and z >= 6 else ""
for j, z in enumerate(v)]
f.add_bar(x=x, y=v, name=str(c), marker_color=colour, text=txt,
textposition="inside", insidetextanchor="middle",
textfont=dict(size=8.5, color="white"),
hovertemplate="%{x}<br>" + str(c) + ": %{y:.2f}%<extra></extra>")
f.update_layout(title="· Отраслевая структура портфеля (ОКЭД), % от портфеля",
barmode="stack", height=540,
yaxis=dict(title="% портфеля", range=[0, 104]))
unit_note(f, "доли внутри месяца, сумма = 100%"); kind_tag(f, "СТОК")
return f, ("Отрасли отобраны по объёму на выбранную дату. Если одна полоса "
"неуклонно расширяется — портфель концентрируется в одной отрасли, "
"и её отраслевой спад станет кредитным риском банка.")
# ---------------------------------------------------------------- B: structure
def b01_dep_filial(A, B, ctx):
d = B["dep_filial"].sort_values("exposure")
deps = sorted([x for x in d["department"].dropna().unique()])
cmap = {dep: SEQ[i % len(SEQ)] for i, dep in enumerate(deps)}
f = go.Figure()
for dep in deps:
s = d[d.department == dep]
f.add_bar(x=money(s.exposure), y=s["name"], orientation="h", name=str(dep),
marker_color=cmap[dep],
text=[f"{fnum(v / UNIT, 1)} ({sh:.1%}) NPL {n:.1f}%"
for v, sh, n in zip(s.exposure, s.share, s.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=10),
customdata=np.stack([s["n_loans"], s["n_clients"]], axis=-1),
hovertemplate="<b>%{y}</b><br>%{x:,.1f} " + UNIT_NAME +
"<br>кредитов %{customdata[0]:,}"
"<br>клиентов %{customdata[1]:,}<extra></extra>")
f.update_layout(title=f"B01 · Портфель по филиалам, {plabel_full(ctx['sel'])}",
height=max(480, 32 * len(d)), xaxis=dict(title=UNIT_NAME),
barmode="stack")
unit_note(f, f"{UNIT_NAME} · доля портфеля банка · NPL филиала")
kind_tag(f, "СРЕЗ")
return f, ("Цвет — департамент, длина — объём. Подпись сразу даёт сумму, долю и "
"качество, поэтому картинку можно вставить в презентацию как есть.")
def b02_buckets(A, B, ctx):
d = B["bucket"].copy()
d["ord"] = d["name"].map({b: i for i, b in enumerate(BUCKET_ORDER)})
d = d.sort_values("ord")
amt = money(d.exposure)
f = go.Figure()
f.add_bar(x=[lab(n) for n in d["name"]], y=amt,
marker_color=[BUCKET_COLORS.get(n, BRAND["muted"]) for n in d["name"]],
text=[f"{fnum(v, 1)}<br>{s:.1%}<br>{int(n):,} кред.".replace(",", " ")
for v, s, n in zip(amt, d["share"], d["n_loans"])],
textposition="outside", cliponaxis=False, textfont=dict(size=10),
hovertemplate="<b>%{x}</b><br>%{y:,.1f} " + UNIT_NAME + "<extra></extra>")
f.update_layout(title=f"B02 · Портфель по корзинам просрочки, "
f"{plabel_full(ctx['sel'])}",
height=480, yaxis=dict(title=UNIT_NAME), showlegend=False)
unit_note(f, f"{UNIT_NAME} · доля портфеля · число кредитов"); kind_tag(f, "СРЕЗ")
clean = float(d[d.name == "CUR"]["share"].sum())
watch = float(d[d.name.isin(["B31_60", "B61_90"])]["share"].sum())
return f, (f"Без просрочки — {clean:.1%} портфеля. Корзины 31-90 дней ({watch:.1%}) — "
"это будущий NPL следующих двух кварталов, а не сегодняшняя проблема.")
def b03_vintage_mix(A, B, ctx):
d = B["vintage_mix"].dropna(subset=["vintage"]).sort_values("vintage")
d = d[d.vintage >= int(d.vintage.max()) - 300]
x = [plabel(v) for v in d.vintage]
amt, nplr = money(d.exposure), d.npl_ratio * 100
f = go.Figure()
f.add_bar(x=x, y=amt, name=f"Остаток на балансе, {UNIT_NAME}",
marker_color=BRAND["gold"], text=thin(amt), textposition="outside",
cliponaxis=False, textfont=dict(size=9.5),
customdata=np.stack([d["share"] * 100, d["n_loans"]], axis=-1),
hovertemplate="выдача %{x}<br>%{y:,.1f} " + UNIT_NAME +
"<br>%{customdata[0]:.1f}% портфеля"
"<br>кредитов %{customdata[1]:,}<extra></extra>")
f.add_scatter(x=x, y=nplr, name="NPL 90+ этой когорты, %", yaxis="y2",
mode="lines+text", line=dict(color=BRAND["bad"], width=2.4),
text=thin(nplr, lambda v: f"{v:.1f}"), textposition="bottom center",
textfont=dict(size=9, color=BRAND["bad"]))
f.update_layout(title=f"B03 · Из выдач каких месяцев состоит портфель "
f"на {plabel_full(ctx['sel'])}",
height=500, xaxis=dict(title="месяц выдачи"),
yaxis=dict(title=UNIT_NAME),
yaxis2=dict(overlaying="y", side="right", title="NPL, %",
showgrid=False, **headroom(_safemax(nplr), 2.8)))
unit_note(f); kind_tag(f, "СРЕЗ")
return f, ("Столбец — сколько денег из выдач того месяца ещё на балансе сегодня. "
"Красная линия — насколько плохо ведёт себя именно эта когорта. "
"Высокий столбец с высокой линией — недавняя крупная выдача, которая "
"уже портится.")
def b04_mob_profile(A, B, ctx):
d = B["mob_band"].copy()
d["ord"] = d["name"].map({m: i for i, m in enumerate(MOB_LABELS)})
d = d.sort_values("ord")
nplr = d.npl_ratio * 100
f = go.Figure()
f.add_bar(x=d["name"], y=d["share"] * 100, marker_color=BRAND["gold"],
name="Доля портфеля, %",
text=[f"{s:.1%}<br>{fnum(e / UNIT, 1)} {UNIT_NAME}"
for s, e in zip(d["share"], d["exposure"])],
textposition="outside", cliponaxis=False, textfont=dict(size=10))
f.add_scatter(x=d["name"], y=nplr, name="NPL 90+, %", yaxis="y2",
mode="lines+markers+text", line=dict(color=BRAND["bad"], width=2.6),
marker=dict(size=8), text=[f"{v:.2f}%" for v in nplr],
textposition="top center", textfont=dict(size=10, color=BRAND["bad"]))
f.update_layout(title=f"B04 · Возраст кредитов в портфеле, {plabel_full(ctx['sel'])}",
height=470, yaxis=dict(title="% портфеля"),
yaxis2=dict(overlaying="y", side="right", title="NPL, %",
showgrid=False, **headroom(_safemax(nplr), 1.9)))
unit_note(f, "возраст = число месяцев с даты выдачи"); kind_tag(f, "СРЕЗ")
return f, ("Молодой портфель с высоким NPL — проблема правил выдачи. Старый "
"портфель с высоким NPL — накопленное наследство, которое изменением "
"правил выдачи уже не лечится.")
def b05_products(A, B, ctx):
d = B["passport"].head(15).sort_values("exposure")
f = go.Figure()
f.add_bar(x=money(d.exposure), y=d["name"], orientation="h", name="Портфель",
marker=dict(color=d.npl_ratio * 100,
colorscale=[[0, BRAND["good"]], [.4, BRAND["gold"]],
[1, BRAND["bad"]]],
colorbar=dict(title="NPL, %", thickness=14)),
text=[f"{fnum(v / UNIT, 1)} · {sh:.1%} · NPL {n:.1f}%"
for v, sh, n in zip(d.exposure, d.share, d.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=9.5),
customdata=np.stack([d["n_loans"], d["avg_ticket"] / 1e6], axis=-1),
hovertemplate="<b>%{y}</b><br>%{x:,.1f} " + UNIT_NAME +
"<br>кредитов %{customdata[0]:,}"
"<br>средний чек %{customdata[1]:,.1f} млн<extra></extra>")
f.update_layout(title=f"B05 · Топ-15 продуктов по объёму, {plabel_full(ctx['sel'])}",
height=620, xaxis=dict(title=UNIT_NAME), showlegend=False)
unit_note(f, f"{UNIT_NAME} · доля портфеля · NPL продукта"); kind_tag(f, "СРЕЗ")
return f, ("Цвет — качество, длина — объём. Длинный красный столбец важнее "
"короткого красного: он означает большие деньги под риском.")
def b06_passport_dep(A, B, ctx):
d = B["passport_dep"].copy()
keep = [g.nlargest(TOP_PASSPORTS, "exposure")
for _, g in d.groupby("department", observed=True)]
d = pd.concat(keep).sort_values(["department", "exposure"])
d["tag"] = d["passport"].astype(str) + " · " + d["department"].astype(str)
deps = sorted([x for x in d["department"].dropna().unique()])
cmap = {dep: SEQ[i % len(SEQ)] for i, dep in enumerate(deps)}
f = go.Figure()
for dep in deps:
s = d[d.department == dep]
f.add_bar(x=money(s.exposure), y=s["tag"], orientation="h", name=str(dep),
marker_color=cmap[dep],
text=[f"{fnum(v / UNIT, 1)} · NPL {n:.1f}%"
for v, n in zip(s.exposure, s.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=9.5),
customdata=np.stack([s["n_loans"], s["share"] * 100], axis=-1),
hovertemplate="<b>%{y}</b><br>%{x:,.1f} " + UNIT_NAME +
"<br>кредитов %{customdata[0]:,}"
"<br>%{customdata[1]:.1f}% портфеля банка<extra></extra>")
f.update_layout(title=f"B06 · Топ-{TOP_PASSPORTS} продуктов каждого департамента, "
f"{plabel_full(ctx['sel'])}",
height=max(520, 28 * len(d)), xaxis=dict(title=UNIT_NAME))
unit_note(f, f"{UNIT_NAME} · NPL продукта"); kind_tag(f, "СРЕЗ")
return f, ("Продукты сгруппированы по департаментам, поэтому видно и абсолютный "
"вес продукта, и то, чей он. Подпись несёт объём и качество сразу.")
def b07_filial_bucket(A, B, ctx):
piv = B["filial_bucket"].pivot_table(index="filial", columns="bucket",
values="share", aggfunc="sum").fillna(0)
keep = [b for b in BUCKET_ORDER if b in piv.columns]
piv = piv[keep]
bad_cols = [c for c in keep if c in DEFAULT_BUCKETS]
if bad_cols:
piv = piv.loc[piv[bad_cols].sum(axis=1).sort_values().index]
f = go.Figure()
for b in keep:
v = piv[b] * 100
f.add_bar(y=list(piv.index), x=v, orientation="h", name=lab(b),
marker_color=BUCKET_COLORS.get(b, BRAND["muted"]),
text=[f"{z:.0f}" if z >= 5 else "" for z in v],
textposition="inside", insidetextanchor="middle",
textfont=dict(size=9, color="white"),
hovertemplate="%{y}<br>" + lab(b) + ": %{x:.2f}%<extra></extra>")
f.update_layout(title=f"B07 · Структура просрочки внутри каждого филиала, "
f"{plabel_full(ctx['sel'])}",
barmode="stack", height=max(480, 32 * len(piv)),
xaxis=dict(title="% портфеля филиала", range=[0, 100]))
unit_note(f, "каждая полоса = 100% портфеля своего филиала"); kind_tag(f, "СРЕЗ")
return f, ("Каждая полоса нормирована на 100%, поэтому крупные и мелкие филиалы "
"сравнимы напрямую. Отсортировано по доле глубокой просрочки: худшие "
"сверху.")
def b08_concentration(A, B, ctx):
lz, c = B["lorenz"], B["concentration"]
lzs = lz.iloc[::max(1, len(lz) // 2000)] if len(lz) else lz
f = go.Figure()
f.add_scatter(x=lzs.client_pct * 100, y=lzs.exposure_pct * 100, mode="lines",
name="Фактическое распределение",
line=dict(color=BRAND["gold_deep"], width=3), fill="tozeroy",
fillcolor="rgba(242,194,48,.22)",
hovertemplate="%{x:.1f}% клиентов держат %{y:.1f}% портфеля"
"<extra></extra>")
f.add_scatter(x=[0, 100], y=[0, 100], mode="lines", name="Равномерное распределение",
line=dict(color=BRAND["ink"], width=1.2, dash="dot"))
marks = [(1, c["top1pct"], "топ-1%"), (5, c["top5pct"], "топ-5%"),
(10, c["top10pct"], "топ-10%")]
f.add_scatter(x=[m[0] for m in marks], y=[m[1] * 100 for m in marks],
mode="markers+text", showlegend=False,
marker=dict(size=10, color=BRAND["ink"]),
text=[f" {m[2]} клиентов → {m[1]:.0%} портфеля" for m in marks],
textposition="middle right", textfont=dict(size=11))
f.update_layout(title=f"B08 · Концентрация портфеля по клиентам, "
f"{plabel_full(ctx['sel'])}",
height=520,
xaxis=dict(title="% клиентов, отсортированных от крупнейших",
range=[0, 100]),
yaxis=dict(title="% портфеля", range=[0, 104]))
unit_note(f, f"клиентов на дату: {c['n_clients']:,}".replace(",", " "))
kind_tag(f, "СРЕЗ")
return f, (f"Топ-10 клиентов держат {c['top10']:.1%} портфеля, топ-50 — "
f"{c['top50']:.1%}, топ-100 — {c['top100']:.1%}. Чем выше кривая над "
"диагональю, тем сильнее концентрация — и тем дороже обходится дефолт "
"одного крупного заёмщика.")
def b09_npl_contribution(A, B, ctx):
d = B["filial"].sort_values("npl_exposure", ascending=False).head(12).iloc[::-1]
f = go.Figure()
f.add_bar(x=d.npl_contrib * 100, y=d["name"], orientation="h",
marker_color=BRAND["bad"],
text=[f"{v:.1f}% общего NPL · {fnum(e / UNIT, 1)} {UNIT_NAME} · "
f"свой NPL {r:.1f}%"
for v, e, r in zip(d.npl_contrib * 100, d.npl_exposure,
d.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=9.5),
hovertemplate="<b>%{y}</b><br>вклад в общий NPL: %{x:.1f}%<extra></extra>")
f.update_layout(title=f"B09 · Кто формирует общий NPL банка, {plabel_full(ctx['sel'])}",
height=500, xaxis=dict(title="% от общего NPL банка"),
showlegend=False)
unit_note(f, f"общий NPL банка на дату: {fnum(B['_tot_npl'] / UNIT, 1)} {UNIT_NAME}")
kind_tag(f, "СРЕЗ")
top3 = float(d.tail(3).npl_contrib.sum())
return f, (f"Три филиала дают {top3:.0%} всего NPL банка. Это не то же самое, что "
"«худшее качество»: крупный филиал со средним NPL весит больше, чем "
"маленький с плохим. Здесь видно, где лежат деньги, а не проценты.")
def b10_auto_court(A, B, ctx):
a_ = B["avto"].copy()
order = ["pervichka", "vtorichka", "non_auto"]
a_["ord"] = a_["name"].map({k: i for i, k in enumerate(order)})
a_ = a_.sort_values("ord")
s_ = B["sud"].copy()
f = go.Figure()
f.add_bar(x=[lab(n) for n in a_["name"]], y=a_["share"] * 100, name="Автосегмент",
marker_color=[BRAND["gold"], BRAND["gold_deep"], BRAND["muted"]][:len(a_)],
text=[f"{s:.1%}<br>{fnum(e / UNIT, 1)} {UNIT_NAME}<br>NPL {n:.2f}%"
for s, e, n in zip(a_["share"], a_["exposure"], a_.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=10))
f.add_bar(x=[lab(n) for n in s_["name"]], y=s_["share"] * 100, name="Судебная стадия",
marker_color=BRAND["cool"],
text=[f"{s:.1%}<br>{fnum(e / UNIT, 1)} {UNIT_NAME}<br>NPL {n:.2f}%"
for s, e, n in zip(s_["share"], s_["exposure"], s_.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=10))
f.update_layout(title=f"B10 · Автосегмент и судебная стадия, {plabel_full(ctx['sel'])}",
height=500, yaxis=dict(title="% портфеля", range=[0, 122]))
unit_note(f, "доля портфеля · объём · NPL сегмента"); kind_tag(f, "СРЕЗ")
return f, ("Автокредиты делятся ровно на три группы: новые авто, авто с пробегом "
"и всё остальное. Судебная стадия показывает, сколько работы уже стоит "
"в очереди у юристов.")
def b11_client_profile(A, B, ctx):
d = B["burden"].copy()
labels = [f"{t} кредит(ов)" for t in d["burden"].astype(str)]
avg = d["avg_exposure"] / 1e6
f = go.Figure()
f.add_bar(x=labels, y=avg, name="Средний долг одного клиента, млн сум",
marker_color=BRAND["gold"], text=[f"{fnum(v, 1)} млн" for v in avg],
textposition="outside", cliponaxis=False, textfont=dict(size=11))
f.add_scatter(x=labels, y=d["client_share"] * 100, name="Доля клиентов, %",
yaxis="y2", mode="lines+markers+text",
line=dict(color=BRAND["cool"], width=2.4), marker=dict(size=9),
text=[f"{v:.1%}" for v in d["client_share"]],
textposition="top center", textfont=dict(size=10, color=BRAND["cool"]))
f.add_scatter(x=labels, y=d["exposure_share"] * 100, name="Доля портфеля, %",
yaxis="y2", mode="lines+markers+text",
line=dict(color=BRAND["bad"], width=2.4), marker=dict(size=9),
text=[f"{v:.1%}" for v in d["exposure_share"]],
textposition="bottom center", textfont=dict(size=10, color=BRAND["bad"]))
for i, r in enumerate(d.itertuples()):
f.add_annotation(x=labels[i], y=0, yshift=-38, showarrow=False,
text=f"{int(r.n_clients):,} клиентов".replace(",", " "),
font=dict(size=10, color=BRAND["muted"]))
f.update_layout(title=f"B11 · Портрет клиента по числу кредитов, "
f"{plabel_full(ctx['sel'])}",
height=520, xaxis=dict(title=""),
yaxis=dict(title="средний долг клиента, млн сум"),
yaxis2=dict(overlaying="y", side="right", title="%", showgrid=False,
range=[0, 100]))
unit_note(f, "средний долг — млн сум на одного клиента", extra_bottom=.04)
kind_tag(f, "СРЕЗ")
multi = d[d["burden"].astype(str) != "1"]
return f, (f"Столбец отвечает на вопрос «сколько должен такой клиент». Клиенты "
f"с несколькими кредитами — {multi.client_share.sum():.1%} базы, но "
f"{multi.exposure_share.sum():.1%} портфеля: их мало, а денег на них "
"много. Расхождение синей и красной линий и есть кредитная нагрузка.")
def b12_oked(A, B, ctx):
d = B["oked_name"].head(TOP_OKED).sort_values("exposure")
f = go.Figure()
f.add_bar(x=money(d.exposure), y=d["name"], orientation="h",
marker=dict(color=d.npl_ratio * 100,
colorscale=[[0, BRAND["good"]], [.4, BRAND["gold"]],
[1, BRAND["bad"]]],
colorbar=dict(title="NPL, %", thickness=14)),
text=[f"{fnum(v / UNIT, 1)} · {sh:.1%} · NPL {n:.1f}%"
for v, sh, n in zip(d.exposure, d.share, d.npl_ratio * 100)],
textposition="outside", cliponaxis=False, textfont=dict(size=9.5),
customdata=np.stack([d["n_loans"], d["n_clients"]], axis=-1),
hovertemplate="<b>%{y}</b><br>%{x:,.1f} " + UNIT_NAME +
"<br>кредитов %{customdata[0]:,}"
"<br>клиентов %{customdata[1]:,}<extra></extra>")
f.update_layout(title=f"B12 · Топ-{TOP_OKED} отраслей по объёму (ОКЭД), "
f"{plabel_full(ctx['sel'])}",
height=max(480, 36 * len(d)), xaxis=dict(title=UNIT_NAME),
showlegend=False)
unit_note(f, f"{UNIT_NAME} · доля портфеля · NPL отрасли"); kind_tag(f, "СРЕЗ")
top3 = float(d.tail(3)["share"].sum())
return f, (f"Три крупнейшие отрасли — {top3:.0%} портфеля. Отраслевая концентрация "
"опаснее филиальной: филиалы независимы, а одна отрасль падает целиком "
"и сразу по всем филиалам.")
# ---------------------------------------------------------------- registry
CHARTS_A_BASE = [
("A01", "СТОК", "Портфель и качество по отчётным датам", a01_portfolio),
("A02", "ПОТОК", "Из чего складывается изменение портфеля", a02_stock_flow),
("A03", "СТОК", "Структура просрочки по месяцам", a03_bucket_mix),
("A04", "СТОК", "Доля NPL 90+ по департаментам", a04_npl_departments),
("A05", "СТОК", "Филиалы и месяцы — карта качества", a05_branch_heatmap),
("A06", "ПОТОК", "Почему изменился NPL", a06_bridge),
("A07", "КОГОРТЫ", "Качество выдач по месяцу выдачи", a07_vintage_quality),
("A08", "КОГОРТЫ", "Контрольные точки качества выдач", a08_vintage_heat),
("A09", "ПОТОК", "Скорость ухода в дефолт", a09_migration),
("A10", "СТОК", "Доля валют в портфеле", a10_currency),
("A11", "ПОТОК", "Выдачи за месяц по департаментам", a11_origination),
("A12", "СТОК", "Покрытие резервами и судебная работа", a12_coverage),
("A13", "СТОК", "Эффект страхового покрытия в динамике", a13_insurance),
]
CHARTS_B_BASE = [
("B01", "СРЕЗ", "Портфель по филиалам и департаментам", b01_dep_filial),
("B02", "СРЕЗ", "Портфель по корзинам просрочки", b02_buckets),
("B03", "СРЕЗ", "Из выдач каких месяцев состоит портфель", b03_vintage_mix),
("B04", "СРЕЗ", "Возраст кредитов в портфеле", b04_mob_profile),
("B05", "СРЕЗ", "Топ продуктов по объёму", b05_products),
("B06", "СРЕЗ", "Продукты по департаментам", b06_passport_dep),
("B07", "СРЕЗ", "Структура просрочки внутри филиала", b07_filial_bucket),
("B08", "СРЕЗ", "Концентрация портфеля по клиентам", b08_concentration),
("B09", "СРЕЗ", "Кто формирует общий NPL банка", b09_npl_contribution),
("B10", "СРЕЗ", "Автосегмент и судебная стадия", b10_auto_court),
("B11", "СРЕЗ", "Портрет клиента по числу кредитов", b11_client_profile),
]
def chart_registry(ctx):
"""
Build the chart list for this run. Per-department charts can only be known
once we have seen the data, so the list is assembled here rather than at
import time.
"""
a = list(CHARTS_A_BASE)
n = len(a)
for i, dep in enumerate(ctx.get("departments", [])):
cid = f"A{n + 1 + i:02d}"
a.append((cid, "СТОК", f"«{dep}»: портфель по продуктам",
_passport_chart(dep, cid)))
if ctx.get("has_oked"):
cid = f"A{len(a) + 1:02d}"
def oked_fn(A, B, c, _cid=cid):
f, note = a_oked_dynamics(A, B, c)
f.update_layout(title=f"{_cid} " + f.layout.title.text)
return f, note
a.append((cid, "СТОК", "Отраслевая структура портфеля (ОКЭД)", oked_fn))
b = list(CHARTS_B_BASE)
if ctx.get("has_oked"):
b.append(("B12", "СРЕЗ", "Отраслевая концентрация (ОКЭД)", b12_oked))
return a, b
# ==========================================================================
# HTML (notebook cell 12)
# ==========================================================================
CSS = """
:root{--ink:#0B0B0C;--soft:#2A2A2E;--gold:#F2C230;--gd:#C9971A;--rule:#E4E1D8;
--muted:#8A867C;--bad:#B3341F;--good:#2F6B4F}
*{box-sizing:border-box}
body{margin:0;background:#fff;color:var(--ink);font-family:Cambria,Georgia,serif;
font-size:15px;line-height:1.55}
.wrap{max-width:1240px;margin:0 auto;padding:0 32px}
.cover{background:var(--ink);color:#fff;padding:64px 0 50px}
.cover h1{font-weight:400;font-size:44px;margin:12px 0 8px}
.eyebrow{font-size:11px;letter-spacing:.24em;text-transform:uppercase;color:var(--gold)}
.meta{color:#B9B5AB;font-size:13px;margin-top:8px}
.kpis{display:grid;grid-template-columns:repeat(auto-fit,minmax(190px,1fr));gap:1px;
background:#26262A;margin-top:34px;border-top:2px solid var(--gold)}
.kpi{background:var(--ink);padding:18px 16px}
.kpi .k{font-size:10px;letter-spacing:.14em;text-transform:uppercase;color:#8E8A80}
.kpi .v{font-size:28px;margin-top:7px}
.kpi .d{font-size:12px;margin-top:3px;color:#8E8A80}
.dirhead{background:#FAF8F2;border-top:1px solid var(--rule);
border-bottom:1px solid var(--rule);padding:38px 0;margin-top:10px}
.dirhead .tag{font-size:11px;letter-spacing:.2em;text-transform:uppercase;color:var(--gd)}
.dirhead h2{font-weight:400;font-size:33px;margin:8px 0 10px}
.dirhead p{color:var(--soft);max-width:84ch;margin:0}
.chart{padding:40px 0 18px;border-top:1px solid var(--rule)}
.chart .cid{font-size:11px;letter-spacing:.16em;color:var(--gd);font-weight:600}
.chart h3{font-weight:400;font-size:21px;margin:4px 0 14px}
.note{border-left:2px solid var(--gold);padding:11px 0 11px 16px;margin:14px 0 0;
font-size:13.5px;color:var(--soft);background:#FCFAF4;max-width:96ch}
footer{background:var(--ink);color:#8E8A80;padding:32px 0;margin-top:52px;font-size:12px}
@media print{.cover,.dirhead{-webkit-print-color-adjust:exact;print-color-adjust:exact}
.chart{break-inside:avoid}}
"""
def _kpis(A, B):
t = A["total"]
cur = t.iloc[-1]
prev = t.iloc[-2] if len(t) > 1 else cur
return [
(f"ПОРТФЕЛЬ, {UNIT_NAME.upper()}", fnum(cur.exposure/UNIT, 1), "", False),
("ДОЛЯ NPL 90+", f"{cur.npl_ratio:.2%}",
f"{(cur.npl_ratio-prev.npl_ratio)*100:+.2f} п.п. за месяц",
cur.npl_ratio > prev.npl_ratio),
("ЗОНА РИСКА 31-90", f"{cur.watch_ratio:.2%}",
f"{(cur.watch_ratio-prev.watch_ratio)*100:+.2f} п.п.",
cur.watch_ratio > prev.watch_ratio),
("ПОКРЫТИЕ РЕЗЕРВАМИ", f"{cur.coverage:.0%}", "", False),
("КЛИЕНТОВ НА ДАТУ", f"{B['_n_clients']:,}".replace(",", " "),
f"кредитов {B['_n_loans']:,}".replace(",", " "), False),
("КРЕДИТОВ НА КЛИЕНТА",
f"{B['_n_loans']/max(B['_n_clients'],1):.2f}", "", False),
]
def build_html(A, B, ctx, out_path, progress=None):
first = [True]
def render(charts):
parts = []
for i, (cid, kind, title, fn) in enumerate(charts, 1):
if progress:
progress(f"график {cid}")
try:
fig, note = fn(A, B, ctx)
body = fig.to_html(full_html=False, config=PCFG,
include_plotlyjs="cdn" if first[0] else False)
first[0] = False
except Exception as e:
body, note = (f"<p class='note'>не построен: "
f"{type(e).__name__}: {e}</p>"), ""
nt = f"<div class='note'>{_html.escape(note)}</div>" if note else ""
parts.append(f"<div class='chart' id='{cid}'><div class='cid'>{cid}</div>"
f"<h3>{_html.escape(title)}</h3>{body}{nt}</div>")
return "".join(parts)
charts_a, charts_b = chart_registry(ctx)
sec_a, sec_b = render(charts_a), render(charts_b)
kpi = "".join(
f'<div class="kpi"><div class="k">{k}</div><div class="v">{v}</div>'
f'<div class="d" style="color:{"#E6A6A6" if w else "#9FD3B6"}">{d or " "}</div></div>'
for k, v, d, w in _kpis(A, B))
stamp = dt.datetime.now().strftime("%d.%m.%Y %H:%M")
periods = ctx["periods"]
doc = f"""<!doctype html><html lang="ru"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{_html.escape(TITLE)} — {plabel_full(ctx['sel'])}</title>
<style>{CSS}</style></head><body>
<div class="cover"><div class="wrap">
<div class="eyebrow">{_html.escape(BANK)} · аналитический отчёт</div>
<h1>{_html.escape(TITLE)}</h1>
<div class="meta">Динамика: {plabel_full(periods[0])} — {plabel_full(periods[-1])}
· Структура на: {plabel_full(ctx['sel'])}
· Сформировано: {stamp}</div>
<div class="kpis">{kpi}</div></div></div>
<div class="dirhead"><div class="wrap"><div class="tag">Направление A</div>
<h2>Динамика</h2><p>Период {plabel_full(periods[0])} — {plabel_full(periods[-1])},
{len(periods)} отчётных дат. Остаток портфеля — сток, по месяцам не складывается.
Выдачи и погашения — потоки, складываются.</p></div></div>
<div class="wrap">{sec_a}</div>
<div class="dirhead"><div class="wrap"><div class="tag">Направление B</div>
<h2>Структура на {plabel_full(ctx['sel'])}</h2><p>Одна отчётная дата. Число клиентов,
средние суммы и концентрация посчитаны исключительно по ней.</p></div></div>
<div class="wrap">{sec_b}</div>
<footer><div class="wrap">{_html.escape(TITLE)} · {plabel_full(ctx['sel'])}
· {len(charts_a)+len(charts_b)} графиков · {UNIT_NAME}</div></footer>
</body></html>"""
Path(out_path).write_text(doc, encoding="utf-8")
return out_path
# ==========================================================================
# ENTRY POINT — the only thing Django calls
# ==========================================================================
def run(data_dir, out_path, period_from=None, period_to=None,
selected_period=None, progress=None):
"""
Full pipeline. Returns a dict of facts the caller can store.
progress: optional callable(str) so the web layer can show a live status.
"""
say = progress or (lambda *_: None)
_ensure_template()
say("чтение файлов")
raw = read_months(data_dir, period_from, period_to)
has_oked = "oked_name" in raw.columns
say("построение панели")
panel = build_panel(raw)
periods = sorted(panel["period"].unique())
if not periods:
raise ValueError("нет данных за выбранный период")
sel = int(selected_period) if selected_period else int(periods[-1])
if sel not in periods:
raise ValueError(f"период {sel} отсутствует в выбранном диапазоне "
f"{periods[0]}–{periods[-1]}")
# Closed loans linger at zero balance: they inflate client counts and averages.
snap = panel[(panel["period"] == sel) & (panel["exposure"] > MIN_BALANCE)].copy()
dropped = int((panel["period"] == sel).sum()) - len(snap)
say("расчёт витрин")
A, B, top_pass = build_marts(panel, snap, periods, sel, has_oked)
ctx = {"sel": sel, "periods": periods, "has_oked": has_oked,
"departments": sorted(top_pass.keys()), "top_pass": top_pass}
n_charts = sum(len(x) for x in chart_registry(ctx))
say("сборка отчёта")
Path(out_path).parent.mkdir(parents=True, exist_ok=True)
build_html(A, B, ctx, out_path, progress=say)
cur = A["total"].iloc[-1]
return {
"html_path": str(out_path),
"selected_period": sel,
"period_from": int(periods[0]),
"period_to": int(periods[-1]),
"n_periods": len(periods),
"n_rows": int(len(panel)),
"n_loans": int(B["_n_loans"]),
"n_clients": int(B["_n_clients"]),
"exposure": float(B["_tot_exp"]),
"npl_ratio": float(cur.npl_ratio),
"dropped_zero_balance": dropped,
"n_charts": n_charts,
}
PO
powerty
Muallif
· Staff
Okt. 8, 2026
Okt. 8, 2026
22
Ko'rishlar
0
Yoqishlar
35m
O'qildi