The Principia Machina/TPC-PY/tpc cluster.py: Difference between revisions

Latest revision as of 04:37, 28 September 2026

"""tpc_cluster 0.3.0 (DRAFT, pending D143, D144): the top-performer
cluster rule, with the paired noise input under FIXED-SET and the
RESOLVING STEP.

Reference implementation, Python standard library only. DATA for landing
round 8 of branch tpm_project_closeout (0.2.0: the tpm_cluster
speculation 0.2.0 section 8.3, TC-7 to TC-12; 0.3.0: record D143.1 (e)
and D144.1 (i), PMP-TPM 0.12.0 THE TOP-PERFORMER CLUSTER, THE CLUSTER
RULE and RESOLUTION AND FLAGS; the cluster docs 0.2.0 TP-9 and its
specimen figure); not a declared TPC-PY version. The 26 vectors of 0.2.0
are unchanged (cluster() without a paired input computes exactly as
0.2.0; the corpus's 0.2.0 cases are byte-identical, proved by
gen_cluster_corpus.py).

cluster(rows, in_max, cusp_max, z, spacing_floor, paired=None)
    -> (spacing, spacing_source, out)
  rows: list of (model_id, config_id, score, se); non-empty str ids,
        (model_id, config_id) unique; score and se finite, se >= 0.
  in_max, cusp_max: step thresholds in spacings, 0 < in_max < cusp_max.
  z >= 0: the one-sided noise threshold; spacing_floor >= 0 (a
        declaration requires it > 0; 0 is allowed here for vectors).
  paired: None under ITEM-SAMPLING; under FIXED-SET the paired SEs of
        the differences, a list of (model_a, config_a, model_b, config_b,
        se_diff), se_diff finite and >= 0, each unordered pair at most
        once (a second entry for one pair, in either order, is refused:
        "paired"), every pair the rule tests present (a missing pair is
        refused: "paired"; spec note PD-OC-1). A row's own se is still
        used to order representatives (R1, R2).
  spacing_source: "GAP", "FLOOR" or "NOISE" (which term set the
        spacing), or "NONE" with fewer than three models.
  out: list of (rank, model_id, config_id, score, se, step, spacings,
        class); class is IN, CUSP or OUT; rank is None for a row that is
        not its model's representative; step and spacings None where
        not defined.

Rule (tpm_cluster speculation 8.3, TC-7 to TC-12; D144.1 (a)-(h)):
  R1 Each model's REPRESENTATIVE is its row with the highest score (then
     the lower se, then config_id). The rule ranks representatives, so
     with two or more models ranks 1 and 2 are two distinct models.
  R2 Sort representatives by score desc, se asc, model_id asc.
     Ranks 1 and 2 are IN.
  R3 g1 = s1 - s2, g2 = s2 - s3. The SPACING is
     u = max(min(g1, g2), spacing_floor, z * max(sd(1, 2), sd(2, 3))):
     never smaller than a top gap could be by noise.
  R4 For rank k >= 3 the STEP is s_(k-1) - s_k; its noise is
     n_k = z * sd(k-1, k). The row is OUT if step - cusp_max * u > n_k;
     else CUSP if step - in_max * u > n_k; else IN. The scan closes at
     the first CUSP or OUT.
  R5 After a CUSP, each following representative, down to the first
     that is significantly below the CUSP row (s_c - s > z * sd(c, row)),
     is CUSP; that one and every one after it are OUT. After an OUT,
     every following representative is OUT.
  R6 A row that is not its model's representative is IN if it is not
     significantly below the lowest-scoring IN representative, else
     CUSP if it is not significantly below the lowest-scoring CUSP
     representative, else OUT.
  R7 THE SE OF A DIFFERENCE sd(a, b) (D143.1 (e); PMP-TPM 0.12.0: "Under
     FIXED-SET each noise term takes the paired SE of its difference"):
     under ITEM-SAMPLING sqrt(e_a^2 + e_b^2), as 0.2.0; under FIXED-SET
     the paired SE the input carries for the pair (a, b). Every noise
     term of R3-R6 reads sd.

resolving_step(spacing, ses, in_max, cusp_max, z, paired=None) -> dict
  The RESOLVING STEP (D144.1 (i); VOCABULARY "the least adjacent drop
  that would be OUT, stated each month"; the cluster docs 0.2.0 "the
  least adjacent drop that is OUT; for two rows of SE e_a, e_b it exceeds
  CUSP MAX u + Z sqrt(e_a^2 + e_b^2); shown in points and in spacings").
  spacing: the month's u as cluster() returned it; ses: the ranked
  representatives' SEs in rank order; paired: None, or the paired SE of
  each adjacent difference (rank k-1, k), a list of len(ses) - 1 values.
  RS1 For each rank k >= 3, out_k = cusp_max * u + z * sd(k-1, k) and
      cusp_k = in_max * u + z * sd(k-1, k): a drop at rank k larger than
      out_k is OUT and one larger than cusp_k is at least CUSP (R4).
  RS2 THE RESOLVING STEP is the least out_k over k >= 3 (the docs'
      specimen: u 1.6, SEs 0.6 with rank 6 at 0.7, gives 6.996 points,
      4.37 spacings, the 0.6/0.6 pairs, which the least reproduces; the
      greatest, 7.116, would not); shown in points and in spacings
      (points / u; None when u is 0). Beside it the least cusp_k, and the
      greatest of each (the drop that would be OUT, or CUSP, at every
      rank), for the page's optional use (spec note PD-OC-2).
  RS3 Fewer than three ranks: status NONE, nothing computed (SPARSE
      MONTH is the page's flag, outside this function).
"""

import math

VERSION = "0.3.0"


def _check(rows, in_max, cusp_max, z, spacing_floor):
    seen = set()
    for r in rows:
        if not (isinstance(r[0], str) and r[0] and isinstance(r[1], str)
                and r[1]):
            raise ValueError("ids")
        if (r[0], r[1]) in seen:
            raise ValueError("ids")
        seen.add((r[0], r[1]))
        if not (math.isfinite(r[2]) and math.isfinite(r[3]) and r[3] >= 0):
            raise ValueError("score or se")
    for x in (in_max, cusp_max, z, spacing_floor):
        if not math.isfinite(x):
            raise ValueError("parameters")
    if not (0 < in_max < cusp_max and z >= 0 and spacing_floor >= 0):
        raise ValueError("parameters")


def _check_paired(paired, rows):
    """The paired-SE table as a dict keyed by the unordered pair."""
    if paired is None:
        return None
    keys = {(r[0], r[1]) for r in rows}
    table = {}
    for p in paired:
        if len(p) != 5:
            raise ValueError("paired")
        a = (p[0], p[1])
        b = (p[2], p[3])
        if a not in keys or b not in keys or a == b:
            raise ValueError("paired")
        if not (isinstance(p[4], (int, float)) and not isinstance(p[4], bool)
                and math.isfinite(p[4]) and p[4] >= 0):
            raise ValueError("paired")
        k = (a, b) if a < b else (b, a)
        if k in table:
            raise ValueError("paired")
        table[k] = float(p[4])
    return table


class _Noise(object):
    """sd(a, b): the SE of the difference of two rows (R7)."""

    def __init__(self, table):
        self.table = table

    def sd(self, ra, rb):
        if self.table is None:
            return math.sqrt(ra[3] * ra[3] + rb[3] * rb[3])
        a = (ra[0], ra[1])
        b = (rb[0], rb[1])
        k = (a, b) if a < b else (b, a)
        if k not in self.table:
            raise ValueError("paired")
        return self.table[k]

    def below(self, ref, row, z):
        """True when row is significantly below ref."""
        return ref[2] - row[2] > z * self.sd(ref, row)


def cluster(rows, in_max=2.5, cusp_max=3.5, z=1.645, spacing_floor=0.0,
            paired=None):
    _check(rows, in_max, cusp_max, z, spacing_floor)
    noise = _Noise(_check_paired(paired, rows))
    best = {}
    for r in rows:
        k = (-r[2], r[3], r[1])
        if r[0] not in best or k < best[r[0]][0]:
            best[r[0]] = (k, r)
    reps = sorted((v[1] for v in best.values()),
                  key=lambda r: (-r[2], r[3], r[0]))
    n = len(reps)
    s = [r[2] for r in reps]
    e = [r[3] for r in reps]
    std = []
    steps = []
    spacing = 0.0
    source = "NONE"
    for i in range(min(n, 2)):
        std.append("IN")
        steps.append((0.0, 0.0))
    if n >= 3:
        g1 = s[0] - s[1]
        g2 = s[1] - s[2]
        gmin = g1 if g1 < g2 else g2
        d1 = noise.sd(reps[0], reps[1])
        d2 = noise.sd(reps[1], reps[2])
        top = z * (d1 if d1 > d2 else d2)
        spacing = gmin
        source = "GAP"
        if spacing_floor > spacing:
            spacing = spacing_floor
            source = "FLOOR"
        if top > spacing:
            spacing = top
            source = "NOISE"
        closed = None
        cusp_row = -1
        for i in range(2, n):
            step = s[i - 1] - s[i]
            spacings = step / spacing if spacing > 0.0 else None
            steps.append((step, spacings))
            if closed == "CUSP":
                if noise.below(reps[cusp_row], reps[i], z):
                    std.append("OUT")
                    closed = "OUT"
                else:
                    std.append("CUSP")
                continue
            if closed == "OUT":
                std.append("OUT")
                continue
            nk = z * noise.sd(reps[i - 1], reps[i])
            if step - cusp_max * spacing > nk:
                std.append("OUT")
                closed = "OUT"
            elif step - in_max * spacing > nk:
                std.append("CUSP")
                closed = "CUSP"
                cusp_row = i
            else:
                std.append("IN")
    rep_key = {(r[0], r[1]) for r in reps}
    low_in = None
    low_cusp = None
    for i in range(n):
        if std[i] == "IN":
            low_in = i
        elif std[i] == "CUSP":
            low_cusp = i
    out = []
    for i in range(n):
        out.append((i + 1, reps[i][0], reps[i][1], s[i], e[i],
                    steps[i][0], steps[i][1], std[i]))
    others = sorted((r for r in rows if (r[0], r[1]) not in rep_key),
                    key=lambda r: (-r[2], r[3], r[0], r[1]))
    for r in others:
        if low_in is not None and not noise.below(reps[low_in], r, z):
            st = "IN"
        elif low_cusp is not None and not noise.below(reps[low_cusp], r, z):
            st = "CUSP"
        else:
            st = "OUT"
        out.append((None, r[0], r[1], r[2], r[3], None, None, st))
    return spacing, source, out


def fmt(x):
    return "-" if x is None else "%.6f" % x


def render(result):
    spacing, source, out = result
    lines = ["spacing|%s|%s" % (fmt(spacing), source)]
    for r in out:
        lines.append("%s|%s|%s|%s|%s|%s|%s|%s" % (
            "-" if r[0] is None else str(r[0]), r[1], r[2], fmt(r[3]),
            fmt(r[4]), fmt(r[5]), fmt(r[6]), r[7]))
    return "\n".join(lines)


# ------------------------------------------------------- resolving_step

def resolving_step(spacing, ses, in_max=2.5, cusp_max=3.5, z=1.645,
                   paired=None):
    if not (isinstance(spacing, (int, float)) and not isinstance(spacing, bool)
            and math.isfinite(spacing) and spacing >= 0):
        raise ValueError("spacing")
    if not isinstance(ses, list):
        raise ValueError("ses")
    for x in ses:
        if not (isinstance(x, (int, float)) and not isinstance(x, bool)
                and math.isfinite(x) and x >= 0):
            raise ValueError("ses")
    for x in (in_max, cusp_max, z):
        if not (isinstance(x, (int, float)) and not isinstance(x, bool)
                and math.isfinite(x)):
            raise ValueError("parameters")
    if not (0 < in_max < cusp_max and z >= 0):
        raise ValueError("parameters")
    n = len(ses)
    if paired is not None:
        if not isinstance(paired, list) or len(paired) != max(n - 1, 0):
            raise ValueError("paired")
        for x in paired:
            if not (isinstance(x, (int, float)) and not isinstance(x, bool)
                    and math.isfinite(x) and x >= 0):
                raise ValueError("paired")
    if n < 3:
        return {"status": "NONE"}
    u = float(spacing)
    ranks = []
    for k in range(2, n):
        if paired is None:
            sd = math.sqrt(ses[k - 1] * ses[k - 1] + ses[k] * ses[k])
        else:
            sd = float(paired[k - 1])
        nk = z * sd
        ranks.append((k + 1, cusp_max * u + nk, in_max * u + nk))
    out_least = min(r[1] for r in ranks)
    out_most = max(r[1] for r in ranks)
    cusp_least = min(r[2] for r in ranks)
    cusp_most = max(r[2] for r in ranks)

    def sp(x):
        return x / u if u > 0.0 else None

    return {"status": "OK", "out": out_least, "out_spacings": sp(out_least),
            "cusp": cusp_least, "cusp_spacings": sp(cusp_least),
            "out_most": out_most, "cusp_most": cusp_most, "ranks": ranks}


def render_resolving(d):
    if d["status"] == "NONE":
        return "resolving|NONE"
    lines = ["resolving|%s|%s|%s|%s" % (fmt(d["out"]), fmt(d["out_spacings"]),
                                        fmt(d["cusp"]),
                                        fmt(d["cusp_spacings"])),
             "greatest|%s|%s" % (fmt(d["out_most"]), fmt(d["cusp_most"]))]
    for k, o, c in d["ranks"]:
        lines.append("rank|%d|%s|%s" % (k, fmt(o), fmt(c)))
    return "\n".join(lines)


# --------------------------------------------------------------- corpus

def run_case(case):
    """One corpus case (0.2.0's shape, plus fn and paired): the render, or
    "ERROR:" and the message."""
    fn = case.get("fn", "cluster")
    p = case["params"]
    try:
        if fn == "cluster":
            paired = case.get("paired")
            if paired is not None:
                paired = [tuple(x) for x in paired]
            return render(cluster([tuple(x) for x in case["rows"]],
                                  p["in_max"], p["cusp_max"], p["z"],
                                  p["spacing_floor"], paired))
        if fn == "resolving":
            return render_resolving(resolving_step(
                case["spacing"], case["ses"], p["in_max"], p["cusp_max"],
                p["z"], case.get("paired")))
        raise ValueError("fn")
    except ValueError as ex:
        return "ERROR:" + str(ex)