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

scan.py: real bid-depth liquidity flag, real-variance flow bound, profit/area

Browse Source 
- max_units_at_impact(): walks the LIVE bid book for a genuine, ticker-
  specific instant-exit liquidity check, replacing a flat 25%-of-flow
  guess (a price-vs-volume elasticity regression was tried first and
  abandoned - confounded by demand-driven volume on real data). Surfaced
  as an [exit-thin] WARNING flag, not a hard cap - a single order-book
  snapshot is less reliable than the 30-day flow aggregate that still
  drives n_lim/filtering (using it as a hard cap once silently zeroed
  out a real recipe from one noisy snapshot).
- tref_precise(): real variance of daily traded volume instead of a
  Poisson unit-independence bound, wired into stage 2 only (bounded
  candidate set, cached). Also fixed: was averaging only over trading
  days, overstating flow ~3x for sparse tickers.
- n_out(): dropped the demand-coverage term, which penalized thin
  resting-bid books even though production sells via patient asks taken
  by walk-up buyers with no resting bid at all.
- --sort area: profit/day per tile of TOTAL footprint, now including the
  single-tier hab area a building's workforce drags along (was building
  area only).
- --deprec/--reclaim-frac: amortizes unreclaimed capex, not full capex -
  real FIO reclaim data showed 50-86% returned even at near-100%
  condition, contradicting a 100%->0 decay-curve assumption.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
dodox
2026-10-10 23:06:49 +02:00
co-authored by Claude Sonnet 5
parent 11f54276b3
commit 3f67cc7ed5
3 changed files with 135 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
puga/saturation.py
+71 -9
View File
@@ -1,34 +1,96 @@
"""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
Idea: N buildings is limited by (a) share of the market's steady traded flow and (b) the queue of
existing sell orders (supply days). Price is a haircut model, not a book walk.
is_thin() checks demand depth as an independent red flag (a genuinely dead bid book is worth
knowing about); n_out() does not multiply by it (see n_out's own docstring for why)."""
import math, statistics
from .history import trading_days
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_D = 7 # days of flow standing demand should cover (used by is_thin only)
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."""
"""Conservative daily flow: min(7d, 30d); with a Poisson lower bound when 30d volume is small.
CAVEAT: the Poisson term treats each traded UNIT as an independent event, which overstates
confidence for lumpy real trades (one 500-unit trade is one event, not 500) - see
`tref_precise()` for the real-variance version, used once a candidate reaches stage 2 (this
cheap version stays for stage-1 screening across every recipe, where fetching daily history
for every ticker would be too expensive)."""
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 tref_precise(tk: str, cx: str, traded7: float, traded30: float, days: int = 30) -> float:
"""Like tref(), but the lower bound comes from the REAL variance of daily traded volume
(`puga.history.trading_days`) instead of a Poisson unit-independence assumption - one 500-unit
trade is one data point here, not 500 independent ones. Falls back to tref() if there's too
little real history to estimate variance from (<=2 trading days).
`trading_days()` only returns days with a trade - zero-volume days are absent, not zero - so
the volume list is padded with explicit zeros for the missing calendar days before computing
mean/stdev, and `days` (not the trading-day count) is used as n. Averaging only over the rows
that have a trade would compute the mean volume ON TRADING DAYS, overstating real
per-CALENDAR-day flow by 1/(fraction of days traded) - worst precisely for the sparse tickers
this function exists to protect."""
td = trading_days(tk, cx, days)
if len(td) <= 2:
return tref(traded7, traded30)
vols = [v for _, v in td] + [0.0] * (days - len(td))
mean = statistics.mean(vols)
sd = statistics.stdev(vols)
bound = max(0.0, mean - 1.96 * sd / math.sqrt(days))
return max(0.0, min(tref(traded7, traded30), bound))
def max_units_at_impact(bids: list[tuple[float, float]], ref_price: float, max_impact_pct: float = 10.0) -> float:
"""Real, ticker-specific price-impact cap, replacing the flat S_SHARE guess for THIS check:
how many units could be sold into the LIVE bid book before the price paid degrades more than
`max_impact_pct` below `ref_price`. Walks bids highest-first (puga.market.walk's own
mechanics) and stops at the first tier priced below the threshold.
A day-over-day price-vs-volume regression does not give a usable general elasticity estimate
here: volume and price co-move mostly from DEMAND bursts (more buyers -> more volume AND
steady/higher price), which swamps and often reverses the sign of the SUPPLY-side effect
actually wanted (does MY added output push price down). Walking the real book sidesteps that
confounding, but only captures the static liquidity snapshot, not how the market reprices
over weeks of sustained extra output (nothing in available data answers that cleanly). See
docs/decisions.md."""
if not ref_price or ref_price <= 0:
return 0.0
threshold = ref_price * (1 - max_impact_pct / 100)
cum = 0.0
for price, qty in sorted(bids, key=lambda x: -x[0]):
if price < threshold:
break
cum += qty
return cum
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."""
def n_out(tr: float, supply: float, q_out: float) -> float:
"""Buildings the output market absorbs: flow share x queue penalty. 0 if no flow.
Deliberately has NO resting-bid-depth term: production sells via patient ASKS (`p_patient()`)
that get taken by buyers crossing the spread, including walk-up buyers with no resting bid at
all - a thin resting bid book doesn't mean they won't show up, since `tr` (real traded flow,
already in `base`) is the actual evidence they do. A market with healthy ask-side flow and
thin standing bids is normal (most buyers don't leave a bid sitting around), not evidence of
absent demand. Real resting-bid demand is still checked independently by `is_thin()` as a red
flag, not folded into this score."""
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
return base * queue
def p_patient(asks: list[tuple[float, float]], tr: float, produced_per_day: float, bid: float, vwap7: float | None,
tests/test_saturation.py
+24 -5
View File
@@ -1,3 +1,4 @@
import datetime
import math
from puga import saturation as s, market, fio
@@ -8,12 +9,22 @@ def test_tref_uses_min_and_poisson_floor():
assert s.tref(0, 0) == 0
def test_n_out_penalises_queue_and_thin_demand():
free = s.n_out(100, 0, 10_000, 5) # no queue, ample demand
def test_tref_precise_pads_missing_calendar_days_not_just_trading_days(monkeypatch):
# 10 trading days out of 30, each with volume 30 -> real per-calendar-day mean is 10, not 30.
# The original bug averaged only over the 10 trading-day rows (mean 30), overstating the
# bound 3x for exactly the sparse tickers it's meant to protect.
base = datetime.date(2026, 1, 1)
fake_td = [(base + datetime.timedelta(days=i * 3), 30.0) for i in range(10)]
monkeypatch.setattr(s, "trading_days", lambda tk, cx, days: fake_td)
bound = s.tref_precise("FAKE", "AI1", traded7=30, traded30=30, days=30)
assert bound < 15.0 # well below the naive trading-day mean of 30, reflecting real sparsity
def test_n_out_penalises_queue():
free = s.n_out(100, 0, 5) # no queue
assert math.isclose(free, 0.25 * 100 / 5)
assert s.n_out(100, 7000, 10_000, 5) < free # 70 days of queue cuts it
assert s.n_out(100, 0, 100, 5) < free # demand only 1 day of flow cuts it
assert s.n_out(0, 0, 100, 5) == 0
assert s.n_out(100, 7000, 5) < free # 70 days of queue cuts it
assert s.n_out(0, 0, 5) == 0
def test_thin_flag():
@@ -57,3 +68,11 @@ def test_effective_demand_excludes_stale_bids_below_the_band():
bids = [(4550, 8), (4200, 250), (2810, 8), (334, 19)]
assert s.effective_demand(bids, band=(663, 16575)) == 8 + 250 + 8 # the 334 bid is stale, excluded
assert s.effective_demand(bids, band=None) == 8 + 250 + 8 + 19
def test_max_units_at_impact_stops_at_the_price_threshold():
bids = [(100, 10), (95, 20), (85, 1000)] # ref 100, 10% floor = 90
assert s.max_units_at_impact(bids, 100, max_impact_pct=10.0) == 30 # 100 and 95 tiers only
assert s.max_units_at_impact(bids, 100, max_impact_pct=20.0) == 1030 # all three tiers now qualify
assert s.max_units_at_impact([], 100) == 0.0
assert s.max_units_at_impact(bids, 0) == 0.0
tools/scan.py
+40 -19
View File
@@ -33,7 +33,9 @@ def main():
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("--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")
@@ -45,7 +47,7 @@ def main():
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"])
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")
@@ -83,10 +85,11 @@ def main():
return None
return sum(n * p for n, p in parts) + a.mcg * b["AreaCost"] * mcg
hab_head = {}
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
@@ -156,7 +159,7 @@ def main():
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
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"):
@@ -170,7 +173,8 @@ def main():
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))
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 = []
@@ -184,6 +188,9 @@ def main():
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)
@@ -192,21 +199,29 @@ def main():
bandcache[t] = (lo, hi)
return bandcache[t]
def demand_of(t):
bids = ((o["ItemCost"], o["ItemCount"] or 0) for o in raw_book(t)["BuyingOrders"])
return sat.effective_demand(bids, band_of(t))
for c in cands[:a.k]:
io, mm = c["io"], c["mm"]
# stage 2: queue penalty from competing asks near market price only, and demand within the tradeable band
n2 = float("inf")
# 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(x.traded7, x.traded30), se, demand_of(t), qo))
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
@@ -217,7 +232,7 @@ def main():
if t in mm:
p = mm[t]
else:
p = sat.p_patient(asks_of(t), sat.tref(x.traded7, x.traded30), N * qo, x.bid, x.vwap7, x.vwap30, x.ask, band_of(t)[1])
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():
@@ -232,7 +247,7 @@ def main():
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
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:
@@ -247,22 +262,28 @@ def main():
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"],
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"}[a.sort]
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(f"{'yld/d':>6} {'pback':>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")
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"
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['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 ''}")
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__":