"""
Measure characters-per-token ratio across text types using tiktoken.

Samples are self-authored for this measurement (not copied from any
copyrighted source). Ten segments per category:
  - english_prose
  - chinese_traditional
  - chinese_simplified
  - python_code
  - json

Two OpenAI tokenizers are measured: cl100k_base (GPT-4 / GPT-3.5-turbo family)
and o200k_base (GPT-4o-and-later family). Claude and Gemini tokenizers
are NOT measured here -- no numbers for them are reported or implied
anywhere in this script or the article that uses its output.

Run:
    pip install tiktoken
    python measure_chars_per_token.py

Writes: measure_chars_per_token.results.json (next to this script).
"""

import json
import statistics
from pathlib import Path

import tiktoken

ENGLISH_PROSE = [
    "A token counter estimates how many units a language model will bill "
    "for a piece of text before you send it. Developers use this to avoid "
    "surprise costs on long prompts.",
    "Most API pricing is quoted per million tokens, split into an input "
    "rate and a higher output rate. Reading the fine print on both numbers "
    "matters more than picking a vendor by reputation alone.",
    "Caching the stable part of a prompt can cut the effective input cost "
    "by a wide margin. The trick only works if the cached prefix does not "
    "change between requests.",
    "Retrieval systems fetch a handful of passages and stuff them into the "
    "context window before the model ever sees the user's question. That "
    "retrieved text is usually the largest single cost driver.",
    "A chat session accumulates history with every turn, so message ten "
    "carries the token weight of the nine before it. Long conversations "
    "get expensive quietly, not suddenly.",
    "Batch endpoints trade latency for a lower per-token rate, which suits "
    "workloads like nightly classification jobs that do not need an answer "
    "in real time.",
    "Reasoning models bill their internal thinking steps as output tokens "
    "even when the visible answer is short. A one-sentence reply can still "
    "carry a large invisible cost.",
    "Rate limits are usually expressed as both requests per minute and "
    "tokens per minute, and whichever ceiling you hit first is the real "
    "constraint on throughput.",
    "Embedding a document collection is a one-time cost, but re-embedding "
    "after every edit adds up fast if the pipeline is not selective about "
    "what actually changed.",
    "Four characters per token is a rule of thumb for English prose, not a "
    "law. It breaks down for code, dense identifiers, and any language "
    "that is not English.",
]

CHINESE_TRADITIONAL = [
    "計算 API 費用時，最容易被忽略的是輸出 token 通常比輸入貴上好幾倍，"
    "所以回覆的長度往往比提示本身更影響帳單。",
    "上下文視窗指的是模型一次能處理的 token 總量，超過這個上限，"
    "多餘的內容就會被截斷或直接被拒絕。",
    "快取機制可以讓重複出現的系統提示以較低的價格計費，"
    "但前提是每次請求送出的內容前綴必須完全相同。",
    "檢索增強生成會把搜尋到的文件片段塞進提示裡，"
    "這些片段通常才是整個請求裡佔比最大的部分。",
    "對話歷史會隨著回合數增加而持續變長，"
    "第十句話的成本其實包含了前面九句的全部內容。",
    "批次處理犧牲即時性換取較低的單價，"
    "適合像是每日分類這種不需要立刻拿到結果的工作。",
    "推理模型會把內部的思考過程當成輸出 token 計費，"
    "即使畫面上只顯示一句簡短的答案，背後仍可能耗費大量隱藏成本。",
    "速率限制通常同時用每分鐘請求數與每分鐘 token 數表示，"
    "先撞到哪一個上限，那個就是實際的吞吐瓶頸。",
    "把整個文件集合做嵌入是一次性的成本，"
    "但如果每次小幅修改都重新嵌入，累積下來也會相當可觀。",
    "四個字元約等於一個 token 這個經驗法則只適用於英文散文，"
    "對中文、程式碼或大量代號來說完全不成立。",
]

CHINESE_SIMPLIFIED = [
    "计算 API 费用时，最容易被忽略的是输出 token 通常比输入贵上好几倍，"
    "所以回复的长度往往比提示本身更影响账单。",
    "上下文窗口指的是模型一次能处理的 token 总量，超过这个上限，"
    "多余的内容就会被截断或直接被拒绝。",
    "缓存机制可以让重复出现的系统提示以较低的价格计费，"
    "但前提是每次请求发送的内容前缀必须完全相同。",
    "检索增强生成会把搜索到的文档片段塞进提示里，"
    "这些片段通常才是整个请求里占比最大的部分。",
    "对话历史会随着轮数增加而持续变长，"
    "第十句话的成本其实包含了前面九句的全部内容。",
    "批处理牺牲即时性换取较低的单价，"
    "适合像是每日分类这种不需要立刻拿到结果的工作。",
    "推理模型会把内部的思考过程当成输出 token 计费，"
    "即使画面上只显示一句简短的答案，背后仍可能耗费大量隐藏成本。",
    "速率限制通常同时用每分钟请求数与每分钟 token 数表示，"
    "先撞到哪一个上限，那个就是实际的吞吐瓶颈。",
    "把整个文档集合做嵌入是一次性的成本，"
    "但如果每次小幅修改都重新嵌入，累积下来也会相当可观。",
    "四个字符约等于一个 token 这个经验法则只适用于英文散文，"
    "对中文、代码或大量代号来说完全不成立。",
]

PYTHON_CODE = [
    "def estimate_tokens(text: str) -> int:\n"
    "    \"\"\"Rough heuristic: 4 characters per token.\"\"\"\n"
    "    return max(1, len(text) // 4)\n",
    "class PricingRow:\n"
    "    def __init__(self, name, in_rate, out_rate):\n"
    "        self.name = name\n"
    "        self.in_rate = in_rate\n"
    "        self.out_rate = out_rate\n"
    "\n"
    "    def monthly_cost(self, in_tok, out_tok, requests):\n"
    "        per_req = in_tok * self.in_rate + out_tok * self.out_rate\n"
    "        return per_req / 1_000_000 * requests\n",
    "import json\n"
    "\n"
    "def load_pricing(path):\n"
    "    with open(path, 'r', encoding='utf-8') as fh:\n"
    "        data = json.load(fh)\n"
    "    return {row['id']: row for row in data['chat']}\n",
    "def chunk_text(text, size=500, overlap=50):\n"
    "    chunks = []\n"
    "    start = 0\n"
    "    while start < len(text):\n"
    "        end = start + size\n"
    "        chunks.append(text[start:end])\n"
    "        start = end - overlap\n"
    "    return chunks\n",
    "async def call_model(client, prompt, model='gpt-5.6'):\n"
    "    response = await client.chat.completions.create(\n"
    "        model=model,\n"
    "        messages=[{'role': 'user', 'content': prompt}],\n"
    "    )\n"
    "    return response.choices[0].message.content\n",
    "def cheapest_model(rows, in_tok, out_tok, requests):\n"
    "    scored = [\n"
    "        (row.monthly_cost(in_tok, out_tok, requests), row.name)\n"
    "        for row in rows\n"
    "    ]\n"
    "    scored.sort()\n"
    "    return scored[0]\n",
    "def retry_with_backoff(fn, attempts=3, base_delay=1.0):\n"
    "    for i in range(attempts):\n"
    "        try:\n"
    "            return fn()\n"
    "        except Exception:\n"
    "            if i == attempts - 1:\n"
    "                raise\n"
    "            time.sleep(base_delay * (2 ** i))\n",
    "def embed_batch(client, texts, model='text-embedding-3-small'):\n"
    "    resp = client.embeddings.create(model=model, input=texts)\n"
    "    return [item.embedding for item in resp.data]\n",
    "def tokens_per_dollar(rate_per_million):\n"
    "    if rate_per_million <= 0:\n"
    "        raise ValueError('rate must be positive')\n"
    "    return 1_000_000 / rate_per_million\n",
    "for model in pricing['chat']:\n"
    "    cost = model['in'] * 2 + model['out'] * 0.5\n"
    "    print(f\"{model['name']}: ${cost:.2f}\")\n",
]

JSON_SAMPLES = [
    '{"id": "gpt-5.6", "name": "GPT-5.6", "vendor": "OpenAI", "in": 4.0, "out": 20.0, "context": 922000}',
    '{"date": "2026-08-22", "field": "in", "from": 5.0, "to": 4.0, "type": "changed"}',
    '{"chunk_id": 17, "doc_id": "handbook-03", "text_len": 512, "overlap": 50, "embedded": true}',
    '{"request": {"input_tokens": 2000, "output_tokens": 500, "requests_per_month": 100000}}',
    '{"model": "text-embedding-3-small", "vendor": "OpenAI", "rate_per_million": 0.02}',
    '{"tracking_since": "2026-07-25", "last_checked": "2026-09-14", "events_count": 8}',
    '{"query": "what changed this quarter", "top_k": 5, "rerank": false, "max_tokens": 300}',
    '{"pipeline": "rag", "steps": ["embed", "retrieve", "rerank", "generate"], "cache": false}',
    '{"vendor_totals": {"OpenAI": 5, "Google": 2, "Anthropic": 1}, "period": "2026-Q3"}',
    '{"result": {"cheapest": "Gemini 3.6 Flash", "monthly_cost_usd": 3.38, "requests": 100000}}',
]

CATEGORIES = {
    "english_prose": ENGLISH_PROSE,
    "chinese_traditional": CHINESE_TRADITIONAL,
    "chinese_simplified": CHINESE_SIMPLIFIED,
    "python_code": PYTHON_CODE,
    "json": JSON_SAMPLES,
}

ENCODINGS = ["cl100k_base", "o200k_base"]


def measure():
    encoders = {name: tiktoken.get_encoding(name) for name in ENCODINGS}
    results = {"encodings_measured": ENCODINGS, "categories": {}}

    for category, samples in CATEGORIES.items():
        cat_result = {"sample_count": len(samples), "by_encoding": {}}
        for enc_name in ENCODINGS:
            enc = encoders[enc_name]
            ratios = []
            per_sample = []
            for text in samples:
                n_chars = len(text)
                n_tokens = len(enc.encode(text))
                ratio = n_chars / n_tokens if n_tokens else 0
                ratios.append(ratio)
                per_sample.append({
                    "chars": n_chars,
                    "tokens": n_tokens,
                    "chars_per_token": round(ratio, 3),
                })
            cat_result["by_encoding"][enc_name] = {
                "avg_chars_per_token": round(statistics.mean(ratios), 3),
                "min_chars_per_token": round(min(ratios), 3),
                "max_chars_per_token": round(max(ratios), 3),
                "samples": per_sample,
            }
        results["categories"][category] = cat_result

    return results


def main():
    results = measure()
    out_path = Path(__file__).with_name("measure_chars_per_token.results.json")
    out_path.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8")

    print(f"Wrote {out_path}")
    print()
    header = f"{'category':<22}{'encoding':<14}{'avg':>8}{'min':>8}{'max':>8}"
    print(header)
    print("-" * len(header))
    for category, cat_result in results["categories"].items():
        for enc_name, enc_result in cat_result["by_encoding"].items():
            print(
                f"{category:<22}{enc_name:<14}"
                f"{enc_result['avg_chars_per_token']:>8}"
                f"{enc_result['min_chars_per_token']:>8}"
                f"{enc_result['max_chars_per_token']:>8}"
            )


if __name__ == "__main__":
    main()
