Files
PuGa/tests/test_saturation.py
dodoxandClaude Sonnet 5 7a538cb300 Initial PuGa toolkit: data layer, econ, depth-aware scan, state sync, plan push
- puga/: cached FIO + PRUNplanner clients, market view with order-book walk,
  econ formulas ported from PRUNplanner (tested against its suite and live FIO),
  saturation model v1 (reviewed by Opus)
- tools/: scan (depth-aware), price, book, chain, state sync, plan_push
  (dry run default, [PuGa]-prefixed plans only), legacy prun_scan/prun_cxarb
- docs/: mechanics (PRUNplanner is source of truth), roadmap, decisions,
  saturation design, archived handoff
- secrets stay in .env (gitignored); ref/ holds PRUNplanner source (ignored)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-18 23:00:21 +02:00

41 lines
1.7 KiB
Python

import math
from puga import saturation as s, market, fio
def test_tref_uses_min_and_poisson_floor():
assert s.tref(60, 40) == 40
assert s.tref(15, 13) < 13 # low volume gets a lower bound
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
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
def test_thin_flag():
assert s.is_thin(1.3, 100, 5) and not s.is_thin(60, 500, 5)
def test_p_patient_clamps_and_falls_to_bid_when_oversupplied():
asks = [(100, 50), (110, 50), (120, 50)]
assert s.p_patient(asks, 10, 20, bid=90, vwap7=105, vwap30=105, ask=100) == 90 # we out-produce flow: sell at bid
p = s.p_patient(asks, 30, 1, 90, 200, 200, 100) # ample flow: capped by ask level 100
assert p == 100
def test_walk_treats_mm_null_count_as_unlimited(monkeypatch):
monkeypatch.setattr(fio, "order_book", lambda m, c, r=False: {
"SellingOrders": [], "BuyingOrders": [{"ItemCount": 4, "ItemCost": 90}, {"ItemCount": None, "ItemCost": 80}]})
r = market.walk("X", "AI1", 1000, "sell")
assert r["filled"] == 1000 and not r["short"] and r["worst"] == 80
def test_effective_supply_ignores_stale_far_asks():
asks = [(100, 10), (110, 20), (150, 1500)]
assert s.effective_supply(asks, 100) == 30 # 1500 units at 1.5x are not competition
assert s.effective_supply(asks, None) == 1530