Gpt Weekly Limit Estimation

GPT Weekly Limit Estimation

Compute token usage and estimate weekly limits from Codex CLI / Claude Code SSE logs.

Quick Start

# 1. Extract billing week data (OpenAI resets at 14:47 local time)
python extract_codex_week.py \
  --src ~/centralized-logs/codex/sse_lines.jsonl \
  --dst /tmp/billing_week.jsonl \
  --start "2026-01-15 14:47" \
  --end "2026-01-22 14:47"

# 2. Compute token costs
python compute_token_costs.py \
  --codex /tmp/billing_week.jsonl \
  --claude ~/centralized-logs/claude/sse_lines.jsonl

# 3. Estimate weekly limit from partial usage
# If you've used $147 and dashboard shows 28%:
# Weekly limit = $147 / 0.28 = $525

Critical Knowledge

OpenAI Billing Week Reset

OpenAI resets weekly usage every THURSDAY at 14:47 LOCAL TIME.

To extract a billing week:

from datetime import datetime, timedelta

# OpenAI resets every Thursday at 14:47 local time
reset_day = "Thursday"
reset_time = "14:47"

# Calculate week boundaries (Thursday to Thursday)
end = datetime(2026, 1, 23, 14, 47, 0)  # Thursday reset
start = end - timedelta(days=7)          # Previous Thursday

# Extract: start (inclusive) to end (exclusive)

Token Counting Methodology

CORRECT: Sum last_token_usage (per-call tokens)

# Each turn.token_count event has:
# - info.total_token_usage  <- CUMULATIVE snapshot (DO NOT SUM)
# - info.last_token_usage   <- Per-call tokens (SUM THIS)

for event in events:
    if event["event"] == "turn.token_count":
        last = event["payload"]["info"]["last_token_usage"]
        total_input += last["input_tokens"]
        total_cached += last["cached_input_tokens"]
        total_output += last["output_tokens"]

WRONG: Sum total_token_usage (causes 2-3x overcount)

# DON'T DO THIS - CodexMonitor's bug
# Summing cumulative totals overcounts when sessions reset/mix

Deduplication

Each turn.token_count may appear twice. Dedupe by tracking total_tokens snapshot per session:

last_total_by_sid = {}

for event in events:
    sid = event["sid"]
    total_snapshot = event["payload"]["info"]["total_token_usage"]["total_tokens"]

    # Skip if same snapshot already seen for this session
    if last_total_by_sid.get(sid) == total_snapshot:
        continue
    last_total_by_sid[sid] = total_snapshot

    # Now safe to count
    process_event(event)

Live Pricing (as of 2026-01-24)

OpenAI ($/MTok)

Cached discount: 90% off input price (cached = 0.1 × input)

Anthropic ($/MTok)

Cache multipliers: read=0.1×, write_5m=1.25×, write_1h=2.0×

Weekly Limit Estimation

If OpenAI dashboard shows you've used X% of your weekly limit:

# From partial usage
cost_so_far = 147.20  # From compute_token_costs.py
percent_used = 0.28   # From OpenAI dashboard (28%)

weekly_limit = cost_so_far / percent_used
# $147.20 / 0.28 = $525.71

Effective Cost Per Million Tokens

With typical usage patterns (95% cache hit rate):

File Locations

Ground Truth (Original)

• /Users/sotola/swe/token-computation/compute_token_costs.py - Main computation script
• /Users/sotola/swe/token-computation/pull_pricing.py - Live pricing fetcher
• /Users/sotola/swe/token-computation/token_cost_method.md - Methodology docs

Centralized Logs

• ~/centralized-logs/codex/sse_lines.jsonl - Codex CLI logs (~18GB)
• ~/centralized-logs/claude/sse_lines.jsonl - Claude Code logs (~3.5GB)

Attached Code

extract_codex_week.py

Fast time-range extractor for large JSONL files.

#!/usr/bin/env python
"""[Created by Codex: 019bebac-ddc1-7a53-9923-6e5590ddca25]"""

from __future__ import annotations
import argparse
from collections import Counter
from pathlib import Path
from typing import Optional, Sequence


def _normalize_iso_local(s: str) -> str:
    s = s.strip()
    if " " in s and "T" not in s:
        s = s.replace(" ", "T", 1)
    if len(s) == 16:
        return s + ":00"
    return s


def _extract_str_field(line: str, field: str) -> Optional[str]:
    """Fast field extraction without json.loads"""
    needle = f'"{field}":"'
    start = line.find(needle)
    if start < 0:
        return None
    start += len(needle)
    end = line.find('"', start)
    if end < 0:
        return None
    return line[start:end]


def extract_range(
    *,
    src_path: Path,
    dst_path: Path,
    start_t: str,
    end_t: str,
    events: Sequence[str],
    assume_sorted_by_t: bool,
) -> None:
    start_t = _normalize_iso_local(start_t)
    end_t = _normalize_iso_local(end_t)
    allowed_events = set(events)

    dst_path.parent.mkdir(parents=True, exist_ok=True)

    total_lines = 0
    selected_lines = 0
    event_counts: Counter[str] = Counter()

    with src_path.open("r", encoding="utf-8", errors="replace") as src, \
         dst_path.open("w", encoding="utf-8") as dst:
        for raw in src:
            total_lines += 1
            if not raw:
                continue

            t = _extract_str_field(raw, "t")
            if not t:
                continue

            if t < start_t:
                continue
            if t >= end_t:
                if assume_sorted_by_t:
                    break
                continue

            ev = _extract_str_field(raw, "event")
            if allowed_events and ev not in allowed_events:
                continue

            dst.write(raw)
            selected_lines += 1
            if ev:
                event_counts[ev] += 1

    print(f"Extracted {selected_lines:,} / {total_lines:,} lines")
    print(f"  {start_t} → {end_t}")
    for ev, n in event_counts.most_common():
        print(f"  {ev}: {n:,}")


def main(argv: Optional[Sequence[str]] = None) -> int:
    ap = argparse.ArgumentParser(description="Extract time window from SSE JSONL")
    ap.add_argument("--src", required=True)
    ap.add_argument("--dst", required=True)
    ap.add_argument("--start", required=True, help='e.g. "2026-01-15 14:47"')
    ap.add_argument("--end", required=True, help='e.g. "2026-01-22 14:47"')
    ap.add_argument("--events", default="turn.token_count,turn.session_configured")
    ap.add_argument("--assume-sorted-by-t", action="store_true")
    args = ap.parse_args(argv)

    events = [e.strip() for e in args.events.split(",") if e.strip()]
    extract_range(
        src_path=Path(args.src).expanduser(),
        dst_path=Path(args.dst).expanduser(),
        start_t=args.start,
        end_t=args.end,
        events=events,
        assume_sorted_by_t=args.assume_sorted_by_t,
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

compute_token_costs.py (core logic)

#!/usr/bin/env python3
"""Token cost computation - core aggregation logic
[Original: /Users/sotola/swe/token-computation/compute_token_costs.py]
"""

import json
from collections import defaultdict
from dataclasses import dataclass
from typing import Dict, Iterator, Any, Optional


def iter_jsonl(path: str) -> Iterator[Dict[str, Any]]:
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                yield json.loads(line)
            except json.JSONDecodeError:
                continue


def parse_sse_data_line(s: str) -> Optional[Dict[str, Any]]:
    if not isinstance(s, str) or not s:
        return None
    if s.startswith("data: "):
        s = s[6:]
    try:
        return json.loads(s)
    except json.JSONDecodeError:
        return None


@dataclass
class CodexTotals:
    model: Optional[str]
    calls: int
    input_tokens: int
    cached_input_tokens: int
    output_tokens: int
    reasoning_output_tokens: int
    total_tokens: int


def compute_codex_totals(path: str) -> CodexTotals:
    model: Optional[str] = None
    last_total_by_sid: Dict[str, Optional[int]] = {}
    sums = defaultdict(int)
    calls = 0

    for outer in iter_jsonl(path):
        ev = outer.get("event")
        inner = parse_sse_data_line(outer.get("line", ""))
        if not inner:
            continue

        if ev == "turn.session_configured":
            payload = inner.get("payload", {})
            if isinstance(payload, dict) and payload.get("model"):
                model = payload.get("model")

        if ev != "turn.token_count":
            continue

        info = inner.get("payload", {}).get("info", {})
        if info is None:
            continue

        total = info.get("total_token_usage", {})
        last = info.get("last_token_usage", {})
        sid = outer.get("sid") or "(unknown)"

        # Dedupe by total_tokens snapshot
        total_tokens_snapshot = total.get("total_tokens")
        if not isinstance(total_tokens_snapshot, int):
            continue

        if last_total_by_sid.get(sid) == total_tokens_snapshot:
            continue
        last_total_by_sid[sid] = total_tokens_snapshot
        calls += 1

        # Sum from last_token_usage (per-call tokens)
        for k in ("input_tokens", "cached_input_tokens", "output_tokens",
                  "reasoning_output_tokens", "total_tokens"):
            v = last.get(k)
            if isinstance(v, int):
                sums[k] += v

    return CodexTotals(
        model=model,
        calls=calls,
        input_tokens=int(sums["input_tokens"]),
        cached_input_tokens=int(sums["cached_input_tokens"]),
        output_tokens=int(sums["output_tokens"]),
        reasoning_output_tokens=int(sums["reasoning_output_tokens"]),
        total_tokens=int(sums["total_tokens"]),
    )


def cost_openai(totals: CodexTotals, tier: str = "standard") -> Dict[str, float]:
    """Compute cost using GPT-5.2 pricing"""
    pricing = {
        "standard": (1.75, 0.175, 14.0),
        "flex": (0.875, 0.0875, 7.0),
        "priority": (3.50, 0.35, 28.0),
    }
    inp, cached, out = pricing.get(tier, pricing["standard"])

    noncached = max(0, totals.input_tokens - totals.cached_input_tokens)
    return {
        "input_noncached": (noncached / 1e6) * inp,
        "input_cached": (totals.cached_input_tokens / 1e6) * cached,
        "output": (totals.output_tokens / 1e6) * out,
        "total": (noncached / 1e6) * inp +
                 (totals.cached_input_tokens / 1e6) * cached +
                 (totals.output_tokens / 1e6) * out,
    }

Common Tasks

Extract and compute for a billing week

# Step 1: Find your reset time from OpenAI dashboard
# Step 2: Calculate week boundaries
python -c "
from datetime import datetime, timedelta
end = datetime(2026, 1, 22, 14, 47)
start = end - timedelta(days=7)
print(f'--start \"{start.strftime(\"%Y-%m-%d %H:%M\")}\" --end \"{end.strftime(\"%Y-%m-%d %H:%M\")}\"')
"

# Step 3: Extract
python extract_codex_week.py \
  --src ~/centralized-logs/codex/sse_lines.jsonl \
  --dst /tmp/billing_week.jsonl \
  --start "2026-01-15 14:47" \
  --end "2026-01-22 14:47" \
  --assume-sorted-by-t

# Step 4: Compute
cd /Users/sotola/swe/token-computation
python compute_token_costs.py \
  --codex /tmp/billing_week.jsonl \
  --claude claude_sse_lines.jsonl

Estimate weekly limit from current usage

# Read from compute_token_costs.py JSON output
import json

# Get cost from script output
cost_so_far = 147.20  # dollars

# Get percentage from OpenAI dashboard
percent_used = 0.28  # 28%

weekly_limit = cost_so_far / percent_used
print(f"Estimated weekly limit: ${weekly_limit:.2f}")
# $525.71

Related Files

• Full original script: /Users/sotola/swe/token-computation/compute_token_costs.py
• Pricing fetcher: /Users/sotola/swe/token-computation/pull_pricing.py
• Methodology docs: /Users/sotola/swe/token-computation/token_cost_method.md
• Investigation report: /tmp/investigation-token-count-discrepancy.md