Buckets:
| #!/usr/bin/env python3 | |
| """exp63c — 单格 τ 的零模型重抽稳定性。 | |
| 同一批单元、同一估计量，只换零模型的随机移位 -> τ 能变多少？ | |
| 这是判定「单格 τ 是否可复现」的直接证据。 | |
| 用法: python exp63c_repeat.py [n_cells] [n_rep] | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| import time | |
| import numpy as np | |
| import pandas as pd | |
| import exp63_region_tau as E | |
| N_CELLS = int(sys.argv[1]) if len(sys.argv) > 1 else 40 | |
| N_REP = int(sys.argv[2]) if len(sys.argv) > 2 else 4 | |
| def main(): | |
| t0 = time.time() | |
| d = pd.read_csv(E.OUT / "per_cell.csv") | |
| k = d[d.kept == True] | |
| # 每个 session 最多取 2 格，避免被大 session 主导 | |
| sel = k.groupby("session", group_keys=False).head(2).head(N_CELLS) | |
| print(f"cells={len(sel)} reps={N_REP} sessions={sel.session.nunique()}", flush=True) | |
| u = pd.read_parquet(E.SEG / "units.parquet") | |
| rows = [] | |
| for sid, g in sel.groupby("session"): | |
| f = E.SEG / f"spikes_session_{int(sid):03d}.npz" | |
| raw = np.load(f)["spikes"] | |
| x = (np.round(raw * 8).T).astype(np.float32) | |
| del raw | |
| nb, nu = x.shape | |
| ml = min(E.MAX_LAG, nb - 2) | |
| sub = u[u.session == sid].sort_values("unit_idx") | |
| for _, c in g.iterrows(): | |
| idx = sub[sub.region == c.region].unit_idx.to_numpy() | |
| taus, snrs = [], [] | |
| for rep in range(N_REP): | |
| rng = np.random.default_rng(1000 + rep) | |
| rho, noise, snr, usable, shared = E.estimate_session(x, idx, ml, rng) | |
| rec = {"snr": snr, "kept": False} | |
| if snr >= E.SNR_MIN and len(usable) >= E.MIN_POINTS: | |
| rec = E.add_fits(rec, rho, usable) | |
| taus.append(float(rec["tau_ms"]) if rec.get("kept") else np.nan) | |
| snrs.append(float(snr)) | |
| rows.append({"session": int(sid), "region": c.region, | |
| "n_units": int(len(idx)), | |
| "tau_orig": float(c.tau_ms), | |
| **{f"tau_r{j}": taus[j] for j in range(N_REP)}, | |
| "snr_min": float(np.min(snrs))}) | |
| print(f" sess{int(sid):03d} done rows={len(rows)} [{time.time()-t0:.0f}s]", flush=True) | |
| del x | |
| out = pd.DataFrame(rows) | |
| out.to_csv(E.OUT / "repeatability.csv", index=False) | |
| tau_cols = [f"tau_r{j}" for j in range(N_REP)] | |
| ok = out.dropna(subset=tau_cols) | |
| print(f"\nwrote repeatability.csv rows={len(out)} 全 {N_REP} 次都成功的格={len(ok)}") | |
| if len(ok): | |
| vals = ok[tau_cols].to_numpy(float) | |
| ratio = vals.max(axis=1) / vals.min(axis=1) | |
| print(f"同一格跨 {N_REP} 次零模型抽样的 τ max/min 比值:") | |
| print(f" median={np.median(ratio):.2f}× p75={np.quantile(ratio,.75):.2f}× " | |
| f"p90={np.quantile(ratio,.9):.2f}× max={ratio.max():.2f}×") | |
| print(f" 比值 > 2× 的格: {int((ratio>2).sum())}/{len(ratio)} " | |
| f"({100*(ratio>2).mean():.0f}%)") | |
| print(f" 比值 > 5× 的格: {int((ratio>5).sum())}/{len(ratio)}") | |
| # 相对离散度 | |
| print(f" log τ 跨重复的 SD（格内）median=" | |
| f"{np.median(np.log(vals).std(axis=1, ddof=1)):.3f} log 单位") | |
| (E.OUT / "report_repeatability.json").write_text( | |
| json.dumps({"n_cells": int(len(out)), "n_rep": N_REP, | |
| "n_all_ok": int(len(ok)), | |
| "ratio_median": float(np.median(ratio)) if len(ok) else None, | |
| "ratio_p90": float(np.quantile(ratio, .9)) if len(ok) else None, | |
| "frac_ratio_gt2": float((ratio > 2).mean()) if len(ok) else None}, | |
| indent=2), encoding="utf-8") | |
| print(f"[{time.time()-t0:.0f}s]") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 3.81 kB
- Xet hash:
- 0ddefe25e69d294af0157608ea06a89f91a944a7705afd3c54ca9f52ed51fb15
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.