feat: CODEX.md items 1+4 — turn-delta anchor + board estimated marker

Item 1 (anchor refinement): parse_codex now uses turn-delta method when
daily granularity is present — estimates cache_create from per-day context
growth deltas instead of fixed 2:1. sigrank.py --codex first fetches
Claude's measured I/O ratio and passes it as fallback anchor, so the
estimate is grounded in real data rather than a provisional constant.

Item 4 (board marker): estimated rows (Codex-anchored) now show a dimmed
~ marker next to the operator name in the leaderboard HTML so measured
vs estimated is visible at a glance.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app.py

@@ -4,6 +4,7 @@ Operator pastes ccusage/codex output -> ingestion -> full profile + board placem
 Board ranks by Net Volumetric Yield (Υ). Four raw integers drive everything.
 """
 import gradio as gr
+import html as _html
 import math as _math
 import re as _re
 from metrics import compute, SEED
@@ -60,9 +61,11 @@ def board_html(extra=None):
         d=f"{m['dev10x']:.2f}" if m['dev10x'] is not None else "\u2014"
         rank_cls = f"mb-rank-{i}" if i <= 3 else ""
         cls = "mb-row you" if you else ("mb-row rank1" if i==1 else "mb-row")
+        ne = _html.escape(n)
+        est_mark = " <span class='mb-est' title='estimated (Codex anchor)'>~</span>" if m.get("cost_estimated") else ""
         out.append(f'<div class="{cls}">'
             f'<span class="mb-rank {rank_cls}">{i}</span>'
-            f'<span class="mb-op"><b>{n}</b><br><span class="mb-raw">R {_fmt_int(m["raw"]["cache_read"])} \u00b7 C {_fmt_int(m["raw"]["cache_create"])} \u00b7 I {_fmt_int(m["raw"]["input"])} \u00b7 O {_fmt_int(m["raw"]["output"])}</span></span>'
+            f'<span class="mb-op"><b>{ne}{est_mark}</b><br><span class="mb-raw">R {_fmt_int(m["raw"]["cache_read"])} \u00b7 C {_fmt_int(m["raw"]["cache_create"])} \u00b7 I {_fmt_int(m["raw"]["input"])} \u00b7 O {_fmt_int(m["raw"]["output"])}</span></span>'
             f'<span class="mb-num">{m["snr"]:.3f}</span>'
             f'<span class="mb-num">{d}</span>'
             f'<span class="mb-num">{m["velocity"]:.2f}</span>'

ingest.py

@@ -53,62 +53,96 @@ def is_codex_shape(d):
     return ("cached_input_tokens" in keys or "cachedInputTokens" in keys or
             "reasoning_output_tokens" in keys or "reasoningOutputTokens" in keys)
 
-def parse_codex(text):
+def parse_codex(text, io_ratio=None):
     """
     Codex reports combined input_tokens (incl. cached) + cached_input_tokens +
-    output + reasoning. No cache_create exists. We estimate the split using the
-    measured field anchor input:output ~ 2:1 (provisional, being refined):
-      est_fresh_input = 2 * output
-      cache_create    = combined_input - est_fresh_input   (clamped >=0)
-      cache_read      = cached_input_tokens                 (measured)
-    Returns (i, o, cache_create, cache_read, meta) with a directional caveat.
+    output + reasoning. cache_create is never reported by OpenAI.
+
+    Anchor strategy (turn-delta-first, ratio fallback):
+    - If daily/session granularity is present, estimate cache_create from
+      per-day context deltas (turn-delta method): each day's input growth above
+      the previous day's total \u2248 new cache writes.
+    - Else, fall back to io_ratio anchor: est_fresh = io_ratio * output.
+    - io_ratio defaults to 2.0 (provisional); pass Claude's measured I/O ratio
+      for better accuracy.
+
+    cache_read = cachedInputTokens (measured directly).
     """
     d = json.loads(text) if isinstance(text, str) else text
     tot = {"in":0, "cached":0, "out":0, "reason":0, "cost":0.0}
-    def add(e):
+    days = []  # for turn-delta
+    def add(e, track=False):
         if not isinstance(e, dict): return
-        tot["in"]     += e.get("input_tokens", e.get("inputTokens",0)) or 0
-        tot["cached"] += e.get("cached_input_tokens", e.get("cachedInputTokens",0)) or 0
-        tot["out"]    += e.get("output_tokens", e.get("outputTokens",0)) or 0
-        tot["reason"] += e.get("reasoning_output_tokens", e.get("reasoningOutputTokens",0)) or 0
-        tot["cost"]   += e.get("costUSD", e.get("cost",0)) or 0.0
+        i   = e.get("input_tokens", e.get("inputTokens",0)) or 0
+        ca  = e.get("cached_input_tokens", e.get("cachedInputTokens",0)) or 0
+        o   = e.get("output_tokens", e.get("outputTokens",0)) or 0
+        r   = e.get("reasoning_output_tokens", e.get("reasoningOutputTokens",0)) or 0
+        c   = e.get("costUSD", e.get("cost",0)) or 0.0
+        tot["in"] += i; tot["cached"] += ca
+        tot["out"] += o; tot["reason"] += r; tot["cost"] += c
+        if track and i > 0:
+            days.append({"date": e.get("date",""), "in": i, "cached": ca})
     if isinstance(d, list):
         for e in d: add(e)
     else:
         for key in ("daily","session","sessions","data","entries","events"):
             v = d.get(key) if isinstance(d, dict) else None
             if isinstance(v, list):
-                for e in v: add(e)
+                for e in v: add(e, track=True)
                 break
             if isinstance(v, dict):
                 for e in v.values(): add(e)
                 break
         else:
             add(d)
+
     combined_in = tot["in"]; read = tot["cached"]; O = tot["out"] + tot["reason"]
-    est_fresh = 2 * O
-    create = combined_in - est_fresh
-    if create >= 0:
-        I = est_fresh
-        caveat = ("estimated \u2191 input-heavy vs 2:1 baseline" if combined_in > est_fresh
-                  else "estimated \u2248 2:1 baseline")
-    else:
+    anchor_used = "2:1 fixed"
+
+    # --- turn-delta method (when we have daily granularity) ---
+    if len(days) >= 2:
+        days.sort(key=lambda x: x["date"])
+        # Each day's fresh input above the previous day's floor \u2248 new cache writes.
+        # Heuristic: new context added each day = max(0, today.in - prev.in)
         create = 0
-        I = max(combined_in - read, 0) or combined_in
-        caveat = "estimated \u2193 output-rich (2:1 anchor inverted)"
+        prev = days[0]["in"]
+        for day in days[1:]:
+            delta = max(0, day["in"] - prev)
+            create += delta
+            prev = day["in"]
+        create += days[0]["in"]  # first day's full input is new cache
+        I = combined_in  # fresh input IS inputTokens (not combined with reads)
+        anchor_used = "turn-delta"
+        caveat = "estimated via turn-delta (cache_create from daily context growth)"
+    else:
+        # --- ratio anchor fallback ---
+        ratio = io_ratio if io_ratio and io_ratio > 0 else 2.0
+        est_fresh = ratio * O
+        create = combined_in - est_fresh
+        if create >= 0:
+            I = est_fresh
+            label = f"{ratio:.2f}:1 anchor (Claude-measured)" if io_ratio else "2:1 anchor (fixed)"
+            anchor_used = label
+            caveat = f"estimated via {label}"
+        else:
+            create = 0
+            I = max(combined_in - read, 0) or combined_in
+            anchor_used = "fallback (anchor inverted)"
+            caveat = "estimated \u2193 output-rich (anchor inverted)"
+
     meta = {"source":"codex", "estimated":True, "caveat":caveat,
-            "anchor":"input:output 2:1 (provisional)",
+            "anchor": anchor_used,
             "cost": tot["cost"] if tot["cost"] > 0 else None}
-    return I, O, create, read, meta
+    return I, O, int(create), read, meta
 
-def ingest_meta(text):
+def ingest_meta(text, io_ratio=None):
     """Returns (i,o,cw,cr,meta) with estimated/caveat/cost."""
     text=text.strip()
     if not text: raise ValueError("empty")
     if text[0] in "{[":
         d=json.loads(text)
         if is_codex_shape(d):
-            return parse_codex(d)
+            return parse_codex(d, io_ratio=io_ratio)
         i,o,cw,cr,cost = parse_ccusage(text)
         return i,o,cw,cr,{"source":"ccusage","estimated":False,"caveat":None,"cost":cost}
     i,o,cw,cr = parse_four(text)

sigrank.py

@@ -143,9 +143,23 @@ def main(argv=None):
     args.stdin = args.stdin_dash == "-"
 
     color = sys.stdout.isatty() and not args.no_color
+
+    # For Codex: compute Claude's real I/O ratio first and use it as anchor.
+    io_ratio = None
+    if args.codex:
+        try:
+            from ingest import parse_ccusage as _pcc
+            _c_args = type("a", (), {"file": None, "stdin": False, "codex": False})()
+            _c_raw, _ = _grab_usage(_c_args)
+            _ci, _co, _, _, _ = _pcc(_c_raw)
+            if _co > 0:
+                io_ratio = _ci / _co
+        except Exception:
+            pass  # fall back to 2:1
+
     try:
         raw, how = _grab_usage(args)
-        i, o, cw, cr, meta = ingest_meta(raw)
+        i, o, cw, cr, meta = ingest_meta(raw, io_ratio=io_ratio)
     except Exception as e:
         print(f"sigrank: {e}", file=sys.stderr)
         return 1