【我在L站搞科研】浅析 Kimi 会员计划模型计费制度：以 Allegro 套餐 Code 额度为例

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

"""

Kimi Coding Plan 轻量级档位额度与倍率快速测算脚本

用法：

  export KIMI_API_KEY="sk-kimi-..."

  python kimi_plan_quick_probe.py

"""



import argparse

import json

import os

import sys

import time

import requests



BASE_URL = "https://api.kimi.com/coding/v1"



# 老版 Allegro (699元) 套餐实测权重 (5h 窗口每 M 消耗比例，2026-10-25 凌晨 UTC+8 复测)

# K3.1 系列在 2026-10-24 早上 UTC+8 首轮为 (0.144, 0.00075, 1.156) / (0.288, 0.0015, 2.312)，之后被下调

# K3 / K3-256k 按实际总长 prompt_tokens + completion_tokens 是否 > 256K 分段，越过后整段输出 ×2

ALLEGRO_BASE_5H = {

    "K3.1 Pro":                  {"w_in": 0.0936,     "w_cr": 0.000473,     "w_out": 0.751,     "cn_api": (13.0, 1.30, 68.0)},

    "K3.1 Pro Highspeed":        {"w_in": 0.1872,     "w_cr": 0.000946,     "w_out": 1.502,     "cn_api": (26.0, 1.30, 136.0)},

    "K3 / K3-256k (总长≤256K)":  {"w_in": 0.1100,     "w_cr": 0.000549,     "w_out": 0.8836,    "cn_api": (20.0, 2.00, 100.0)},

    "K3 / K3-256k (总长>256K)":  {"w_in": 0.1100,     "w_cr": 0.000549,     "w_out": 1.7672,    "cn_api": (20.0, 2.00, 100.0)},

    "K2.8 Preview":              {"w_in": 1.0 / 15.0, "w_cr": 1.0 / 3000.0, "w_out": 2.0 / 3.0, "cn_api": (6.8,  1.36, 28.8)},

    "K2.7 Code Highspeed":       {"w_in": 0.20000,    "w_cr": 0.001000,     "w_out": 2.00000,   "cn_api": (13.7, 2.74, 57.6)},

}



def get_usages(api_key: str):

    headers = {"Authorization": f"Bearer {api_key}"}

    r = requests.get(f"{BASE_URL}/usages", headers=headers, timeout=10)

    r.raise_for_status()

    data = r.json()

    u5 = data["usages"]["limit_5h"]["used_ratio"]

    u7 = data["usages"]["limit_7d"]["used_ratio"]

    r5 = data["usages"]["limit_5h"]["reset_time"]

    r7 = data["usages"]["limit_7d"]["reset_time"]

    return u5, u7, r5, r7



def stream_call(api_key: str, model: str, messages: list, max_tokens: int):

    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}

    payload = {

        "model": model,

        "messages": messages,

        "max_tokens": max_tokens,

        "stream": True,

        "stream_options": {"include_usage": True},

    }

    t0 = time.perf_counter()

    r = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload, stream=True, timeout=180)

    r.raise_for_status()



    t_first = None

    t_last = None

    usage = {}

    for line in r.iter_lines():

        if not line:

            continue

        s = line.decode("utf-8", errors="replace")

        if not s.startswith("data: "):

            continue

        d = s[6:].strip()

        if d == "[DONE]":

            break

        try:

            obj = json.loads(d)

        except Exception:

            continue

        if obj.get("usage"):

            usage = obj["usage"]

        choices = obj.get("choices") or []

        if choices:

            delta = choices[0].get("delta") or {}

            if delta.get("reasoning_content") or delta.get("content"):

                if t_first is None:

                    t_first = time.perf_counter()

                t_last = time.perf_counter()



    t_end = time.perf_counter()

    ttft = (t_first - t0) if t_first else (t_end - t0)

    gen_t = (t_last - t_first) if (t_first and t_last and t_last > t_first) else 0.0

    pt = usage.get("prompt_tokens", 0)

    ct = usage.get("cached_tokens", 0) or (usage.get("prompt_tokens_details") or {}).get("cached_tokens", 0)

    out = usage.get("completion_tokens", 0)

    tps = (out / gen_t) if (gen_t > 0.05 and out > 0) else 0.0

    return pt - ct, ct, out, ttft, tps



def main():

    parser = argparse.ArgumentParser(description="Kimi Coding Plan 轻量级额度与倍率快速测算脚本")

    parser.add_argument("--api-key", default=os.environ.get("KIMI_API_KEY", ""), help="Kimi Coding API Key (sk-kimi-...)")

    parser.add_argument("--model", default="kimi-for-coding", choices=["k3.1-pro", "kimi-for-coding"],

                        help="用于测算的基准模型（kimi-for-coding 的权重两轮实测一致，作为默认基准）")

    args = parser.parse_args()



    api_key = args.api_key.strip()

    if not api_key:

        print("错误: 请通过 --api-key sk-kimi-... 或环境变量 KIMI_API_KEY 提供 API Key")

        sys.exit(1)



    u5_init, u7_init, r5, r7 = get_usages(api_key)

    print("=" * 76)

    print("  Kimi Coding Plan 轻量级额度与计费倍率测算")

    print("=" * 76)

    print(f"当前 5h 已用比例: {u5_init*100:.2f}% (重置时间: {r5})")

    print(f"当前 7d 已用比例: {u7_init*100:.2f}% (重置时间: {r7})")

    print("-" * 76)



    model = args.model

    offset = 96 if model == "k3.1-pro" else 97

    uid = str(int(time.time() * 1000))[-8:]

    sys_40k = f"Session {uid}: " + " ".join(["apple"] * (160 * 256 - offset))



    # Step 1: 40.96K 未命中输入测试

    b5, b7, _, _ = get_usages(api_key)

    in1, cr1, out1, ttft_pre, _ = stream_call(

        api_key, model,

        [{"role": "system", "content": sys_40k}, {"role": "user", "content": "Reply with the single word OK. [0]"}],

        max_tokens=32,

    )

    time.sleep(1.8)

    m5_1, m7_1, _, _ = get_usages(api_key)

    d5_in = m5_1 - b5

    print(f"[测试 1/3: 未缓存输入] in_miss={in1:6d}, cache={cr1:6d}, out={out1:4d} | delta_5h={d5_in:+.4f} | Prefill TTFT={ttft_pre:.2f}s")



    # Step 2: 2 次 40.96K 连续缓存命中测试 (逐次累加实际 Token 计数)

    in2_sum = cr2_sum = out2_sum = 0

    ttft_hit = 0.0

    for i in range(1, 3):

        in_c, cr_c, out_c, ttft_hit, _ = stream_call(

            api_key, model,

            [{"role": "system", "content": sys_40k}, {"role": "user", "content": f"Reply with the single word OK. [{i}]"}],

            max_tokens=16,

        )

        in2_sum += in_c

        cr2_sum += cr_c

        out2_sum += out_c

        time.sleep(1.5)

    m5_2, m7_2, _, _ = get_usages(api_key)

    d5_cr = m5_2 - m5_1

    print(f"[测试 2/3: 连续缓存x2] in_miss={in2_sum:6d}, cache={cr2_sum:6d}, out={out2_sum:4d} | delta_5h={d5_cr:+.4f} | CacheHit TTFT={ttft_hit:.2f}s")



    # Step 3: 2048 Token 输出测试

    uid_out = str(int(time.time() * 1000))[-8:]

    in3, cr3, out3, ttft_short, tps_dec = stream_call(

        api_key, model,

        [

            {"role": "system", "content": f"Session {uid_out}: Output benchmark."},

            {"role": "user", "content": "Write a continuous numbered list of short technical definitions in English from 1 to 2500, one per line."},

        ],

        max_tokens=2048,

    )

    time.sleep(1.8)

    u5_end, u7_end, _, _ = get_usages(api_key)

    d5_out = u5_end - m5_2

    print(f"[测试 3/3: 输出测试  ] in_miss={in3:6d}, cache={cr3:6d}, out={out3:4d} | delta_5h={d5_out:+.4f} | Short TTFT={ttft_short:.2f}s, Decode TPS={tps_dec:.1f} tok/s")



    tot_in = in1 + in2_sum + in3

    tot_cr = cr1 + cr2_sum + cr3

    tot_out = out1 + out2_sum + out3

    tot_d5 = u5_end - u5_init

    tot_d7 = u7_end - u7_init



    base = ALLEGRO_BASE_5H["K3.1 Pro" if model == "k3.1-pro" else "K2.8 Preview"]

    expected_allegro_5h = (base["w_in"] * tot_in + base["w_cr"] * tot_cr + base["w_out"] * tot_out) / 1e6



    capacity_scale = (expected_allegro_5h / tot_d5) if tot_d5 > 0 else 1.0

    if tot_d7 >= 0.0005:

        ratio_5h_to_7d = tot_d5 / tot_d7

        ratio_note = f"实测估计值 {ratio_5h_to_7d:.1f} : 1"

    else:

        ratio_5h_to_7d = 5.0

        ratio_note = "7d 增量较小，采用基准比例 5.0 : 1"



    print("\n" + "=" * 76)

    print("  测算结果汇总与当前订阅内模型额度推算")

    print("=" * 76)

    print(f"本次测试总消耗: 5h 额度 {tot_d5*100:.2f}% | 7d 额度 {tot_d7*100:.2f}%")

    print(f"Allegro (699元) 基准理论消耗 (2026-10-25 权重): 5h 额度 {expected_allegro_5h*100:.3f}%")

    print(f"--> 当前套餐 5h 额度容量约为老版 Allegro (699元) 套餐的: 【 {capacity_scale:.2f} 倍 】")

    print(f"--> 7d / 5h 窗口容量比: 【 {ratio_note} 】")

    print("-" * 76)

    print(f"{'模型名称':26s} | {'5h纯输入上限':12s} | {'5h纯输出上限':12s} | {'参考工况5h等效API':18s} | {'7d满额度等效API':14s}")

    print("-" * 76)



    r_in_out = 2.5

    h_cache = 0.975

    c_cr_out = r_in_out * h_cache / (1.0 - h_cache)



    for m_name, cfg in ALLEGRO_BASE_5H.items():

        w_in_user = cfg["w_in"] / capacity_scale

        w_cr_user = cfg["w_cr"] / capacity_scale

        w_out_user = cfg["w_out"] / capacity_scale



        max_in_m = 1.0 / w_in_user

        max_out_m = 1.0 / w_out_user



        p_in, p_cr, p_out = cfg["cn_api"]

        val_5h = (p_in * r_in_out + p_cr * c_cr_out + p_out) / (w_in_user * r_in_out + w_cr_user * c_cr_out + w_out_user)

        val_7d = val_5h * ratio_5h_to_7d



        print(f"{m_name:26s} | {max_in_m:9.2f} M | {max_out_m:9.2f} M | {val_5h:14.1f} 元/5h | {val_7d:11.1f} 元/7d")

    print("=" * 76)



if __name__ == "__main__":

    main()

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

"""

Kimi Coding Plan 多模型计费权重测定与流式吞吐基准测试脚本

用法：

  export KIMI_API_KEY="sk-kimi-..."

  python kimi_plan_full_benchmark.py --phase all

  python kimi_plan_full_benchmark.py --phase analyze --log bench_all_models_log.jsonl

"""



import argparse

import json

import os

import sys

import time

import requests

from scipy.optimize import linprog



BASE_URL = "https://api.kimi.com/coding/v1"

K3_THRESHOLD = 256000



OFFSETS = {

    "k3.1-pro": 96,

    "k3.1-pro-highspeed": 96,

    "k3-256k": 97,

    "k3": 97,

    "kimi-for-coding": 97,

    "kimi-for-coding-highspeed": 11,

}



def get_headers(api_key: str) -> dict:

    return {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}



def get_usages(api_key: str):

    headers = get_headers(api_key)

    for attempt in range(5):

        try:

            r = requests.get(f"{BASE_URL}/usages", headers=headers, timeout=10)

            r.raise_for_status()

            data = r.json()

            return data["usages"]["limit_5h"]["used_ratio"], data["usages"]["limit_7d"]["used_ratio"], data

        except Exception:

            if attempt == 4:

                raise

            time.sleep(1.0)



def stream_chat(api_key: str, model: str, messages: list, max_tokens: int = 64):

    headers = get_headers(api_key)

    payload = {

        "model": model,

        "messages": messages,

        "max_tokens": max_tokens,

        "stream": True,

        "stream_options": {"include_usage": True},

    }

    t_start = time.perf_counter()

    r = requests.post(f"{BASE_URL}/chat/completions", headers=headers, json=payload, stream=True, timeout=300)

    if r.status_code != 200:

        return {

            "status_code": r.status_code,

            "error": r.text[:500],

            "ttft": None,

            "gen_time": None,

            "total_time": time.perf_counter() - t_start,

            "tps_decode": None,

            "tps_e2e": None,

            "usage": {},

        }



    t_first_token = None

    t_last_token = None

    chunk_count = 0

    usage = {}



    for line in r.iter_lines():

        if not line:

            continue

        s = line.decode("utf-8", errors="replace")

        if not s.startswith("data: "):

            continue

        data_str = s[6:].strip()

        if data_str == "[DONE]":

            break

        try:

            obj = json.loads(data_str)

        except Exception:

            continue

        if obj.get("usage"):

            usage = obj["usage"]

        choices = obj.get("choices") or []

        if choices:

            delta = choices[0].get("delta") or {}

            if delta.get("reasoning_content") or delta.get("content"):

                if t_first_token is None:

                    t_first_token = time.perf_counter()

                t_last_token = time.perf_counter()

                chunk_count += 1



    t_end = time.perf_counter()

    ttft = (t_first_token - t_start) if t_first_token else (t_end - t_start)

    gen_time = (t_last_token - t_first_token) if (t_first_token and t_last_token and t_last_token > t_first_token) else 0.0

    total_time = t_end - t_start

    comp_tokens = usage.get("completion_tokens", 0)

    tps_decode = (comp_tokens / gen_time) if (gen_time > 0.05 and comp_tokens > 0) else None

    tps_e2e = (comp_tokens / total_time) if (total_time > 0 and comp_tokens > 0) else None



    return {

        "status_code": 200,

        "ttft": round(ttft, 4),

        "gen_time": round(gen_time, 4),

        "total_time": round(total_time, 4),

        "chunks": chunk_count,

        "tps_decode": round(tps_decode, 2) if tps_decode else None,

        "tps_e2e": round(tps_e2e, 2) if tps_e2e else None,

        "usage": usage,

    }



def make_aligned_sys(model: str, blocks_256: int, uid: str = None):

    if uid is None:

        uid = str(int(time.time() * 1000))[-8:]

    offset = OFFSETS.get(model, 97)

    n_apples = blocks_256 * 256 - offset

    return f"Session {uid}: " + " ".join(["apple"] * n_apples), uid



def short_output_msgs(uid: str, target_words: int = 3000):

    return [

        {"role": "system", "content": f"Session {uid}: Output generator."},

        {

            "role": "user",

            "content": (

                f"Write a continuous numbered list of short technical definitions in English from 1 to {target_words}, "

                "one per line, without stopping early."

            ),

        },

    ]



def run_step(api_key: str, log_path: str, tag: str, model: str, messages: list, max_tokens: int = 64, sleep_after: float = 1.8):

    b5, b7, _ = get_usages(api_key)

    res = stream_chat(api_key, model, messages, max_tokens=max_tokens)

    time.sleep(sleep_after)

    a5, a7, _ = get_usages(api_key)



    u = res.get("usage") or {}

    pt = u.get("prompt_tokens", 0)

    ct = u.get("cached_tokens", 0) or (u.get("prompt_tokens_details") or {}).get("cached_tokens", 0)

    in_miss = pt - ct

    out_tok = u.get("completion_tokens", 0)

    reas_tok = (u.get("completion_tokens_details") or {}).get("reasoning_tokens", 0)



    d5, d7 = round(a5 - b5, 4), round(a7 - b7, 4)

    record = {

        "ts": time.time(),

        "tag": tag,

        "model": model,

        "status_code": res["status_code"],

        "error": res.get("error"),

        "prompt_tokens": pt,

        "cached_tokens": ct,

        "in_miss": in_miss,

        "completion_tokens": out_tok,

        "reasoning_tokens": reas_tok,

        "ttft": res.get("ttft"),

        "gen_time": res.get("gen_time"),

        "total_time": res.get("total_time"),

        "tps_decode": res.get("tps_decode"),

        "tps_e2e": res.get("tps_e2e"),

        "before_5h": b5,

        "after_5h": a5,

        "delta_5h": d5,

        "before_7d": b7,

        "after_7d": a7,

        "delta_7d": d7,

    }

    with open(log_path, "a", encoding="utf-8") as f:

        f.write(json.dumps(record, ensure_ascii=False) + "\n")



    print(

        f"[{tag:30s}] model={model:25s} | in_miss={in_miss:7d} cache={ct:7d} out={out_tok:5d} (reas={reas_tok:4d}) "

        f"| 5h: {b5:.4f}->{a5:.4f} ({d5:+.4f}) | 7d: {b7:.4f}->{a7:.4f} ({d7:+.4f}) "

        f"| TTFT={str(res.get('ttft')):6s}s TPS={str(res.get('tps_decode')):6s}",

        flush=True,

    )

    return record



def solve_lp_segment(name: str, seg_rows: list, window: str = "5h"):

    """

    基于连续实验段的线性规划区间拟合求解器：

    自动按相邻检查点配额读数衔接关系拆分为连续无间断子段（避免跨模型交替调用时直接累加造成混杂），

    并在每个连续子段内引入该子段的初始未四舍五入潜变量，联立求解最大安全裕度中心解与各权重的可行区间。

    """

    valid = [r for r in seg_rows if r.get("status_code", 200) == 200]

    if not valid:

        return



    b_key = f"before_{window}"

    a_key = f"after_{window}"

    delta = 0.00005



    # 将记录按配额衔接拆分为连续子段 (sub-segments)

    sub_segments = []

    cur_seg = [valid[0]]

    for r in valid[1:]:

        if abs(r[b_key] - cur_seg[-1][a_key]) < 1e-6:

            cur_seg.append(r)

        else:

            sub_segments.append(cur_seg)

            cur_seg = [r]

    sub_segments.append(cur_seg)



    num_segs = len(sub_segments)

    # 决策变量顺序: [x0_0, ..., x0_{S-1}, w_in, w_cr, w_out, eps]

    dim_total = num_segs + 4

    idx_win = num_segs

    idx_wcr = num_segs + 1

    idx_wout = num_segs + 2

    idx_eps = num_segs + 3



    A_ub = []

    b_ub = []

    A_ub_proj = []

    b_ub_proj = []



    tot_in = tot_cr = tot_out = 0.0

    bounds_main = []



    for s_idx, seg in enumerate(sub_segments):

        b0 = seg[0][b_key]

        bounds_main.append((b0 - delta, b0 + delta))



        # k = 0 初始检查点约束: |x0_s - b0| + eps <= delta

        row_pos = [0.0] * dim_total

        row_neg = [0.0] * dim_total

        row_pos[s_idx], row_pos[idx_eps] = 1.0, 1.0

        row_neg[s_idx], row_neg[idx_eps] = -1.0, 1.0

        A_ub.extend([row_pos, row_neg])

        b_ub.extend([b0 + delta, -b0 + delta])



        A_ub_proj.extend([row_pos[:num_segs + 3], row_neg[:num_segs + 3]])

        b_ub_proj.extend([b0 + delta, -b0 + delta])



        cum_in = cum_cr = cum_out = 0.0

        for r in seg:

            cum_in += r["in_miss"] / 1e6

            cum_cr += r["cached_tokens"] / 1e6

            cum_out += r["completion_tokens"] / 1e6

            y = r[a_key]



            rp = [0.0] * dim_total

            rn = [0.0] * dim_total

            rp[s_idx], rp[idx_win], rp[idx_wcr], rp[idx_wout], rp[idx_eps] = 1.0, cum_in, cum_cr, cum_out, 1.0

            rn[s_idx], rn[idx_win], rn[idx_wcr], rn[idx_wout], rn[idx_eps] = -1.0, -cum_in, -cum_cr, -cum_out, 1.0

            A_ub.extend([rp, rn])

            b_ub.extend([y + delta, -y + delta])



            A_ub_proj.extend([rp[:num_segs + 3], rn[:num_segs + 3]])

            b_ub_proj.extend([y + delta, -y + delta])



        tot_in += cum_in

        tot_cr += cum_cr

        tot_out += cum_out



    bounds_proj = list(bounds_main) + [(0.0, None), (0.0, None), (0.0, None)]

    bounds_main.extend([(0.0, None), (0.0, None), (0.0, None), (0.0, delta)])



    c_obj = [0.0] * dim_total

    c_obj[idx_eps] = -1.0

    res_margin = linprog(c_obj, A_ub=A_ub, b_ub=b_ub, bounds=bounds_main, method="highs")



    print(

        f"\n=== [{name} | window={window}] ({len(valid)} steps across {num_segs} contiguous segments | "

        f"in={tot_in*1e6:.0f}, cr={tot_cr*1e6:.0f}, out={tot_out*1e6:.0f}) ==="

    )

    if res_margin.success:

        win = res_margin.x[idx_win]

        wcr = res_margin.x[idx_wcr]

        wout = res_margin.x[idx_wout]

        eps = res_margin.x[idx_eps]

        print(f"  Max-Safety-Margin Solution ({window} / M): w_in={win:.6f}, w_cr={wcr:.6f}, w_out={wout:.6f} (margin eps*={eps:.6f})")

        for idx_w, wname in [(idx_win, "w_in"), (idx_wcr, "w_cr"), (idx_wout, "w_out")]:

            c_min = [0.0] * (num_segs + 3)

            c_max = [0.0] * (num_segs + 3)

            c_min[idx_w], c_max[idx_w] = 1.0, -1.0

            rmin = linprog(c_min, A_ub=A_ub_proj, b_ub=b_ub_proj, bounds=bounds_proj, method="highs")

            rmax = linprog(c_max, A_ub=A_ub_proj, b_ub=b_ub_proj, bounds=bounds_proj, method="highs")

            if rmin.success and rmax.success:

                print(f"  {wname:6s} feasible interval: [{rmin.x[idx_w]:.6f}, {rmax.x[idx_w]:.6f}]")

    else:

        print("  Warning: No strictly feasible solution with eps >= 0 under current rounding bound.")



def run_all_benchmarks(api_key: str, log_path: str):

    for m, prefix in [

        ("k3.1-pro", "k31pro"),

        ("k3.1-pro-highspeed", "k31hs"),

        ("kimi-for-coding", "kfc"),

        ("kimi-for-coding-highspeed", "kfchs"),

        ("k3-256k", "k3_256k"),

        ("k3", "k3_lt256k"),

    ]:

        sys_51k, _ = make_aligned_sys(m, 200)

        run_step(api_key, log_path, f"{prefix}_warm_51k", m,

                 [{"role": "system", "content": sys_51k}, {"role": "user", "content": "Reply with the single word OK. [0]"}], max_tokens=32)

        for i in range(1, 7):

            run_step(api_key, log_path, f"{prefix}_hit_51k_{i}", m,

                     [{"role": "system", "content": sys_51k}, {"role": "user", "content": f"Reply with the single word OK. [{i}]"}], max_tokens=16)

        for i in range(1, 3):

            uid = str(int(time.time() * 1000))[-8:]

            run_step(api_key, log_path, f"{prefix}_out_{i}", m, short_output_msgs(uid), max_tokens=1500 if "hs" in prefix else 2048)



    # k3 >256K 区间测试

    sys_271k, _ = make_aligned_sys("k3", 1060)

    run_step(api_key, log_path, "k3_gt256k_warm_271k", "k3",

             [{"role": "system", "content": sys_271k}, {"role": "user", "content": "Reply with the single word OK. [0]"}], max_tokens=16)

    for i in range(1, 6):

        run_step(api_key, log_path, f"k3_gt256k_hit_271k_{i}", "k3",

                 [{"role": "system", "content": sys_271k}, {"role": "user", "content": f"Reply with the single word OK. [{i}]"}], max_tokens=16)

    for i in range(1, 3):

        run_step(api_key, log_path, f"k3_gt256k_out_2048_{i}", "k3",

                 [{"role": "system", "content": sys_271k},

                  {"role": "user", "content": f"Ignore the apples above. Write a numbered list of short definitions from 1 to 3000. [{i}]"}],

                 max_tokens=2048)



if __name__ == "__main__":

    parser = argparse.ArgumentParser(description="Kimi Coding Plan 多模型计费权重测定与吞吐测试")

    parser.add_argument("--api-key", default=os.environ.get("KIMI_API_KEY", ""), help="Kimi Coding API Key (sk-kimi-...)")

    parser.add_argument("--phase", default="analyze", choices=["all", "analyze"], help="执行测试序列 (all) 或分析已有日志 (analyze)")

    parser.add_argument("--log", default="bench_all_models_log.jsonl", help="JSONL (JSON Lines) 日志路径")

    parser.add_argument("--window", default="both", choices=["5h", "7d", "both"], help="拟合窗口选择")

    args = parser.parse_args()



    if args.phase == "all":

        if not args.api_key:

            print("请通过 --api-key 或环境变量 KIMI_API_KEY 提供 API Key", file=sys.stderr)

            sys.exit(1)

        run_all_benchmarks(args.api_key, args.log)



    if os.path.exists(args.log):

        rows = [json.loads(line) for line in open(args.log, encoding="utf-8")]

        windows = ["5h", "7d"] if args.window == "both" else [args.window]

        for w in windows:

            for m in ["k3.1-pro", "k3.1-pro-highspeed", "kimi-for-coding", "kimi-for-coding-highspeed"]:

                solve_lp_segment(m, [r for r in rows if r["model"] == m], window=w)

            # k3 / k3-256k 按实际总长 prompt + completion 是否 > 256K 分段（越过后整段输出 ×2）

            k3_rows = [r for r in rows if r["model"] in ("k3", "k3-256k")]

            solve_lp_segment("k3 / k3-256k (总长<=256K)", [r for r in k3_rows if r["prompt_tokens"] + r["completion_tokens"] <= K3_THRESHOLD], window=w)

            solve_lp_segment("k3 / k3-256k (总长>256K)", [r for r in k3_rows if r["prompt_tokens"] + r["completion_tokens"] > K3_THRESHOLD], window=w)