dodox/PuGa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add simulator replica, persistence, planet scan, README; split empire state out of the repo

Browse Source 
- tools/simulate.py + puga/simulate.py: replica of PRUNplanner's simulator (flows and
  efficiency verified against screenshots), reports real new capex (planned minus built)
- tools/scan.py: staffing variants, freight, HQ/experts, --planet mode, demolish-later,
  --min-n as a pure market-size filter, --json output
- tools/history.py, tools/persistence.py: margin history and short-horizon payback checks
- tools/plan_push.py: guarded delete; tools/state.py: syncs to empire/
- README with features and setup; CLAUDE.md made generic
- Own-empire material (profile, state, plans, notes) moved to gitignored empire/;
  generic examples in plans/examples and state/company.example.yaml

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
dodox
2026-09-19 00:11:02 +02:00
co-authored by Claude Sonnet 5
parent 7a538cb300
commit 3bbf524ebb
26 changed files with 778 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/history.py
Executable
+69
View File
@@ -0,0 +1,69 @@
#!/usr/bin/env python3
"""Margin history of a recipe from daily exchange candles (FIO cxpc DAY_ONE): output value minus input cost per batch, per month.
Daily price = value traded / units traded (VWAP), forward-filled over days without trades.
tools/history.py KV # AI1, the recipe producing KV
tools/history.py BHP --cx AI1 --eff 1.18 --overhead 3200 --capex 94000 --months 12
Building/day = batches/day at --eff; profit/day = batches/day * margin_per_batch - overhead (wages etc, --overhead)."""
import argparse, datetime, statistics, sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from puga import config, fio
def daily_prices(tk, cx):
out = {}
for e in fio.cxpc(tk, cx):
if e.get("Interval") == "DAY_ONE" and e.get("Traded"):
out[datetime.datetime.fromtimestamp(e["DateEpochMs"] / 1000, datetime.timezone.utc).date()] = (e["Volume"] / e["Traded"], e["Traded"])
return out
def main():
ap = argparse.ArgumentParser()
ap.add_argument("ticker")
ap.add_argument("--cx", default=config.DEFAULT_CX)
ap.add_argument("--recipe", type=int, default=0, help="index if several recipes produce it")
ap.add_argument("--eff", type=float, default=1.0)
ap.add_argument("--overhead", type=float, default=0.0, help="daily wages/other per building")
ap.add_argument("--capex", type=float, help="per building; gives ROI/day")
ap.add_argument("--months", type=int, default=14)
a = ap.parse_args()
t = a.ticker.upper()
recs = [r for r in fio.recipes() if any(o["Ticker"] == t for o in r["Outputs"])]
r = recs[a.recipe]
ins = {i["Ticker"]: i["Amount"] for i in r["Inputs"]}
outs = {o["Ticker"]: o["Amount"] for o in r["Outputs"]}
per_day = 24 / (r["TimeMs"] / 3.6e6) * a.eff
print(f"{r['BuildingTicker']}: {ins} -> {outs}, {r['TimeMs']/3.6e6:.1f}h, eff {a.eff}: {per_day:.3f} batches/day; recipe {a.recipe + 1} of {len(recs)}")
series = {m: daily_prices(m, a.cx) for m in list(ins) + list(outs)}
days = sorted(set.intersection(*[set(s) for s in series.values()])) if False else sorted(set().union(*[set(s) for s in series.values()]))
last = {}
rows = []
for d in days:
for m, s in series.items():
if d in s:
last[m] = s[d][0]
if len(last) == len(series):
rev = sum(last[m] * n for m, n in outs.items())
cost = sum(last[m] * n for m, n in ins.items())
rows.append((d, rev, cost, series[t].get(d, (0, 0))[1]))
if not rows:
sys.exit("no overlapping history")
by = {}
for d, rev, cost, tr in rows:
by.setdefault((d.year, d.month), []).append((rev, cost, tr, per_day * (rev - cost) - a.overhead))
print(f"\n{'month':8} {'out/batch':>10} {'in/batch':>10} {'margin':>9} {'margin%':>7} {'profit/d':>9} {'ROI/d%':>7} {'units/d':>8} {'days>0':>7}")
for (y, m), v in list(sorted(by.items()))[-a.months:]:
rev = statistics.mean(x[0] for x in v); cost = statistics.mean(x[1] for x in v); pr = statistics.mean(x[3] for x in v)
roi = f"{100 * pr / a.capex:7.1f}" if a.capex else f"{'-':>7}"
print(f"{y}-{m:02d} {rev:10.0f} {cost:10.0f} {rev-cost:9.0f} {100*(rev-cost)/cost if cost else 0:7.1f} {pr:9.0f} {roi} "
f"{statistics.mean(x[2] for x in v):8.1f} {100*sum(1 for x in v if x[3] > 0)/len(v):6.0f}%")
prof = [per_day * (rev - cost) - a.overhead for _, rev, cost, _ in rows]
for label, n in (("last 30d", 30), ("last 90d", 90), ("last 180d", 180), ("all", len(prof))):
p = prof[-n:]
print(f"{label:9} profit/d mean {statistics.mean(p):8.0f} min {min(p):8.0f} max {max(p):8.0f} days>0 {100*sum(1 for x in p if x > 0)/len(p):4.0f}%"
+ (f" ROI/d {100*statistics.mean(p)/a.capex:5.1f}%" if a.capex else ""))
if __name__ == "__main__":
main()
tools/persistence.py
Executable
+71
View File
@@ -0,0 +1,71 @@
#!/usr/bin/env python3
"""How stable are scan opportunities? Re-prices each scan row over its exchange history (daily VWAP candles) and reports
mean ROI/day over 30/90/180 days and all history, plus the share of days with positive profit.
tools/scan.py --min-n 3 --json /tmp/rows.json --top 20 ; tools/persistence.py /tmp/rows.json [--only KV,BHP]
Method: profit(period) = profit_now + batches/day * (margin_period - margin_now), margin = output value - input cost per batch at daily VWAP,
margin_now = mean of the last 7 days. So wages/freight/efficiency in the scan row carry over; only prices move."""
import argparse, json, re, statistics, sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
sys.path.insert(0, str(Path(__file__).resolve().parent))
from puga import config
from history import daily_prices
def parse(side):
return {m.group(2): float(m.group(1)) for m in re.finditer(r"([\d.]+)([A-Z0-9]+)", side)}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("rows")
ap.add_argument("--cx", default=config.DEFAULT_CX)
ap.add_argument("--only", help="comma list of output tickers")
ap.add_argument("--top", type=int, default=25)
ap.add_argument("--lag", type=float, default=3.0, help="days from decision to first output (build, hauling, workers)")
ap.add_argument("--min-mkt", type=float, default=0.0)
a = ap.parse_args()
only = {x.strip().upper() for x in a.only.split(",")} if a.only else None
rows = json.loads(Path(a.rows).read_text())
cache = {}
out = []
seen = set()
for r in rows:
left, right = r["rec"].replace(" [thin]", "").split("->")
ins, outs = parse(left), parse(right)
if only and not (set(outs) & only):
continue
key = (r["rec"], r["bld"], r["staff"])
if key in seen:
continue
seen.add(key)
toks = list(ins) + list(outs)
for t in toks:
if t not in cache:
cache[t] = daily_prices(t, a.cx)
last, series = {}, []
for d in sorted(set().union(*[set(cache[t]) for t in toks])):
for t in toks:
if d in cache[t]:
last[t] = cache[t][d][0]
if len(last) == len(toks):
series.append(sum(last[t] * n for t, n in outs.items()) - sum(last[t] * n for t, n in ins.items()))
if len(series) < 60:
continue
per_day = 24 / r["h"] * r["eff"]
now = statistics.mean(series[-7:])
prof = [r["total"] / r["n"] + per_day * (m - now) for m in series] # per building
capex = r["capex"] / r["n"]
f = lambda n: 100 * statistics.mean(prof[-n:]) / capex
p14 = statistics.mean(prof[-14:])
out.append((f(14), dict(pb=capex / p14 if p14 > 0 else 999, net7=p14 * (7 - a.lag) - capex, net14=p14 * (14 - a.lag) - capex, d7=f(7), d14=f(14), capex=capex, profit_now=r["total"] / r["n"], rec=r["rec"], bld=r["bld"], staff=r["staff"], now=100 * (r["total"] / r["n"]) / capex, d30=f(30), d90=f(90), d180=f(180), all=f(len(prof)),
pos=100 * sum(1 for x in prof[-180:] if x > 0) / len(prof[-180:]), days=len(prof), market=r["n_lim"])))
out.sort(key=lambda x: -x[0]) # ranked by 14-day mean ROI
out = [x for x in out if x[1]["market"] >= a.min_mkt]
print(f"{'now':>5} {'7d':>5} {'14d':>5} {'30d':>5} {'payback':>7} {'net@7d':>8} {'net@14d':>8} {'90d':>5} {'180d':>5} {'pos%':>5} {'days':>5} {'mkt':>5} bld staff recipe (ROI/day % per building; pos% = days profitable in last 180d)")
for _, o in out[:a.top]:
print(f"{o['now']:5.1f} {o['d7']:5.1f} {o['d14']:5.1f} {o['d30']:5.1f} {o['pb']:6.1f}d {o['net7']:8.0f} {o['net14']:8.0f} {o['d90']:5.1f} {o['d180']:5.1f} {o['pos']:5.0f} {o['days']:5d} {o['market']:5.1f} {o['bld']:4} {o['staff']:5} {o['rec']}")
if __name__ == "__main__":
main()
tools/plan_push.py
+18 -8
View File
@@ -1,13 +1,14 @@
#!/usr/bin/env python3
"""Turn a YAML plan spec (plans/*.yaml) into a PRUNplanner plan, via the Api-Key API.
DEFAULT IS DRY RUN. Guardrails (docs/decisions.md): only plans named '[PuGa] ...' are created/updated; never delete;
--update refuses if the existing plan's name does not start with '[PuGa]'. Show the dry-run to Dominik and get a yes before --apply.
DEFAULT IS DRY RUN. Guardrails (docs/decisions.md): only plans named '[PuGa] ...' are created/updated; deletes only [PuGa] plans and only when asked;
--update refuses if the existing plan's name does not start with '[PuGa]'. Show the dry-run to the user and get a yes before --apply.
tools/plan_push.py list
tools/plan_push.py plans/deimos_bhp.yaml # dry run: validated payload summary
tools/plan_push.py plans/deimos_bhp.yaml --json # full JSON payload
tools/plan_push.py plans/deimos_bhp.yaml --apply # create (after user says yes)
tools/plan_push.py plans/deimos_bhp.yaml --apply --update <uuid>
tools/plan_push.py plans/examples/base_plus_hwp.yaml # dry run: validated payload summary
tools/plan_push.py plans/examples/base_plus_hwp.yaml --json # full JSON payload
tools/plan_push.py plans/examples/base_plus_hwp.yaml --apply # create (after user says yes)
tools/plan_push.py plans/examples/base_plus_hwp.yaml --apply --update <uuid>
tools/plan_push.py delete <uuid> # only [PuGa] plans, only when the user asks
Spec: name, planet (natural id), permits, cogc (e.g. METALLURGY or null), hq, experts {METALLURGY: 2}, lux {pioneer: [true,true]},
infrastructure {HB1: 4, HB2: 1}, buildings: [{building: SME, amount: 5, recipes: ["ALO,FLX,C,O=>4AL"|"AL,STL,HE=>BHP"|"EXT#ALO"]}]
@@ -79,7 +80,8 @@ def build_payload(spec: dict, recipes: list[dict], building_tickers: set[str]) -
def main():
ap = argparse.ArgumentParser()
ap.add_argument("spec", help="plans/*.yaml or 'list'")
ap.add_argument("spec", help="plans/*.yaml, 'list', or 'delete'")
ap.add_argument("target", nargs="?", help="uuid for 'delete'")
ap.add_argument("--apply", action="store_true", help="actually write to PRUNplanner (creates a new plan)")
ap.add_argument("--update", metavar="UUID", help="with --apply: update this existing [PuGa] plan instead of creating")
ap.add_argument("--json", action="store_true")
@@ -89,6 +91,14 @@ def main():
for p in pp.request("GET", "/planning/plan/"):
print(p["uuid"], p.get("plan_name"), p.get("planet_natural_id"))
return
if a.spec == "delete":
# delete is allowed only on request, and only for plans this tool made ([PuGa] prefix).
cur = pp.request("GET", f"/planning/plan/{a.target}/")
if not str(cur.get("plan_name", "")).startswith(PREFIX):
sys.exit(f"REFUSED: '{cur.get('plan_name')}' is not a {PREFIX} plan")
pp.request("DELETE", f"/planning/plan/{a.target}/")
print("DELETED:", cur["plan_name"], a.target)
return
spec = yaml.safe_load(Path(a.spec).read_text())
payload = build_payload(spec, pp.recipes(), {b["building_ticker"] for b in pp.buildings()})
d = payload["plan_data"]
@@ -100,7 +110,7 @@ def main():
if a.json:
print(json.dumps(payload, indent=1))
if not a.apply:
print("\nDRY RUN: nothing sent. Re-run with --apply after Dominik confirms.")
print("\nDRY RUN: nothing sent. Re-run with --apply after the user confirms.")
return
if a.update:
cur = pp.request("GET", f"/planning/plan/{a.update}/")
tools/scan.py
+103 -50
View File
@@ -4,21 +4,21 @@ For each recipe: buildings the market can absorb (N*), and ROI/day at N=1 and N*
(not top-of-book) and ask-walked inputs. Excludes thin markets by default.
tools/scan.py # AI1, all tiers, top 30 by ROI at N*
tools/scan.py --cogc METALLURGY --tier S # Deimos-like: metallurgy COGC, up to settlers
tools/scan.py --cogc METALLURGY --tier S # e.g. a Metallurgy-COGC base, up to settlers
tools/scan.py --sort total --min-n 3 # rank by absorbable profit/day
"""
import argparse, sys
import argparse, itertools, sys
from pathlib import Path
import yaml
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from puga import ROOT, config, econ, fio, market, saturation as sat
from puga import ROOT, config, econ, fio, market, prunplanner as pp, saturation as sat
TIER_CODE = {"pioneer": "P", "settler": "S", "technician": "T", "engineer": "E", "scientist": "Sc"}
TIER_RANK = {"P": 0, "S": 1, "T": 2, "E": 3, "Sc": 4}
def load_state():
p = ROOT / "state" / "company.yaml"
p = config.state_path()
return yaml.safe_load(p.read_text()) if p.exists() else {}
@@ -27,10 +27,18 @@ def main():
ap.add_argument("--cx", default=config.DEFAULT_CX)
ap.add_argument("--tier", help="max workforce tier: P S T E Sc")
ap.add_argument("--cogc", help="COGC programme on the base, e.g. METALLURGY, or SETTLERS")
ap.add_argument("--staffing", default="both", choices=["full", "under", "both"], help="evaluate fully staffed, understaffed (drop tiers), or both (default)")
ap.add_argument("--trip-cost", type=float, default=9250.0, help="AIC per loaded round trip AI1<->base (handoff est. 8.5-10k); 0 disables freight")
ap.add_argument("--cargo", type=float, default=500.0, help="t and m3 per trip (starter ship)")
ap.add_argument("--no-hq", action="store_true", help="ignore HQ from state (new base without the HQ)")
ap.add_argument("--permits-used", type=int, help="override permits used (affects faction bonus multiplier), e.g. 2 for a second base")
ap.add_argument("--deprec", type=float, default=0, help="demolish-later mode: building value decays linearly to 0 over this many days (game: ~60); subtracts capex/deprec per day")
ap.add_argument("--planet", help="planet natural id: adds extraction (EXT/COL/RIG) from its resources, uses its fertility and active COGC; new base (not in state) => no HQ, permits+1")
ap.add_argument("--own", type=int, default=1, help="how many buildings WE would run; ROI is measured at this size (default 1). --min-n is only a market-size filter")
ap.add_argument("--skip", default="", help="tiers left unstaffed, e.g. technician (no housing/wages; efficiency = staffed headcount share)")
ap.add_argument("--experts", default="", help="expert counts, e.g. METALLURGY=2 (bonus x1.0306..1.284 for 1..5)")
ap.add_argument("--hq", action="store_true", help="corp HQ bonus x1.1")
ap.add_argument("--no-faction", action="store_true", help="ignore faction bonus from state/company.yaml")
ap.add_argument("--no-faction", action="store_true", help="ignore faction bonus from empire/state/company.yaml")
ap.add_argument("--min-n", type=float, default=1.0, help="min buildings the market absorbs (default 1)")
ap.add_argument("--max-n", type=int, default=50, help="cap N* (capital/attention limit)")
ap.add_argument("--budget", type=float, help="capital limit in AIC: caps N and drops recipes whose single-building capex exceeds it")
@@ -39,6 +47,7 @@ def main():
ap.add_argument("--top", type=int, default=30)
ap.add_argument("--k", type=int, default=80, help="candidates refined with order books")
ap.add_argument("--show-thin", action="store_true")
ap.add_argument("--json", help="also write all rows to this JSON file (for tools/persistence.py)")
a = ap.parse_args()
snap = market.snapshot()
@@ -47,8 +56,22 @@ def main():
st = load_state()
faction = None if a.no_faction else (st.get("company") or {}).get("faction")
pu, pt = (st.get("permits") or {}).get("used", 1), (st.get("permits") or {}).get("total", 2)
if a.permits_used:
pu = a.permits_used
planet = None
if a.planet:
planet = pp._g(f"/data/planet/{a.planet}/", 3600)
if not a.cogc:
a.cogc = (planet.get("active_cogc_program_type") or "").replace("ADVERTISING_", "") or None
if a.planet not in [b.get("planet") for b in st.get("bases", [])]: # new base: no HQ, one more permit used
a.no_hq = True
if not a.permits_used:
pu += 1
print(f"# planet {a.planet} {planet['planet_name']}: fertility {planet['fertility']:.2f}, COGC {a.cogc}, resources "
+ ", ".join(f"{r['material_ticker']} {r['daily_extraction']:.0f}/d" for r in planet["resources"]))
blds = {b["Ticker"]: b for b in fio.buildings()}
mat = {m["Ticker"]: m for m in fio.materials()}
mcg = ask("MCG") or 0
def bcost(tk):
@@ -76,53 +99,74 @@ def main():
skip = {x.strip() for x in a.skip.split(",") if x.strip()}
wages = {t: wage(t) for t in econ.TIERS}
cands = []
for rec in fio.recipes():
all_recipes = list(fio.recipes())
if planet: # extraction pseudo-recipes: one 24h cycle yielding the daily extraction
bt = {"MINERAL": "EXT", "GASEOUS": "COL", "LIQUID": "RIG"}
for r in planet["resources"]:
all_recipes.append(dict(BuildingTicker=bt[r["resource_type"]], Inputs=[], TimeMs=econ.TOTAL_MS_DAY,
Outputs=[dict(Ticker=r["material_ticker"], Amount=r["daily_extraction"])]))
for rec in all_recipes:
b = blds.get(rec["BuildingTicker"])
if not b or not rec["Outputs"] or b["Ticker"].startswith("HB"):
continue
heads = {t: b[t.capitalize() + "s"] for t in econ.TIERS}
used = [t for t in econ.TIERS if heads[t] and t not in skip] # staffed tiers
if not used:
continue
top = TIER_CODE[used[-1]]
if a.tier and TIER_RANK[top] > TIER_RANK[a.tier]:
continue
if any(hab_head[t] is None or wages[t] is None for t in used):
need = [t for t in econ.TIERS if heads[t]]
if not need:
continue
ins = {i["Ticker"]: i["Amount"] for i in rec["Inputs"]}
outs = {o["Ticker"]: o["Amount"] for o in rec["Outputs"]}
if any(not Q(t) or not Q(t).ask for t in ins) or any(not Q(t) or not Q(t).bid for t in outs):
continue
bd = {t + "s": heads[t] for t in econ.TIERS}
eff, _ = econ.building_efficiency(bd, {t: (0.0 if t in skip else 1.0) for t in econ.TIERS}, expertise=b["Expertise"] or None, cogc=a.cogc,
hq=a.hq, experts=experts, faction=faction, permits_used=pu, permits_total=pt)
io = econ.production_io([dict(time_ms=rec["TimeMs"], inputs=ins, outputs=outs)], eff, 1)
# saturation per output
n_lim, lim, thin, mm = float("inf"), "", False, {}
for t, qo in io["out"].items():
x = Q(t)
tr = sat.tref(x.traded7, x.traded30)
if x.mm_buy and x.bid and x.mm_buy >= 0.9 * x.bid:
mm[t] = x.mm_buy # market maker floor: unlimited depth at mm_buy
capex0 = bcost(b["Ticker"])
if capex0 is None:
continue
# staffing variants: fully staffed and every partial subset (efficiency = staffed headcount share; housing/wages only for staffed tiers)
variants = []
if a.staffing in ("full", "both"):
variants.append(tuple(need))
if a.staffing in ("under", "both") and len(need) > 1:
for k in range(1, len(need)):
variants += list(itertools.combinations(need, k))
for keep in variants:
keep = tuple(t for t in keep if t not in skip)
if not keep:
continue
thin |= sat.is_thin(tr, x.demand, qo)
n = sat.n_out(tr, 0, x.demand, qo) # stage 1: flow + demand only; queue penalty applied in stage 2 with the book
if n < n_lim:
n_lim, lim = n, t
if n_lim == float("inf"):
n_lim, lim = float(a.max_n), "MM"
if (thin or n_lim < a.min_n) and not a.show_thin:
continue
capex = bcost(b["Ticker"])
if capex is None:
continue
capex += sum(heads[t] * hab_head[t] for t in used)
wcost = sum(heads[t] * wages[t] for t in used)
price0 = {t: mm.get(t) or min(x for x in (Q(t).vwap7, Q(t).vwap30, Q(t).ask) if x) if (Q(t).vwap7 or Q(t).vwap30 or Q(t).ask) else Q(t).bid
for t in io["out"]}
rough = sum(io["out"][t] * price0[t] for t in io["out"]) - sum(io["in"][t] * ask(t) for t in io["in"]) - wcost
cands.append(dict(rec=rec, b=b, io=io, eff=eff, capex=capex, wcost=wcost, n_lim=n_lim, lim=lim, thin=thin, mm=mm,
rough_roi=100 * rough / capex, top=top, ins=ins, outs=outs))
top = TIER_CODE[keep[-1]]
if a.tier and TIER_RANK[top] > TIER_RANK[a.tier]:
continue
if any(hab_head[t] is None or wages[t] is None for t in keep):
continue
bd = {t + "s": heads[t] for t in econ.TIERS}
eff, _ = econ.building_efficiency(bd, {t: (1.0 if t in keep else 0.0) for t in econ.TIERS}, expertise=b["Expertise"] or None,
cogc=a.cogc, hq=a.hq or (bool(st.get("hq")) and not a.no_hq), experts=experts, faction=faction,
permits_used=pu, permits_total=pt,
fertility=planet["fertility"] if planet else None, is_farm=b["Ticker"] in ("FRM", "ORC"))
if eff <= 0:
continue # e.g. farms on fertility -1 planets
io = econ.production_io([dict(time_ms=rec["TimeMs"], inputs=ins, outputs=outs)], eff, 1)
n_lim, lim, thin, mm = float("inf"), "", False, {}
for t, qo in io["out"].items():
x = Q(t)
tr = sat.tref(x.traded7, x.traded30)
if x.mm_buy and x.bid and x.mm_buy >= 0.9 * x.bid:
mm[t] = x.mm_buy # market maker floor: unlimited depth at mm_buy
continue
thin |= sat.is_thin(tr, x.demand, qo)
n = sat.n_out(tr, 0, x.demand, qo) # stage 1: flow + demand only; queue penalty applied in stage 2 with the book
if n < n_lim:
n_lim, lim = n, t
if n_lim == float("inf"):
n_lim, lim = float(a.max_n), "MM"
if (thin or n_lim < a.min_n) and not a.show_thin:
continue
capex = capex0 + sum(heads[t] * hab_head[t] for t in keep)
wcost = sum(heads[t] * wages[t] for t in keep)
price0 = {t: mm.get(t) or min(x for x in (Q(t).vwap7, Q(t).vwap30, Q(t).ask) if x) if (Q(t).vwap7 or Q(t).vwap30 or Q(t).ask) else Q(t).bid
for t in io["out"]}
rough = sum(io["out"][t] * price0[t] for t in io["out"]) - sum(io["in"][t] * ask(t) for t in io["in"]) - wcost
staff = "".join(TIER_CODE[t] for t in keep) + ("" if len(keep) == len(need) else "-")
cands.append(dict(rec=rec, b=b, io=io, eff=eff, capex=capex, wcost=wcost, n_lim=n_lim, lim=lim, thin=thin, mm=mm,
rough_roi=100 * rough / capex, top=top, staff=staff, ins=ins, outs=outs))
cands.sort(key=lambda c: c["rough_roi"], reverse=True)
rows = []
@@ -162,13 +206,19 @@ def main():
if w["short"]:
return None
cost += w["total"]
return rev - cost - N * c["wcost"]
frt = 0.0
if a.trip_cost: # imports and exports share round trips: trips = worst of tonnes/m3 in either direction
wi = sum(N * qi * mat[t]["Weight"] for t, qi in io["in"].items()); vi = sum(N * qi * mat[t]["Volume"] for t, qi in io["in"].items())
wo = sum(N * qo * mat[t]["Weight"] for t, qo in io["out"].items()); vo = sum(N * qo * mat[t]["Volume"] for t, qo in io["out"].items())
c["tons"] = max(wi, wo)
frt = max(wi, wo, vi, vo) / a.cargo * a.trip_cost
c["frt"] = frt
dep = N * c["capex"] / a.deprec if a.deprec else 0.0 # value lost per day if demolished after holding
return rev - cost - N * c["wcost"] - frt - dep
if a.budget and c["capex"] > a.budget:
continue
n_star = max(1, min(int(c["n_lim"]), a.max_n))
if a.budget:
n_star = max(1, min(n_star, int(a.budget // c["capex"])))
n_star = max(1, a.own) # our size; the market-size filter (--min-n) already applied above
net1 = evaluate(1)
if net1 is None:
continue
@@ -180,15 +230,18 @@ def main():
continue
rows.append(dict(roi1=100 * net1 / c["capex"], roiN=100 * netN / n_star / c["capex"], total=netN, n=n_star,
n_lim=c["n_lim"], lim=c["lim"], capex=c["capex"] * n_star, eff=c["eff"], bld=c["b"]["Ticker"],
top=c["top"], h=c["rec"]["TimeMs"] / 3.6e6, thin=c["thin"],
top=c["top"], staff=c["staff"], tons=c.get("tons", 0), frt=c.get("frt", 0), h=c["rec"]["TimeMs"] / 3.6e6, thin=c["thin"],
rec=" ".join(f"{v:g}{k}" for k, v in c["ins"].items()) + " -> " + " ".join(f"{v:g}{k}" for k, v in c["outs"].items())))
key = {"roi": "roiN", "roi1": "roi1", "total": "total"}[a.sort]
rows.sort(key=lambda r: r[key], reverse=True)
if a.json:
import json
Path(a.json).write_text(json.dumps(rows))
print(f"# {a.cx} tier<={a.tier or 'any'} cogc={a.cogc or '-'} faction={faction or '-'} ({pu}/{pt} permits) min-n>={a.min_n} sorted by {a.sort}")
print("# price: patient asks near VWAP (clamped), inputs ask-walked; N* = market-absorbable buildings (cap %d); all estimates" % a.max_n)
print(f"{'ROI@1':>6} {'ROI@N*':>7} {'N*':>4} {'lim':4} {'profit/d@N*':>11} {'capex@N*':>9} {'eff':>4} {'bld':4} {'tier':4} {'h':>5} recipe")
print("# ROI/day is for OUR size (--own buildings, default 1) incl. our own price impact; market = buildings the market could absorb (--min-n filters on it, nothing else). Net of freight. Patient asks near VWAP, inputs ask-walked; all estimates")
print(f"{'ROI/d':>6} {'own':>3} {'market':>6} {'lim':4} {'profit/d':>11} {'capex':>9} {'eff':>4} {'bld':4} {'staff':5} {'t/d':>5} {'frt/d':>6} {'h':>5} recipe")
for r in rows[:a.top]:
print(f"{r['roi1']:6.1f} {r['roiN']:7.1f} {r['n']:4d} {r['lim']:4} {r['total']:11.0f} {r['capex']:9.0f} {r['eff']*100:4.0f} {r['bld']:4} {r['top']:4} "
print(f"{r['roiN']:6.1f} {r['n']:3d} {r['n_lim']:6.1f} {r['lim']:4} {r['total']:11.0f} {r['capex']:9.0f} {r['eff']*100:4.0f} {r['bld']:4} {r['staff']:5} {r['tons']:5.0f} {r['frt']:6.0f} "
f"{r['h']:5.1f} {r['rec']}{' [thin]' if r['thin'] else ''}")
tools/simulate.py
Executable
+87
View File
@@ -0,0 +1,87 @@
#!/usr/bin/env python3
"""Read a plan the way PRUNplanner's simulator does: efficiencies, workforce, material I/O, profit.
Source: a local spec (plans/*.yaml) or the plan stored in his PRUNplanner account (--uuid; picks up his UI edits).
tools/simulate.py plans/examples/base_plus_hwp.yaml
tools/simulate.py --uuid <plan-uuid> --basis ask --cx AI1
Prices: --basis real (buy at ask, sell at 7d VWAP at --cx: what a patient trader gets) | uni30 (default: volume-weighted 30d VWAP across all exchanges, = PRUNplanner 'Universe 30D') | vwap30 | vwap7 | ask | bid | mid at --cx."""
import argparse, sys
from pathlib import Path
import yaml
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from puga import ROOT, config, market, prunplanner as pp
from puga.simulate import simulate
sys.path.insert(0, str(Path(__file__).resolve().parent))
import plan_push
def main():
ap = argparse.ArgumentParser()
ap.add_argument("spec", nargs="?")
ap.add_argument("--uuid")
ap.add_argument("--cx", default=config.DEFAULT_CX)
ap.add_argument("--basis", default="uni30", choices=["real", "uni30", "vwap30", "vwap7", "ask", "bid", "mid"])
ap.add_argument("--off", default="", help="building tickers whose recipes are switched off (quantity 0), e.g. EXT,SME")
ap.add_argument("--no-faction", action="store_true")
ap.add_argument("--cm-free", action="store_true", help="treat the core module as free/already there (e.g. new base where CM cost is ignored)")
ap.add_argument("--no-hq", action="store_true", help="ignore hq: true from empire/state/company.yaml")
a = ap.parse_args()
recipes, blds = pp.recipes(), pp.buildings()
if a.uuid:
plan = pp.request("GET", f"/planning/plan/{a.uuid}/")
elif a.spec:
plan = plan_push.build_payload(yaml.safe_load(Path(a.spec).read_text()), recipes, {b["building_ticker"] for b in blds})
else:
sys.exit("give a spec file or --uuid")
off = {x.strip() for x in a.off.split(",") if x.strip()}
for b in plan["plan_data"]["buildings"]:
if b["name"] in off:
for ar in b["active_recipes"]:
ar["amount"] = 0
planet = pp._g(f"/data/planet/{plan['planet_natural_id']}/", 3600)
snap = market.snapshot()
def price(t, side="both"):
if a.basis == "real":
q = snap.get((t, a.cx))
return None if not q else (q.ask if side == "buy" else (q.vwap7 or q.vwap30 or q.bid))
if a.basis == "uni30":
return market.uni30(snap, t) or ((snap.get((t, a.cx)) or market.Quote(t, a.cx)).ask)
q = snap.get((t, a.cx))
if not q:
return None
v = {"vwap30": q.vwap30 or q.vwap7 or q.ask, "vwap7": q.vwap7 or q.vwap30 or q.ask, "ask": q.ask, "bid": q.bid,
"mid": (q.ask + q.bid) / 2 if q.ask and q.bid else None}[a.basis]
return v
import yaml as _y
st = _y.safe_load(config.state_path().read_text())
faction = None if a.no_faction else (st.get("company") or {}).get("faction")
perm = (st.get("permits", {}).get("used", 1), st.get("permits", {}).get("total", 2))
if not a.no_hq and st.get("hq"):
plan["plan_corphq"] = True
built = next((b.get("buildings", {}) for b in st.get("bases", []) if b.get("planet") == plan["planet_natural_id"]), {})
if a.cm_free:
built = {**built, "CM": 1}
r = simulate(plan, recipes, blds, planet["resources"], planet["fertility"], price, faction, perm, built=built)
print(f"{plan['plan_name']} {plan['planet_natural_id']} COGC {plan.get('plan_cogc')} prices: {a.cx} {a.basis}")
print(f"Area {r['area']:.0f}/500 Profit/day {r['profit']:,.0f} gross {r['gross']:,.0f} degradation {r['degradation']:,.0f} plan cost {r['plan_cost']:,.0f} ROI {r['roi_days'] and round(r['roi_days'], 2)} d")
nc = r["new_capex"]
print(f"NEW CAPEX = planned minus already built ({built or 'nothing built'}): {nc:,.0f} -> payback {nc / r['profit']:.2f} d = {100 * r['profit'] / nc:.1f}%/day" if r["profit"] > 0 and nc > 0 else ("NEW CAPEX: nothing to build" if nc <= 0 else "NEW CAPEX: profit <= 0"))
print("\nWORKFORCE need supply open eff%")
for t, w in r["workforce"].items():
if w["need"] or w["supply"]:
print(f" {t:10} {w['need']:5.0f} {w['supply']:6.0f} {w['open']:5.0f} {w['eff']*100:7.2f}")
print("\nBUILDINGS")
for b in r["buildings"]:
print(f" {b['amount']:3} x {b['building']:4} eff {b['efficiency']*100:7.2f}% {b['recipes']}")
print(f"\n{'MATERIAL':8} {'in/d':>9} {'out/d':>9} {'delta':>9} {'price':>7} {'value/d':>10}")
for tk, f in sorted(r["flows"].items(), key=lambda kv: -abs(kv[1]["value"])):
print(f"{tk:8} {f['inp']:9.2f} {f['out']:9.2f} {f['delta']:9.2f} {f['price']:7.0f} {f['value']:10.0f}")
if r["missing_prices"]:
print("no price for:", r["missing_prices"])
if __name__ == "__main__":
main()
tools/state.py
+12 -8
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Sync state/company.yaml from live FIO (own data via FIO_REST_KEY) and show it.
"""Sync empire/state/company.yaml from live FIO (own data via FIO_REST_KEY) and show it.
tools/state.py sync # overwrite live fields (buildings, production efficiency, storage, ships, cash, permits)
tools/state.py show
Requires the FIO extension to have uploaded recently; check `as_of`. Manual keys (company, notes) are preserved."""
@@ -8,23 +8,27 @@ from collections import Counter
from pathlib import Path
import yaml
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from puga import ROOT, fio
from puga import ROOT, config, fio
P = ROOT / "state" / "company.yaml"
P = config.EMPIRE_DIR / "state" / "company.yaml" # gitignored
def sync():
u = fio.me()
P.parent.mkdir(parents=True, exist_ok=True)
st = yaml.safe_load(P.read_text()) if P.exists() else {}
company = fio.private(f"/company/code/{(st.get('company') or {}).get('ticker', 'GBI')}", ttl=0)
code = (st.get('company') or {}).get('ticker') or config.get('COMPANY_CODE')
if not code:
sys.exit('set COMPANY_CODE=<your company code> in .env (first sync), or put company: {ticker: ..} in empire/state/company.yaml')
company = fio.own(f"/company/code/{code}", ttl=0)
for stale in ("cash_aic",):
st.pop(stale, None)
st["as_of"] = datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%MZ")
st["cash"] = {b["Currency"]: b["Amount"] for b in company["Balances"] if b["Amount"]}
sites = fio.private(f"/sites/{u}", ttl=0)
prod = fio.private(f"/production/{u}", ttl=0)
stores = fio.private(f"/storage/{u}", ttl=0)
ships = fio.private(f"/ship/ships/{u}", ttl=0)
sites = fio.own(f"/sites/{u}", ttl=0)
prod = fio.own(f"/production/{u}", ttl=0)
stores = fio.own(f"/storage/{u}", ttl=0)
ships = fio.own(f"/ship/ships/{u}", ttl=0)
st["permits"] = {"used": sites[0]["InvestedPermits"] if sites else 0, "total": sites[0]["MaximumPermits"] if sites else 0}
old = {b["planet"]: b for b in st.get("bases", [])}
bases = []