""" classical.py - non-neural, non-boosting models that use the shots. python tools/nnlab/classical.py [--smoke] [--only ] Methods: cls_lr cls_lr_feda cls_svm cls_svm_feda cls_nys_lr cls_knn cls_lda (+ zero-shot cls_lr_zs cls_svm_zs saved once at n=0, rep=0) All hyper-parameters are fixed in advance (nothing is chosen on Amsterdam, nothing was tuned at all). Inputs: lab.robustz per city; PCA(64) fitted on Madrid+Amsterdam features together (label-free); Nystroem fitted on Madrid+Amsterdam (label-free). Amsterdam labels are read only via lab.shot_labels(lab.shots_for(...)). """ from __future__ import annotations import argparse import sys import time import warnings from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import lab # noqa: E402 import numpy as np # noqa: E402 warnings.filterwarnings("ignore") T0 = time.time() SHOT_W = 30.0 KNN_K = 25 ALL = ["cls_lr_zs", "cls_svm_zs", "cls_lr", "cls_lr_feda", "cls_svm", "cls_svm_feda", "cls_nys_lr", "cls_knn", "cls_lda"] def log(msg): print(f"[{time.time() - T0:7.1f}s] {msg}", flush=True) ap = argparse.ArgumentParser() ap.add_argument("--smoke", action="store_true") ap.add_argument("--only", default=None) args = ap.parse_args() SMOKE = args.smoke SFX = "_SMOKE" if SMOKE else "" # configuration (smoke = tiny) ------------------------------------------------ # LR_C = 1.0 LR_ITER = 15 if SMOKE else 100 SVM_SUB = 1500 if SMOKE else 6000 # class-balanced Madrid subsample for SVC SVM_C = 3.0 NYS_COMP = 60 if SMOKE else 500 PCA_K = 64 SEED = 0 CFG = dict(shot_weight=SHOT_W, robustz_clip=10.0, pca=PCA_K, lr_C=LR_C, lr_max_iter=LR_ITER, svm_sub=SVM_SUB, svm_C=SVM_C, svm_gamma="scale", nystroem=NYS_COMP, nys_gamma="1/(393*var(Zm))", knn_k=KNN_K, lda="lsqr auto shrinkage, shots duplicated 30x", madrid_choices="none - everything fixed in advance", smoke=SMOKE) # data ------------------------------------------------------------------------ # d = lab.load(official60=False) ym, pm = d["ym"], d["pm"] Zm = lab.robustz(d["Xm"]) Za = lab.robustz(d["Xa"]) nM, nA = len(Zm), len(Za) log(f"loaded Madrid {Zm.shape} Amsterdam {Za.shape}") def cached(name, fn): p = lab.CACHE / f"cls_{name}.npz" if p.exists(): z = np.load(p) return tuple(z[k] for k in z.files) out = fn() np.savez(p, *out) return out def _pca(): from sklearn.decomposition import PCA pca = PCA(PCA_K, svd_solver="randomized", random_state=SEED).fit(np.vstack([Zm, Za])) return pca.transform(Zm).astype(np.float32), pca.transform(Za).astype(np.float32) _pcs = {} def pca64(): if not _pcs: _pcs["v"] = cached("pca64", _pca) return _pcs["v"] def nys(): def f(): from sklearn.kernel_approximation import Nystroem gamma = 1.0 / (Zm.shape[1] * float(Zm.var())) ny = Nystroem(gamma=gamma, n_components=NYS_COMP, random_state=SEED).fit(np.vstack([Zm, Za])) return ny.transform(Zm).astype(np.float32), ny.transform(Za).astype(np.float32) return cached(f"nys{NYS_COMP}", f) def feda(X, is_madrid): z = np.zeros_like(X) return np.hstack([X, X, z]) if is_madrid else np.hstack([X, z, X]) def weights(y_all, n_src): w = np.r_[np.ones(n_src), np.full(len(y_all) - n_src, SHOT_W)] return w * lab.balanced_class_weights(y_all)[y_all - 1] def lr_fit(X, y, w): from sklearn.linear_model import LogisticRegression return LogisticRegression(C=LR_C, max_iter=LR_ITER, tol=1e-3).fit(X, y, sample_weight=w) # SVC Madrid subsample (fixed seed, class balanced) --------------------------- # def svm_subsample(): rng = np.random.default_rng(SEED) q = SVM_SUB // 4 idx = [] for c in lab.CLASSES: ii = np.flatnonzero(ym == c) idx.append(rng.choice(ii, min(q, len(ii)), replace=False)) return np.sort(np.concatenate(idx)) def svm_fit_predict(Xs, ys, Xt, shots_X=None, ys_shots=None): from sklearn.svm import SVC if shots_X is not None: X = np.vstack([Xs, shots_X]) y = np.r_[ys, ys_shots] w = weights(y, len(ys)) else: X, y = Xs, ys w = lab.balanced_class_weights(y)[y - 1] m = SVC(C=SVM_C, kernel="rbf", gamma="scale", decision_function_shape="ovr", cache_size=1000) m.fit(X, y, sample_weight=w) return lab.softmax(m.decision_function(Xt)) # unit-of-work functions ------------------------------------------------------ # # each: f(idx, ys) -> P (nA, 4) ; idx/ys None for zero-shot def f_lr(idx, ys): if idx is None: m = lr_fit(Zm, ym, lab.balanced_class_weights(ym)[ym - 1]) else: y = np.r_[ym, ys] m = lr_fit(np.vstack([Zm, Za[idx]]), y, weights(y, nM)) return m.predict_proba(Za) def f_lr_feda(idx, ys): y = np.r_[ym, ys] X = np.vstack([feda(Zm, True), feda(Za[idx], False)]) m = lr_fit(X, y, weights(y, nM)) return m.predict_proba(feda(Za, False)) def f_svm(idx, ys): Pm, Pa = pca64() s = svm_subsample() if idx is None: return svm_fit_predict(Pm[s], ym[s], Pa) return svm_fit_predict(Pm[s], ym[s], Pa, Pa[idx], ys) def f_svm_feda(idx, ys): Pm, Pa = pca64() s = svm_subsample() return svm_fit_predict(feda(Pm[s], True), ym[s], feda(Pa, False), feda(Pa[idx], False), ys) def f_nys_lr(idx, ys): Nm, Na = nys() y = np.r_[ym, ys] m = lr_fit(np.vstack([Nm, Na[idx]]), y, weights(y, nM)) return m.predict_proba(Na) def _knn_madrid(): def f(): from sklearn.neighbors import NearestNeighbors Pm, Pa = pca64() nn = NearestNeighbors(n_neighbors=KNN_K, algorithm="brute", n_jobs=lab.THREADS).fit(Pm) dist, ind = nn.kneighbors(Pa) return dist.astype(np.float32), ind.astype(np.int32) return cached("knn_madrid", f) def f_knn(idx, ys): # Madrid neighbours are shot-independent (cached); merge with distances to the shots, # keep the 25 closest of the union, vote with weight (1 or 30) / distance. Pm, Pa = pca64() dm, im = _knn_madrid() S = Pa[idx] ds = np.sqrt(np.maximum(((Pa ** 2).sum(1)[:, None] + (S ** 2).sum(1)[None, :] - 2.0 * Pa @ S.T), 0.0)).astype(np.float32) D = np.hstack([dm, ds]) lbl = np.hstack([ym[im], np.broadcast_to(ys, (nA, len(ys)))]) wt = np.hstack([np.ones_like(dm), np.full_like(ds, SHOT_W)]) o = np.argpartition(D, KNN_K - 1, axis=1)[:, :KNN_K] r = np.arange(nA)[:, None] D, lbl, wt = D[r, o], lbl[r, o], wt[r, o] v = wt / np.maximum(D, 1e-6) P = np.zeros((nA, 4)) for c in range(4): P[:, c] = (v * (lbl == c + 1)).sum(1) P += 1e-3 * P.sum(1, keepdims=True) / 4 return P / P.sum(1, keepdims=True) def f_lda(idx, ys): from sklearn.discriminant_analysis import LinearDiscriminantAnalysis Pm, Pa = pca64() X = np.vstack([Pm] + [Pa[idx]] * int(SHOT_W)) y = np.r_[ym, np.tile(ys, int(SHOT_W))] m = LinearDiscriminantAnalysis(solver="lsqr", shrinkage="auto").fit(X, y) return m.predict_proba(Pa) METHODS = { "cls_lr_zs": (f_lr, True), "cls_svm_zs": (f_svm, True), "cls_lr": (f_lr, False), "cls_lr_feda": (f_lr_feda, False), "cls_svm": (f_svm, False), "cls_svm_feda": (f_svm_feda, False), "cls_nys_lr": (f_nys_lr, False), "cls_knn": (f_knn, False), "cls_lda": (f_lda, False), } # main loop ------------------------------------------------------------------- # todo = [m for m in ALL if args.only is None or args.only in m] draw_list = [(5, 0), (200, 0)] if SMOKE else list(lab.draws()) for name in todo: fn, zs = METHODS[name] out_name = name + SFX units = [(0, 0)] if zs else draw_list for n, rep in units: if lab.have_pred(out_name, n, rep): log(f"skip {out_name} n={n} r={rep}") continue t = time.time() if zs: P = fn(None, None) else: idx = lab.shots_for(d, n, rep) ys = lab.shot_labels(d, idx) P = fn(idx, ys) lab.save_pred(out_name, n, rep, P, meta=dict(CFG, method=out_name)) log(f"{out_name} n={n} r={rep} done in {time.time() - t:.1f}s") log("all done")