#!/usr/bin/env python3 """P34 API quickstart — validate, submit, poll, save. Python 3.8+, standard library only. curl -O https://hyperc.com/lp/sample/p34_quickstart.py python3 p34_quickstart.py # 1. validate the sample request (free, no key) export P34_API_KEY=... # never hardcode it; see "Keys" below python3 p34_quickstart.py # 2. + an input test (mock: true) — placeholders, no charge python3 p34_quickstart.py --input mine.json --real # 3. an actual fit on YOUR records python3 p34_quickstart.py --resume SESSION_ID # keep polling a fit that is still running Keys none POST /validate checks structure, columns and volume floors. Nothing is queued or charged. API key pay as you go from the API platform (https://hyperc.com/api.html), or a membership key (test-...) from https://api.hyperc.com/app/. Input tests (mock: true) are free; actual fits are metered in tokens. free-... a website workspace's free key also works: input tests on any market, actual fits only on free-tier markets such as t5market.com. What is saved (./p34-out/ by default) validate.json the /validate response result-mock.json, plan-mock.csv an input test: PLACEHOLDERS shaped like a result, never predictions result-actual.json, plan-actual.csv an actual fit: qty > 0 rows are the recommended plan, for review The bundled sample request (p34-sample-request-v1.json, fetched next to this file on first use) is GENERATED data for learning the format — a small wholesale reseller with 300 offers — not anyone's business history. This client therefore refuses --real on it: an actual fit would spend compute on numbers that mean nothing. Contract: GET https://api.hyperc.com/v1/ (authoritative) and https://github.com/hyperc-ai/P34-API-DOCS. This client places no orders: P34 returns quantities and predicted profit; what to buy is your decision. """ import argparse import csv import hashlib import json import os import sys import time import urllib.error import urllib.request from pathlib import Path VERSION = "v1.2 · 2026-10-07" API = os.environ.get("P34_API_URL", "https://api.hyperc.com/v1").rstrip("/") SAMPLE_NAME = "p34-sample-request-v1.json" SAMPLE_URL = "https://hyperc.com/lp/sample/" + SAMPLE_NAME class ApiError(Exception): def __init__(self, status, body): self.status, self.body = status, body detail = body.get("detail", body) if isinstance(body, dict) else body if isinstance(detail, dict): # e.g. {"code": "free_markets_only", "message": ...} self.code, self.message = detail.get("code", ""), detail.get("message", json.dumps(detail)) else: self.code, self.message = "", str(detail) super().__init__(f"HTTP {status}: {self.message}") def call(method, path, key=None, body=None, timeout=120): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(API + path, data=data, method=method) req.add_header("Accept", "application/json") if data is not None: req.add_header("Content-Type", "application/json") if key: req.add_header("Authorization", "Bearer " + key) try: with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) except urllib.error.HTTPError as e: raw = e.read().decode(errors="replace") try: parsed = json.loads(raw) except ValueError: parsed = raw raise ApiError(e.code, parsed) from None def explain(err): """What an API refusal means for the person running this script.""" hints = { 401: "The key is missing, unknown or revoked. Check P34_API_KEY.", 403: ("A free key runs actual fits only on free-tier markets (such as t5market.com). " "Use mock: true to input-test this market, or an API key (pay as you go) for an actual fit." if err.code == "free_markets_only" else "This key may not do that."), 422: ("The request was refused as it stands — often a missing business description " "(describe the business and how its unit economics are computed)."), 429: ("No tokens to cover the call — top up (pay as you go) or subscribe — or a rate limit. " "Nothing was queued or charged; input tests (mock: true) still work on every tier."), } return hints.get(err.status, "") def load_request(path, here): if path: return json.loads(Path(path).read_text(encoding="utf-8")), False local = here / SAMPLE_NAME if not local.exists(): print(f"fetching the sample request -> {local}") try: with urllib.request.urlopen(SAMPLE_URL, timeout=60) as r: local.write_bytes(r.read()) except (urllib.error.URLError, OSError) as e: sys.exit(f"could not fetch {SAMPLE_URL} ({e}). Download it next to this file, or pass --input.") return json.loads(local.read_text(encoding="utf-8")), True def save_result(res, kind, out): (out / f"result-{kind}.json").write_text(json.dumps(res, indent=1), encoding="utf-8") rows = [m for m in res.get("menu", []) if (m.get("qty") or 0) > 0] with open(out / f"plan-{kind}.csv", "w", newline="", encoding="utf-8") as f: f.write(f"# {kind} result, session {res.get('session_id', '?')}, saved by p34_quickstart {VERSION}\n") if kind == "mock": f.write("# MOCK: deterministic placeholders from an input test. NOT predictions; do not act on them.\n") else: f.write("# qty = recommended quantity; profit = P34's PREDICTED profit at that quantity, not a result.\n") w = csv.writer(f) w.writerow(["key", "qty", "predicted_profit"]) for m in rows: w.writerow([m.get("key"), m.get("qty"), m.get("profit")]) return rows def poll(session_id, key, out, every, timeout_min): deadline = time.time() + timeout_min * 60 last, misses = None, 0 while True: try: res = call("GET", f"/result/{session_id}", key, timeout=30) misses = 0 except ApiError as e: if e.status == 404: sys.exit(f"session {session_id} is unknown to this server (or was removed).") if e.status == 429: # the poll was refused, not the fit sys.exit(f"polling refused: HTTP 429 {e.message}\n" f" Only the poll was refused: session {session_id} is not canceled, and its result waits on the server.\n" f" Top up (pay as you go) or subscribe, then resume with: " f"python3 {Path(sys.argv[0]).name} --resume {session_id}") raise except (urllib.error.URLError, TimeoutError, ConnectionError) as e: misses += 1 # the fit keeps running server-side; just retry if misses >= 5: sys.exit(f"network trouble ({e}); resume later with --resume {session_id}") time.sleep(min(60, every * misses)) continue status = res.get("status") if status != last: print(f" {status}") last = status if status == "done": res.setdefault("session_id", session_id) kind = "mock" if res.get("mock") else "actual" # the server's label wins over our flag rows = save_result(res, kind, out) print(f"done: {len(rows)} rows with qty > 0 -> {out}/plan-{kind}.csv") if kind == "mock": print("MOCK result: placeholders shaped like a real answer. Not predictions — do not act on them.") else: print(f"predicted profit (sum): {res.get('predicted_profit_sum')} — a forecast, not a result. " "Review the plan; nothing has been bought.") return res if status == "failed": print("FAILED:", res.get("error") or res.get("feedback") or "no reason given") for entry in (res.get("feedback_log") or [])[-5:]: print(" feedback:", json.dumps(entry)[:300]) sys.exit(1) if time.time() > deadline: print(f"still {status} after {timeout_min} min. The fit keeps running; " f"resume with: python3 {Path(sys.argv[0]).name} --resume {session_id}") return None time.sleep(every) def main(): ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--input", help="your request JSON (menus, sales, business_description)") ap.add_argument("--real", action="store_true", help="actual fit instead of an input test (needs your own --input)") ap.add_argument("--resume", metavar="SESSION_ID", help="poll an existing session instead of submitting") ap.add_argument("--model", help="model version (GET https://api.hyperc.com/v1/ lists them)") ap.add_argument("--out", default="p34-out", help="output folder (default: ./p34-out)") ap.add_argument("--poll", type=int, default=None, help="seconds between polls (default 5 mock, 30 actual)") ap.add_argument("--timeout-min", type=int, default=30) args = ap.parse_args() key = os.environ.get("P34_API_KEY", "").strip() out = Path(args.out) out.mkdir(parents=True, exist_ok=True) kind = "actual" if args.real else "mock" every = args.poll or (30 if args.real else 5) if args.resume: if not key: sys.exit("set P34_API_KEY to poll a session") poll(args.resume, key, out, every, args.timeout_min) return body, is_sample = load_request(args.input, Path(__file__).resolve().parent) body.pop("mock", None) digest = hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()[:12] print(f"p34_quickstart {VERSION} · {API} · request {digest} " f"({len(body.get('menus', []))} menu rows, {len(body.get('sales', []))} sales rows)") # 1. Validate: free, no key, nothing queued or charged. v = call("POST", "/validate", body=body) (out / "validate.json").write_text(json.dumps(v, indent=1), encoding="utf-8") print(f"1 validate: ok={v.get('ok')} · {len(v.get('errors', []))} errors · {len(v.get('warnings', []))} warnings") for e in v.get("errors", []): print(f" error {e.get('code')}: {e.get('detail')}") for w in v.get("warnings", []): print(f" warning {w.get('code')}: {w.get('detail')}") if not v.get("ok"): sys.exit("fix the errors above and run again — nothing was queued and nothing was charged") if not key: print("\nNext: set P34_API_KEY to input-test the request (mock: true, free on every tier).\n" " API key, pay as you go: https://hyperc.com/api.html\n" " membership key: https://api.hyperc.com/app/") return if args.real and is_sample: sys.exit("refusing --real on the bundled sample: it is generated data, so an actual fit would spend " "compute on meaningless numbers. Pass --input with your own records.") # 2. Submit. mock: true = input test (placeholders, no grounding, no charge). body = dict(body) if not args.real: body["mock"] = True if args.model: body["model"] = args.model try: ack = call("POST", "/fit", key, body) except ApiError as e: hint = explain(e) sys.exit(f"2 fit refused: HTTP {e.status} {e.code} {e.message}" + (f"\n {hint}" if hint else "")) sid = ack["session_id"] print(f"2 fit accepted ({kind}): session {sid} · status {ack.get('status')}" + (" · replayed an identical earlier request" if ack.get("replayed") else "")) (out / f"session-{kind}.txt").write_text(sid + "\n", encoding="utf-8") # 3. Poll until done or failed. An actual fit takes minutes: grounding -> queued -> processing -> done. print(f"3 polling every {every}s (Ctrl-C is safe: the fit keeps running; --resume {sid})") try: poll(sid, key, out, every, args.timeout_min) except KeyboardInterrupt: print(f"\nstopped polling. The session keeps running; resume with --resume {sid}") if __name__ == "__main__": try: main() except ApiError as e: sys.exit(f"API error: HTTP {e.status} {e.code} {e.message}\n {explain(e)}".rstrip()) except (urllib.error.URLError, TimeoutError, ConnectionError) as e: sys.exit(f"could not reach {API} ({e}). Nothing was submitted; run again when the network is back.")