diff --git a/CLAUDE.md b/CLAUDE.md index b33cac1..eb1ff1b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,6 +32,9 @@ Agent-first toolkit for advising a Prosperous Universe player. The player asks q - `puga/` library; `tools/` CLIs; `plans/examples/` generic plan specs (also test fixtures); `state/company.example.yaml` example state; `tests/`; `data/` (gitignored fetched data cache); `ref/` (gitignored PRUNplanner source). - Rule: game facts and generic methods go in `docs/`; anything about the player's own bases, cash, decisions or preferences goes in `empire/`. +## Library use (not just CLI) +For anything the CLI tools don't expose (custom what-ifs, cost breakdowns with wages/capex, ad-hoc market checks) import `puga.*` directly in a short inline script instead of stretching a tool's flags. See `docs/library.md` for the module map and patterns; most non-trivial answers in this project came from a `python - < ...` (e.g. `puga scan --min-n 3`, `puga plan pull `); the `tools/x.py` paths below are the same programs. - `tools/state.py sync|show` refreshes `empire/state/company.yaml` from live FIO (cash, permits, buildings, real production efficiency, storage, ships). @@ -39,6 +42,8 @@ Run as `puga ...` (e.g. `puga scan --min-n 3`, `puga plan pull `); - `tools/simulate.py [--off EXT,SME] [--basis real|uni30|vwap30|ask|...]` replica of PRUNplanner's simulator (efficiency, workforce, material I/O, profit). Flows and efficiency verified exact vs screenshots; profit within ~2%. The NEW CAPEX line = planned minus already built (from state), the honest payback; ignore the plan-level ROI (PRUNplanner always adds a core module). `--uuid` reads the SAVED plan from the account (UI edits must be saved first). - `tools/scan.py` DEPTH-AWARE recipe scan (use this): buildings the market could absorb, ROI for our own size, patient prices, ask-walked inputs. Fully staffed AND understaffed variants by default (`--staffing`), freight netted out (`--trip-cost`, `--cargo`), HQ/COGC/experts/faction, `--planet ID` adds that planet's extraction, fertility and COGC (a planet not in state is a new base: no HQ, permits+1), `--deprec 60` prices demolish-later. `--min-n` is ONLY a noise filter on market capacity (use 3, ideally 10); ROI is always for our own size (`--own`, default 1 building), never at the filtered scale. Below 1 lets sub-building junk in. `--json rows.json` feeds `persistence.py`. Library: `puga/saturation.py`. - `tools/history.py TICKER` monthly margin history of the recipe producing TICKER; `tools/persistence.py rows.json` re-prices scan rows over history (mean ROI 7/14/30/90/180d, payback, net gain over 7 and 14 days, % days profitable). ALWAYS check persistence before recommending: current margins are often a spike. +- `tools/buy.py --days N` cash-aware shopping list: construction gap (plan vs empire state's built buildings) plus N days of the plan's NET operating stock (via `simulate()`, so self-produced inputs like AL net against consumption), priced at real fill cost, checked against cash/reserve. Pass a plan covering the WHOLE base for restocking, not just a new addition (see `docs/library.md`). +- `tools/sell.py TICKER QTY` where to post an ask (undercutting the current best) vs hitting the bids now, plus expected hours to clear from 30-day traded-volume percentiles. - `tools/chain.py TICKER` make-vs-buy cost tree plus sourcing depth of inputs. - `tools/price.py` prices across exchanges with VWAP, volume, fill price for a quantity; `tools/book.py` order-book ladder. - `tools/prun_scan.py`, `tools/prun_cxarb.py` legacy top-of-book scans; overstate thin markets. Baseline only. `tools/arb.py` replacement is on the roadmap. diff --git a/docs/library.md b/docs/library.md new file mode 100644 index 0000000..657fc58 --- /dev/null +++ b/docs/library.md @@ -0,0 +1,110 @@ +# Using `puga` as a library (not the CLI) + +The `tools/*.py` scripts cover the common questions. For anything more involved — a custom +what-if, a one-off cost breakdown with wages and capex, a market check the CLI doesn't expose — +import the modules directly and write a short script inline. This is normal; most answers in +this project's history came from a `python - < tool), `docs/decisions.md`. +13. [x] `tools/buy.py`: cash-aware shopping list (construction gap vs empire state's built count, plus N days of net operating stock via simulate(), priced at real fill cost, checked against cash/reserve). +14. [x] `tools/sell.py`: ask-post-price + expected-clear-time advisor from 30d traded-volume percentiles, vs hitting the bids now. + 12. [ ] `tools/portfolio.py`: choose the set of 1-building opportunities for one base that maximises total profit given shared whole-building housing (HB1/HB2/...), capex, permits and (for demolish-later) 60-day value decay. Persistence check exists (`tools/persistence.py`); wire it into the ranking. ## Open questions diff --git a/tests/test_buy.py b/tests/test_buy.py new file mode 100644 index 0000000..b6f151a --- /dev/null +++ b/tests/test_buy.py @@ -0,0 +1,39 @@ +import importlib.util +from pathlib import Path + +spec = importlib.util.spec_from_file_location("buy", Path(__file__).resolve().parent.parent / "tools" / "buy.py") +buy = importlib.util.module_from_spec(spec) +spec.loader.exec_module(buy) + +BUILDINGS = [ + {"building_ticker": "HWP", "costs": [{"material_ticker": "BBH", "material_amount": 4}, {"material_ticker": "LTA", "material_amount": 2}], "area_cost": 25}, + {"building_ticker": "HB2", "costs": [{"material_ticker": "BTA", "material_amount": 2}], "area_cost": 12}, +] +PLAN = {"plan_data": {"buildings": [{"name": "HWP", "amount": 1}], "infrastructure": [{"building": "HB2", "amount": 1}]}} + + +def test_construction_items_include_mcg_and_full_amount_when_nothing_built(): + need = buy.construction_items(PLAN, BUILDINGS, built={}) + assert need == {"BBH": 4, "LTA": 2, "MCG": 100 + 48, "BTA": 2} + + +def test_construction_items_gap_only_counts_unbuilt_units(): + need = buy.construction_items(PLAN, BUILDINGS, built={"HWP": 1, "HB2": 1}) + assert need == {} # already fully built: nothing to buy + + +def test_construction_items_partial_gap(): + plan = {"plan_data": {"buildings": [{"name": "HWP", "amount": 3}], "infrastructure": []}} + need = buy.construction_items(plan, BUILDINGS, built={"HWP": 1}) + assert need == {"BBH": 8, "LTA": 4, "MCG": 200} # 2 more HWP needed, not 3 + + +def test_stock_items_only_keeps_net_consumption(): + flows = {"C": {"inp": 11.13, "out": 0}, "AL": {"inp": 42.78, "out": 44.57}, "BHP": {"inp": 0, "out": 14.26}} + got = buy.stock_items(flows, days=2) + assert got == {"C": 22.26} # AL nets to +1.78/day (own smelters cover the HWP): not bought + + +def test_net_of_on_hand_floors_at_zero(): + got = buy.net_of_on_hand({"C": 20, "O": 5}, {"C": 30, "O": 2}) + assert got == {"C": 0.0, "O": 3.0} diff --git a/tools/buy.py b/tools/buy.py new file mode 100755 index 0000000..73c6a6f --- /dev/null +++ b/tools/buy.py @@ -0,0 +1,144 @@ +#!/usr/bin/env python3 +"""Cash-aware shopping list: construction gap (plan vs what's already built) plus N days of the +plan's steady-state operating stock (from the simulator), priced at real order-book fill cost, +checked against cash. Formalizes the by-hand calc used repeatedly for buildouts and restocks. + + puga buy plans/examples/hwp_buildout.yaml --days 2 # nothing built yet: full construction + 2d stock + puga buy plans/examples/hwp_buildout.yaml --days 4 # once built: construction gap is empty, just restock + puga buy --uuid --days 3 --cash 72200 --reserve 3000 + +Materials already sitting in the base's storage (empire state) are subtracted from both sections. +IMPORTANT for --days (stock): pass a plan that lists EVERY production building at the base, not +just a new addition, or its own production won't net against consumption (e.g. a standalone HWP +plan will say "buy 42.8 AL/day" because it can't see your smelters; the full base+HWP plan nets +that to ~0). Use the full plan for restocking; a partial plan is fine for construction-only. +Limitation: on-hand storage in empire/state/company.yaml is not scoped per base; with more than +one base this will misattribute inventory. Fine while there is only one. +""" +import argparse, math, sys +from pathlib import Path +import yaml +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from puga import config, fio, market, prunplanner as pp +from puga.simulate import simulate + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import plan_push + + +def construction_items(plan: dict, buildings: list, built: dict) -> dict: + """Materials needed for the plan's buildings/infra beyond what `built` (ticker -> count) already has.""" + bl = {b["building_ticker"]: b for b in buildings} + d = plan["plan_data"] + need: dict[str, float] = {} + for tk, amt in [(b["name"], b["amount"]) for b in d["buildings"]] + [(i["building"], i["amount"]) for i in d["infrastructure"]]: + gap = amt - built.get(tk, 0) + if gap <= 0: + continue + info = bl[tk] + for c in info["costs"]: + need[c["material_ticker"]] = need.get(c["material_ticker"], 0) + c["material_amount"] * gap + need["MCG"] = need.get("MCG", 0) + 4 * info["area_cost"] * gap + return need + + +def stock_items(flows: dict, days: float) -> dict: + """days worth of NET shortfall (consumption minus own production) from simulate()'s flows. + A material the plan both consumes and produces (e.g. AL: SME makes it, HWP eats it) nets out; + only buy what production doesn't cover.""" + return {tk: -delta * days for tk, f in flows.items() if (delta := f["out"] - f["inp"]) < 0} + + +def net_of_on_hand(items: dict, on_hand: dict) -> dict: + return {tk: max(0.0, q - on_hand.get(tk, 0)) for tk, q in items.items()} + + +def price(items: dict, cx: str): + mat = {m["Ticker"]: m for m in fio.materials()} + rows, total, weight, volume = [], 0.0, 0.0, 0.0 + for tk, qty in sorted(items.items()): + qty = math.ceil(qty) + if qty <= 0: + continue + w = market.walk(tk, cx, qty, "buy") + rows.append((tk, qty, w["avg"] or 0, w["total"], w["short"])) + total += w["total"] + m = mat.get(tk) + if m: + weight += qty * m["Weight"] + volume += qty * m["Volume"] + return rows, total, weight, volume + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("spec", nargs="?", help="plans/*.yaml describing the target setup") + ap.add_argument("--uuid", help="read the plan from the PRUNplanner account instead") + ap.add_argument("--days", type=float, default=3, help="days of operating stock on top of construction") + ap.add_argument("--cx", default=config.DEFAULT_CX) + ap.add_argument("--cash", type=float, help="override cash (default: empire state)") + ap.add_argument("--reserve", type=float, default=0, help="AIC to keep unspent") + ap.add_argument("--cargo", type=float, default=500.0, help="ship capacity t/m3, to flag multi-trip") + 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") + + st = yaml.safe_load(config.state_path().read_text()) + cash = a.cash if a.cash is not None else st.get("cash", {}).get("AIC", 0) + base = next((b for b in st.get("bases", []) if b.get("planet") == plan["planet_natural_id"]), {}) + built = base.get("buildings", {}) + on_hand = next((s["items"] for s in st.get("storage", []) if s.get("type") == "STORE"), {}) + + if not st.get("hq"): + plan["plan_corphq"] = False + faction = st.get("company", {}).get("faction") + perm = (st.get("permits", {}).get("used", 1), st.get("permits", {}).get("total", 2)) + planet = pp._g(f"/data/planet/{plan['planet_natural_id']}/", 3600) + price_ask = lambda t, side="buy": (lambda q: q.ask if q else None)(market.snapshot().get((t, a.cx))) + r = simulate(plan, recipes, blds, planet["resources"], planet["fertility"], price_ask, faction, perm) + + con = net_of_on_hand(construction_items(plan, blds, built), on_hand) + stk = net_of_on_hand(stock_items(r["flows"], a.days), on_hand) + con_rows, con_total, con_w, con_v = price(con, a.cx) + stk_rows, stk_total, stk_w, stk_v = price(stk, a.cx) + + def show(title, rows): + if not rows: + return + print(f"\n{title}") + for tk, qty, avg, cost, short in rows: + have = on_hand.get(tk, 0) + print(f" {tk:5} {qty:8.0f} (have {have:5.0f}) avg {avg:7.0f} = {cost:9,.0f}{' SHORT BOOK' if short else ''}") + + show(f"CONSTRUCTION (plan vs built {built or 'nothing'})", con_rows) + if con_rows: + print(f" subtotal {con_total:9,.0f} ({con_w:.0f} t, {con_v:.0f} m3)") + show(f"OPERATING STOCK, {a.days:g} days", stk_rows) + if stk_rows: + print(f" subtotal {stk_total:9,.0f} ({stk_w:.0f} t, {stk_v:.0f} m3)") + + total, weight, volume = con_total + stk_total, con_w + stk_w, con_v + stk_v + budget = cash - a.reserve + print(f"\nTOTAL {total:,.0f} AIC ({weight:.0f} t, {volume:.0f} m3{' -- fits one trip' if max(weight, volume) <= a.cargo else ' -- NEEDS MULTIPLE TRIPS'})") + print(f"CASH {cash:,.0f} minus reserve {a.reserve:,.0f} = budget {budget:,.0f}") + if total <= budget: + print(f"AFFORDABLE, {budget - total:,.0f} left over") + elif con_total <= budget: + lo, hi = 0.0, a.days + for _ in range(20): + mid = (lo + hi) / 2 + _, t, _, _ = price(net_of_on_hand(stock_items(r["flows"], mid), on_hand), a.cx) + lo, hi = (mid, hi) if con_total + t <= budget else (lo, mid) + print(f"SHORT by {total - budget:,.0f}: construction alone fits; stock affordable at ~{lo:.2f} days instead of {a.days:g}") + else: + print(f"SHORT by {total - budget:,.0f}: even construction alone ({con_total:,.0f}) exceeds budget") + + +if __name__ == "__main__": + main() diff --git a/tools/sell.py b/tools/sell.py new file mode 100755 index 0000000..e9d515e --- /dev/null +++ b/tools/sell.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Where to post a sell order and how long it should take to clear, vs hitting the bids now. +Formalizes the by-hand check done before every AL/BHP sale. + + puga sell BHP 8 --cx AI1 + puga sell AL 16 --undercut 5 +""" +import argparse, datetime, statistics, sys +from pathlib import Path +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from puga import config, fio, market + + +def daily_traded(tk: str, cx: str, days: int = 30) -> list[float]: + rows = [(datetime.datetime.fromtimestamp(e["DateEpochMs"] / 1000, datetime.timezone.utc).date(), e["Traded"]) + for e in fio.cxpc(tk, cx) if e.get("Interval") == "DAY_ONE" and e.get("Traded")] + return [t for _, t in rows[-days:]] + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("ticker") + ap.add_argument("qty", type=float) + ap.add_argument("--cx", default=config.DEFAULT_CX) + ap.add_argument("--undercut", type=float, default=10, help="AIC to undercut the current best ask by") + a = ap.parse_args() + t = a.ticker.upper() + ob = fio.order_book(t, a.cx) + asks = sorted((o["ItemCost"], o["ItemCount"] or 0) for o in ob["SellingOrders"] if o.get("ItemCost")) + best_ask = asks[0][0] if asks else None + post = round((best_ask - a.undercut) if best_ask else (ob.get("Ask") or 0)) + ahead = sum(u for p, u in asks if p < post) + + hit = market.walk(t, a.cx, a.qty, "sell") + trades = daily_traded(t, a.cx) + med = statistics.median(trades) if trades else 0 + lo = sorted(trades)[int(0.2 * (len(trades) - 1))] if trades else 0 + hi = sorted(trades)[int(0.8 * (len(trades) - 1))] if trades else 0 + + print(f"{t}.{a.cx} best ask {best_ask} bid {ob.get('Bid')} vwap7 {ob.get('PriceAverage')}") + print(f"\nOPTION A: post {a.qty:g} at {post:.0f} ({ahead:.0f} units ahead of you at a lower price)") + print(f" revenue if filled: {post * a.qty:,.0f}") + if med: + print(f" expected clear time (last {len(trades)}d volume): busy day {24 * a.qty / hi:.1f}h | " + f"median day {24 * a.qty / med:.1f}h | slow day {24 * a.qty / max(lo, 1):.1f}h") + else: + print(" no recent trade history to estimate clear time") + + print(f"\nOPTION B: hit the bids now (instant)") + print(f" revenue: {hit['total']:,.0f} (avg {hit['avg']:.0f}, worst {hit['worst']:.0f}" + + (f", SHORT: book only fills {hit['filled']:.0f}" if hit["short"] else "") + ")") + + print(f"\nposting patiently gains {post * a.qty - hit['total']:,.0f} over hitting the bids, at the cost of waiting") + + +if __name__ == "__main__": + main() diff --git a/tools/simulate.py b/tools/simulate.py index 62eb219..548f80b 100755 --- a/tools/simulate.py +++ b/tools/simulate.py @@ -58,7 +58,9 @@ def main(): 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"): + if a.no_hq: + plan["plan_corphq"] = False + elif 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: diff --git a/tools/state.py b/tools/state.py index 5803c33..8831750 100755 --- a/tools/state.py +++ b/tools/state.py @@ -30,6 +30,8 @@ def sync(): 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} + if st.get("permits_total_override"): # e.g. the HQ screen (base permits x / y) disagrees with FIO's MaximumPermits + st["permits"]["total"] = st["permits_total_override"] old = {b["planet"]: b for b in st.get("bases", [])} bases = [] for s in sites: