"""Spend ledger for the training pod (pure Python; runs on this host). Run as ``python -m training.ledger``. Classifier-plan §3.4. The pod bills from creation until it is terminated, whether or not a stage runs, so spend is wall-clock hours since ``started_at`` times the hourly price (an argument: the price checked at rent time) plus disk. Guards: cap **$50** per training round; no new stage once the projected spend after it passes **$35** (replan); at **$45** everything stops. Smoke replan rule: measured training throughput below 1,900 tok/s (half the §4.5 assumption) or a projection above $35 -> drop the screen (use B-last), then 1.5 epochs. start --ledger L --price 2.69 --started-at 2026-11-20T09:00:00+09:00 [--disk-gb 150] check --ledger L --stage NAME --hours H exit 0 = go, 1 = no new stage (replan line), 2 = stop line record --ledger L --stage NAME --event start|end|note [--note TEXT] status --ledger L replan --ledger L --tokens-per-second X --train-tokens N [--epochs 2] [--screen-hours 0.8] [--eval-hours 0.8] """ from __future__ import annotations import argparse import json import os from datetime import datetime, timezone from pathlib import Path VERSION = "ledger-v1" BUDGET = 50.0 NO_NEW_STAGE = 35.0 STOP = 45.0 MIN_TOKENS_PER_SECOND = 1900.0 REPLAN_EPOCHS = 1.5 DISK_PRICE_GB_MONTH = 0.10 HOURS_PER_MONTH = 730.0 DEFAULT_LEDGER = Path.home() / "naver_psy_models" / "ledger.json" def parse_time(text: str) -> datetime: value = datetime.fromisoformat(text.replace("Z", "+00:00")) if value.tzinfo is None: raise ValueError("times must carry a UTC offset: " + text) return value.astimezone(timezone.utc) def now() -> datetime: return datetime.now(timezone.utc) def rate(ledger: dict) -> float: """$ per wall-clock hour: GPU plus disk.""" return ledger["price_per_hour"] + ledger["disk_gb"] * DISK_PRICE_GB_MONTH / HOURS_PER_MONTH def spent(ledger: dict, at: datetime) -> dict: hours = max(0.0, (at - parse_time(ledger["started_at"])).total_seconds() / 3600) gpu = hours * ledger["price_per_hour"] disk = hours * ledger["disk_gb"] * DISK_PRICE_GB_MONTH / HOURS_PER_MONTH return {"hours": round(hours, 3), "gpu": round(gpu, 2), "disk": round(disk, 2), "total": round(gpu + disk, 2)} def check(ledger: dict, stage_hours: float, at: datetime) -> dict: """Go / no new stage / stop for a stage of the given estimated length.""" s = spent(ledger, at) projected = round(s["total"] + stage_hours * rate(ledger), 2) if s["total"] >= STOP: verdict, code = "stop: spend is at the $45 stop line; stop all stages, sync down and terminate", 2 elif projected > NO_NEW_STAGE: verdict, code = "no new stage: projected spend passes the $35 replan line", 1 else: verdict, code = "go", 0 return {"spent": s, "stage_hours": stage_hours, "projected": projected, "verdict": verdict, "code": code, "lines": {"no_new_stage": NO_NEW_STAGE, "stop": STOP, "budget": BUDGET}} def train_hours(train_tokens: int, tokens_per_second: float, epochs: float) -> float: return train_tokens * epochs / tokens_per_second / 3600 def replan(ledger: dict, tokens_per_second: float, train_tokens: int, at: datetime, epochs: float = 2.0, screen_hours: float = 0.8, eval_hours: float = 0.8) -> dict: """The smoke rule: projection of the remaining plan, and the reduced plan when it is needed.""" s = spent(ledger, at) r = rate(ledger) final = train_hours(train_tokens, tokens_per_second, epochs) full = round(s["total"] + (screen_hours + final + eval_hours) * r, 2) slow = tokens_per_second < MIN_TOKENS_PER_SECOND reasons = [] if slow: reasons.append(f"throughput {tokens_per_second:.0f} tok/s < {MIN_TOKENS_PER_SECOND:.0f}") if full > NO_NEW_STAGE: reasons.append(f"projection ${full:.2f} > ${NO_NEW_STAGE:.0f}") out = {"spent": s, "tokens_per_second": tokens_per_second, "train_tokens": train_tokens, "plan": {"screen": True, "epochs": epochs, "final_hours": round(final, 2), "projected": full}, "replan": bool(reasons), "reasons": reasons} if reasons: reduced_final = train_hours(train_tokens, tokens_per_second, REPLAN_EPOCHS) reduced = round(s["total"] + (reduced_final + eval_hours) * r, 2) out["reduced_plan"] = {"screen": False, "arm": "B-last", "epochs": REPLAN_EPOCHS, "final_hours": round(reduced_final, 2), "projected": reduced, "fits": reduced <= NO_NEW_STAGE} return out def load(path) -> dict: return json.loads(Path(path).read_text()) def save(path, ledger: dict) -> None: path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(path.name + ".tmp") tmp.write_text(json.dumps(ledger, indent=1) + "\n") os.chmod(tmp, 0o600) tmp.replace(path) def new_ledger(price: float, started_at: str, disk_gb: float, at: datetime) -> dict: if price <= 0: raise ValueError("price must be positive") parse_time(started_at) return {"version": VERSION, "started_at": started_at, "price_per_hour": price, "disk_gb": disk_gb, "events": [{"at": at.isoformat(), "event": "ledger-start"}]} def main(argv=None) -> int: p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) sub = p.add_subparsers(dest="command", required=True) s = sub.add_parser("start") s.add_argument("--price", type=float, required=True, help="GPU price in $ per hour (checked at rent time)") s.add_argument("--started-at", required=True, help="When the pod was created (ISO time with offset)") s.add_argument("--disk-gb", type=float, default=150.0, help="Container disk plus network volume") c = sub.add_parser("check") c.add_argument("--stage", required=True) c.add_argument("--hours", type=float, required=True, help="Estimated hours of the next stage") r = sub.add_parser("record") r.add_argument("--stage", required=True) r.add_argument("--event", required=True, choices=["start", "end", "note"]) r.add_argument("--note", default="") sub.add_parser("status") q = sub.add_parser("replan") q.add_argument("--tokens-per-second", type=float, required=True) q.add_argument("--train-tokens", type=int, required=True) q.add_argument("--epochs", type=float, default=2.0) q.add_argument("--screen-hours", type=float, default=0.8) q.add_argument("--eval-hours", type=float, default=0.8) for sp in (s, c, r, q, sub.choices["status"]): sp.add_argument("--ledger", default=str(DEFAULT_LEDGER)) args = p.parse_args(argv) at = now() if args.command == "start": if Path(args.ledger).exists(): raise SystemExit("refusing to overwrite " + args.ledger) save(args.ledger, new_ledger(args.price, args.started_at, args.disk_gb, at)) print(json.dumps(spent(load(args.ledger), at))) return 0 ledger = load(args.ledger) if args.command == "check": result = check(ledger, args.hours, at) ledger["events"].append({"at": at.isoformat(), "event": "check", "stage": args.stage, **result}) save(args.ledger, ledger) print(json.dumps(result)) return result["code"] if args.command == "replan": result = replan(ledger, args.tokens_per_second, args.train_tokens, at, args.epochs, args.screen_hours, args.eval_hours) ledger["events"].append({"at": at.isoformat(), "event": "replan", **result}) save(args.ledger, ledger) print(json.dumps(result)) return 1 if result["replan"] else 0 if args.command == "record": ledger["events"].append({"at": at.isoformat(), "event": args.event, "stage": args.stage, "note": args.note, "spent": spent(ledger, at)}) save(args.ledger, ledger) print(json.dumps({"spent": spent(ledger, at), "events": len(ledger["events"]), "price_per_hour": ledger["price_per_hour"]})) return 0 if __name__ == "__main__": raise SystemExit(main())