""" nnlab / transductive - refine probabilities ALREADY in the prediction store with the unlabelled Amsterdam pool, the pixel coordinates and the n labelled shots. python tools/nnlab/transductive.py --base lgbm_upw [--base zs_xgb_dann,zs_lgbm] [--smoke] [--only lp] [--madrid-check] For every base B and protocol draw (n, rep) it writes: B__clamp shot rows -> one-hot of their known label, then the lab's neighbour smoothing (r=6, alpha=0.5) over ALL pixels, shots included as neighbours. The smoothing is already inside the prediction, so READ THE "rule" COLUMN of `lab.py score` for these methods: its "rule_smooth" column would smooth a second time. B__lp label propagation (Zhou et al. 2004): F <- a S F + (1-a) Y0, 20 power iterations from F = Y0 (-> (1-a)(I-aS)^-1 Y0), S = D^-1/2 A D^-1/2, A = spatial graph (lab.adjacency(pa, r=3)) UNION feature kNN graph (k=10, cosine, PCA-32 of lab.robustz(Xa)), self-loops removed, Y0 = base probabilities with shot rows = 10 x one-hot, a = 0.8. F row-normalised. B__lapshot LaplacianShot-style (Ziko et al. 2020): y <- softmax(log p + lam * G y), 10 iterations from y = p, lam = 1.0, shots clamped to one-hot every iteration. G = the same union graph, ROW-NORMALISED (D^-1 A, i.e. the mean over neighbours) - see DEVIATIONS. B__em label-shift EM (Saerens et al. 2002), training prior = Madrid prior, 50 iterations on the non-shot rows, estimated target prior shrunk 50% toward the Madrid prior, posteriors re-weighted with that prior. For a shot-independent base it is label-free and saved once (n=0). B__st (only B == lgbm_upw) one round of self-training: per class c, among non-shot pixels whose base argmax is c, the top 15% by base p(c) become pseudo-labels (weight 3.0); Madrid weight 1, shots weight 30, then class-balanced over the union; refit with winners.train_retrained's robust-z scaling and LightGBM parameters; predict all Amsterdam rows. ~1-2 min per draw, run LAST. The "rule" column is the one to read for __clamp; for the others both columns are legitimate (rule_smooth is a further, separate smoothing). DEVIATIONS (all fixed in advance, none chosen on Amsterdam): * lapshot uses the row-normalised graph. With the raw 0/1 union graph a pixel has ~22 neighbours on average (min 10; measured), so lam=1 puts a pairwise logit of up to ~22 against a unary of a few nats: the unary term is drowned and the result is a hard spatial majority vote. `--madrid-check` compares both on the Madrid pseudo-target. * Self-loops are dropped from the union graph (Zhou et al. set W_ii = 0). * LightGBM in __st uses n_jobs = NNLAB_THREADS instead of -1 (shared laptop). * __st reads the features from lab.load() (float32 parquet columns -> float64), not winners.load_data (which also loads Amsterdam age_class). Hard rules kept: Amsterdam labels only via lab.shot_labels(lab.shots_for(...)). """ from __future__ import annotations import argparse import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import lab # noqa: E402 import numpy as np # noqa: E402 from scipy import sparse # noqa: E402 W = lab.W T0 = time.time() # fixed knobs (spec) ---------------------------------------------------------- # CLAMP_R, CLAMP_ALPHA = lab.SMOOTH_R, lab.SMOOTH_ALPHA # 6, 0.5 GRAPH_R, KNN_K, PCA_K = 3, 10, 32 LP_ALPHA, LP_ITERS, LP_SEED_W = 0.8, 20, 10.0 LS_LAM, LS_ITERS = 1.0, 10 LS_NORM = "row" # see DEVIATIONS EM_ITERS, EM_SHRINK = 50, 0.5 ST_SHARE, ST_W, ST_SHOT_W = 0.15, 3.0, 30.0 FAST = ("clamp", "lp", "lapshot", "em") SHOT_FREE_BASES = {"zs_lgbm", "zs_xgb_dann", "zs_dann"} def log(msg): print(f"[{time.time() - T0:7.1f}s] {msg}", flush=True) def onehot(y): Y = np.zeros((len(y), 4)) Y[np.arange(len(y)), np.asarray(y) - 1] = 1.0 return Y def rownorm(P): P = np.clip(P, 1e-12, None) return P / P.sum(1, keepdims=True) # --------------------------------------------------------------------------- # # Graphs (draw independent) # --------------------------------------------------------------------------- # def knn_graph(Z, k=KNN_K): """Symmetric 0/1 kNN graph, cosine similarity (= euclidean on unit vectors).""" from scipy.spatial import cKDTree U = Z / np.clip(np.linalg.norm(Z, axis=1, keepdims=True), 1e-12, None) _, nb = cKDTree(U).query(U, k=k + 1) n = len(U) rows = np.repeat(np.arange(n), k) cols = nb[:, 1:].ravel() A = sparse.coo_matrix((np.ones(len(rows)), (rows, cols)), shape=(n, n)).tocsr() return A.maximum(A.T).tocsr() def union_graph(X, coords, tag=None): """Spatial r=3 graph UNION feature kNN graph on PCA-32 of robustz(X); no self-loops. X must be ONE city's (or one pseudo-city's) raw features.""" path = lab.CACHE / f"transd_graph_{tag}.npz" if tag else None if path is not None and path.exists(): return sparse.load_npz(path).tocsr() from sklearn.decomposition import PCA Z = PCA(PCA_K, random_state=0).fit_transform(lab.robustz(X)) A_sp = lab.adjacency(coords, GRAPH_R) A = A_sp.maximum(knn_graph(Z)).tocsr() A.setdiag(0) A.eliminate_zeros() A.data[:] = 1.0 if path is not None: sparse.save_npz(path, A) return A class Graph: def __init__(self, A, coords): self.A = A deg = np.asarray(A.sum(1)).ravel() self.deg = deg dinv = 1.0 / np.sqrt(np.clip(deg, 1e-12, None)) self.S = sparse.diags(dinv) @ A @ sparse.diags(dinv) # Zhou self.R = sparse.diags(1.0 / np.clip(deg, 1e-12, None)) @ A # mean over nbrs self.coords = coords # --------------------------------------------------------------------------- # # Refinements (all take base P (N,4), shot idx, shot labels) # --------------------------------------------------------------------------- # def r_clamp(P, shot, ys, g): Q = P.copy() Q[shot] = onehot(ys) return lab.smooth(Q, g.coords, r=CLAMP_R, alpha=CLAMP_ALPHA) def r_lp(P, shot, ys, g, a=LP_ALPHA, iters=LP_ITERS): Y0 = P.copy() Y0[shot] = LP_SEED_W * onehot(ys) F = Y0.copy() for _ in range(iters): F = a * (g.S @ F) + (1 - a) * Y0 return rownorm(F) def r_lapshot(P, shot, ys, g, lam=LS_LAM, iters=LS_ITERS, norm=LS_NORM): G = g.R if norm == "row" else g.A U = np.log(np.clip(P, 1e-6, None)) Ys = onehot(ys) Y = P.copy() Y[shot] = Ys for _ in range(iters): Y = lab.softmax(U + lam * (G @ Y)) Y[shot] = Ys return Y def em_prior(P, prior_s, iters=EM_ITERS): pi_s = np.asarray(prior_s, float) / np.sum(prior_s) pi = pi_s.copy() for _ in range(iters): Q = rownorm(P * (pi / pi_s)) pi = Q.mean(0) return pi def r_em(P, shot, prior_s): pi_s = np.asarray(prior_s, float) / np.sum(prior_s) keep = np.setdiff1d(np.arange(len(P)), shot) if shot is not None else slice(None) pi_t = em_prior(P[keep], pi_s) pi_hat = (1 - EM_SHRINK) * pi_t + EM_SHRINK * pi_s return rownorm(P * (pi_hat / pi_s)), pi_t, pi_hat def pseudo_labels(P, shot, share=ST_SHARE): """Per class c: among non-shot rows with argmax c, the top `share` by p(c).""" free = np.ones(len(P), bool) free[shot] = False am = P.argmax(1) idx, lab_ = [], [] for k in range(4): cand = np.flatnonzero(free & (am == k)) m = int(round(share * len(cand))) if m == 0: continue top = cand[np.argsort(-P[cand, k], kind="stable")[:m]] idx.append(top) lab_.append(np.full(m, k + 1)) return np.concatenate(idx), np.concatenate(lab_) def r_st(P, shot, ys, Xm, ym, Xa, params): pidx, plab = pseudo_labels(P, shot) ms, ams = W.robustz_fit(Xm), W.robustz_fit(Xa) Za = W.robustz(Xa, ams) X = np.vstack([W.robustz(Xm, ms), Za[shot], Za[pidx]]) y = np.r_[ym, ys, plab] w = np.r_[np.ones(len(ym)), np.full(len(ys), ST_SHOT_W), np.full(len(plab), ST_W)] w = w * W.balanced_class_weights(y)[y - 1] m = W._fit_lgbm(X, y, params, sample_weight=w) del X return W.predict(m, Za), np.bincount(plab, minlength=5)[1:] # --------------------------------------------------------------------------- # # Madrid validation (pseudo-target, Madrid labels only) # --------------------------------------------------------------------------- # def madrid_check(d, smoke): """Train a retrained-style LightGBM on the Madrid source split, predict the spatial pseudo-target, draw pseudo-shots there, apply every refinement and print the rule macro F1 on the pseudo-target's non-shot rows.""" src, tgt = lab.madrid_split(d["pm"]) Xm = d["Xm"].astype(np.float64) ym = d["ym"] tag = "SMOKE" if smoke else "full" cp = lab.CACHE / f"transd_madrid_base_{tag}.npy" params = {**W.LGBM_PARAMS, **W.RETRAIN_OVERRIDES, "n_jobs": lab.THREADS} if smoke: params.update(n_estimators=40) if cp.exists(): Pt = np.load(cp) else: ss, ts = W.robustz_fit(Xm[src]), W.robustz_fit(Xm[tgt]) cw = W.balanced_class_weights(ym[src]) m = W._fit_lgbm(W.robustz(Xm[src], ss), ym[src], params, sample_weight=cw[ym[src] - 1]) Pt = W.predict(m, W.robustz(Xm[tgt], ts)) np.save(cp, Pt) log(f"madrid-check: base model fitted on {len(src)} src rows") yt, pt = ym[tgt], d["pm"][tgt] prior = np.array([(ym[src] == c).mean() for c in lab.CLASSES]) g = Graph(union_graph(Xm[tgt], pt, tag=f"madrid_tgt_{len(tgt)}"), pt) log(f"madrid-check: pseudo-target {len(tgt)} rows, mean degree {g.deg.mean():.1f}") variants = { "base": lambda P, s, y: P, "clamp": lambda P, s, y: r_clamp(P, s, y, g), "lp": lambda P, s, y: r_lp(P, s, y, g), "lapshot_row": lambda P, s, y: r_lapshot(P, s, y, g, norm="row"), "lapshot_raw": lambda P, s, y: r_lapshot(P, s, y, g, norm="raw"), "em": lambda P, s, y: r_em(P, s, prior)[0], } res = {k: {} for k in variants} for n in ((5, 200) if smoke else lab.N_LIST): for rep in range(1 if smoke else 3): rng = np.random.default_rng(7 + 1000 * n + rep) shot = np.concatenate([rng.choice(np.flatnonzero(yt == c), n, replace=False) for c in lab.CLASSES]) keep = np.setdiff1d(np.arange(len(yt)), shot) for k, f in variants.items(): Q = f(Pt, shot, yt[shot]) res[k].setdefault(n, []).append(W.macro_f1(yt[keep], W.decide(Q[keep], prior))) log("madrid-check: rule macro F1 on pseudo-target (non-shot rows), mean over reps") out = {} for k, r in res.items(): m = {n: float(np.mean(v)) for n, v in r.items()} out[k] = m print(f" {k:<12s} " + " ".join(f"n{n}={v:.4f}" for n, v in m.items()) + f" avg={np.mean(list(m.values())):.4f}", flush=True) return out # --------------------------------------------------------------------------- # def main(): ap = argparse.ArgumentParser() ap.add_argument("--base", action="append", default=None, help="base method(s); repeatable or comma-separated") ap.add_argument("--smoke", action="store_true") ap.add_argument("--only", default=None, help="substring filter on method names") ap.add_argument("--madrid-check", action="store_true", help="validate the refinements on the Madrid pseudo-target and exit") a = ap.parse_args() bases = [b for s in (a.base or ["lgbm_upw"]) for b in s.split(",") if b] sfx = "_SMOKE" if a.smoke else "" draws = [(5, 0), (200, 0)] if a.smoke else list(lab.draws()) d = lab.load(official60=False) log(f"loaded: Madrid {d['Xm'].shape}, Amsterdam {d['Xa'].shape}") if a.madrid_check: madrid_check(d, a.smoke) return g = Graph(union_graph(d["Xa"], d["pa"], tag=f"ams_r{GRAPH_R}_k{KNN_K}_pca{PCA_K}"), d["pa"]) log(f"graph: {g.A.nnz} edges, mean degree {g.deg.mean():.1f}, min {g.deg.min():.0f}") knobs = dict(clamp=dict(r=CLAMP_R, alpha=CLAMP_ALPHA), graph=dict(spatial_r=GRAPH_R, knn_k=KNN_K, knn_metric="cosine", pca=PCA_K, on="lab.robustz(Xa)", self_loops=False, weights="0/1 union"), lp=dict(a=LP_ALPHA, iters=LP_ITERS, seed_weight=LP_SEED_W, S="D^-1/2 A D^-1/2"), lapshot=dict(lam=LS_LAM, iters=LS_ITERS, graph_norm=LS_NORM, clamp=True), em=dict(iters=EM_ITERS, shrink_to_madrid_prior=EM_SHRINK, train_prior="Madrid"), st=dict(share=ST_SHARE, pseudo_w=ST_W, shot_w=ST_SHOT_W, madrid_w=1.0, class_balanced=True, params="winners LGBM_PARAMS+RETRAIN_OVERRIDES")) choice_note = ("all knobs fixed in advance by the task spec; lapshot row-normalisation " "fixed a priori (raw 0/1 graph with lam=1 drowns the unary term); " "checked on Madrid pseudo-target via --madrid-check, never on Amsterdam") def want(name): return a.only is None or a.only in name def base_pred(B, n, rep): """Base probabilities for a draw. In smoke mode a missing per-draw base falls back to zs_xgb_dann (lgbm_upw is still being written).""" if lab.have_pred(B, n, rep) or lab.have_pred(B, 0, 0): return lab.load_pred(B, n, rep), B if a.smoke: return lab.load_pred("zs_xgb_dann", 0, 0), "zs_xgb_dann" return None, None # ---- fast methods ---------------------------------------------------- # for B in bases: for kind in FAST: name = f"{B}__{kind}{sfx}" if not want(name): continue zero = kind == "em" and B in SHOT_FREE_BASES for n, rep in ([(0, 0)] if zero else draws): if lab.have_pred(name, n, rep): continue t = time.time() P, used = base_pred(B, max(n, 0), rep) if not zero else (lab.load_pred(B, 0, 0), B) if P is None: log(f"{name} n={n} rep={rep}: base {B} missing, skipped") continue meta = dict(base=B, base_used=used, kind=kind, knobs=knobs[kind] if kind in knobs else None, graph=knobs["graph"] if kind in ("lp", "lapshot") else None, choices=choice_note, read_column="rule" if kind == "clamp" else "rule or rule_smooth") if zero: Q, pi_t, pi_hat = r_em(P, None, d["prior"]) meta.update(pi_t=pi_t.tolist(), pi_hat=pi_hat.tolist()) else: shot = lab.shots_for(d, n, rep) ys = lab.shot_labels(d, shot) if kind == "clamp": Q = r_clamp(P, shot, ys, g) elif kind == "lp": Q = r_lp(P, shot, ys, g) elif kind == "lapshot": Q = r_lapshot(P, shot, ys, g) else: Q, pi_t, pi_hat = r_em(P, shot, d["prior"]) meta.update(pi_t=pi_t.tolist(), pi_hat=pi_hat.tolist()) lab.save_pred(name, n, rep, Q, meta) log(f"{name} n={n} rep={rep} base={used} done in {time.time() - t:.1f}s") # ---- self-training (last, slow) -------------------------------------- # st_bases = [B for B in bases if B == "lgbm_upw" or B in __import__("os").environ.get("NNLAB_ST_BASES", "").split(",")] # pseudo-labels may come from any stored method if not st_bases and a.smoke: st_bases = [bases[0]] # exercise the path in smoke mode params = {**W.LGBM_PARAMS, **W.RETRAIN_OVERRIDES, "n_jobs": lab.THREADS} if a.smoke: params.update(n_estimators=40) Xm = Xa = None for B in st_bases: name = f"{B}__st{sfx}" if not want(name): continue for n, rep in draws: if lab.have_pred(name, n, rep): continue P, used = base_pred(B, n, rep) if P is None: log(f"{name} n={n} rep={rep}: base {B} missing, skipped") continue if Xm is None: Xm, Xa = d["Xm"].astype(np.float64), d["Xa"].astype(np.float64) t = time.time() shot = lab.shots_for(d, n, rep) ys = lab.shot_labels(d, shot) Q, counts = r_st(P, shot, ys, Xm, d["ym"], Xa, params) meta = dict(base=B, base_used=used, kind="st", knobs=knobs["st"], n_estimators=params["n_estimators"], n_jobs=params["n_jobs"], choices=choice_note, pseudo_counts_last=counts.tolist()) lab.save_pred(name, n, rep, Q, meta) log(f"{name} n={n} rep={rep} base={used} pseudo={counts.tolist()} " f"done in {time.time() - t:.1f}s") log("done") if __name__ == "__main__": main()