PuGa/tools/scan.py

#!/usr/bin/env python3
"""Depth-aware single-step recipe scan at one exchange (saturation model v1, see docs/saturation-design.md).
For each recipe: buildings the market can absorb (N*), and payback/yield at N=1 and N* using a patient sell
price (not top-of-book) and ask-walked inputs. Excludes thin markets by default. yld/d is profit/day over
capex as a %, NOT a compoundable rate - it's 1/payback; see docs/decisions.md's "Drop ROI/day %" entry.

  tools/scan.py                              # AI1, all tiers, top 30 by yield at N*
  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, 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, prunplanner as pp, saturation as sat
from puga import planet as pl

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 = config.state_path()
    return yaml.safe_load(p.read_text()) if p.exists() else {}


def main():
    ap = argparse.ArgumentParser()
    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: assumed holding days before demolition; amortizes the UNRECLAIMED portion of capex (capex x (1 - --reclaim-frac)) over that period, NOT full capex - real FIO reclaimables data (this account's own buildings, 2026-10-10) showed ~50-86%% returned even at near-100%% condition, contradicting a 100%%->0 time-decay story. See docs/decisions.md.")
    ap.add_argument("--reclaim-frac", type=float, default=0.65, help="fraction of capex recovered on demolition - real observed range 50-86%% across 7 owned building types (CM a low outlier at ~7%%, excluded); single base/age, small sample, treat as a rough placeholder not a calibrated curve")
    ap.add_argument("--max-impact", type=float, default=10.0, help="max %% the realized price may degrade below vwap7/bid when walking the LIVE bid book for your planned daily output (puga.saturation.max_units_at_impact) - feeds the [exit-thin] warning flag: can you instantly DUMP a day of output into resting bids if you needed cash now, not whether your patient asks clear (that's the market/lim columns). See docs/decisions.md")
    ap.add_argument("--input", help="only recipes consuming this ticker as an input, e.g. --input AL for recipes built on top of your own AL chain")
    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; yield/payback 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 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")
    ap.add_argument("--mcg", type=float, default=4.0, help="MCG per area (4 on normal planets)")
    ap.add_argument("--sort", default="roi", choices=["roi", "total", "roi1", "area"])
    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()
    Q = lambda t: snap.get((t, a.cx))
    ask = lambda t: (Q(t).ask if Q(t) else None)
    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):
        b = blds[tk]
        parts = [(x["Amount"], ask(x["CommodityTicker"])) for x in b["BuildingCosts"]]
        if any(p is None for _, p in parts):
            return None
        cost = sum(n * p for n, p in parts) + a.mcg * b["AreaCost"] * mcg
        if planet is not None:
            extras = pl.environment_extras(planet, b["AreaCost"])
            prices = {etk: ask(etk) for etk in extras}
            if any(p is None for p in prices.values()):
                return None
            cost += sum(qty * prices[etk] for etk, qty in extras.items())
        return cost

    hab_head, hab_area = {}, {}
    for tier, hab in (("pioneer", "HB1"), ("settler", "HB2"), ("technician", "HB3"), ("engineer", "HB4"), ("scientist", "HB5")):
        c = bcost(hab)
        hab_head[tier] = None if c is None else c / econ.HAB_CAP[hab][tier]
        hab_area[tier] = econ.HAB_AREA[hab] / econ.HAB_CAP[hab][tier]

    def wage(tier):  # per worker per day, both luxuries supplied, at ask
        tot = 0.0
        for tk, need, _, _ in econ.CONSUMPTION[tier]:
            p = ask(tk)
            if p is None:
                return None
            tot += need / 100 * p
        return tot

    experts = {k.strip().upper(): int(v) for k, v in (x.split("=") for x in a.experts.split(",") if "=" in x)}
    skip = {x.strip() for x in a.skip.split(",") if x.strip()}
    wages = {t: wage(t) for t in econ.TIERS}
    cands = []
    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}
        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 a.input and a.input.upper() not in ins:
            continue
        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
        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
            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, qo)      # stage 1: flow 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,
                              hab_area_per_bld=sum(heads[t] * hab_area[t] for t in keep)))

    cands.sort(key=lambda c: c["rough_roi"], reverse=True)
    rows = []
    book, bandcache = {}, {}

    def raw_book(t):
        if t not in book:
            book[t] = fio.order_book(t, a.cx)
        return book[t]

    def asks_of(t):
        return sorted((o["ItemCost"], (o["ItemCount"] or 0)) for o in raw_book(t)["SellingOrders"])

    def bids_of(t):
        return [(o["ItemCost"], (o["ItemCount"] or 0)) for o in raw_book(t)["BuyingOrders"] if o.get("ItemCost")]

    def band_of(t):
        if t not in bandcache:
            ob = raw_book(t)
            lo = ob.get("WidePriceBandLow") or ob.get("NarrowPriceBandLow")
            hi = ob.get("WidePriceBandHigh") or ob.get("NarrowPriceBandHigh")
            bandcache[t] = (lo, hi)
        return bandcache[t]

    for c in cands[:a.k]:
        io, mm = c["io"], c["mm"]
        # stage 2: queue penalty from competing asks near market price, flow from REAL daily-volume
        # variance (tref_precise) now that we're down to a bounded candidate set worth the extra fetch
        n2, bid_thin = float("inf"), False
        for t, qo in io["out"].items():
            if t in mm:
                continue
            x = Q(t)
            se = sat.effective_supply(asks_of(t), x.vwap7 or x.ask, band=band_of(t))
            n2 = min(n2, sat.n_out(sat.tref_precise(t, a.cx, x.traded7, x.traded30), se, qo))
            # Bid-book depth is a single LIVE SNAPSHOT (orders get added/withdrawn constantly) -
            # not reliable enough to use as a hard cap like the flow-based n_out above, which
            # aggregates real traded volume over 30 days. Surfaced as a warning flag instead, and
            # measures a DIFFERENT thing than market/lim: not "can my patient asks clear" (that's
            # p_patient()/n_out, ask-side, the real production-selling mechanism), but "could I
            # instantly dump a day of output into resting BIDS right now if I needed cash" - panic-
            # exit liquidity, not production price impact. A "yes" here doesn't guarantee it holds
            # up over time, and a "no" might just be an unlucky instant - re-check before acting.
            depth_units = sat.max_units_at_impact(bids_of(t), x.vwap7 or x.bid, a.max_impact)
            bid_thin |= depth_units < qo
        c["n_lim"] = float(a.max_n) if n2 == float("inf") else n2
        c["bid_thin"] = bid_thin
        if c["n_lim"] < a.min_n and not a.show_thin:
            continue

        def evaluate(N):
            rev = 0.0
            for t, qo in io["out"].items():
                x = Q(t)
                if t in mm:
                    p = mm[t]
                else:
                    p = sat.p_patient(asks_of(t), sat.tref_precise(t, a.cx, x.traded7, x.traded30), N * qo, x.bid, x.vwap7, x.vwap30, x.ask, band_of(t)[1])
                rev += N * qo * p
            cost = 0.0
            for t, qi in io["in"].items():
                w = market.walk(t, a.cx, N * qi, "buy")
                if w["short"]:
                    return None
                cost += w["total"]
            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"] * (1 - a.reclaim_frac) / a.deprec if a.deprec else 0.0     # unreclaimed capex 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, a.own)                  # our size; the market-size filter (--min-n) already applied above
        net1 = evaluate(1)
        if net1 is None:
            continue
        netN = evaluate(n_star)
        while netN is None and n_star > 1:
            n_star = max(1, n_star // 2)
            netN = evaluate(n_star)
        if netN is None:
            continue
        area = n_star * ((c["b"].get("AreaCost") or 0) + c.get("hab_area_per_bld", 0.0))
        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"], staff=c["staff"], tons=c.get("tons", 0), frt=c.get("frt", 0), h=c["rec"]["TimeMs"] / 3.6e6,
                         thin=c["thin"], bid_thin=c.get("bid_thin", False),
                         area=area, profit_area=netN / area if area else 0.0,
                         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", "area": "profit_area"}[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("# yld/d is profit/day / capex as a %, for OUR size (--own buildings, default 1) incl. our own price impact; this is NOT a compoundable rate (1/payback, see docs/decisions.md) - pback is the number to actually reason about. 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("# profit/area ranks by AIC/day per tile of TOTAL footprint - production building AreaCost PLUS the single-tier hab area its own workforce drags along (heads/100 x hab area, e.g. a 100-head pioneer building pulls a full HB1) - once a base's area cap binds (not cash), this is the real allocation criterion, not yld/d. --sort area to rank by it.")
    print("# [exit-thin] = INSTANT-DUMP liquidity check, not production price impact: right now, the LIVE bid book can't absorb even 1 day of 1 building's output within --max-impact%% if you needed to sell immediately. A single snapshot (orders change constantly), NOT folded into market/lim (which measures patient ask-side selling via real 30d flow, the actual production mechanism). Re-check before acting on it.")
    print(f"{'yld/d':>6} {'pback':>6} {'own':>3} {'market':>6} {'lim':4} {'profit/d':>11} {'capex':>9} {'area':>5} {'$/area':>7} {'eff':>4} {'bld':4} {'staff':5} {'t/d':>5} {'frt/d':>6} {'h':>5}  recipe")
    for r in rows[:a.top]:
        pback = f"{100/r['roiN']:6.1f}" if r['roiN'] > 0 else "   inf"
        flags = ('  [thin]' if r['thin'] else '') + ('  [exit-thin]' if r['bid_thin'] else '')
        print(f"{r['roiN']:6.1f} {pback} {r['n']:3d} {r['n_lim']:6.1f} {r['lim']:4} {r['total']:11.0f} {r['capex']:9.0f} {r['area']:5.0f} {r['profit_area']:7.1f} {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']}{flags}")


if __name__ == "__main__":
    main()