- puga/network.py + tools/network.py: model a multi-base chain (e.g. mine LST at one base, ship it, consume it at another). Simulates each base independently, nets a transferred material's producer-surplus against consumer-need, charges real freight only on what's moved, cm_free for a founding covered by a Core Module Kit. Worked example in plans/chains/. - puga/saturation.py: real price-band filtering. FIO's order_book NarrowPriceBandLow/High and WidePriceBandLow/High match APEX's own displayed Price Band exactly (verified live) - an order outside it is a stale artifact, not just uncompetitive. Added in_band() and effective_demand(); effective_supply() gained the same hard band filter alongside its existing soft vwap-proximity filter (renamed that param mult to free up `band`). Wired into tools/scan.py stage 2 in place of the raw, unfiltered demand figure. Was flagged as an unimplemented refinement in saturation-design.md since the original design review. - tools/sell.py: self-contained tranche split (aggressive tranche capped at a volume quantile, median normally or 80th pct with --tight when a payment is imminent and stockout risk outweighs margin; patient tranche priced under the next competitor tier). Instant-bid comparison moved behind --show-bid (off by default). - Docs and roadmap updated accordingly. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
78 lines
4.0 KiB
Python
78 lines
4.0 KiB
Python
"""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 in_band(price: float, band: tuple[float | None, float | None] | None) -> bool:
|
|
"""band = (low, high), the exchange's tradeable Price Band (FIO order_book's Narrow/WidePriceBandLow/High;
|
|
verified live 2026-09-25 to match APEX's own displayed 'Price Band' field exactly). An order outside it is
|
|
not just uncompetitive, it is not currently placeable at all, and existing ones there are stale artifacts
|
|
(e.g. a leftover 334 AIC bid sitting under a 663 floor) that will never fill; not just a heuristic cutoff."""
|
|
if not band:
|
|
return True
|
|
lo, hi = band
|
|
return (lo is None or price >= lo) and (hi is None or price <= hi)
|
|
|
|
|
|
def effective_supply(asks: list[tuple[float, float]], ref_price: float | None, mult: float = 1.25,
|
|
band: tuple[float | None, float | None] | None = None) -> float:
|
|
"""Units of standing sell orders that actually compete: in the tradeable price band (hard filter, if given)
|
|
AND price <= mult x reference (vwap7 or ask; soft filter for "active competition" vs merely valid-but-stale
|
|
high asks like OCK's top tier). asks = [(price, units)]."""
|
|
asks = [(p, u) for p, u in asks if in_band(p, band)]
|
|
if not ref_price:
|
|
return sum(u for _, u in asks)
|
|
return sum(u for p, u in asks if p <= mult * ref_price)
|
|
|
|
|
|
def effective_demand(bids: list[tuple[float, float]], band: tuple[float | None, float | None] | None) -> float:
|
|
"""Units of standing buy orders that are actually tradeable: within the exchange's price band. Filters out
|
|
stale/abandoned bids placed under a different band or long since left behind by the market (e.g. a 334 AIC
|
|
bid when the floor is 663) rather than genuine, fillable demand. bids = [(price, units)]."""
|
|
return sum(u for p, u in bids if in_band(p, band))
|