Spaces:
Running
Running
| """ | |
| Geometry core for DiscourseGeometry (no UI code). | |
| Implements the spatial-geometric framework of | |
| Vallego, J. (2026). A spatial-geometric framework for discourse comparison: Centroid projection, | |
| Frobenius norm, and eigenanalysis in a constructed semantic measurement space. SSRN. | |
| Conventions | |
| - Every axis is defined by two sets of exemplar texts: pole A (positive) and pole B (negative). | |
| - The first axis is the primary axis; any further axes are secondary axes, orthogonalised | |
| with Gram-Schmidt in the order given (the first secondary axis is the anchor). | |
| - Score = E · v, with v the unit vector from the pole B centre to the pole A centre | |
| (positive = toward pole A), as in the paper. | |
| - Position (%) = 100 × ((E − cB) · (cA − cB)) / ‖cA − cB‖² | |
| 0% = pole B centre, 100% = pole A centre (higher = closer to pole A). | |
| - Eigenvector signs are arbitrary: only angles (0–90°) and variance shares are interpreted. | |
| """ | |
| import math | |
| import re | |
| import numpy as np | |
| from scipy import stats | |
| # ------------------------------------------------------------------------------------------ text units | |
| SENT_SPLIT = re.compile(r'(?<=[.!?])\s+(?=[A-Z“"‘(\[])') | |
| ABBREV = re.compile(r'\b(St|Sts|Mt|Mts|Ste|Dr|Mr|Mrs|Ms|Prof|Rev|Fr|Sr|Jr|Gen|Col|Capt|No|Nos|Vol|vs|etc|' | |
| r'approx|ca|cf|e\.g|i\.e|Ave|Rd|Ft|Fig|Eq|al)\.(?=\s)') | |
| _DOT = "․" | |
| def split_units(text, mode="auto"): | |
| """ | |
| Split a text box into units (exemplars, responses, passages). | |
| 'dash' : units separated by a line containing only --- | |
| 'blank' : units separated by blank lines | |
| 'line' : one unit per line | |
| 'auto' : --- if present, else blank lines if present, else one per line | |
| """ | |
| text = (text or "").replace("\r\n", "\n").strip() | |
| if not text: | |
| return [] | |
| if mode == "auto": | |
| mode = "dash" if re.search(r"^\s*---\s*$", text, re.M) else "blank" if re.search(r"\n\s*\n", text) else "line" | |
| if mode == "dash": | |
| parts = re.split(r"^\s*---\s*$", text, flags=re.M) | |
| elif mode == "blank": | |
| parts = re.split(r"\n\s*\n", text) | |
| else: | |
| parts = text.split("\n") | |
| return [re.sub(r"\s+", " ", p).strip() for p in parts if p.strip()] | |
| def split_sentences(text, min_words=4): | |
| out = [] | |
| for block in re.split(r"\n\s*\n|\n", text or ""): | |
| block = ABBREV.sub(lambda m: m.group(1) + _DOT, block.strip()) | |
| for s in SENT_SPLIT.split(block): | |
| s = s.replace(_DOT, ".").strip() | |
| if len(s.split()) >= min_words: | |
| out.append(s) | |
| return out | |
| # ------------------------------------------------------------------------------------------ helpers | |
| def unit(v): | |
| n = np.linalg.norm(v) | |
| return v / n if n > 0 else v | |
| def gram_schmidt(vs): | |
| basis = [] | |
| for v in vs: | |
| w = v - sum((v @ b) * b for b in basis) | |
| basis.append(unit(w)) | |
| return np.array(basis) if basis else np.zeros((0, 0)) | |
| def angle(u, v): | |
| """Sign-invariant angle between two directions, 0–90°.""" | |
| c = abs(float(unit(u) @ unit(v))) | |
| return math.degrees(math.acos(min(1.0, c))) | |
| def cov_eig(C): | |
| X = C - C.mean(0) | |
| w, V = np.linalg.eigh(X.T @ X / max(len(C) - 1, 1)) | |
| o = np.argsort(w)[::-1] | |
| return w[o], V[:, o] | |
| def pc1_full(E): | |
| X = E - E.mean(0) | |
| _, s, Vt = np.linalg.svd(X, full_matrices=False) | |
| var = s ** 2 / max(len(E) - 1, 1) | |
| return Vt[0], float(var[0] / var.sum()) if var.sum() > 0 else 0.0 | |
| def sign_fixed(v): | |
| """Present an eigenvector with its largest-magnitude loading positive (signs are arbitrary).""" | |
| return v if v[np.argmax(np.abs(v))] >= 0 else -v | |
| def size_label(d): | |
| if d != d: | |
| return "n/a" | |
| d = abs(d) | |
| return "negligible" if d < 0.2 else "small" if d < 0.5 else "medium" if d < 0.8 else "large" | |
| # ------------------------------------------------------------------------------------------ axes | |
| def build_axes(specs, embed): | |
| """ | |
| specs: list of dicts {name, pos_label, neg_label, pos: [texts], neg: [texts]}; first = primary. | |
| Returns a dict with per-axis vectors and validation statistics, plus the orthonormal bases. | |
| """ | |
| axes = [] | |
| for s in specs: | |
| if len(s["pos"]) < 2 or len(s["neg"]) < 2: | |
| raise ValueError(f"Axis '{s['name']}' needs at least two exemplars for each pole.") | |
| Ep, En = embed(s["pos"]), embed(s["neg"]) | |
| cp, cn = Ep.mean(0), En.mean(0) | |
| v = unit(cp - cn) | |
| sp, sn = Ep @ v, En @ v | |
| sd_p, sd_n = sp.std(ddof=1), sn.std(ddof=1) | |
| sep = sp.mean() - sn.mean() | |
| # leave-one-out check: is each exemplar still on its own side of the midpoint without itself? | |
| stray = [] | |
| for i in range(len(Ep)): | |
| cp_i = np.delete(Ep, i, 0).mean(0) | |
| vi = unit(cp_i - cn) | |
| if Ep[i] @ vi < (cp_i @ vi + cn @ vi) / 2: | |
| stray.append((s["pos_label"], s["pos"][i])) | |
| for i in range(len(En)): | |
| cn_i = np.delete(En, i, 0).mean(0) | |
| vi = unit(cp - cn_i) | |
| if En[i] @ vi > (cp @ vi + cn_i @ vi) / 2: | |
| stray.append((s["neg_label"], s["neg"][i])) | |
| axes.append(dict(name=s["name"], pos_label=s["pos_label"], neg_label=s["neg_label"], | |
| n_pos=len(Ep), n_neg=len(En), cp=cp, cn=cn, v=v, Ep=Ep, En=En, | |
| pos_texts=list(s["pos"]), neg_texts=list(s["neg"]), | |
| mean_pos=float(sp.mean()), mean_neg=float(sn.mean()), sd_pos=float(sd_p), sd_neg=float(sd_n), | |
| separation=float(sep), snr=float(sep / ((sd_p + sd_n) / 2)) if sd_p + sd_n > 0 else float("inf"), | |
| dprime=float(sep / math.sqrt((sd_p ** 2 + sd_n ** 2) / 2)) if sd_p + sd_n > 0 else float("inf"), | |
| stray=stray)) | |
| prim, secs = axes[0], axes[1:] | |
| sec_basis = gram_schmidt([a["v"] for a in secs]) if secs else np.zeros((0, len(prim["v"]))) | |
| full_basis = gram_schmidt([prim["v"]] + [a["v"] for a in secs]) | |
| angles = [dict(a=axes[i]["name"], b=axes[j]["name"], angle=angle(axes[i]["v"], axes[j]["v"]), | |
| cosine=float(axes[i]["v"] @ axes[j]["v"])) | |
| for i in range(len(axes)) for j in range(i + 1, len(axes))] | |
| overlap = float(np.linalg.norm(sec_basis @ prim["v"])) if len(secs) else float("nan") | |
| return dict(axes=axes, primary=prim, secondary=secs, sec_basis=sec_basis, full_basis=full_basis, | |
| angles=angles, overlap=overlap, dim=len(prim["v"])) | |
| def position(E, ax): | |
| d = ax["cp"] - ax["cn"] | |
| return 100.0 * ((E - ax["cn"]) @ d) / (d @ d) | |
| # ------------------------------------------------------------------------------------------ tests | |
| def two_sample(a, b, paired, n_perm, rng): | |
| """b − a. Paired: paired t, d_z, sign-flip permutation. Independent: Welch t, Cohen's d, label permutation.""" | |
| if paired: | |
| d = b - a | |
| t, p = stats.ttest_rel(b, a) | |
| sd = d.std(ddof=1) | |
| eff = d.mean() / sd if sd > 1e-12 else float("nan") | |
| obs = abs(d.mean()) | |
| cnt = sum(abs((d * rng.choice([-1, 1], len(d))).mean()) >= obs for _ in range(n_perm)) | |
| else: | |
| t, p = stats.ttest_ind(b, a, equal_var=False) | |
| pooled = math.sqrt((a.var(ddof=1) + b.var(ddof=1)) / 2) | |
| eff = (b.mean() - a.mean()) / pooled if pooled > 1e-12 else float("nan") | |
| allv, n = np.concatenate([a, b]), len(a) | |
| obs = abs(b.mean() - a.mean()) | |
| cnt = 0 | |
| for _ in range(n_perm): | |
| rng.shuffle(allv) | |
| if abs(allv[n:].mean() - allv[:n].mean()) >= obs: | |
| cnt += 1 | |
| return dict(mean_a=float(a.mean()), mean_b=float(b.mean()), delta=float(b.mean() - a.mean()), | |
| t=float(t), p=float(p), d=float(eff), p_perm=(cnt + 1) / (n_perm + 1)) | |
| def _resample_pair(EA, EB, paired, rng): | |
| if paired: | |
| m = rng.random(len(EA)) < 0.5 | |
| return np.where(m[:, None], EB, EA), np.where(m[:, None], EA, EB) | |
| pool = np.vstack([EA, EB]) | |
| idx = rng.permutation(len(pool)) | |
| return pool[idx[:len(EA)]], pool[idx[len(EA):]] | |
| def _boot(EA, EB, paired, rng): | |
| if paired: | |
| i = rng.integers(0, len(EA), len(EA)) | |
| return EA[i], EB[i] | |
| return EA[rng.integers(0, len(EA), len(EA))], EB[rng.integers(0, len(EB), len(EB))] | |
| # ------------------------------------------------------------------------------------------ analysis | |
| def analyse(G, EA, EB, paired, n_perm=999, n_boot=999, seed=0, E_base=None): | |
| """ | |
| The six analytical layers of the framework for collections A and B (B is compared with A). | |
| G = output of build_axes. EA, EB = unit-normalised embeddings (rows = units). | |
| """ | |
| if paired and len(EA) != len(EB): | |
| raise ValueError(f"Paired mode needs the same number of units in both collections ({len(EA)} vs {len(EB)}).") | |
| if len(EA) < 3 or len(EB) < 3: | |
| raise ValueError("Each collection needs at least three units.") | |
| rng = np.random.default_rng(seed) | |
| P, secs, S, F4 = G["primary"], G["secondary"], G["sec_basis"], G["full_basis"] | |
| v, dim = P["v"], G["dim"] | |
| R = dict(paired=paired, n_a=len(EA), n_b=len(EB), dim=dim, n_perm=n_perm, n_boot=n_boot) | |
| # Layer 1 – primary axis | |
| sa, sb = EA @ v, EB @ v | |
| R["L1"] = two_sample(sa, sb, paired, n_perm, rng) | |
| R["L1"].update(pos_a=float(position(EA, P).mean()), pos_b=float(position(EB, P).mean()), | |
| sd_a=float(sa.std(ddof=1)), sd_b=float(sb.std(ddof=1))) | |
| if E_base is not None and len(E_base): | |
| R["baseline"] = dict(score=float((E_base @ v).mean()), position=float(position(E_base, P).mean()), | |
| n=len(E_base)) | |
| k = 1 + len(secs) | |
| R["bonferroni_alpha"] = 0.05 / k | |
| # Layers 2–3 – secondary axes, GS-projected and raw | |
| R["L2"] = [] | |
| if len(secs): | |
| CA, CB = EA @ S.T, EB @ S.T | |
| for i, ax in enumerate(secs): | |
| gs = two_sample(CA[:, i], CB[:, i], paired, n_perm, rng) | |
| raw = two_sample(EA @ ax["v"], EB @ ax["v"], paired, 0, rng) | |
| ratio = abs(gs["delta"]) / abs(raw["delta"]) if abs(raw["delta"]) > 1e-12 else float("inf") | |
| if gs["p"] > 0.05 and raw["p"] > 0.05: | |
| flag = "both ≈ 0: ratio not meaningful" if i else "anchor (unchanged by GS)" | |
| elif i == 0: | |
| flag = "anchor (unchanged by GS)" | |
| elif ratio > 1.5 or ratio < 1 / 1.5: | |
| flag = "possible ordering effect" | |
| else: | |
| flag = "anchor-independent" | |
| R["L2"].append(dict(name=ax["name"], gs=gs, raw_delta=raw["delta"], raw_p=raw["p"], ratio=ratio, flag=flag, | |
| significant=gs["p"] < R["bonferroni_alpha"])) | |
| # Layers 4–5 – Frobenius | |
| if paired: | |
| D = EB - EA | |
| per = np.linalg.norm(D, axis=1) | |
| fro = float(np.linalg.norm(D)) | |
| L5 = dict(fro=fro, per_mean=float(per.mean()), per_sd=float(per.std(ddof=1)), | |
| consistency=float(per.mean() / per.std(ddof=1)) if per.std(ddof=1) > 1e-12 else float("nan"), | |
| per_unit=per.tolist()) | |
| tot2 = fro ** 2 | |
| L5["primary_norm"] = float(np.linalg.norm(D @ v)) | |
| L5["primary_frac_var"] = L5["primary_norm"] ** 2 / tot2 if tot2 else 0.0 | |
| if len(secs): | |
| Ssc = D @ S.T | |
| L4 = dict(fro=float(np.linalg.norm(Ssc)), | |
| dims=[dict(name=ax["name"], col_norm=float(np.linalg.norm(Ssc[:, i])), | |
| share=float(np.linalg.norm(Ssc[:, i]) ** 2 / np.linalg.norm(Ssc) ** 2) | |
| if np.linalg.norm(Ssc) > 0 else 0.0, | |
| raw_norm=float(np.linalg.norm(D @ ax["v"]))) | |
| for i, ax in enumerate(secs)]) | |
| L5["sec_norm"] = L4["fro"] | |
| L5["sec_frac_var"] = L4["fro"] ** 2 / tot2 if tot2 else 0.0 | |
| R["L4"] = L4 | |
| F = D @ F4.T | |
| L5["full_norm"] = float(np.linalg.norm(F)) | |
| L5["full_frac_var"] = L5["full_norm"] ** 2 / tot2 if tot2 else 0.0 | |
| L5["full_axes"] = [dict(name=a["name"], norm=float(np.linalg.norm(F[:, i]))) | |
| for i, a in enumerate([P] + list(secs))] | |
| L5["residual_norm"] = math.sqrt(max(tot2 - L5["full_norm"] ** 2, 0.0)) | |
| top = np.argsort(per)[::-1][:5] | |
| L5["top_units"] = [int(i) for i in top] | |
| else: | |
| dmu = EB.mean(0) - EA.mean(0) | |
| sep = float(np.linalg.norm(dmu)) | |
| Sa, Sb = np.cov(EA, rowvar=False), np.cov(EB, rowvar=False) | |
| size = (np.linalg.norm(Sa) + np.linalg.norm(Sb)) / 2 | |
| tot2 = sep ** 2 | |
| L5 = dict(centroid_sep=sep, shape_diff=float(np.linalg.norm(Sa - Sb) / size) if size > 0 else 0.0, | |
| primary_frac_var=float((dmu @ v) ** 2 / tot2) if tot2 else 0.0) | |
| if len(secs): | |
| cs = S @ dmu | |
| L5["sec_frac_var"] = float(cs @ cs / tot2) if tot2 else 0.0 | |
| R["L4"] = dict(dims=[dict(name=ax["name"], component=float(cs[i]), | |
| share=float(cs[i] ** 2 / (cs @ cs)) if cs @ cs > 0 else 0.0) | |
| for i, ax in enumerate(secs)]) | |
| cf = F4 @ dmu | |
| L5["full_frac_var"] = float(cf @ cf / tot2) if tot2 else 0.0 | |
| L5["random_1"] = 1 / dim | |
| L5["random_sec"] = len(secs) / dim | |
| L5["random_full"] = (1 + len(secs)) / dim | |
| R["L5"] = L5 | |
| # Layer 6 – eigenanalysis | |
| L6 = {} | |
| ua, sha = pc1_full(EA) | |
| ub, shb = pc1_full(EB) | |
| L6["full"] = dict(rotation=angle(ua, ub), pc1_share_a=sha, pc1_share_b=shb, | |
| pc1_to_primary_a=angle(ua, v), pc1_to_primary_b=angle(ub, v)) | |
| if len(secs) >= 2: | |
| CA, CB = EA @ S.T, EB @ S.T | |
| p3 = unit(S @ v) | |
| wa, Va = cov_eig(CA) | |
| wb, Vb = cov_eig(CB) | |
| def desc(w, V, C): | |
| return dict(eigenvalues=w.tolist(), explained=(w / w.sum()).tolist() if w.sum() > 0 else [], | |
| isotropy=float(w[-1] / w[0]) if w[0] > 0 else float("nan"), | |
| eff_dim=float(w.sum() ** 2 / (w ** 2).sum()) if (w ** 2).sum() > 0 else float("nan"), | |
| total_var=float(w.sum()), pc1=sign_fixed(V[:, 0]).tolist(), | |
| pc1_to_primary=angle(V[:, 0], p3) if np.linalg.norm(p3) > 0 else float("nan")) | |
| L6["sec"] = dict(a=desc(wa, Va, CA), b=desc(wb, Vb, CB), rotation=angle(Va[:, 0], Vb[:, 0]), | |
| primary_in_sec=float(np.linalg.norm(S @ v))) | |
| ge_s = ge_f = 0 | |
| for _ in range(n_perm): | |
| Xa, Xb = _resample_pair(EA, EB, paired, rng) | |
| if angle(cov_eig(Xa @ S.T)[1][:, 0], cov_eig(Xb @ S.T)[1][:, 0]) >= L6["sec"]["rotation"]: | |
| ge_s += 1 | |
| if angle(pc1_full(Xa)[0], pc1_full(Xb)[0]) >= L6["full"]["rotation"]: | |
| ge_f += 1 | |
| L6["sec"]["rotation_p"] = (ge_s + 1) / (n_perm + 1) | |
| L6["full"]["rotation_p"] = (ge_f + 1) / (n_perm + 1) | |
| rots, ang_a, ang_b, chg = [], [], [], [] | |
| for _ in range(n_boot): | |
| Xa, Xb = _boot(EA, EB, paired, rng) | |
| va = cov_eig(Xa @ S.T)[1][:, 0] | |
| vb = cov_eig(Xb @ S.T)[1][:, 0] | |
| rots.append(angle(va, vb)) | |
| aa, bb = angle(va, p3), angle(vb, p3) | |
| ang_a.append(aa); ang_b.append(bb); chg.append(bb - aa) | |
| ci = lambda x: (float(np.percentile(x, 2.5)), float(np.percentile(x, 97.5))) | |
| L6["sec"].update(rotation_ci=ci(rots), pc1_to_primary_a_ci=ci(ang_a), pc1_to_primary_b_ci=ci(ang_b), | |
| pc1_to_primary_change_ci=ci(chg)) | |
| else: | |
| ge_f = 0 | |
| for _ in range(n_perm): | |
| Xa, Xb = _resample_pair(EA, EB, paired, rng) | |
| if angle(pc1_full(Xa)[0], pc1_full(Xb)[0]) >= L6["full"]["rotation"]: | |
| ge_f += 1 | |
| L6["full"]["rotation_p"] = (ge_f + 1) / (n_perm + 1) | |
| R["L6"] = L6 | |
| # per-unit coordinates | |
| R["units_a"] = dict(score=sa.tolist(), position=position(EA, P).tolist(), | |
| sec=(EA @ S.T).tolist() if len(secs) else []) | |
| R["units_b"] = dict(score=sb.tolist(), position=position(EB, P).tolist(), | |
| sec=(EB @ S.T).tolist() if len(secs) else []) | |
| return R | |