Layer 1 core: Wright-Fisher knowledge-transmission model with E1-E2
Scaffold plus the Layer 1 analytical core and the first two experiments. - knowledge/: truth, metrics, teachers (2.7.1 shared-switch construction), step, lineage, experiment, config, seeding (imported as `knowledge`). - Validation spine green: neutral decay (Pred 1), fixation (Pred 2), exact mutation-drift equilibrium (Pred 3), union coverage (Pred 5). 68 tests pass. - E1 reproduces tail-first collapse. E2 delivers the headline: a grounding phase boundary g* << 1, with stationary H tracking the exact H_eq closed form (g=0.005 -> 68% of truth diversity; g=0.05 -> 96%). - Reproducibility: uv venv from a hash-pinned uv.lock is the source of truth; every run writes results.parquet + resolved_config.yaml + manifest.json (lib versions, git commit, sha256). Figures and manifests tracked; the large regenerable parquet is gitignored. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
commit
a6eb9b7512
33 changed files with 4356 additions and 0 deletions
110
src/knowledge/metrics.py
Normal file
110
src/knowledge/metrics.py
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
"""Per-generation metrics (blueprint 2.3).
|
||||
|
||||
All metrics operate on a probability vector ``p`` (a distribution over the K knowledge
|
||||
items / alleles). Global and per-region variants are provided; E3 needs the per-region
|
||||
forms. Every function is a pure function of its inputs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
def heterozygosity(p: np.ndarray) -> float:
|
||||
"""Expected heterozygosity / Simpson diversity ``H = 1 - sum_i p_i^2``.
|
||||
|
||||
The lineage-health metric and the quantity the re-mint gate reads. ``H=0`` at
|
||||
fixation (one item), ``H=1-1/K`` at the uniform distribution.
|
||||
|
||||
Args:
|
||||
p (np.ndarray): Probability vector over the K items.
|
||||
|
||||
Returns:
|
||||
float: Heterozygosity in [0, 1).
|
||||
"""
|
||||
p = np.asarray(p, dtype=float)
|
||||
return float(1.0 - np.sum(p * p))
|
||||
|
||||
|
||||
def forward_kl(p_star: np.ndarray, p: np.ndarray, eps: float) -> float:
|
||||
"""Forward KL to truth ``D_KL(p_star || p) = sum_i p*_i log(p*_i / p_i)``.
|
||||
|
||||
The correct primary collapse metric: it *diverges* when ``p`` drops mass that
|
||||
``p_star`` has, i.e. it explicitly punishes forgetting the improbable (blueprint 2.3;
|
||||
reverse KL is deliberately not used). ``p`` is floored at ``eps`` to stay finite.
|
||||
|
||||
Args:
|
||||
p_star (np.ndarray): True distribution.
|
||||
p (np.ndarray): Current distribution.
|
||||
eps (float): Floor applied to ``p`` before the log.
|
||||
|
||||
Returns:
|
||||
float: Forward KL divergence (nats).
|
||||
"""
|
||||
p_star = np.asarray(p_star, dtype=float)
|
||||
p = np.asarray(p, dtype=float)
|
||||
p_floored = np.maximum(p, eps)
|
||||
mask = p_star > 0.0 # 0 * log 0 contributes nothing
|
||||
return float(np.sum(p_star[mask] * np.log(p_star[mask] / p_floored[mask])))
|
||||
|
||||
|
||||
def tail_mass(p: np.ndarray, tail_mask: np.ndarray) -> float:
|
||||
"""Total probability mass ``p`` places on the designated tail items.
|
||||
|
||||
The direct measure of collapse: the tail is lost first, so ``tail_mass -> 0`` is the
|
||||
signature of drift-driven collapse (blueprint 2.3).
|
||||
|
||||
Args:
|
||||
p (np.ndarray): Current distribution.
|
||||
tail_mask (np.ndarray): Boolean mask of tail items.
|
||||
|
||||
Returns:
|
||||
float: Mass on tail items.
|
||||
"""
|
||||
p = np.asarray(p, dtype=float)
|
||||
return float(p[np.asarray(tail_mask, dtype=bool)].sum())
|
||||
|
||||
|
||||
def support_size(p: np.ndarray, eps: float) -> int:
|
||||
"""Number of items with mass above ``eps`` (surviving items).
|
||||
|
||||
Args:
|
||||
p (np.ndarray): Current distribution.
|
||||
eps (float): Support threshold.
|
||||
|
||||
Returns:
|
||||
int: Count of items with ``p_i > eps``.
|
||||
"""
|
||||
p = np.asarray(p, dtype=float)
|
||||
return int(np.sum(p > eps))
|
||||
|
||||
|
||||
def per_region(func, p: np.ndarray, regions: np.ndarray, *args) -> dict[int, float]:
|
||||
"""Apply a metric independently to each region's sub-vector.
|
||||
|
||||
Each region's slice of ``p`` is *not* renormalised; the metric sees the raw masses,
|
||||
so per-region ``tail_mass`` and ``support_size`` are directly comparable across
|
||||
regions (needed for E3). ``heterozygosity`` per region is therefore a within-``p``
|
||||
quantity, documented as such.
|
||||
|
||||
Args:
|
||||
func: One of the metric callables above (called as ``func(p_region, *args)``).
|
||||
p (np.ndarray): Current distribution.
|
||||
regions (np.ndarray): Length-K region index per item.
|
||||
*args: Extra positional args forwarded to ``func`` (e.g. eps, or a per-region
|
||||
tail mask which is sliced automatically if it is a length-K boolean array).
|
||||
|
||||
Returns:
|
||||
dict[int, float]: Region index -> metric value.
|
||||
"""
|
||||
p = np.asarray(p, dtype=float)
|
||||
regions = np.asarray(regions)
|
||||
out: dict[int, float] = {}
|
||||
for r in np.unique(regions):
|
||||
sel = regions == r
|
||||
sliced_args = tuple(
|
||||
a[sel] if isinstance(a, np.ndarray) and a.shape == regions.shape else a
|
||||
for a in args
|
||||
)
|
||||
out[int(r)] = func(p[sel], *sliced_args)
|
||||
return out
|
||||
Loading…
Add table
Add a link
Reference in a new issue