|
#!/usr/bin/env python3 -B |
|
# nuff.py: tiny python tricks in one file -- records, io, format, |
|
# rand, stats, columns, distance. (c) 2026 Tim Menzies, MIT. |
|
# No global config: params (p, cliff, conf, rng) as kwargs. |
|
import re, sys, random, traceback |
|
from math import log2, log, exp, sqrt, pi |
|
from bisect import bisect_left, bisect_right |
|
from types import SimpleNamespace as o # o(a=1).a == 1 |
|
isa = isinstance |
|
BIG = 1e32 # "no cut yet" sentinel |
|
|
|
# ---- records, io, format -------------------------------------- |
|
def thing(s): |
|
"Coerce a (trimmed) string to int, float, bool, else str." |
|
if (s[1:] if s[:1] == "-" else s).isdigit(): return int(s) |
|
try: return float(s) |
|
except ValueError: return s=="True" or (s!="False" and s) |
|
|
|
def settings(s): |
|
"Parse every var=val in a string into an o (vals coerced)." |
|
pairs = re.findall(r"(\w+)=(\S+)", s) |
|
return o(**{k: thing(v) for k, v in pairs}) |
|
|
|
def csv(file, clean=lambda s: s.partition("#")[0].split(",")): |
|
"Yield typed rows from a CSV file ('#' starts a comment)." |
|
with open(file, encoding="utf-8") as f: |
|
for line in f: |
|
row = [x.strip() for x in clean(line)] # strip once, here |
|
if any(row): # skip blank/comment lines |
|
yield tuple(thing(x) for x in row) # hashable: rows key caches |
|
|
|
def say(x, dec=2): |
|
"Pretty string: whole floats as ints, else `dec` places." |
|
if isa(x, float): |
|
return str(int(x)) if x == int(x) else f"{x:.{dec}f}" |
|
if isa(x, o): x = vars(x) # unwrap a namespace |
|
if isa(x, dict): return "{" + ", ".join(f"{k}: {say(v,dec)}" |
|
for k, v in sorted(x.items())) + "}" |
|
if isa(x, (list, tuple)): |
|
return "[" + ", ".join(say(v, dec) for v in x) + "]" |
|
return str(x) |
|
|
|
def sho(rows, just): |
|
"Align rows (list[list[str]]) to columns; just='>'/'<' per col." |
|
w = [max(len(r[c]) for r in rows) for c in range(len(just))] |
|
j = lambda c, s: s.rjust(w[c]) if just[c] == ">" else s.ljust(w[c]) |
|
return "\n".join(" ".join(j(c, s) for c, s in enumerate(r)).rstrip() |
|
for r in rows) |
|
|
|
def run(fns, name, seed): |
|
"Reseed, run one test_* fn; return 1 on failure else 0." |
|
random.seed(seed) |
|
try: fns[name](); return 0 |
|
except Exception: traceback.print_exc(); return 1 |
|
|
|
def usage(g, fns, help=None): |
|
"Print `help` (default g's __doc__) then the test_* commands." |
|
print((help if help is not None else g.get("__doc__") or "").strip()) |
|
print("\ncommands:", *(" --" + n for n in fns)) |
|
|
|
def main(g, help=None, argv=None, seed=1): |
|
"""Run g's test_* fns: --name picks some, none=all; reseed. |
|
-h prints usage. --seed=N sets the seed. Pass globals() as g.""" |
|
fns = {k[5:]: v for k, v in g.items() if k.startswith("test_")} |
|
argv = sys.argv[1:] if argv is None else argv |
|
if "-h" in argv or "--help" in argv: |
|
return usage(g, fns, help) or 0 |
|
for a in argv: |
|
if a.startswith("--seed="): seed = int(a.split("=", 1)[1]) |
|
named = [a[2:] for a in argv if a[:2] == "--" and "=" not in a] |
|
todo = [n for n in named if n in fns] or list(fns) |
|
return sum(run(fns, n, seed) for n in todo) |
|
|
|
# ---- rand: own random.Random(seed) for repeatability ---------- |
|
def shuffle(lst, rng=random): |
|
"Shuffled copy via the given RNG (no global config)." |
|
lst = lst[:]; rng.shuffle(lst); return lst |
|
|
|
def some(lst, k=512, rng=random): |
|
"Up to k items sampled without replacement, via rng." |
|
return rng.sample(lst, min(k, len(lst))) |
|
|
|
def one(lst, rng=random): |
|
"One random item via rng." |
|
return rng.choice(lst) |
|
|
|
# ---- stats: are two samples the same? ------------------------- |
|
def cliffs(xs, ys): |
|
"Cliff's delta effect size in 0..1 (0=identical)." |
|
ys = sorted(ys); m = len(ys) |
|
gt = sum(bisect_left(ys, x) for x in xs) # ys below x |
|
lt = sum(m - bisect_right(ys, x) for x in xs) # ys above x |
|
return abs(gt - lt) / (len(xs) * m + 1e-32) |
|
|
|
def ks(xs, ys): |
|
"Kolmogorov-Smirnov: max gap between the two CDFs." |
|
xs, ys = sorted(xs), sorted(ys) |
|
n, m = len(xs), len(ys) |
|
gap = lambda v: abs(bisect_right(xs, v)/n |
|
- bisect_right(ys, v)/m) |
|
return max(map(gap, xs + ys)) |
|
|
|
def same(xs, ys, cliff=0.195, conf=1.36): |
|
"True if xs,ys are statistically indistinguishable." |
|
if cliffs(xs, ys) > cliff: return False |
|
n, m = len(xs), len(ys) |
|
return ks(xs, ys) <= conf * ((n + m) / (n * m)) ** 0.5 |
|
|
|
def top_tier(groups, cliff=0.195, conf=1.36): |
|
"Groups {name:nums} tied for best (lowest median wins)." |
|
rank = lambda lst: sorted(lst)[len(lst) // 2] |
|
items = sorted(groups.items(), key=lambda kv: rank(kv[1])) |
|
best, (k0, lst0) = {}, items[0] |
|
for k, lst in items: |
|
if k == k0 or same(lst0, lst, cliff, conf): best[k] = lst |
|
else: break |
|
return best |
|
|
|
# ---- columns: a Sym is a {value:count} dict, Num a 3-tuple ---- |
|
Sym = dict # isa(col, Sym) reads "col is a Sym" |
|
def Num(n=0, mu=0, m2=0): return (n, mu, m2) # count,mean,sumsq |
|
def n_(x): return x[0] # a Num's count (n, mu, m2) |
|
def mu_(x): return x[1] # a Num's mean |
|
def m2_(x): return x[2] # a Num's sum of squares |
|
|
|
def sd(num): |
|
"Standard deviation of a Num from its m2." |
|
n, mu, m2 = num |
|
return 0 if n < 2 else (max(0, m2) / (n - 1)) ** 0.5 |
|
|
|
def welford(num, v, inc=1): |
|
"Num + v (inc=-1 removes); returns a new (n, mu, m2)." |
|
n, mu, m2 = num |
|
if (n := n + inc) <= 0: return Num() |
|
d = v - mu; mu += inc * d / n |
|
return n, mu, m2 + inc * d * (v - mu) |
|
|
|
def norm(num, v): |
|
"Map v to 0..1 via a logistic on its z-score (Num only)." |
|
if v == "?": return v |
|
z = (v - mu_(num)) / (sd(num) + 1e-32) |
|
return 1 / (1 + exp(-1.7 * max(-3, min(3, z)))) |
|
|
|
def mix(i, j, inc=1): |
|
"Combine two same-type cols; inc=-1 removes j from i." |
|
if isa(i, Sym): |
|
return {k: i.get(k, 0) + inc * j.get(k, 0) for k in i | j} |
|
(ni, mui, m2i), (nj, muj, m2j) = i, j |
|
n = ni + inc * nj |
|
if n <= 0: return Num() |
|
d = muj - mui |
|
mu = (ni * mui + inc * nj * muj) / n |
|
m2 = m2i + inc * m2j + inc * d * d * ni * nj / n |
|
return Num(n, mu, m2) |
|
|
|
# ---- table: roles in Data, columns held by at-index ----------- |
|
def Data(src=None): |
|
"""Table o(names, cols{at:col}, x, y, goal, klass, rows). |
|
Upper=Num lower=Sym; +-! = y goal (+max); ! klass; X skip.""" |
|
src = iter(src or []) |
|
data = o(names=next(src, []), cols={}, x=[], y=[], goal={}, |
|
klass=None, rows=[]) |
|
return adds(src, roles(data)) |
|
|
|
def roles(data): |
|
"Assign column roles from data.names; returns data." |
|
for at, s in enumerate(data.names): |
|
data.cols[at] = Num() if s[0].isupper() else Sym() |
|
if s[-1] == "X": continue |
|
if s[-1] in "+-!": |
|
data.y.append(at) |
|
data.goal[at] = s[-1] == "+" |
|
if s[-1] == "!": data.klass = at |
|
else: data.x.append(at) |
|
return data |
|
|
|
def add(it, v, inc=1): |
|
"Add to a Sym/Num/Data; RETURNS the (new) it. Skips '?'." |
|
if isa(it, Sym): |
|
if v != "?": it[v] = it.get(v, 0) + inc |
|
return it |
|
if isa(it, tuple): |
|
return welford(it, v, inc) if v != "?" else it |
|
(it.rows.append if inc == 1 else it.rows.remove)(v) |
|
for at in it.cols: it.cols[at] = add(it.cols[at], v[at], inc) |
|
return it |
|
|
|
def adds(src, it=None): |
|
"Fold src into it (a fresh Num by default); returns it." |
|
if it is None: it = Num() |
|
for v in src: it = add(it, v) |
|
return it |
|
|
|
def clone(data, src=None): |
|
"New Data with data's columns; optionally seed src rows." |
|
return Data([data.names] + (src or [])) |
|
|
|
def mid(col): |
|
"Central tendency: mode (Sym) or mean (Num)." |
|
return max(col, key=col.get) if isa(col, Sym) else mu_(col) |
|
|
|
def mids(data): |
|
"Central tendency of every column, keyed by at-index." |
|
return {at: mid(col) for at, col in data.cols.items()} |
|
|
|
def spread(col): |
|
"Diversity: entropy (Sym) or stdev (Num)." |
|
if not isa(col, Sym): return sd(col) # Num |
|
n = sum(col.values()) |
|
return -sum(c/n * log2(c/n) for c in col.values() if c) # 0*log0 = 0 |
|
|
|
# ---- distance: exponent `p` is a keyword, never global -------- |
|
def minkowski(vals, p=2): |
|
"Aggregate per-item distances via the p-norm." |
|
total, n = 0, 0 |
|
for v in vals: total += v ** p; n += 1 |
|
return (total / (n or 1)) ** (1 / p) |
|
|
|
def disty(data, row, **kw): |
|
"Distance of a row to the best goals (0 = ideal)." |
|
return minkowski( |
|
(abs(norm(data.cols[at], row[at]) - data.goal[at]) |
|
for at in data.y if row[at] != "?"), **kw) |
|
|
|
def distx(data, r1, r2, **kw): |
|
"Distance between two rows over the x-columns." |
|
return minkowski((gap(data.cols[at], r1[at], r2[at]) |
|
for at in data.x), **kw) |
|
|
|
def gap(col, u, v): |
|
"Distance between two values of one column (0..1)." |
|
if u == v == "?": return 1 |
|
if isa(col, Sym): return u != v # Sym |
|
u, v = norm(col, u), norm(col, v) # "?" passes through |
|
if u == "?": u = 1 if v < 0.5 else 0 # a missing value |
|
if v == "?": v = 1 if u < 0.5 else 0 # goes to the far end |
|
return abs(u - v) |
|
|
|
# ---- bayes: naive-bayes likelihood (m, k as kwargs) ----------- |
|
def like(col, v, n=0, prior=0, k=1): |
|
"How a column likes v (Sym: m-estimate, Num: gauss); n=#rows." |
|
if isa(col, Sym): |
|
return (col.get(v, 0) + k * prior) / (n + k) |
|
s = sd(col) + 1e-32; z = 2 * s * s |
|
return exp(-(v - mu_(col)) ** 2 / z) / sqrt(pi * z) |
|
|
|
def likes(data, row, nrows, nklasses, m=2, k=1): |
|
"Log-likelihood of a row under this data (naive bayes)." |
|
n = len(data.rows) |
|
prior = (n + m) / (nrows + m * nklasses) |
|
ls = [like(data.cols[at], v, n, prior, k) for at in data.x |
|
if (v := row[at]) != "?"] |
|
return log(prior) + sum(log(x) for x in ls if x > 0) |
|
|
|
def confuse(pairs): |
|
"(want, got) pairs -> o(pd, pf, prec, acc) per want." |
|
out, n = {}, len(pairs) |
|
for k in sorted({want for want, _ in pairs}): |
|
tp = sum(want == k and got == k for want, got in pairs) |
|
fn = sum(want == k and got != k for want, got in pairs) |
|
fp = sum(want != k and got == k for want, got in pairs) |
|
tn = n - tp - fn - fp |
|
out[k] = o(label=k, n=tp + fn, pd=tp / (tp + fn + 1e-32), |
|
pf=fp / (fp + tn + 1e-32), |
|
prec=tp / (tp + fp + 1e-32), |
|
acc=(tp + tn) / (n + 1e-32)) |
|
return out |
|
|
|
# ---- tree: build a min-variance binary tree over the x-columns - |
|
def has(v, lo, hi): return v == "?" or lo <= v <= hi |
|
|
|
def _impurity(col): |
|
"Split cost: a Num's sum-of-squares, or a Sym's entropy*count." |
|
return m2_(col) if isa(col, tuple) else spread(col) * sum(col.values()) |
|
|
|
def _separate(data, rows, y, Y=Num): |
|
"Yield each (score, at, lo, hi, yes, no, n) split candidate; Y=y-col kind." |
|
ys = {r: y(r) for r in rows} # cache y per (hashable) row |
|
for at in data.x: |
|
sym = isa(data.cols[at], Sym) |
|
rs = sorted((r for r in rows if r[at] != "?"), key=lambda r: r[at]) |
|
tot = Y() |
|
for r in rs: tot = add(tot, ys[r]) |
|
yes, run, run_n = Y(), Y(), 0 # distinct objs: Sym add mutates in place |
|
for k, r in enumerate(rs): |
|
run = add(run, ys[r]); run_n += 1 |
|
if k+1 < len(rs) and rs[k+1][at] == r[at]: continue # only cut at boundaries |
|
if sym: grp, n = run, run_n |
|
else: grp, n = (yes := mix(yes, run)), k+1 |
|
run = Y(); run_n = 0 |
|
no = mix(tot, grp, -1) |
|
lo = r[at] if sym else -BIG |
|
yield _impurity(grp) + _impurity(no), at, lo, r[at], grp, no, n |
|
|
|
def treeCut(data, rows, y, leaf=3, Y=Num): |
|
"The (at, lo, hi) of the lowest-impurity cut, or None." |
|
ok = (c for c in _separate(data, rows, y, Y) if c[6] >= leaf) |
|
best = min(ok, key=lambda c: c[0], default=None) |
|
return best and best[1:4] |
|
|
|
def tree(data, rows=None, y=None, leaf=3, lvl=0, maxDepth=12, Y=Num): |
|
"""Min-impurity binary tree; yes=match. Y=Num+y=disty -> regression |
|
(leaf mu=mean); Y=Sym+y=class label -> classification (leaf mu=mode).""" |
|
rows = data.rows if rows is None else rows |
|
y = y or (lambda r: disty(data, r)) |
|
yc = adds((y(r) for r in rows), Y()) |
|
what = lambda a: Num() if isa(data.cols[a], tuple) else Sym() |
|
ymid = [mid( adds( (r[a] for r in rows), what(a))) for a in data.y] |
|
t = o(at=None, mu=mid(yc), n=len(rows), ymid=ymid) |
|
if len(rows) >= 2*leaf and lvl < maxDepth and _impurity(yc) > 0 and \ |
|
(cut := treeCut(data, rows, y, leaf, Y)): # stop on a pure node |
|
at, lo, hi = cut |
|
yes, no = [], [] |
|
for r in rows: # one pass |
|
(yes if has(r[at], lo, hi) else no).append(r) |
|
if len(yes) >= leaf and len(no) >= leaf: |
|
t.at, t.lo, t.hi, t.yes = at, lo, hi, True # go left on match |
|
t.left = tree(data, yes, y, leaf, lvl+1, maxDepth, Y) |
|
t.right = tree(data, no, y, leaf, lvl+1, maxDepth, Y) |
|
return t |
|
|
|
def treePredict(t, row): |
|
"Walk a tree OR an FFT to a leaf; return its value (disty mean / class mode)." |
|
while t.at is not None: # yes-side = left |
|
t = t.left if has(row[t.at], t.lo, t.hi) == t.yes else t.right |
|
return t.mu |
|
|
|
# ---- FFTs: `ffts` is the only fft-aware code; `fft` picks one - |
|
def ffts(t): |
|
"Fan a tree into FFTs: each level, one child exits as a leaf." |
|
if t.at is None: |
|
yield "", t |
|
else: |
|
for yes in (False, True): # which child exits |
|
ex, cont = (t.left, t.right) if yes else (t.right, t.left) |
|
lf = o(at=None, mu=ex.mu, n=ex.n) # exit collapsed to leaf |
|
for bias, rest in ffts(cont): |
|
yield str(int(yes)) + bias, o(at=t.at, lo=t.lo, hi=t.hi, |
|
yes=yes, left=lf, right=rest) |
|
|
|
def fftLeaves(t): |
|
"An FFT's leaves: the exit leaf per cue, plus the final leaf." |
|
while t.at is not None: |
|
yield t.left; t = t.right |
|
yield t |
|
|
|
def fft(t, guard=3): |
|
"From t's fan, the FFT with the lowest leaf-mu (leaves n>=guard)." |
|
def lo(f): |
|
ms = [lf.mu for lf in fftLeaves(f) if lf.n >= guard] |
|
return min(ms) if ms else BIG |
|
return min((f for _, f in ffts(t)), key=lo) |
|
|
|
# ---- tree: pretty-print as an indented table ------------------ |
|
def _treeShow1(data, t, best, worst, out, lvl=0, edge=""): |
|
"Rows of [mark,d2h,n,goal-means,text]; edge=test that reached node." |
|
def cue(t, yes): # edge label into a child |
|
nm = data.names[t.at] # 'Kloc <= 182' / 'Kloc > 182' |
|
if t.lo == t.hi: |
|
return f"{nm} == {say(t.lo)}" if yes else f"{nm} != {say(t.lo)}" |
|
return f"{nm} <= {say(t.hi)}" if yes else f"{nm} > {say(t.hi)}" |
|
num = isa(t.mu, (int, float)) # +/- only ranks regression leaves |
|
mark = "+" if t.at is None and num and t.mu == best else \ |
|
"-" if t.at is None and num and t.mu == worst else "" |
|
out += [[mark, say(t.mu), say(t.n)] |
|
+ [say(v) for v in t.ymid] + ["| "*max(0, lvl-1) + edge]] |
|
if t.at is not None: |
|
kids = [(t.left, cue(t, True)), (t.right, cue(t, False))] |
|
kids.sort(key=lambda kt: kt[0].mu) # better first |
|
for kid, e in kids: |
|
_treeShow1(data, kid, best, worst, out, lvl+1, e) |
|
|
|
def treeShow(data, t): |
|
"+/- best/worst leaf, d2h, n, goal means, then the tree." |
|
mus = [] |
|
def leaves(t): |
|
if t.at is None: mus.append(t.mu) |
|
else: leaves(t.left); leaves(t.right) |
|
leaves(t) |
|
ynm = [data.names[a] for a in data.y] |
|
head = ["", "d2h", "n"] + ynm + ["tree"] |
|
out = [head] |
|
_treeShow1(data, t, min(mus), max(mus), out) |
|
print(sho(out, ">"*(len(head)-1) + "<")) # nums>, tree< |
