Text rule trends
skill · published by agentleFS · version 1 · approved by frdchang@gmail.com on 2026-10-12
Measure how the rules in versioned text files change over time: extract, cluster into themes, and compare periods.
Mountscripts/analyze.py
#!/usr/bin/env python3
"""Turn item lives into trends: read each file at an age, bin by period, pool samples, map
clusters to themes, label rise/fall/steady, compare before/after a date, and count the share
of one file kind among new files.
Usage:
python3 analyze.py --sample NAME=DIR [--sample NAME=DIR ...] --out OUT
[--config config.json] [--themes themes.json] [options]
Every option can also be set in --config (a JSON object whose keys are the option names with
underscores, e.g. {"age": 30, "period": "quarter", "from": "2025Q4"}). The command line wins.
Options (defaults in brackets):
--mode cohort|snapshot [cohort] cohort: bin files by the period they were created in, and read
each file as it stood --age days after creation. snapshot: for every
period, read every file that existed as it stood on the period's last
day (deleted files drop out).
--age N [30] days after creation (cohort mode)
--created first-item|first-version [first-item] a file's creation date: the birth of its
earliest item, or its first commit
--period quarter|month|year [quarter]
--periods P1,P2,... [every period with data] the bins to report, e.g. 2025Q2,2025Q3
--from P --to P [first and last bin] the two bins compared for rise/fall
--threshold X [0.15] a theme rises if its share of a file's items grows by more than
X (relative), falls if it shrinks by more than X, else steady. A theme
that starts at 0 and grows is "rise". If either bin holds fewer than
--min-files files, every theme is "too few".
--min-files N [20]
--before A:B --after C:D date windows [A, B) and [C, D) by creation date for the before/after
comparison. Or --split DATE: before = every file created earlier,
after = every file created on DATE or later. In both modes each file in a
window is read at creation + --age (snapshot mode too: a window compares
files by when they started, not calendar periods).
--kinds K1,K2 [every kind, most files first] file kinds to report separately, in this
order (also the order of the bootstrap draws)
--share A:B kind A's share of new files among A and B, per period
--boot N --seed S [3000, 3] bootstrap resamples and numpy default_rng seed
--today YYYY-MM-DD [the current date] files younger than --age on this day are left out
of the bins. Set it to make a run repeatable.
Each sample DIR holds rules.json (extract.py) and labels.json (cluster.py). The first --sample
wins when a file is in more than one (it is counted once when pooling).
--force-themes apply --themes even when its centroids_id differs from the labels'.
Without it, analyze refuses: cluster numbers from another fit name
different clusters, and the themes would silently be wrong.
Writes to OUT (empty bins are null, never 0):
pooled.json {Q, nfiles, tot, med, pf, files_with_rules, overlap}: all samples pooled; per
period, files, mean and median items per file, and mean items per file in each
cluster (pf[i][cluster]). Q is the list of periods, whatever the unit.
themes.json {q, nfiles, tot, med, from_q, to_q, threshold, groups: [{name, kind, v,
share_from, share_to}], rest, files_with_items, items, samples: {name:
{history_files, files, files_with_items, items}}}. share_from/share_to are
FRACTIONS (0.181 = 18.1%). The input to tufte-charts and watercolor-data-art.
compare.json per sample and kind: "periods" (n, mean, median, pf per period) and "window"
(means, medians, and a bootstrap 95% CI of the difference in means, after minus
before; "too_few": true when a side has fewer than --min-files files).
share.json per sample (with --share): new files per period by kind, counted by each file's
first commit, whether it has items or not, leaving out files first committed after
--today; over --periods if given, else every period. "share" is in PERCENT
(87.2 = 87.2%). Also "files" (files with a version) and "history_files".
"""
import argparse
import collections
import json
import os
import sys
from datetime import date, timedelta
import numpy as np
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from common import period_end, period_of, period_range, period_start # noqa: E402
def load(root, k, a, today):
R = json.load(open(os.path.join(root, "rules.json")))
life = R["life"]
labels = json.load(open(os.path.join(root, "labels.json")))["labels"]
if len(labels) != len(life):
raise SystemExit(f"{root}: labels.json does not match rules.json")
files = R.get("files", {})
first, kind = {}, {}
for r in life:
first[r["repo"]] = min(first.get(r["repo"], "9999"), r["born"][:10])
kind[r["repo"]] = r["kind"]
if a.created == "first-version":
for f in first:
key = f[: -len(".tsv.gz")] if f.endswith(".tsv.gz") else f[: -len(".tsv")]
if key in files:
first[f] = files[key]["first"][:10]
deleted = {}
for f in first:
key = f[: -len(".tsv.gz")] if f.endswith(".tsv.gz") else f[: -len(".tsv")]
deleted[f] = (files.get(key, {}).get("deleted") or "9999")[:10]
return {"first": first, "kind": kind, "life": life, "labels": labels, "deleted": deleted,
"files": files, "nrules": len(life), "k": k, "age": a.age, "today": today}
def count_at(D, when):
"""Items alive per file on a date: when(file) -> 'YYYY-MM-DD' or None (file not counted)."""
k = D["k"]
cnt = {f: 0 for f in D["first"]}
cl = {f: np.zeros(k) for f in D["first"]}
at = {f: when(f) for f in D["first"]}
for r, l in zip(D["life"], D["labels"]):
t = at[r["repo"]]
if t and r["born"][:10] <= t and (r["died"] or "9999")[:10] > t:
cnt[r["repo"]] += 1
cl[r["repo"]][l] += 1
return cnt, cl
def window(s):
a, b = s.split(":")
return a, b
def main():
pre = argparse.ArgumentParser(add_help=False)
pre.add_argument("--config")
c, _ = pre.parse_known_args()
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--config")
ap.add_argument("--sample", action="append", required=True, help="NAME=DIR, repeatable")
ap.add_argument("--themes")
ap.add_argument("--out", required=True)
ap.add_argument("--mode", default="cohort", choices=["cohort", "snapshot"])
ap.add_argument("--today", default=date.today().isoformat())
ap.add_argument("--age", type=int, default=30)
ap.add_argument("--created", default="first-item", choices=["first-item", "first-version"])
ap.add_argument("--period", default="quarter", choices=["quarter", "month", "year"])
ap.add_argument("--periods")
ap.add_argument("--from", dest="from_", metavar="P")
ap.add_argument("--to", metavar="P")
ap.add_argument("--threshold", type=float, default=0.15)
ap.add_argument("--min-files", type=int, default=20)
ap.add_argument("--before")
ap.add_argument("--after")
ap.add_argument("--split")
ap.add_argument("--kinds")
ap.add_argument("--share")
ap.add_argument("--boot", type=int, default=3000)
ap.add_argument("--seed", type=int, default=3)
ap.add_argument("--force-themes", action="store_true", help="apply the themes file even if its centroids_id differs from the labels'")
if c.config:
cfg = json.load(open(c.config))
cfg = {("from_" if k == "from" else k.replace("-", "_")): (",".join(v) if isinstance(v, list) else v)
for k, v in cfg.items() if not k.startswith("_")}
ap.set_defaults(**cfg)
a = ap.parse_args()
os.makedirs(a.out, exist_ok=True)
today = date.fromisoformat(a.today)
samples = [(s.split("=", 1)[0], s.split("=", 1)[1]) for s in a.sample]
for name, root in samples:
for f, step in (("rules.json", "extract.py"), ("labels.json", "cluster.py fit or assign")):
if not os.path.exists(os.path.join(root, f)):
raise SystemExit(f"{root}/{f} is missing: run {step} on this sample first")
labs = {name: json.load(open(os.path.join(root, "labels.json"))) for name, root in samples}
k_lab = max(l["k"] for l in labs.values())
ids = {name: l.get("centroids_id") for name, l in labs.items()}
if len(set(ids.values())) > 1:
msg = "samples were labelled against different centroids: " + ", ".join(f"{n} {i}" for n, i in ids.items())
if not a.force_themes:
raise SystemExit(msg + ". Assign every sample to the same centroids file (cluster.py assign), or pass --force-themes.")
print("warning: " + msg)
if a.themes:
cfg = json.load(open(a.themes))
K = max(k_lab, 1 + max(c for t in cfg["themes"] for c in t["clusters"]))
theme_list = cfg["themes"]
want, have = cfg.get("centroids_id"), next(iter(ids.values()))
if want and have and want != have:
msg = (f"{a.themes} names the clusters of centroids {want}, but the labels come from {have}. "
"Cluster numbers from another fit mean different things, so the themes would be wrong")
if not a.force_themes:
raise SystemExit(msg + ". Assign the samples to the centroids the themes file was written for "
"(cluster.py assign --centroids ...), or write a themes file for this fit. "
"--force-themes overrides.")
print("warning: " + msg)
elif not want:
print(f"note: {a.themes} has no centroids_id, so it cannot be checked against the labels")
else:
K = k_lab
theme_list = [{"name": f"cluster {c}", "clusters": [c]} for c in range(K)]
data = {name: load(root, K, a, today) for name, root in samples}
per = lambda d: period_of(d, a.period)
# cohort: each file at creation + age; snapshot: each file at the end of every period
for D in data.values():
if a.mode == "cohort":
when = lambda f, D=D: (date.fromisoformat(D["first"][f]) + timedelta(days=a.age)).isoformat()
D["cnt"], D["cl"] = count_at(D, when)
D["ok"] = {f for f, d in D["first"].items() if date.fromisoformat(d) + timedelta(days=a.age) <= today}
if a.periods:
Q = a.periods.split(",")
else:
firsts = [d for D in data.values() for d in D["first"].values()]
if a.mode == "snapshot": # periods that have ended by --today
p = per(today.isoformat())
if period_end(p) > today:
p = per((period_start(p) - timedelta(days=1)).isoformat())
last = p
else:
ok = [d for D in data.values() for f, d in D["first"].items() if f in D["ok"]]
if not ok:
raise SystemExit(f"no file is {a.age} days old by {a.today}")
last = per(max(ok))
Q = period_range(per(min(firsts)), last, a.period)
def bin_files(D, q):
"""(files, counts, cluster counts) for one sample in one period."""
if a.mode == "cohort":
fs = [f for f in D["first"] if per(D["first"][f]) == q and f in D["ok"]]
return fs, D["cnt"], D["cl"]
end = period_end(q).isoformat()
fs = [f for f in D["first"] if D["first"][f] <= end and D["deleted"][f] > end]
cnt, cl = count_at(D, lambda f: end)
return fs, cnt, cl
# ---- per sample, per kind: periods and the before/after windows
rng = np.random.default_rng(a.seed)
def ci(x, y):
bs = [rng.choice(y, len(y)).mean() - rng.choice(x, len(x)).mean() for _ in range(a.boot)]
return [round(float(v), 1) for v in np.percentile(bs, [2.5, 97.5])]
if a.split:
b0, b1, a0, a1 = "0000-00-00", a.split, a.split, "9999-99-99" # open-ended; printed as "before/from DATE"
elif a.before and a.after:
(b0, b1), (a0, a1) = window(a.before), window(a.after)
else:
b0 = b1 = a0 = a1 = None
if a.kinds:
KINDS = a.kinds.split(",")
else:
kc = collections.Counter(D["kind"][f] for D in data.values() for f in D["first"])
KINDS = sorted(kc, key=lambda x: (-kc[x], x))
if a.split:
wb, wa = f"created before {a.split}", f"created {a.split} or later"
else:
wb, wa = f"{b0}..{b1}", f"{a0}..{a1}"
compare = {"Q": Q, "today": a.today, "mode": a.mode, "age_days": a.age, "before": [None, a.split] if a.split else [b0, b1], "after": [a.split, None] if a.split else [a0, a1]}
for name, _ in samples:
D = data[name]
res = {"files_with_rules": len(D["first"]), "rules": D["nrules"]}
print(f"== {name}: {len(D['first'])} files with items, {D['nrules']} items")
for kd in KINDS:
rows = []
for q in Q:
allf, cnt, cl = bin_files(D, q)
fs = [f for f in allf if D["kind"][f] == kd]
n = len(fs)
rows.append({"n": n,
"tot": sum(cnt[f] for f in fs) / n if n else None,
"med": float(np.median([cnt[f] for f in fs])) if n else None,
"pf": (sum((cl[f] for f in fs), np.zeros(K)) / n).tolist() if n else None})
w = {}
if b0 is not None:
if a.mode == "cohort":
cnt = D["cnt"]
else: # snapshot mode still compares files by creation, each read at creation + age
cnt, _ = count_at(D, lambda f: (date.fromisoformat(D["first"][f]) + timedelta(days=a.age)).isoformat())
x = np.array([cnt[f] for f in D["first"] if D["kind"][f] == kd and b0 <= D["first"][f] < b1])
y = np.array([cnt[f] for f in D["first"] if D["kind"][f] == kd and a0 <= D["first"][f] < a1])
w = {"n_before": len(x), "n_after": len(y)}
if len(x) and len(y):
w.update(mean_before=float(x.mean()), mean_after=float(y.mean()),
median_before=float(np.median(x)), median_after=float(np.median(y)),
diff=round(float(y.mean() - x.mean()), 1), ci95=ci(x, y))
print(f" {kd} {wb} -> {wa}: n {len(x)} -> {len(y)}, mean {x.mean():.1f} -> {y.mean():.1f}, "
f"median {np.median(x)} -> {np.median(y)}, diff {w['diff']:+.1f} CI {w['ci95']}")
if min(len(x), len(y)) < a.min_files:
w["too_few"] = True
print(f" note: fewer than {a.min_files} files on one side of the {kd} window; too few to tell")
res[kd] = {"periods": rows, "window": w}
compare[name] = res
json.dump(compare, open(os.path.join(a.out, "compare.json"), "w"), indent=1)
# ---- pooled across samples, a file counted once (first sample wins)
owner = {}
for name, _ in samples:
for f in data[name]["first"]:
owner.setdefault(f, name)
overlap = sum(len(data[n]["first"]) for n, _ in samples) - len(owner)
pooled = {"Q": Q, "nfiles": [], "tot": [], "med": [], "pf": [], "files_with_rules": len(owner), "overlap": overlap}
for q in Q:
vals, cls = [], []
for name, _ in samples:
D = data[name]
fs, cnt, cl = bin_files(D, q)
for f in fs:
if owner[f] == name:
vals.append(cnt[f])
cls.append(cl[f])
n = len(vals)
pooled["nfiles"].append(n)
pooled["tot"].append(sum(vals) / n if n else None)
pooled["med"].append(float(np.median(vals)) if n else None)
pooled["pf"].append((sum(cls, np.zeros(K)) / n).tolist() if n else None)
json.dump(pooled, open(os.path.join(a.out, "pooled.json"), "w"))
print(f"== pooled ({a.mode}): {len(owner)} files with items ({overlap} in more than one sample)")
for i, q in enumerate(Q):
if pooled["nfiles"][i]:
print(f" {q}: files {pooled['nfiles'][i]:5d} mean {pooled['tot'][i]:5.2f} median {pooled['med'][i]}")
else:
print(f" {q}: no files")
# ---- themes
qf, qt = a.from_ or Q[0], a.to or Q[-1]
i0, i1 = Q.index(qf), Q.index(qt)
few = min(pooled["nfiles"][i0], pooled["nfiles"][i1]) < a.min_files
named, groups = set(), []
for t in theme_list:
v = [sum(pf[c] for c in t["clusters"]) if pf is not None else None for pf in pooled["pf"]]
t0, t1 = pooled["tot"][i0] or 0.0, pooled["tot"][i1] or 0.0
# an empty compared bin has no share at all: null, never 0
s0 = v[i0] / t0 if pooled["nfiles"][i0] and t0 else (0.0 if pooled["nfiles"][i0] else None)
s1 = v[i1] / t1 if pooled["nfiles"][i1] and t1 else (0.0 if pooled["nfiles"][i1] else None)
if few or s0 is None or s1 is None:
kind = "too few"
elif s0 == 0:
kind = "rise" if s1 > 0 else "steady"
else:
change = s1 / s0 - 1
kind = "rise" if change > a.threshold else "fall" if change < -a.threshold else "steady"
groups.append({"name": t["name"], "kind": kind, "v": v, "share_from": s0, "share_to": s1})
named |= set(t["clusters"])
rest = [pooled["tot"][i] - sum(g["v"][i] for g in groups) if pooled["nfiles"][i] else None for i in range(len(Q))]
samples_info = {}
for name, root in samples:
R_ = json.load(open(os.path.join(root, "rules.json")))
samples_info[name] = {"history_files": R_.get("nhistory", R_["nfiles"]), "files": R_["nfiles"],
"files_with_items": len(data[name]["first"]), "items": data[name]["nrules"]}
themes = {"q": Q, "nfiles": pooled["nfiles"], "tot": pooled["tot"], "med": pooled["med"],
"files_with_items": len(owner), "items": sum(d["nrules"] for d in data.values()), "samples": samples_info,
"from_q": qf, "to_q": qt, "threshold": a.threshold, "groups": groups, "rest": rest}
json.dump(themes, open(os.path.join(a.out, "themes.json"), "w"), indent=1)
print(f"== themes, share of a file's items {qf} -> {qt}" + (f" (too few files: fewer than {a.min_files} in a compared bin)" if few else ""))
for g in groups[::-1]:
pct = lambda x: " – " if x is None else f"{x * 100:5.1f}%"
print(f" {g['kind']:7} {g['name'][:42]:42} {pct(g['share_from'])} -> {pct(g['share_to'])}")
if len(named) < K:
print(f" ({K - len(named)} clusters in no theme; their items are in 'rest')")
# ---- share of kind A among new files of kinds A and B
if a.share:
A, B = a.share.split(":")
share = {}
for name, root in samples:
files = json.load(open(os.path.join(root, "rules.json"))).get("files")
if files is None:
raise SystemExit(f"{root}/rules.json has no 'files'; run extract.py from this skill")
fc = {k: (v["kind"], v["first"]) for k, v in files.items() if v["first"][:10] <= a.today}
qs = Q if a.periods else sorted({per(d) for _, d in fc.values()})
c = collections.Counter((kd, per(d)) for kd, d in fc.values())
row = {"Q": qs, "files": len(fc), "history_files": samples_info[name]["history_files"]}
for kd in dict.fromkeys(KINDS + [A, B]):
row[kd] = [c[(kd, q)] for q in qs]
row["share_of"], row["share_vs"] = A, B
row["share"] = [round(100 * c[(A, q)] / (c[(A, q)] + c[(B, q)]), 1) if c[(A, q)] + c[(B, q)] else None for q in qs]
share[name] = row
print(f"== {name}: {A} share of new {A} + {B} files: " + ", ".join(f"{q} {s}%" if s is not None else f"{q} –" for q, s in zip(qs, row["share"])))
json.dump(share, open(os.path.join(a.out, "share.json"), "w"), indent=1)
if __name__ == "__main__":
main()