Initial PuGa toolkit: data layer, econ, depth-aware scan, state sync, plan push

- puga/: cached FIO + PRUNplanner clients, market view with order-book walk,
  econ formulas ported from PRUNplanner (tested against its suite and live FIO),
  saturation model v1 (reviewed by Opus)
- tools/: scan (depth-aware), price, book, chain, state sync, plan_push
  (dry run default, [PuGa]-prefixed plans only), legacy prun_scan/prun_cxarb
- docs/: mechanics (PRUNplanner is source of truth), roadmap, decisions,
  saturation design, archived handoff
- secrets stay in .env (gitignored); ref/ holds PRUNplanner source (ignored)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

puga/__init__.py

@@ -0,0 +1,4 @@
+"""PuGa: Prosperous Universe advisory toolkit."""
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent

puga/cache.py

@@ -0,0 +1,17 @@
+"""Tiny disk cache for JSON over HTTP, keyed by URL, expiry by TTL seconds."""
+import hashlib, json, time, urllib.request
+from .config import CACHE_DIR
+
+UA = "PuGa/0.1 (personal advisory toolkit)"
+
+
+def get_json(url: str, ttl: int, headers: dict[str, str] | None = None, refresh: bool = False):
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    f = CACHE_DIR / (hashlib.sha1(url.encode()).hexdigest()[:16] + ".json")
+    if not refresh and f.exists() and time.time() - f.stat().st_mtime < ttl:
+        return json.loads(f.read_text())
+    req = urllib.request.Request(url, headers={"User-Agent": UA, **(headers or {})})
+    with urllib.request.urlopen(req, timeout=60) as r:
+        data = json.load(r)
+    f.write_text(json.dumps(data))
+    return data

puga/config.py

@@ -0,0 +1,26 @@
+"""Settings from .env (gitignored) and process environment. Never log secret values."""
+import os
+from . import ROOT
+
+
+def _load_env() -> dict[str, str]:
+    out: dict[str, str] = {}
+    p = ROOT / ".env"
+    if p.exists():
+        for line in p.read_text().splitlines():
+            line = line.split("#", 1)[0].strip()
+            if "=" in line:
+                k, v = line.split("=", 1)
+                out[k.strip()] = v.strip().strip("'\"")
+    return out
+
+
+_ENV = _load_env()
+
+
+def get(key: str, default: str | None = None) -> str | None:
+    return os.environ.get(key) or _ENV.get(key) or default
+
+
+DEFAULT_CX = get("DEFAULT_CX", "AI1")
+CACHE_DIR = ROOT / "data" / "cache"

puga/econ.py

@@ -0,0 +1,179 @@
+"""Game economics ported from PRUNplanner (ref/frontend/src/features/planning/calculations/*).
+PRUNplanner is the source of truth; each function names its source file. Pure functions, no I/O."""
+import itertools, math
+
+TOTAL_MS_DAY = 24 * 3600 * 1000
+TIERS = ["pioneer", "settler", "technician", "engineer", "scientist"]
+
+# --- bonusCalculations.ts -------------------------------------------------------------
+EXPERT_BONUS = {0: 0.0, 1: 0.0306, 2: 0.0696, 3: 0.1248, 4: 0.1974, 5: 0.284}
+FACTION_BONUS = {
+    "ANTARES": {"ELECTRONICS": 0.05},
+    "BENTEN": {"MANUFACTURING": 0.05},
+    "HORTUS": {"AGRICULTURE": 0.03, "FOOD_INDUSTRIES": 0.02},
+    "MORIA": {"METALLURGY": 0.02, "CONSTRUCTION": 0.03},
+    "OUTSIDEREGION": {"CHEMISTRY": 0.02, "FUEL_REFINING": 0.02, "RESOURCE_EXTRACTION": 0.02},
+}
+WORKFORCE_COGC = {"PIONEERS": "pioneer", "SETTLERS": "settler", "TECHNICIANS": "technician",
+                  "ENGINEERS": "engineer", "SCIENTISTS": "scientist"}
+
+
+def expert_bonus(n: int) -> float:
+    return EXPERT_BONUS.get(n, 0.0)          # out of range (<0 or >5) gives 0, as in the source
+
+
+def faction_multiplier(faction: str | None, expertise: str | None, permits_used: float, permits_total: float):
+    """Returns the factor (1 + bonus*m) or None if no bonus applies."""
+    if not expertise or not faction:
+        return None
+    b = FACTION_BONUS.get(faction, {}).get(expertise)
+    if not b:
+        return None
+    m = 2 * (-2 * (permits_used / permits_total) + 3)
+    return 1 + b * m
+
+
+# --- workforceCalculations.ts ---------------------------------------------------------
+BASE_SAT = 0.02 * (1 + 10 / 3) * (1 + 4) * (1 + 5 / 6)     # 0.7944; both luxuries met gives exactly 1.0
+LUX1, LUX2 = 1 + 1 / 11, 1 + 2 / 13
+
+
+def tier_efficiency(capacity: float, required: float, lux1: bool, lux2: bool) -> float:
+    """calculateSatisfaction: min(1, capacity/required) * base * luxury multipliers; 0 if nothing required."""
+    if required <= 0:
+        return 0.0
+    sat = 1.0 if required < capacity else capacity / required
+    eff = BASE_SAT * (LUX1 if lux1 else 1) * (LUX2 if lux2 else 1)
+    return sat * eff
+
+
+# (ticker, need per worker per day, lux1?, lux2?)
+CONSUMPTION = {
+    "pioneer": [("DW", 4, 0, 0), ("RAT", 4, 0, 0), ("OVE", .5, 0, 0), ("PWO", .2, 1, 0), ("COF", .5, 0, 1)],
+    "settler": [("DW", 5, 0, 0), ("RAT", 6, 0, 0), ("EXO", .5, 0, 0), ("PT", .5, 0, 0), ("REP", .2, 1, 0), ("KOM", 1, 0, 1)],
+    "technician": [("DW", 7.5, 0, 0), ("RAT", 7, 0, 0), ("MED", .5, 0, 0), ("HMS", .5, 0, 0), ("SCN", .1, 0, 0), ("SC", .1, 1, 0), ("ALE", 1, 0, 1)],
+    "engineer": [("DW", 10, 0, 0), ("MED", .5, 0, 0), ("FIM", 7, 0, 0), ("HSS", .2, 0, 0), ("PDA", .1, 0, 0), ("VG", .2, 1, 0), ("GIN", 1, 0, 1)],
+    "scientist": [("DW", 10, 0, 0), ("MED", .5, 0, 0), ("MEA", 7, 0, 0), ("LC", .2, 0, 0), ("WS", .05, 0, 0), ("NST", .1, 1, 0), ("WIN", 1, 0, 1)],
+}   # needs are per 100 workers per day in the table above; divided below
+
+
+def workforce_consumption(tier: str, required: float, capacity: float, lux1: bool, lux2: bool) -> dict[str, float]:
+    """calculateSingleWorkforceConsumption: units/day by ticker. Consumers = min(required, capacity)."""
+    n = min(required, capacity)
+    if n <= 0:
+        return {}
+    out = {}
+    for tk, need, l1, l2 in CONSUMPTION[tier]:
+        if (not l1 and not l2) or (l1 and lux1) or (l2 and lux2):
+            out[tk] = need / 100 * n
+    return out
+
+
+# --- bonusCalculations.ts: building efficiency ---------------------------------------
+def workforce_factor(building: dict, tier_eff: dict[str, float]) -> float:
+    """building: {'pioneers': n, 'settlers': n, ...} required heads; tier_eff: {'pioneer': eff, ...}."""
+    heads = {t: building.get(t + "s", 0) for t in TIERS}
+    total = sum(heads.values())
+    return sum(heads[t] / total * tier_eff.get(t, 0.0) for t in TIERS) if total else 0.0
+
+
+def building_efficiency(building: dict, tier_eff: dict[str, float], *, expertise: str | None = None,
+                        cogc: str | None = None, hq: bool = False, experts: dict[str, int] | None = None,
+                        faction: str | None = None, permits_used: float = 0, permits_total: float = 1,
+                        fertility: float | None = None, is_farm: bool = False,
+                        condition: float | None = None) -> tuple[float, dict[str, float]]:
+    """Product of factors. Returns (total, elements). `expertise` uses upper snake (METALLURGY);
+    `experts` keys are the same names ('METALLURGY': 3). fertility only applies when is_farm (FRM, ORC)."""
+    el: dict[str, float] = {}
+    if is_farm and fertility is not None:
+        el["FERTILITY"] = 1 + fertility * (10 / 33) if fertility != -1.0 else 0.0
+    if hq:
+        el["HQ"] = 1.1
+    if expertise:
+        if cogc == expertise:
+            el["COGC"] = 1.25
+        n = (experts or {}).get(expertise, 0)
+        if n > 0:
+            el["EXPERT"] = 1 + expert_bonus(n)
+    if cogc in WORKFORCE_COGC and building.get(WORKFORCE_COGC[cogc] + "s", 0) > 0:
+        el["COGC_WORKFORCE"] = 1.1
+    el["WORKFORCE"] = workforce_factor(building, tier_eff)
+    if condition is not None:      # building wear; not in PRUNplanner but present in FIO's live Efficiency (verified 2026-09-18)
+        el["CONDITION"] = condition
+    fb = faction_multiplier(faction, expertise, permits_used, permits_total)
+    if fb is not None:
+        el["FACTION"] = fb
+    total = 1.0
+    for v in el.values():
+        total *= v
+    return total, el
+
+
+# --- extraction: extractionCalculations.ts + backend gamedata/fio/importers.py --------
+CYCLE_MS = {"MINERAL": 12 * 3600e3, "GASEOUS": 6 * 3600e3, "LIQUID": 4.8 * 3600e3}
+
+
+def daily_extraction(factor: float, resource_type: str) -> float:
+    """Backend: factor*60 for GASEOUS, else factor*70 (factor = concentration as fraction). Before efficiency."""
+    return factor * (60.0 if resource_type == "GASEOUS" else 70.0)
+
+
+def extraction_cycle(resource_type: str, daily: float) -> tuple[float, int]:
+    """(time_ms, amount) per cycle; amount is rounded up, time scaled to keep the daily rate."""
+    amt = math.ceil(daily * CYCLE_MS[resource_type] / TOTAL_MS_DAY)
+    return amt * (TOTAL_MS_DAY / daily), amt
+
+
+# --- production: usePlanCalculation.ts / buildingCalculations.ts ----------------------
+def production_io(recipes: list[dict], efficiency: float, n_buildings: float = 1) -> dict[str, dict[str, float]]:
+    """recipes: [{'time_ms', 'inputs': {tk: amt}, 'outputs': {tk: amt}, 'amount': repeats (default 1)}].
+    Each recipe's time = time_ms * amount / efficiency; batches/day = day*n / sum(times).
+    Returns {'in': {tk: per day}, 'out': {tk: per day}} for all buildings."""
+    times = [r["time_ms"] * r.get("amount", 1) / efficiency for r in recipes]
+    runs = TOTAL_MS_DAY * n_buildings / sum(times)
+    io = {"in": {}, "out": {}}
+    for r in recipes:
+        a = r.get("amount", 1)
+        for side, key in (("in", "inputs"), ("out", "outputs")):
+            for tk, amt in r[key].items():
+                io[side][tk] = io[side].get(tk, 0) + amt * a * runs
+    return io
+
+
+# --- habOptimization.ts ---------------------------------------------------------------
+HAB_AREA = {"HB1": 10, "HB2": 12, "HB3": 14, "HB4": 16, "HB5": 18, "HBB": 14, "HBC": 17, "HBM": 20, "HBL": 22}
+HAB_CAP = {  # tier -> capacity per hab
+    "HB1": {"pioneer": 100}, "HB2": {"settler": 100}, "HB3": {"technician": 100},
+    "HB4": {"engineer": 100}, "HB5": {"scientist": 100},
+    "HBB": {"pioneer": 75, "settler": 75}, "HBC": {"settler": 75, "technician": 75},
+    "HBM": {"technician": 75, "engineer": 75}, "HBL": {"engineer": 75, "scientist": 75},
+}
+
+
+def optimize_habs(required: dict[str, float], costs: dict[str, float], goal: str = "cost",
+                  max_area: float | None = None) -> dict | None:
+    """Integer hab mix covering `required` heads per tier, minimising 'cost' or 'area' (optionally area-capped).
+    Exhaustive over combined habs (<=4 vars), single habs fill the rest. Returns {'habs', 'cost', 'area'} or None."""
+    req = {t: max(0.0, required.get(t, 0)) for t in TIERS}
+    combos = [h for h in HAB_CAP if len(HAB_CAP[h]) == 2]
+    single = {next(iter(HAB_CAP[h])): h for h in HAB_CAP if len(HAB_CAP[h]) == 1}
+    limits = [math.ceil(min(req[t] for t in HAB_CAP[h]) / 75) for h in combos]
+    best = None
+    for counts in itertools.product(*[range(l + 1) for l in limits]):
+        cov = dict.fromkeys(TIERS, 0.0)
+        habs = dict(zip(combos, counts))
+        for h, n in habs.items():
+            for t, c in HAB_CAP[h].items():
+                cov[t] += c * n
+        for t in TIERS:
+            rest = req[t] - cov[t]
+            if rest > 0:
+                habs[single[t]] = habs.get(single[t], 0) + math.ceil(rest / 100)
+        cost = sum(costs[h] * n for h, n in habs.items())
+        area = sum(HAB_AREA[h] * n for h, n in habs.items())
+        if goal == "cost" and max_area is not None and area > max_area:
+            continue
+        key = cost if goal == "cost" else area
+        if best is None or key < best[0]:
+            best = (key, {h: n for h, n in habs.items() if n}, cost, area)
+    return None if best is None else {"habs": best[1], "cost": best[2], "area": best[3]}

puga/fio.py

@@ -0,0 +1,39 @@
+"""Public FIO REST (rest.fnar.net). Static game data caches long; market data short."""
+from .cache import get_json
+
+BASE = "https://rest.fnar.net"
+STATIC, MARKET = 24 * 3600, 15 * 60
+
+
+def _g(path: str, ttl: int, refresh: bool = False):
+    return get_json(BASE + path, ttl, refresh=refresh)
+
+
+def exchange_all(refresh=False):      return _g("/exchange/all", MARKET, refresh)
+def order_book(mat: str, cx: str, refresh=False):
+    """Full book: BuyingOrders / SellingOrders lists (price + quantity). Short TTL."""
+    return _g(f"/exchange/{mat}.{cx}", 5 * 60, refresh)
+def cxpc(mat: str, cx: str, refresh=False):
+    return _g(f"/exchange/cxpc/{mat}.{cx}", 3600, refresh)
+def materials(refresh=False):         return _g("/material/allmaterials", STATIC, refresh)
+def recipes(refresh=False):           return _g("/recipes/allrecipes", STATIC, refresh)
+def buildings(refresh=False):         return _g("/building/allbuildings", STATIC, refresh)
+def planets_full(refresh=False):      return _g("/planet/allplanets/full", STATIC, refresh)
+def workforce_needs(refresh=False):   return _g("/global/workforceneeds", STATIC, refresh)
+def systems(refresh=False):           return _g("/systemstars", STATIC, refresh)
+def stations(refresh=False):          return _g("/exchange/station", STATIC, refresh)
+
+
+def private(path: str, ttl: int = 300, refresh: bool = False):
+    """Authenticated FIO REST (own data: /sites/{user}, /storage/{user}, /production/{user}, /ship/ships/{user}...).
+    Auth: `Authorization: <FIO_REST_KEY>` (verified 2026-09-18). FIO_API_KEY does NOT work on rest.fnar.net."""
+    from . import config
+    key = config.get("FIO_REST_KEY")
+    if not key:
+        raise RuntimeError("FIO_REST_KEY missing in .env")
+    return get_json(BASE + path, ttl, headers={"Authorization": key}, refresh=refresh)
+
+
+def me() -> str:
+    from . import config
+    return config.get("FIO_USERNAME", "")

puga/market.py

@@ -0,0 +1,53 @@
+"""Unified market view: FIO book totals merged with PRUNplanner VWAP/volume, plus order-book walks."""
+from dataclasses import dataclass
+from . import fio, prunplanner as pp
+
+CXS = ["AI1", "NC1", "CI1", "IC1", "NC2", "CI2"]
+
+
+@dataclass
+class Quote:
+    tk: str
+    cx: str
+    ask: float | None = None
+    bid: float | None = None
+    supply: float = 0        # units on sell side (order-book total)
+    demand: float = 0        # units on buy side
+    vwap7: float | None = None
+    vwap30: float | None = None
+    traded7: float = 0       # avg units/day over 7d
+    traded30: float = 0      # avg units/day over 30d
+    mm_buy: float | None = None
+    mm_sell: float | None = None
+
+
+def snapshot(refresh=False) -> dict[tuple[str, str], Quote]:
+    q: dict[tuple[str, str], Quote] = {}
+    for e in fio.exchange_all(refresh):
+        k = (e["MaterialTicker"], e["ExchangeCode"])
+        q[k] = Quote(k[0], k[1], e.get("Ask"), e.get("Bid"), e.get("Supply") or 0, e.get("Demand") or 0,
+                     mm_buy=e.get("MMBuy"), mm_sell=e.get("MMSell"))
+    for e in pp.exchanges(refresh):
+        k = (e["ticker"], e["exchange_code"])
+        x = q.setdefault(k, Quote(k[0], k[1], e.get("ask"), e.get("bid"), e.get("supply") or 0, e.get("demand") or 0))
+        x.vwap7, x.vwap30 = e.get("vwap_7d"), e.get("vwap_30d")
+        x.traded7, x.traded30 = e.get("avg_traded_7d") or 0, e.get("avg_traded_30d") or 0
+    return q
+
+
+def walk(mat: str, cx: str, qty: float, side: str, refresh=False) -> dict:
+    """Walk the live order book. side='buy' takes sell orders (you pay asks); side='sell' hits buy orders (you get bids).
+    Returns avg price, units filled, worst price reached, total cost/proceeds. Shallow books give filled < qty."""
+    ob = fio.order_book(mat, cx, refresh)
+    orders = ob["SellingOrders"] if side == "buy" else ob["BuyingOrders"]
+    # MM orders come with ItemCount None = unlimited depth; treat as infinite
+    orders = sorted((o for o in orders if o.get("ItemCount") != 0), key=lambda o: o["ItemCost"], reverse=(side == "sell"))
+    left, total, worst = qty, 0.0, None
+    for o in orders:
+        take = min(left, float("inf") if o.get("ItemCount") is None else o["ItemCount"])
+        total += take * o["ItemCost"]
+        worst, left = o["ItemCost"], left - take
+        if left <= 0:
+            break
+    filled = qty - left
+    return dict(avg=total / filled if filled else None, filled=filled, worst=worst, total=total, short=left > 0)

puga/prunplanner.py

@@ -0,0 +1,36 @@
+"""api.prunplanner.org. Public data endpoints need no auth; planning endpoints use `Authorization: Api-Key <key>`."""
+from .cache import get_json
+from . import config
+
+BASE = "https://api.prunplanner.org"
+MARKET, STATIC = 15 * 60, 24 * 3600
+
+
+def _g(path: str, ttl: int, refresh: bool = False):
+    return get_json(BASE + path, ttl, refresh=refresh)
+
+
+def exchanges(refresh=False):
+    """Per ticker.CX: ask/bid/supply/demand, vwap_daily/7d/30d, traded_daily, sum/avg_traded_7d/30d, exchange_status."""
+    return _g("/data/exchanges/", MARKET, refresh)
+def materials(refresh=False):  return _g("/data/materials/", STATIC, refresh)
+def recipes(refresh=False):    return _g("/data/recipes/", STATIC, refresh)
+def buildings(refresh=False):  return _g("/data/buildings/", STATIC, refresh)
+
+
+def api_key() -> str | None:
+    return config.get("PRUNPLANNER_API_KEY")
+
+
+def request(method: str, path: str, body: dict | None = None):
+    """Authenticated call (Api-Key). Callers must enforce write guardrails (see tools/plan_push.py)."""
+    import json, urllib.request
+    key = api_key()
+    if not key:
+        raise RuntimeError("PRUNPLANNER_API_KEY missing in .env")
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(BASE + path, data=data, method=method, headers={
+        "Authorization": "Api-Key " + key, "Content-Type": "application/json", "User-Agent": "PuGa/0.1"})
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+    return json.loads(raw) if raw else None

puga/saturation.py

@@ -0,0 +1,56 @@
+"""Saturation model v1 (docs/saturation-design.md, after Opus review 2026-09-18). Pure functions.
+Idea: N buildings is limited by (a) share of the market's steady traded flow, (b) the queue of existing
+sell orders (supply days) and (c) whether buy-side stock covers it. Price is a haircut model, not a book walk."""
+import math
+
+S_SHARE = 0.25     # max share of traded flow one producer takes
+T_Q = 7            # days of flow the standing sell queue is compared against
+T_D = 7            # days of flow standing demand should cover
+T_W = 3            # patient-selling window (days)
+
+
+def tref(traded7: float, traded30: float) -> float:
+    """Conservative daily flow: min(7d, 30d); with a Poisson lower bound when 30d volume is small."""
+    t = min(traded7, traded30)
+    if 0 < traded30 < 20:
+        t = min(t, max(0.0, traded30 * (1 - 1.96 / math.sqrt(30 * traded30))))
+    return max(0.0, t)
+
+
+def is_thin(tr: float, demand: float, q_out: float) -> bool:
+    return tr < 3 * q_out or demand < 7 * q_out
+
+
+def n_out(tr: float, supply: float, demand: float, q_out: float) -> float:
+    """Buildings the output market absorbs: flow share x queue penalty x demand coverage. 0 if no flow."""
+    if tr <= 0 or q_out <= 0:
+        return 0.0
+    base = S_SHARE * tr / q_out
+    queue = min(1.0, T_Q * tr / supply) if supply > 0 else 1.0
+    cover = min(1.0, demand / (T_D * tr))
+    return base * queue * cover
+
+
+def p_patient(asks: list[tuple[float, float]], tr: float, produced_per_day: float, bid: float, vwap7: float | None,
+              vwap30: float | None, ask: float | None, wide_high: float | None = None) -> float:
+    """Price we can hold asks at. asks = [(price, units)] ascending. Find the highest ask level whose units-ahead
+    (levels strictly below) <= what buyers will absorb of the queue over T_W days once our output is counted:
+    target = T_W * (tr - produced). target <= 0 => sell at the bid. Clamped to [bid, min(vwap7, vwap30, ask, wide_high)]."""
+    hi = min(x for x in (vwap7, vwap30, ask, wide_high) if x)
+    target = T_W * (tr - produced_per_day)
+    if target <= 0:
+        return bid
+    p, ahead = hi, 0.0
+    for price, units in asks:
+        if ahead <= target:
+            p = price
+        ahead += units
+    return max(bid, min(p, hi))
+
+
+def effective_supply(asks: list[tuple[float, float]], ref_price: float | None, band: float = 1.25) -> float:
+    """Units of standing sell orders that actually compete: price <= band x reference (vwap7 or ask).
+    Stale asks far above market (e.g. 1500 units at 1.5x vwap) are not competition. asks = [(price, units)]."""
+    if not ref_price:
+        return sum(u for _, u in asks)
+    return sum(u for p, u in asks if p <= band * ref_price)