| """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()) |
|
|
