cache writes are paid for and never read back

"""Report Anthropic cache writes that are never read back.

Read only. GET requests and nothing else against the Admin API, which needs an
Admin API key (sk-ant-admin...); a workspace key is rejected by every
/v1/organizations/* path, and an Admin key can be provisioned read-only. The
repair is printed, never performed: moving a cache_control breakpoint is a
change to your own request, not something a script should do to you.

The messages usage report carries token sums per bucket and no request count at
all, so "reads per write" below means read tokens per write token. It is a
proxy for call counts, not a call count, and this API has no call count to
check it against.
"""
import argparse
import datetime
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("anthropic_cache_write_ratio")

API = "https://api.anthropic.com/v1"
VERSION = "2023-06-01"

# Published multipliers on base input.
WRITE_5M = 1.25
WRITE_1H = 2.00
READ = 0.10
BASE = 1.00


def accumulate(results, into=None):
    """Sum token fields across usage-report results, keeping the TTLs apart. Pure.

    The two cache_creation members are priced differently, so summing them here
    would throw away the information the break-even calculation needs. They live
    inside a nested cache_creation object, which is the field a flat parser
    misses entirely.
    """
    total = {"uncached": 0, "cache_read": 0, "write_5m": 0, "write_1h": 0}
    if into:
        total.update(into)
    for result in results or []:
        total["uncached"] += int(result.get("uncached_input_tokens") or 0)
        total["cache_read"] += int(result.get("cache_read_input_tokens") or 0)
        creation = result.get("cache_creation") or {}
        total["write_5m"] += int(creation.get("ephemeral_5m_input_tokens") or 0)
        total["write_1h"] += int(creation.get("ephemeral_1h_input_tokens") or 0)
    return total


def break_even_ratio(write_5m, write_1h):
    """Read tokens per write token at which caching starts to save money. Pure.

    Caching w write tokens and r read tokens costs 1.25*w5 + 2.0*w1h + 0.1*r,
    against w5 + w1h + r for the same tokens uncached. Solving for r gives
    r > ((1.25-1)*w5 + (2.0-1)*w1h) / (1 - 0.1), which is about 0.28 for pure
    5m traffic and about 1.11 for pure 1h. Returns None when nothing was
    written, because a ratio against zero is not a number.
    """
    writes = write_5m + write_1h
    if writes <= 0:
        return None
    premium = (WRITE_5M - BASE) * write_5m + (WRITE_1H - BASE) * write_1h
    return premium / ((BASE - READ) * writes)


def effective_multiplier(write_5m, write_1h, reads):
    """What this cached traffic costs per token relative to not caching. Pure.

    Above 1.0 means the caching is charging you a surcharge: the same tokens
    would have been cheaper with the feature switched off.
    """
    tokens = write_5m + write_1h + reads
    if tokens <= 0:
        return None
    cost = WRITE_5M * write_5m + WRITE_1H * write_1h + READ * reads
    return cost / tokens


def verdict(total, min_writes=100_000, margin=1.5):
    """Classify one key's cache economics over the window. Pure.

    Returns (state, detail). `margin` is how far above break-even a ratio has to
    sit before it is called safe rather than marginal, because a ratio sitting
    on the line will cross it the first week traffic dips.
    """
    reads = int(total.get("cache_read", 0))
    write_5m = int(total.get("write_5m", 0))
    write_1h = int(total.get("write_1h", 0))
    writes = write_5m + write_1h

    if writes == 0 and reads == 0:
        return ("no-caching",
                "no cache reads and no cache writes in this window: caching is "
                "not switched on for this key at all, which is a different "
                "problem from this one")
    if writes == 0:
        return ("reads-only",
                "%d read token(s) against entries written before this window "
                "opened. Widen the window before drawing a ratio from it." % reads)
    if writes < min_writes:
        return ("too-little-traffic",
                "only %d cache write token(s) in the window; too little to draw "
                "a ratio from" % writes)

    ratio = reads / writes
    threshold = break_even_ratio(write_5m, write_1h)
    multiplier = effective_multiplier(write_5m, write_1h, reads)
    shape = ("%.2f read tokens per write token against a break-even of %.2f; "
             "this traffic costs %.2fx what the same tokens would cost with "
             "caching switched off" % (ratio, threshold, multiplier))
    if ratio < threshold:
        return ("losing", shape)
    if ratio < threshold * margin:
        return ("marginal", shape + ", which is barely above the line")
    return ("paying-off", shape)


def get(session, path, params):
    r = session.get(API + path, params=params, timeout=60)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Anthropic: /v1/organizations/* needs an Admin "
                         "API key (sk-ant-admin...), not a workspace key"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def buckets(session, path, params):
    params = dict(params)
    while True:
        page = get(session, path, params)
        for bucket in page.get("data") or []:
            yield bucket
        if not page.get("has_more") or not page.get("next_page"):
            return
        params["page"] = page["next_page"]


def window_start(days):
    """Floor to the hour, because starting_at must sit on a bucket boundary."""
    now = datetime.datetime.now(datetime.timezone.utc)
    top = now.replace(minute=0, second=0, microsecond=0)
    return (top - datetime.timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7, help="days of hourly buckets to read")
    ap.add_argument("--min-writes", type=int, default=100_000,
                    help="cache write tokens below which no ratio is claimed")
    args = ap.parse_args()

    admin = os.environ.get("ANTHROPIC_ADMIN_KEY")
    if not admin:
        log.error("set ANTHROPIC_ADMIN_KEY to an Admin API key (sk-ant-admin...); "
                  "a workspace key cannot read /v1/organizations/*")
        return 2

    s = requests.Session()
    s.headers.update({"x-api-key": admin, "anthropic-version": VERSION})

    params = {"starting_at": window_start(args.days), "bucket_width": "1h",
              "limit": min(args.days * 24, 168), "group_by[]": ["api_key_id"]}

    by_key = {}
    for bucket in buckets(s, "/organizations/usage_report/messages", params):
        for result in bucket.get("results") or []:
            name = result.get("api_key_id") or "unattributed"
            by_key[name] = accumulate([result], by_key.get(name))

    if not by_key:
        log.info("no message usage in the last %d day(s)", args.days)
        return 0

    losing = 0
    for name, total in sorted(by_key.items(),
                              key=lambda kv: -(kv[1]["write_5m"] + kv[1]["write_1h"])):
        state, detail = verdict(total, args.min_writes)
        line = "%-18s %s  %s" % (state, name, detail)
        if state in ("paying-off", "too-little-traffic", "reads-only"):
            log.info(line)
            continue
        if state == "no-caching":
            log.info(line)
            continue
        losing += 1
        log.warning(line)
        log.warning("  repair: move the cache_control breakpoint to the end of the "
                    "stable prefix and keep timestamps, request ids and the user's "
                    "question strictly after it, then re-measure this ratio tomorrow")
        if total["write_1h"] > total["write_5m"]:
            log.warning("  note: most writes here are 1h entries at 2x base input, "
                        "so break-even needs about twice the reads a 5m entry does")
        log.warning("  confirm in money: GET %s/organizations/cost_report"
                    "?starting_at=<T-30d>&group_by[]=description", API)

    log.info("%d key(s), %d losing money on caching", len(by_key), losing)
    return 1 if losing else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report Anthropic cache writes that are never read back.
 *
 * Read only. GET requests and nothing else against the Admin API, which needs
 * an Admin API key (sk-ant-admin...); a workspace key is rejected by every
 * /v1/organizations/* path, and an Admin key can be provisioned read-only.
 * The repair is printed, never performed.
 *
 * The usage report has no request-count field, so "reads per write" means read
 * tokens per write token: a proxy for call counts, not a call count.
 */
const API = 'https://api.anthropic.com/v1';
const VERSION = '2023-06-01';

// Published multipliers on base input.
const WRITE_5M = 1.25;
const WRITE_1H = 2.00;
const READ = 0.10;
const BASE = 1.00;

/** Sum token fields across results, keeping the two TTLs apart. Pure. */
export function accumulate(results, into = null) {
  const total = { uncached: 0, cache_read: 0, write_5m: 0, write_1h: 0, ...(into ?? {}) };
  for (const result of results ?? []) {
    total.uncached += Number(result.uncached_input_tokens ?? 0);
    total.cache_read += Number(result.cache_read_input_tokens ?? 0);
    const creation = result.cache_creation ?? {};
    total.write_5m += Number(creation.ephemeral_5m_input_tokens ?? 0);
    total.write_1h += Number(creation.ephemeral_1h_input_tokens ?? 0);
  }
  return total;
}

/**
 * Read tokens per write token at which caching starts to save money. Pure.
 * About 0.28 for pure 5m traffic, about 1.11 for pure 1h. Null when nothing
 * was written, because a ratio against zero is not a number.
 */
export function breakEvenRatio(write5m, write1h) {
  const writes = write5m + write1h;
  if (writes <= 0) return null;
  const premium = (WRITE_5M - BASE) * write5m + (WRITE_1H - BASE) * write1h;
  return premium / ((BASE - READ) * writes);
}

/**
 * What this cached traffic costs per token relative to not caching. Pure.
 * Above 1.0 means caching is charging a surcharge.
 */
export function effectiveMultiplier(write5m, write1h, reads) {
  const tokens = write5m + write1h + reads;
  if (tokens <= 0) return null;
  return (WRITE_5M * write5m + WRITE_1H * write1h + READ * reads) / tokens;
}

/** Classify one key's cache economics over the window. Pure. */
export function verdict(total, minWrites = 100_000, margin = 1.5) {
  const reads = Number(total.cache_read ?? 0);
  const write5m = Number(total.write_5m ?? 0);
  const write1h = Number(total.write_1h ?? 0);
  const writes = write5m + write1h;

  if (writes === 0 && reads === 0) {
    return ['no-caching',
      'no cache reads and no cache writes in this window: caching is not ' +
      'switched on for this key at all, which is a different problem from this one'];
  }
  if (writes === 0) {
    return ['reads-only',
      `${reads} read token(s) against entries written before this window opened. ` +
      'Widen the window before drawing a ratio from it.'];
  }
  if (writes < minWrites) {
    return ['too-little-traffic',
      `only ${writes} cache write token(s) in the window; too little to draw a ratio from`];
  }

  const ratio = reads / writes;
  const threshold = breakEvenRatio(write5m, write1h);
  const multiplier = effectiveMultiplier(write5m, write1h, reads);
  const shape = `${ratio.toFixed(2)} read tokens per write token against a ` +
    `break-even of ${threshold.toFixed(2)}; this traffic costs ` +
    `${multiplier.toFixed(2)}x what the same tokens would cost with caching switched off`;
  if (ratio < threshold) return ['losing', shape];
  if (ratio < threshold * margin) return ['marginal', `${shape}, which is barely above the line`];
  return ['paying-off', shape];
}

async function get(adminKey, path, params) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) {
    for (const one of Array.isArray(v) ? v : [v]) url.searchParams.append(k, one);
  }
  const res = await fetch(url, {
    headers: { 'x-api-key': adminKey, 'anthropic-version': VERSION },
  });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Anthropic: /v1/organizations/* needs an ` +
                    'Admin API key (sk-ant-admin...), not a workspace key');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

async function* buckets(adminKey, path, params) {
  const q = { ...params };
  for (;;) {
    const page = await get(adminKey, path, q);
    for (const bucket of page.data ?? []) yield bucket;
    if (!page.has_more || !page.next_page) return;
    q.page = page.next_page;
  }
}

/** Floor to the hour: starting_at must sit on a bucket boundary. */
export function windowStart(days, now = new Date()) {
  const top = Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate(),
                       now.getUTCHours());
  return new Date(top - days * 86400000).toISOString().replace(/\.\d{3}Z$/, 'Z');
}

async function main() {
  const adminKey = process.env.ANTHROPIC_ADMIN_KEY;
  if (!adminKey) {
    console.error('set ANTHROPIC_ADMIN_KEY to an Admin API key (sk-ant-admin...); ' +
                  'a workspace key cannot read /v1/organizations/*');
    process.exitCode = 2;
    return;
  }
  const days = Number(process.env.DAYS ?? 7);
  const minWrites = Number(process.env.MIN_WRITES ?? 100_000);

  const params = {
    starting_at: windowStart(days),
    bucket_width: '1h',
    limit: Math.min(days * 24, 168),
    'group_by[]': ['api_key_id'],
  };

  const byKey = new Map();
  for await (const bucket of buckets(adminKey, '/organizations/usage_report/messages',
                                     params)) {
    for (const result of bucket.results ?? []) {
      const name = result.api_key_id ?? 'unattributed';
      byKey.set(name, accumulate([result], byKey.get(name)));
    }
  }

  if (byKey.size === 0) {
    console.log(`no message usage in the last ${days} day(s)`);
    return;
  }

  let losing = 0;
  const ordered = [...byKey.entries()].sort(
    (a, b) => (b[1].write_5m + b[1].write_1h) - (a[1].write_5m + a[1].write_1h));
  for (const [name, total] of ordered) {
    const [state, detail] = verdict(total, minWrites);
    const line = `${state.padEnd(18)} ${name}  ${detail}`;
    if (state !== 'losing' && state !== 'marginal') { console.log(line); continue; }
    losing += 1;
    console.warn(line);
    console.warn('  repair: move the cache_control breakpoint to the end of the stable ' +
                 "prefix and keep timestamps, request ids and the user's question " +
                 'strictly after it, then re-measure this ratio tomorrow');
    if (total.write_1h > total.write_5m) {
      console.warn('  note: most writes here are 1h entries at 2x base input, so ' +
                   'break-even needs about twice the reads a 5m entry does');
    }
    console.warn(`  confirm in money: GET ${API}/organizations/cost_report` +
                 '?starting_at=<T-30d>&group_by[]=description');
  }

  console.log(`${byKey.size} key(s), ${losing} losing money on caching`);
  process.exitCode = losing ? 1 : 0;
}

if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from anthropic_cache_write_ratio import (
    accumulate, break_even_ratio, effective_multiplier, verdict,
)


def test_accumulate_keeps_the_two_ttls_apart():
    # Summing them would destroy the information break-even needs.
    total = accumulate([{
        "cache_read_input_tokens": 5,
        "cache_creation": {"ephemeral_5m_input_tokens": 100,
                           "ephemeral_1h_input_tokens": 20},
    }])
    assert total["write_5m"] == 100
    assert total["write_1h"] == 20
    assert total["cache_read"] == 5


def test_break_even_for_pure_5m_writes():
    # (1.25 - 1) / (1 - 0.1)
    assert round(break_even_ratio(1000, 0), 4) == 0.2778


def test_break_even_for_pure_1h_writes_is_about_four_times_higher():
    # (2.0 - 1) / (1 - 0.1)
    assert round(break_even_ratio(0, 1000), 4) == 1.1111


def test_break_even_of_nothing_written_is_none_not_zero():
    assert break_even_ratio(0, 0) is None


def test_at_break_even_the_effective_multiplier_is_exactly_one():
    # The identity that keeps the two functions from drifting apart.
    for w5, w1h in ((1000, 0), (0, 1000), (600, 400)):
        reads = break_even_ratio(w5, w1h) * (w5 + w1h)
        assert round(effective_multiplier(w5, w1h, reads), 6) == 1.0


def test_writes_with_no_reads_cost_more_than_not_caching():
    assert effective_multiplier(1000, 0, 0) == 1.25
    assert effective_multiplier(0, 1000, 0) == 2.0


def test_a_key_that_writes_and_never_reads_is_losing():
    state, detail = verdict({"cache_read": 0, "write_5m": 5_000_000, "write_1h": 0})
    assert state == "losing"
    assert "1.25x" in detail


def test_a_key_reading_back_many_times_is_paying_off():
    state, _ = verdict({"cache_read": 50_000_000, "write_5m": 5_000_000, "write_1h": 0})
    assert state == "paying-off"


def test_just_above_break_even_is_marginal_not_safe():
    writes = 5_000_000
    reads = int(break_even_ratio(writes, 0) * writes * 1.1)
    assert verdict({"cache_read": reads, "write_5m": writes, "write_1h": 0})[0] == "marginal"


def test_no_writes_and_no_reads_is_the_other_note():
    state, detail = verdict({"cache_read": 0, "write_5m": 0, "write_1h": 0})
    assert state == "no-caching"
    assert "different problem" in detail


def test_reads_with_no_writes_in_the_window_is_not_a_ratio():
    state, detail = verdict({"cache_read": 9_000_000, "write_5m": 0, "write_1h": 0})
    assert state == "reads-only"
    assert "Widen the window" in detail


def test_a_trickle_of_writes_makes_no_claim():
    assert verdict({"cache_read": 0, "write_5m": 10, "write_1h": 0})[0] == "too-little-traffic"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import {
  accumulate, breakEvenRatio, effectiveMultiplier, verdict, windowStart,
} from './anthropic-cache-write-ratio.mjs';

test('accumulate keeps the two TTLs apart', () => {
  const total = accumulate([{
    cache_read_input_tokens: 5,
    cache_creation: { ephemeral_5m_input_tokens: 100, ephemeral_1h_input_tokens: 20 },
  }]);
  assert.equal(total.write_5m, 100);
  assert.equal(total.write_1h, 20);
  assert.equal(total.cache_read, 5);
});

test('break-even for pure 5m writes', () => {
  assert.equal(Number(breakEvenRatio(1000, 0).toFixed(4)), 0.2778);
});

test('break-even for pure 1h writes is about four times higher', () => {
  assert.equal(Number(breakEvenRatio(0, 1000).toFixed(4)), 1.1111);
});

test('break-even of nothing written is null not zero', () => {
  assert.equal(breakEvenRatio(0, 0), null);
});

test('at break-even the effective multiplier is exactly one', () => {
  for (const [w5, w1h] of [[1000, 0], [0, 1000], [600, 400]]) {
    const reads = breakEvenRatio(w5, w1h) * (w5 + w1h);
    assert.equal(Number(effectiveMultiplier(w5, w1h, reads).toFixed(6)), 1);
  }
});

test('writes with no reads cost more than not caching', () => {
  assert.equal(effectiveMultiplier(1000, 0, 0), 1.25);
  assert.equal(effectiveMultiplier(0, 1000, 0), 2.0);
});

test('a key that writes and never reads is losing', () => {
  const [state, detail] = verdict({ cache_read: 0, write_5m: 5_000_000, write_1h: 0 });
  assert.equal(state, 'losing');
  assert.match(detail, /1\.25x/);
});

test('a key reading back many times is paying off', () => {
  assert.equal(
    verdict({ cache_read: 50_000_000, write_5m: 5_000_000, write_1h: 0 })[0],
    'paying-off');
});

test('just above break-even is marginal not safe', () => {
  const writes = 5_000_000;
  const reads = Math.floor(breakEvenRatio(writes, 0) * writes * 1.1);
  assert.equal(verdict({ cache_read: reads, write_5m: writes, write_1h: 0 })[0],
               'marginal');
});

test('no writes and no reads is the other note', () => {
  const [state, detail] = verdict({ cache_read: 0, write_5m: 0, write_1h: 0 });
  assert.equal(state, 'no-caching');
  assert.match(detail, /different problem/);
});

test('reads with no writes in the window is not a ratio', () => {
  const [state, detail] = verdict({ cache_read: 9_000_000, write_5m: 0, write_1h: 0 });
  assert.equal(state, 'reads-only');
  assert.match(detail, /Widen the window/);
});

test('a trickle of writes makes no claim', () => {
  assert.equal(verdict({ cache_read: 0, write_5m: 10, write_1h: 0 })[0],
               'too-little-traffic');
});

test('the window start is floored to the hour', () => {
  assert.equal(windowStart(7, new Date('2026-08-30T13:45:12Z')), '2026-08-23T13:00:00Z');
});