forked from Dan4ick/Lidar_Muxa
212 lines
12 KiB
Python
212 lines
12 KiB
Python
"""Обучение считывания MBON с учителем и честная проверка по бэгам.
|
||
|
||
Метки изготовлены физикой (`tools/make_training_set.py`), поэтому впервые можно
|
||
обучать не «частоту обстановки», а прямо различение «предмет / тоннель». Слои
|
||
при этом те же: разрежённый код клеток Кеньона, торможение APL, один выход
|
||
MBON — меняется только учитель (см. `flyguard/mbon_readout.py`).
|
||
|
||
Проверка — leave-one-bag-out: модель обучается на всех бэгах, кроме
|
||
проверяемого. Рядом считаются три вещи, без которых цифрам верить нельзя:
|
||
|
||
* **развёртка по ёмкости** — сколько клеток Кеньона реально нужно. Отбор
|
||
«победитель забирает всё» стоит дороже матмула, и в худшем кадре (64
|
||
кандидата) 20 000 клеток это 7.3 мс из бюджета в 100 мс, а 8 000 — 2.8 мс.
|
||
Платить за ёмкость имеет смысл, только если она что-то даёт;
|
||
* **важность признаков перестановкой** — если наверх вылезли `lat`/`abs_lat`,
|
||
модель выучила «что у оси, то предмет». В тоннеле у оси полно штатных
|
||
конструкций, и именно они дают нам ложные тревоги, так что такая модель
|
||
сделала бы хуже, показывая красивые цифры;
|
||
* **контрольный градиентный бустинг** по сырым признакам — сколько качества
|
||
стоит сам разрежённый код.
|
||
|
||
python tools/train_mbon.py --device cuda --baseline
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import time
|
||
|
||
import numpy as np
|
||
|
||
import _bootstrap as B # noqa: F401
|
||
from _metrics import auc, fpr_at_tpr
|
||
from flyguard.mbon_readout import MbonConfig, MbonReadout
|
||
|
||
BANDS = ((0, 30), (30, 55), (55, 80), (80, 110), (110, 160), (160, 230))
|
||
|
||
|
||
def loo(names, X, Y, cfg, n_pn, args, keep_models=False):
|
||
"""Обучение на всех бэгах кроме проверяемого. Возвращает список результатов."""
|
||
out = []
|
||
for held in names:
|
||
tr = [n for n in names if n != held]
|
||
Xtr = np.concatenate([X[n] for n in tr])
|
||
ytr = np.concatenate([Y[n] for n in tr])
|
||
m = MbonReadout(cfg, n_pn=n_pn)
|
||
m.fit_normalizer(Xtr)
|
||
m.learn(Xtr, ytr, epochs=args.epochs, lr=args.lr, l2=args.l2,
|
||
device=args.device)
|
||
s = m.score(X[held])
|
||
out.append((held, s, m if keep_models else None, Xtr, ytr))
|
||
return out
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser(description=__doc__)
|
||
ap.add_argument("--data", default=str(B.CACHE / "training_set.npz"))
|
||
ap.add_argument("--out", default=str(B.ARTIFACTS / "mbon_readout.npz"))
|
||
ap.add_argument("--n-kc", type=int, default=0,
|
||
help="ёмкость итоговой модели; 0 — взять лучшую из развёртки")
|
||
ap.add_argument("--sweep-kc", default="4000,8000,20000",
|
||
help="ёмкости для развёртки; пусто — не разворачивать")
|
||
ap.add_argument("--active", type=int, default=100,
|
||
help="активных клеток после торможения APL")
|
||
ap.add_argument("--epochs", type=int, default=60)
|
||
ap.add_argument("--lr", type=float, default=4.0)
|
||
ap.add_argument("--l2", type=float, default=1e-5)
|
||
ap.add_argument("--device", default="cpu")
|
||
ap.add_argument("--baseline", action="store_true",
|
||
help="сравнить с градиентным бустингом по сырым признакам")
|
||
ap.add_argument("--save-folds", default=str(B.ARTIFACTS / "mbon_folds"),
|
||
help="куда сложить модели по складкам: без них сквозная "
|
||
"оценка нечестна — считывание увидит проверяемый бэг")
|
||
args = ap.parse_args()
|
||
|
||
d = np.load(args.data, allow_pickle=True)
|
||
names = [str(n) for n in d["names"]]
|
||
X = {n: d[f"X_{n}"].astype(np.float32) for n in names}
|
||
Y = {n: d[f"y_{n}"].astype(np.int8) for n in names}
|
||
D = {n: d[f"d_{n}"].astype(np.float32) for n in names}
|
||
feats = [str(f) for f in d["features"]]
|
||
n_pn = next(iter(X.values())).shape[1]
|
||
tot = sum(v.shape[0] for v in X.values())
|
||
pos = sum(int(v.sum()) for v in Y.values())
|
||
print(f"выборка: {tot} кандидатов, предметов {pos} ({pos / tot:.1%}), "
|
||
f"признаков {n_pn}")
|
||
for n in names:
|
||
print(f" {n:<42}{X[n].shape[0]:7d} предметов {int(Y[n].sum()):6d} "
|
||
f"({Y[n].mean():5.1%})")
|
||
|
||
# --------------------------------------------- что можно выучить по одному
|
||
# признаку. Эта таблица нужна не для качества, а для ловли артефактов
|
||
# вставки: если один признак сам по себе почти идеально делит предмет и
|
||
# обстановку — особенно с обратным знаком на дальности, — значит модель
|
||
# сенсора врёт, а не детектор хорош. Так была найдена интенсивность
|
||
# (EXPERIMENTS п. 11.1).
|
||
Xa = np.concatenate([X[n] for n in names])
|
||
ya = np.concatenate([Y[n] for n in names])
|
||
Da = np.concatenate([D[n] for n in names])
|
||
single = np.array([auc(Xa[:, j], ya) for j in range(n_pn)])
|
||
print()
|
||
print("что делит выборку само по себе (AUC одного признака):")
|
||
for j in np.argsort(-np.abs(single - 0.5))[:8]:
|
||
print(f" {feats[j]:<14}{single[j]:.3f}")
|
||
worst = []
|
||
for j in range(n_pn):
|
||
for lo, hi in BANDS:
|
||
m = (Da >= lo) & (Da < hi)
|
||
if (ya[m] == 1).sum() < 20:
|
||
continue
|
||
a = auc(Xa[m, j], ya[m])
|
||
if a > 0.97 or a < 0.03:
|
||
worst.append((feats[j], lo, hi, a))
|
||
if worst:
|
||
print(" ВНИМАНИЕ: один признак почти полностью решает задачу — "
|
||
"похоже на артефакт вставки, а не на сигнал:")
|
||
for f, lo, hi, a in worst[:8]:
|
||
print(f" {f:<14}{lo:4d}-{hi:<4d} м AUC {a:.3f}")
|
||
|
||
# ------------------------------------------------------ развёртка по ёмкости
|
||
best_kc = args.n_kc
|
||
if args.sweep_kc.strip():
|
||
print("\nразвёртка по ёмкости (leave-one-bag-out):")
|
||
print(f"{'клеток':>8}{'активных':>10}{'AUC':>9}"
|
||
f"{'обстановки при 95% предметов':>31}")
|
||
scores = {}
|
||
for kc in (int(x) for x in args.sweep_kc.split(",")):
|
||
cfg = MbonConfig(n_kc=kc, sparsity=min(args.active / kc, 1.0))
|
||
res = loo(names, X, Y, cfg, n_pn, args)
|
||
a = float(np.nanmean([auc(s, Y[h]) for h, s, *_ in res]))
|
||
f = float(np.nanmean([fpr_at_tpr(s, Y[h], 0.95) for h, s, *_ in res]))
|
||
scores[kc] = a
|
||
print(f"{kc:8d}{min(args.active, kc):10d}{a:9.4f}{f:30.2%}", flush=True)
|
||
if not best_kc:
|
||
# берём наименьшую ёмкость, отстающую от лучшей не более чем на
|
||
# 0.002 AUC: дальше платим временем кадра за шум
|
||
top = max(scores.values())
|
||
best_kc = min(k for k, v in scores.items() if v >= top - 0.002)
|
||
print(f"выбрана ёмкость {best_kc} (лучшая AUC {top:.4f})")
|
||
if not best_kc:
|
||
best_kc = MbonConfig().n_kc
|
||
|
||
# ------------------------------------------------------------- подробно
|
||
cfg = MbonConfig(n_kc=best_kc, sparsity=min(args.active / best_kc, 1.0))
|
||
print(f"\nподробно при {best_kc} клетках:")
|
||
res = loo(names, X, Y, cfg, n_pn, args, keep_models=True)
|
||
for held, s, m, Xtr, ytr in res:
|
||
line = (f"{held:<42} AUC {auc(s, Y[held]):.4f} "
|
||
f"обстановки при 95% предметов {fpr_at_tpr(s, Y[held], 0.95):6.2%}")
|
||
if args.baseline:
|
||
import lightgbm as lgb
|
||
g = lgb.LGBMClassifier(n_estimators=400, learning_rate=0.05,
|
||
num_leaves=63, verbose=-1)
|
||
g.fit(Xtr, ytr)
|
||
line += f" | бустинг AUC {auc(g.predict_proba(X[held])[:, 1], Y[held]):.4f}"
|
||
print(line, flush=True)
|
||
|
||
if args.save_folds:
|
||
# По одной модели на складку. `evaluate.py --mbon-dir` берёт ту, что
|
||
# НЕ видела проверяемый бэг: иначе сквозные цифры ложные, ровно как
|
||
# было бы с памятью тоннеля, обученной на всём подряд.
|
||
from pathlib import Path
|
||
out_dir = Path(args.save_folds)
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
for held, _, mm, *_ in res:
|
||
mm.save(out_dir / f"mbon_{held}.npz")
|
||
print(f"модели по складкам сохранены: {out_dir} ({len(res)} шт.)")
|
||
|
||
print(f"\nсредний AUC: {np.nanmean([auc(s, Y[h]) for h, s, *_ in res]):.4f}")
|
||
print("AUC по полосам дальности:")
|
||
print(" " + "".join(f"{a}-{b} м".rjust(12) for a, b in BANDS))
|
||
line = " "
|
||
for lo, hi in BANDS:
|
||
ss = np.concatenate([s[(D[h] >= lo) & (D[h] < hi)] for h, s, *_ in res])
|
||
yy = np.concatenate([Y[h][(D[h] >= lo) & (D[h] < hi)] for h, s, *_ in res])
|
||
line += f"{auc(ss, yy):12.3f}" if (yy == 1).sum() >= 20 else " -"
|
||
print(line)
|
||
|
||
print("\nважность признаков (падение AUC при перестановке):")
|
||
rng = np.random.default_rng(0)
|
||
drops = []
|
||
for held, s, m, *_ in res:
|
||
base = auc(s, Y[held])
|
||
row = []
|
||
for j in range(n_pn):
|
||
Xp = X[held].copy()
|
||
Xp[:, j] = rng.permutation(Xp[:, j])
|
||
row.append(base - auc(m.score(Xp), Y[held]))
|
||
drops.append(row)
|
||
imp = np.mean(drops, axis=0)
|
||
for j in np.argsort(-imp)[:10]:
|
||
print(f" {feats[j]:<14}{imp[j]:+.4f}")
|
||
lat_j = [j for j, f in enumerate(feats) if f in ("lat", "abs_lat")]
|
||
if lat_j and max(imp[j] for j in lat_j) >= sorted(imp)[-3]:
|
||
print(" ВНИМАНИЕ: положение в сечении среди главных признаков — "
|
||
"модель могла выучить «что у оси, то предмет»")
|
||
|
||
# -------------------------------------------------------------- итоговая
|
||
Xall = np.concatenate([X[n] for n in names])
|
||
yall = np.concatenate([Y[n] for n in names])
|
||
m = MbonReadout(cfg, n_pn=n_pn)
|
||
m.fit_normalizer(Xall)
|
||
t0 = time.time()
|
||
m.learn(Xall, yall, epochs=args.epochs, lr=args.lr, l2=args.l2,
|
||
device=args.device, verbose=True)
|
||
m.save(args.out)
|
||
print(f"\nитоговая модель: {best_kc} клеток, {m.n_active} активных, "
|
||
f"обучена на {Xall.shape[0]} примерах за {time.time() - t0:.1f} с")
|
||
print("сохранено:", args.out)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|