arborist/tests/test_watt_cogs.py
russell@unturf.com 1aff09f021
feat(#000057): energy-COGS layer for watt_bench — marginal vs gross $/1k-tok
fox 2026-05-21: compute cost-of-goods-sold by kWh vs tokens, with the
three power states (idle / warm-idle / generation) MEASURED per
card×model×server — never hardcoded (his 40/127/380 W were illustrative
of one 3090). The only operator input is --price-per-kwh (default 0.33
USD/kWh, a configurable site rate).

energy_cogs() (pure, unit-tested) decomposes measured generation energy
against the measured warm-idle baseline:
  * gross    — all measured joules over the window (all-in, includes the
               warm-idle cost of keeping the model hot, amortized).
  * marginal — joules ABOVE warm-idle: what one more request's burst
               actually costs (clamped >=0).
kWh = J/3.6e6; $/1k-tok is the unit that compares to API pricing. Both
surface per cell + a COGS print line.

watt_bench's arborist arm now loads the frozen bench.stock_v1 policy
(--answer-mode, drift-guarded on non-reasoning) so cost is measured for
the SAME substrate the campaign grades. Cells record
window_start/end_unix so a post-hoc load_monitor queue-depth cross-ref
can flag organic-traffic contamination on the non-isolated single-slot
endpoints. 6 COGS tests; full suite 2534 passed.
2026-05-21 10:19:41 -04:00

59 lines
2.6 KiB
Python

"""Energy-COGS math for bench/watt_bench.py:energy_cogs (#000057).
Pure function, no GPU / no nvidia-smi. fox 2026-05-21: power states are
MEASURED per card/model/server; the only operator input is price_per_kwh.
These tests pin the joules→kWh→$ arithmetic and the warm-idle marginal
decomposition. See docs/calculator-test-patterns.md.
"""
from bench.watt_bench import energy_cogs
def test_gross_cogs_one_kwh():
# 3.6e6 J == exactly 1 kWh.
c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
idle_mean_w=100.0, price_per_kwh=0.33)
assert c["available"] is True
assert c["gross_usd"] == 0.33 # 1 kWh * $0.33
assert c["gross_usd_per_1k_tok"] == 0.33 # 0.33 / 1000 * 1000
assert c["mean_gen_w_measured"] == 36000.0 # 3.6e6 J / 100 s
def test_marginal_subtracts_measured_warm_idle():
# warm-idle 100 W over 100 s = 10_000 J of the 3.6e6 J is "keep hot".
c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
idle_mean_w=100.0, price_per_kwh=0.33)
assert c["warm_idle_joules"] == 10000.0
assert c["marginal_joules"] == 3.6e6 - 10000.0
# marginal < gross because warm-idle energy is removed.
assert c["marginal_usd_per_1k_tok"] < c["gross_usd_per_1k_tok"]
def test_marginal_clamps_to_zero_when_idle_exceeds_load():
# A noisy idle window can read hotter than a short quiet load window.
c = energy_cogs(gpu_joules=5000.0, window_s=100.0, tot_tok=10,
idle_mean_w=100.0, price_per_kwh=0.33)
assert c["marginal_joules"] == 0.0
assert c["marginal_usd"] == 0.0
def test_no_idle_baseline_gives_gross_only():
c = energy_cogs(gpu_joules=3.6e6, window_s=100.0, tot_tok=1000,
idle_mean_w=None, price_per_kwh=0.33)
assert c["available"] is True
assert c["warm_idle_w_measured"] is None
assert "marginal_usd_per_1k_tok" not in c # can't decompose without it
def test_unavailable_when_inputs_missing():
assert energy_cogs(None, 100.0, 1000, 100.0, 0.33)["available"] is False
assert energy_cogs(3.6e6, None, 1000, 100.0, 0.33)["available"] is False
assert energy_cogs(3.6e6, 100.0, 0, 100.0, 0.33)["available"] is False
def test_price_is_the_only_lever():
# Same measured energy, double the rate -> double the dollar COGS.
a = energy_cogs(3.6e6, 100.0, 1000, 100.0, 0.33)
b = energy_cogs(3.6e6, 100.0, 1000, 100.0, 0.66)
# Tolerate the function's 6-decimal display rounding on the ratio.
assert round(b["gross_usd"] / a["gross_usd"], 3) == 2.0
assert round(b["marginal_usd"] / a["marginal_usd"], 3) == 2.0