aiolabs/satmachineadmin
8
0
Fork 0
Code Issues Pull requests Projects Releases 4 Packages Wiki Activity Actions

feat(v2): principal-based fee split — fixes super under-payment (#38 3/5)

Browse source 
Replaces the broken fraction-of-fee math with fraction-of-principal,
direction-aware. Pre-#38: super_fee_fraction was interpreted as
`round(fee_sats * super_fraction)`, paying super ~13× below intent on
every cashout since the bitspire wire-shape landed. Post-#38: super
and operator shares are computed independently against principal
using the per-direction fractions from SuperConfig + Machine.

Per workspace CLAUDE.md "Backwards-compatibility on pre-public-launch
code" (v2-bitspire hasn't shipped to users), no compat shims:

- calculations.py: delete `split_two_stage_commission` (legacy
  fraction-of-fee). Keep `split_principal_based` as the sole split fn.
- migrations.py m009: extend to also DROP the deprecated
  `super_fee_fraction` column after backfilling its value into the
  new directional fields.
- models.py: drop `super_fee_fraction` from SuperConfig +
  UpdateSuperConfigData entirely.
- bitspire.py parse_settlement: new signature takes `super_config:
  SuperConfig` instead of `super_fee_fraction: float`. Resolves
  directional fractions from super_config + machine by tx_type, then
  computes via split_principal_based. Raises SettlementInvariantError
  on unknown tx_type.
- tasks.py: pass `super_config` through to parse_settlement; assert
  non-None (m001 inserts the singleton at install time — None is an
  impossible state).
- partial-dispense ratio path in distribution.py is unchanged — still
  uses `settlement.platform_fee_sats / settlement.fee_sats` from the
  landed row, which is the right invariant (lock at landing) and
  independent of the per-direction config.

Tests:
- Rename `test_two_stage_split.py` → `test_operator_split_legs.py`.
  Drop the legacy-function test classes. Keep TestAllocateOperatorSplitLegs
  (still-production fn) and TestPartialDispenseSplitRatio (inline ratio
  math in distribution.py).
- New `test_principal_based_fees.py`: pure-math tests for
  `split_principal_based` (six cases including a direct regression
  test pinning the pre-#38 bug at 240→3000 sats per 100k principal at
  3% super), plus parse_settlement directional dispatch tests
  (cash-in routes through cash-in fractions; cash-out through
  cash-out; unknown tx_type raises; zero-zero free-charge ATM; cross-
  direction guard).

Migration verified end-to-end via container restart: super_config
columns post-m009 = id/super_fee_wallet_id/updated_at/
super_cash_in_fee_fraction/super_cash_out_fee_fraction (no
super_fee_fraction). dca_machines + dca_settlements gained the
expected new columns. 156/156 tests green.

Refs: aiolabs/satmachineadmin#37 (parent), #38 (this layer). Closes
the load-bearing super under-payment bug standalone.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-06-01 11:24:09 +02:00
commit 1babdfbf06
8 changed files with 522 additions and 275 deletions
Download patch file Download diff file Expand all files Collapse all files

150
tests/test_operator_split_legs.py Normal file
View file

@ -0,0 +1,150 @@
"""
Tests for `allocate_operator_split_legs` (operator's commission-leg
distribution) and the partial-dispense ratio math in
`apply_partial_dispense_and_redistribute`.
Both are split-arithmetic concerns that survive the post-#38
principal-based-math refactor:
- `allocate_operator_split_legs` slices the operator's share across
their commission legs by their per-leg fractions. Function-level,
no fee-model coupling.
- Partial-dispense ratio math (in distribution.py) preserves the
ORIGINAL platform/operator ratio recorded against a settlement at
land time when an operator partial-dispenses post-hoc. The ratio
comes from the absolute platform_fee_sats / fee_sats recorded on
the settlement row, NOT the current super-config fractions — the
contract is locked at landing.
Pre-#38 tests for `split_two_stage_commission` lived here; that
function was removed when the principal-based math landed
(aiolabs/satmachineadmin#38).
"""
import pytest
from ..calculations import allocate_operator_split_legs
class TestAllocateOperatorSplitLegs:
"""Operator's remaining share split into commission legs by fraction."""
def test_plan_example_50_30_20_on_70(self):
amounts = allocate_operator_split_legs(70, [0.5, 0.3, 0.2])
assert amounts == [35, 21, 14]
def test_realistic_50_30_20_on_5575(self):
amounts = allocate_operator_split_legs(5575, [0.5, 0.3, 0.2])
# Plan-scale: 5575 * (0.5, 0.3, 0.2) = (2787.5, 1672.5, 1115)
# Last leg absorbs rounding remainders so sum == 5575 exactly.
assert sum(amounts) == 5575
assert amounts[0] == round(5575 * 0.5)
assert amounts[1] == round(5575 * 0.3)
# Last leg absorbs the remainder.
assert amounts[2] == 5575 - amounts[0] - amounts[1]
def test_single_leg_full_remainder(self):
amounts = allocate_operator_split_legs(7965, [1.0])
assert amounts == [7965]
def test_zero_operator_fee_zeros_all_legs(self):
amounts = allocate_operator_split_legs(0, [0.5, 0.3, 0.2])
assert amounts == [0, 0, 0]
def test_empty_legs_list_returns_empty(self):
amounts = allocate_operator_split_legs(100, [])
assert amounts == []
def test_last_leg_absorbs_rounding_remainder(self):
# 100 sats split [1/3, 1/3, 1/3] — last leg absorbs the +1 remainder.
amounts = allocate_operator_split_legs(100, [1 / 3, 1 / 3, 1 / 3])
assert sum(amounts) == 100
assert amounts[0] == round(100 / 3) # 33
assert amounts[1] == round(100 / 3) # 33
# Last leg absorbs the rounding (34, not 33) so total == 100.
assert amounts[2] == 100 - amounts[0] - amounts[1]
@pytest.mark.parametrize(
"operator_fee,fractions",
[
(1, [0.5, 0.5]),
(7, [0.5, 0.3, 0.2]),
(100, [0.5, 0.5]),
(5575, [0.5, 0.3, 0.2]),
(1_000_000, [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]),
],
)
def test_invariant_sum_equals_operator_fee(self, operator_fee, fractions):
amounts = allocate_operator_split_legs(operator_fee, fractions)
assert sum(amounts) == operator_fee
assert all(a >= 0 for a in amounts)
class TestPartialDispenseSplitRatio:
"""Partial-dispense recompute (closes #11 H6) must preserve the
ORIGINAL platform/operator ratio recorded on the settlement row at
land time. Super raising or lowering a global rate post-hoc must
NOT retroactively change an existing settlement's share split.
The math is inlined in `apply_partial_dispense_and_redistribute`
(distribution.py) rather than in a standalone function. These tests
mirror the inline math so a future refactor doesn't silently change
the invariant.
"""
def _recompute(self, original_fee, original_platform_fee, new_fee):
"""Mirror of the ratio math in apply_partial_dispense_and_redistribute."""
if original_fee > 0:
ratio = original_platform_fee / original_fee
else:
ratio = 0.0
new_platform = round(new_fee * ratio)
new_platform = max(0, min(new_platform, new_fee))
new_operator = new_fee - new_platform
return new_platform, new_operator
def test_30pct_lands_then_partial(self):
# Landed at platform ratio 30/100 = 0.30; new fee = 50.
# Original ratio preserved → new_platform = round(50 * 0.30) = 15.
new_platform, new_operator = self._recompute(100, 30, 50)
assert new_platform == 15
assert new_operator == 35
assert new_platform + new_operator == 50
def test_super_changed_rate_doesnt_affect_existing_settlement(self):
# Landed with platform=2390, fee=7965 (ratio ≈ 0.30). Super then
# bumps the global rate to 50%. Operator partial-dispenses to
# 50% gross → new_fee = round(7965 * 0.5) = 3982. The 30% ratio
# at land time MUST persist regardless of the new super rate.
new_platform, new_operator = self._recompute(7965, 2390, 3982)
# Expected with original ratio: round(3982 * 0.30006...) = 1195
# With (broken) current rate of 50%: would be 1991 — much higher.
assert 1190 <= new_platform <= 1200
assert new_platform + new_operator == 3982
# Original platform share was ~30%; preserved within rounding.
assert abs(new_platform / 3982 - 2390 / 7965) < 0.001
def test_zero_original_fee_yields_zero_platform(self):
new_platform, new_operator = self._recompute(0, 0, 0)
assert new_platform == 0
assert new_operator == 0
def test_invariant_sum_equals_new_fee(self):
# Random-ish parameter sweep over realistic values.
cases = [
(100, 30, 50),
(100, 0, 50), # original platform_fee was 0
(100, 100, 50), # original platform_fee was full fee
(7965, 2390, 3982),
(7965, 7965, 3982),
(1_000_000, 333_333, 250_000),
]
for orig_comm, orig_plat, new_comm in cases:
new_platform, new_operator = self._recompute(
orig_comm, orig_plat, new_comm
)
assert new_platform + new_operator == new_comm, (
f"sum invariant violated: {orig_comm=} {orig_plat=} "
f"{new_comm=} → {new_platform=} {new_operator=}"
)
assert 0 <= new_platform <= new_comm

270
tests/test_principal_based_fees.py Normal file
View file

@ -0,0 +1,270 @@
"""
Tests for the post-#38 principal-based fee split:
- `calculations.split_principal_based(principal_sats, super_frac,
operator_frac)` — pure-function math
- `bitspire.parse_settlement` — directional dispatch by tx_type
("cash_in" → super_cash_in + operator_cash_in;
"cash_out" → super_cash_out + operator_cash_out)
The bug this layer closes: pre-#38 math interpreted super_fee_fraction
as fraction-of-fee instead of fraction-of-principal, under-paying the
super by ~13× per cashout. Tests below pin the new math to the
intended fraction-of-principal model and verify the per-direction
routing through parse_settlement.
Fee mismatch recording (`fee_mismatch_sats` column, Phase 1
observability per coord-log §2026-06-01T07:00Z) lands in the next
commit; those tests live in `test_fee_mismatch_recording.py`.
"""
from datetime import datetime
import pytest
from ..bitspire import SettlementInvariantError, parse_settlement
from ..calculations import split_principal_based
from ..models import Machine, SuperConfig
# ---------------------------------------------------------------------------
# split_principal_based — pure-function math
# ---------------------------------------------------------------------------
class TestSplitPrincipalBased:
def test_super_fraction_only(self):
"""Operator at 0% — super takes exactly super_frac of principal,
operator gets 0."""
platform, operator = split_principal_based(100_000, 0.03, 0.0)
assert platform == 3_000
assert operator == 0
def test_operator_fraction_only(self):
"""Super at 0% — operator takes exactly operator_frac of
principal, platform gets 0."""
platform, operator = split_principal_based(100_000, 0.0, 0.05)
assert platform == 0
assert operator == 5_000
def test_both_fractions(self):
"""Both shares independently computed against principal — total
is super + operator, not anchored to any fee_sats input."""
platform, operator = split_principal_based(100_000, 0.03, 0.05)
assert platform == 3_000
assert operator == 5_000
def test_zero_principal_yields_zero_shares(self):
platform, operator = split_principal_based(0, 0.03, 0.05)
assert platform == 0
assert operator == 0
def test_negative_principal_yields_zero_shares(self):
"""Defensive: negative principal can't happen in production but
the function should not produce negative outputs if it ever does."""
platform, operator = split_principal_based(-100, 0.03, 0.05)
assert platform == 0
assert operator == 0
def test_rounding_does_not_compound(self):
"""The two shares round independently — there is no carryover.
On a 1_000_000-sat principal with super=0.0333, operator=0.0777,
each share rounds against principal individually."""
platform, operator = split_principal_based(1_000_000, 0.0333, 0.0777)
assert platform == round(1_000_000 * 0.0333) # 33_300
assert operator == round(1_000_000 * 0.0777) # 77_700
def test_super_frac_out_of_range_raises(self):
with pytest.raises(ValueError, match="super_frac"):
split_principal_based(100_000, 1.5, 0.0)
with pytest.raises(ValueError, match="super_frac"):
split_principal_based(100_000, -0.1, 0.0)
def test_operator_frac_out_of_range_raises(self):
with pytest.raises(ValueError, match="operator_frac"):
split_principal_based(100_000, 0.0, 1.5)
with pytest.raises(ValueError, match="operator_frac"):
split_principal_based(100_000, 0.0, -0.1)
def test_super_under_payment_bug_regression(self):
"""Direct regression test for the bug this layer closes.
Pre-#38 math (deleted): `round(fee_sats * super_fraction)` with
fee_sats=8_000 (= 8% of 100_000 principal) and super_fraction=0.03
produced platform_fee_sats=240 — ~13× below intent.
Post-#38 math: split_principal_based(100_000, 0.03, 0.05) gives
platform=3_000, which IS the intended 3% of principal."""
platform, operator = split_principal_based(100_000, 0.03, 0.05)
# Post-#38: super gets intended 3% of principal (3_000 sats)
# Pre-#38 would have produced ~240 sats from round(8000 * 0.03).
assert platform == 3_000
# ---------------------------------------------------------------------------
# parse_settlement — directional dispatch via tx_type
# ---------------------------------------------------------------------------
def _bitspire_extra(
*,
tx_type: str = "cash_out",
principal_sats: int = 100_000,
fee_sats: int = 8_000,
exchange_rate: float = 0.00001,
fiat_amount: float = 100.0,
currency: str = "EUR",
nostr_sender_pubkey: str = "a" * 64,
extra_overrides: dict | None = None,
):
"""Canonical bitspire-stamped Payment.extra dict for tests. Mirrors
the shape required by `is_bitspire_payment` + the canonical sat-
amount invariants in `_assert_sat_invariants`."""
base = {
"source": "bitspire",
"type": tx_type,
"principal_sats": principal_sats,
"fee_sats": fee_sats,
"fee_fraction": fee_sats / principal_sats if principal_sats else 0.0,
"exchange_rate": exchange_rate,
"fiat_amount": fiat_amount,
"currency": currency,
"txid": "fake-txid",
"nostr_sender_pubkey": nostr_sender_pubkey,
}
if extra_overrides:
base.update(extra_overrides)
return base
_NOW = datetime(2026, 6, 1, 12, 0, 0)
def _machine(
machine_id: str = "m1",
machine_npub: str = "a" * 64,
op_in: float = 0.0,
op_out: float = 0.0,
fiat_code: str = "EUR",
) -> Machine:
return Machine(
id=machine_id,
operator_user_id="op1",
machine_npub=machine_npub,
wallet_id="w1",
name="Test",
location=None,
fiat_code=fiat_code,
is_active=True,
operator_cash_in_fee_fraction=op_in,
operator_cash_out_fee_fraction=op_out,
created_at=_NOW,
updated_at=_NOW,
)
def _super_config(in_frac: float = 0.0, out_frac: float = 0.0) -> SuperConfig:
return SuperConfig(
id="default",
super_cash_in_fee_fraction=in_frac,
super_cash_out_fee_fraction=out_frac,
super_fee_wallet_id="super-wallet",
updated_at=_NOW,
)
class TestParseSettlementDirectional:
def test_cash_out_uses_cash_out_fractions(self):
"""tx_type='cash_out' must route to super_cash_out +
operator_cash_out fractions."""
machine = _machine(op_in=0.10, op_out=0.05)
super_cfg = _super_config(in_frac=0.10, out_frac=0.03)
extra = _bitspire_extra(tx_type="cash_out", principal_sats=100_000)
data = parse_settlement(
machine=machine,
payment_hash="ph1",
wire_sats=108_000,
extra=extra,
super_config=super_cfg,
)
# super_cash_out=0.03, operator_cash_out=0.05 against 100_000 principal
assert data.platform_fee_sats == 3_000
assert data.operator_fee_sats == 5_000
assert data.tx_type == "cash_out"
def test_cash_in_uses_cash_in_fractions(self):
"""tx_type='cash_in' must route to super_cash_in +
operator_cash_in fractions (not cash_out)."""
machine = _machine(op_in=0.04, op_out=0.10)
super_cfg = _super_config(in_frac=0.02, out_frac=0.10)
extra = _bitspire_extra(tx_type="cash_in", principal_sats=100_000)
# cash-in wire invariant: wire = principal - fee
data = parse_settlement(
machine=machine,
payment_hash="ph2",
wire_sats=92_000,
extra=extra,
super_config=super_cfg,
)
# super_cash_in=0.02, operator_cash_in=0.04 against 100_000 principal
assert data.platform_fee_sats == 2_000
assert data.operator_fee_sats == 4_000
assert data.tx_type == "cash_in"
def test_unknown_tx_type_raises(self):
machine = _machine()
super_cfg = _super_config()
extra = _bitspire_extra(
tx_type="cash_out",
extra_overrides={"type": "withdrawal"}, # not a known direction
)
with pytest.raises(SettlementInvariantError, match="unknown tx_type"):
parse_settlement(
machine=machine,
payment_hash="ph3",
wire_sats=108_000,
extra=extra,
super_config=super_cfg,
)
def test_zero_fractions_zero_split(self):
"""Free-charge ATM: both super + operator at 0 → platform and
operator fees are both 0, principal is the full take."""
machine = _machine(op_in=0.0, op_out=0.0)
super_cfg = _super_config(in_frac=0.0, out_frac=0.0)
extra = _bitspire_extra(
tx_type="cash_out", principal_sats=100_000, fee_sats=0
)
data = parse_settlement(
machine=machine,
payment_hash="ph4",
wire_sats=100_000,
extra=extra,
super_config=super_cfg,
)
assert data.platform_fee_sats == 0
assert data.operator_fee_sats == 0
assert data.principal_sats == 100_000
def test_cash_in_does_not_use_cash_out_config(self):
"""Cross-direction guard: cash-in must NOT pick up cash-out's
super or operator fractions even when they're set differently.
Pin both directions concretely to prove the dispatch."""
machine = _machine(op_in=0.01, op_out=0.10)
super_cfg = _super_config(in_frac=0.01, out_frac=0.10)
extra = _bitspire_extra(tx_type="cash_in", principal_sats=100_000)
# cash-in wire invariant: wire = principal - fee
data = parse_settlement(
machine=machine,
payment_hash="ph5",
wire_sats=92_000,
extra=extra,
super_config=super_cfg,
)
# Cash-in totals = 0.01 + 0.01 = 0.02; not 0.10 + 0.10 = 0.20
assert data.platform_fee_sats == 1_000 # 100_000 * 0.01
assert data.operator_fee_sats == 1_000 # 100_000 * 0.01

214
tests/test_two_stage_split.py
View file

@ -1,214 +0,0 @@
"""
Tests for the v2 two-stage commission split (super first, operator remainder).
The plan calls out a verification scenario explicitly:
super_fee_fraction=0.30 (i.e. 30%), operator splits [0.5, 0.3, 0.2] on a
100-sat fee → super_wallet gets 30, operator legs get 35 / 21 / 14.
Also covers the edge cases: super_fee_fraction=0.0 (no super takes the
whole fee), super_fee_fraction=1.0 (super takes everything), single-leg
operator ruleset, zero operator fee.
"""
import pytest
from ..calculations import (
allocate_operator_split_legs,
split_two_stage_commission,
)
class TestSplitTwoStageCommission:
"""Stage-1: super takes super_fee_fraction of the fee; operator gets rest."""
def test_plan_example_100sats_30pct(self):
platform, operator = split_two_stage_commission(100, 0.30)
assert platform == 30
assert operator == 70
assert platform + operator == 100
def test_realistic_7965sats_30pct(self):
# From the plan's 2000 GTQ → 266800 sats @ 3% commission example.
platform, operator = split_two_stage_commission(7965, 0.30)
assert platform == 2390 # round(7965 * 0.30) = 2389.5 → 2390
assert operator == 5575 # 7965 - 2390
assert platform + operator == 7965
def test_super_fraction_zero_leaves_all_to_operator(self):
platform, operator = split_two_stage_commission(7965, 0.0)
assert platform == 0
assert operator == 7965
def test_super_fraction_one_takes_everything(self):
platform, operator = split_two_stage_commission(7965, 1.0)
assert platform == 7965
assert operator == 0
def test_zero_commission(self):
platform, operator = split_two_stage_commission(0, 0.30)
assert platform == 0
assert operator == 0
def test_negative_commission_clamps_to_zero(self):
# Defensive: should never happen, but verify we don't go negative.
platform, operator = split_two_stage_commission(-100, 0.30)
assert platform == 0
assert operator == 0
@pytest.mark.parametrize("fee_sats", [1, 7, 100, 7965, 1_000_000])
@pytest.mark.parametrize("super_fraction", [0.0, 0.1, 0.30, 0.5, 0.777, 1.0])
def test_invariant_sum_equals_commission(self, fee_sats, super_fraction):
platform, operator = split_two_stage_commission(fee_sats, super_fraction)
assert platform + operator == fee_sats
assert 0 <= platform <= fee_sats
assert 0 <= operator <= fee_sats
class TestAllocateOperatorSplitLegs:
"""Stage-2: operator's remainder split across N leg wallets per pct rules."""
def test_plan_example_50_30_20_on_70(self):
amounts = allocate_operator_split_legs(70, [0.5, 0.3, 0.2])
assert amounts == [35, 21, 14]
assert sum(amounts) == 70
def test_realistic_50_30_20_on_5575(self):
amounts = allocate_operator_split_legs(5575, [0.5, 0.3, 0.2])
# 50%: round(2787.5) = 2788; 30%: round(1672.5) = 1672; last absorbs
# remainder: 5575 - 2788 - 1672 = 1115.
# Note: round() uses banker's rounding so 2787.5 → 2788 actually
# because 2788 is even. Confirm by total invariant.
assert sum(amounts) == 5575
assert len(amounts) == 3
def test_single_leg_full_remainder(self):
amounts = allocate_operator_split_legs(100, [1.0])
assert amounts == [100]
def test_zero_operator_fee_zeros_all_legs(self):
amounts = allocate_operator_split_legs(0, [0.5, 0.5])
assert amounts == [0, 0]
def test_empty_legs_list_returns_empty(self):
amounts = allocate_operator_split_legs(100, [])
assert amounts == []
def test_last_leg_absorbs_rounding_remainder(self):
# 100 / 3 ≈ 33.33 each; rounding makes the first two 33 and last 34.
amounts = allocate_operator_split_legs(100, [1 / 3, 1 / 3, 1 / 3])
assert sum(amounts) == 100
assert amounts[0] == round(100 / 3) # 33
assert amounts[1] == round(100 / 3) # 33
# Last leg absorbs the rounding (34, not 33) so total == 100.
assert amounts[2] == 100 - amounts[0] - amounts[1]
@pytest.mark.parametrize(
"operator_fee,fractions",
[
(1, [0.5, 0.5]),
(7, [0.5, 0.3, 0.2]),
(100, [0.5, 0.5]),
(5575, [0.5, 0.3, 0.2]),
(1_000_000, [0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1]),
],
)
def test_invariant_sum_equals_operator_fee(self, operator_fee, fractions):
amounts = allocate_operator_split_legs(operator_fee, fractions)
assert sum(amounts) == operator_fee
assert all(a >= 0 for a in amounts)
class TestEndToEndScenarios:
"""The full two-stage split — super then operator legs — composed."""
def test_plan_example_full(self):
# 100 sats fee, super_fee_fraction=0.30, operator splits [0.5, 0.3, 0.2].
platform, operator = split_two_stage_commission(100, 0.30)
legs = allocate_operator_split_legs(operator, [0.5, 0.3, 0.2])
assert platform == 30
assert legs == [35, 21, 14]
assert platform + sum(legs) == 100
def test_super_fraction_zero_full_pipeline(self):
platform, operator = split_two_stage_commission(7965, 0.0)
legs = allocate_operator_split_legs(operator, [1.0])
assert platform == 0
assert legs == [7965]
assert platform + sum(legs) == 7965
def test_super_fraction_one_full_pipeline(self):
platform, operator = split_two_stage_commission(7965, 1.0)
legs = allocate_operator_split_legs(operator, [0.5, 0.5])
assert platform == 7965
# Operator has zero to distribute; both legs get zero.
assert legs == [0, 0]
assert platform + sum(legs) == 7965
class TestPartialDispenseSplitRatio:
"""The partial-dispense recompute (H6 fix) must preserve the ORIGINAL
platform/operator ratio from the landed settlement — NOT re-derive
from the current super_fee_fraction.
These tests cover the math; the actual function lives in distribution.py
and is exercised end-to-end via integration testing. Here we verify the
invariant a future maintainer should never break.
"""
def _recompute(self, original_fee, original_platform_fee, new_fee):
"""Mirror of the ratio math in apply_partial_dispense_and_redistribute."""
if original_fee > 0:
ratio = original_platform_fee / original_fee
else:
ratio = 0.0
new_platform = round(new_fee * ratio)
new_platform = max(0, min(new_platform, new_fee))
new_operator = new_fee - new_platform
return new_platform, new_operator
def test_plan_scenario_30pct_lands_then_partial(self):
# Landed at super_fee_fraction=0.30: 100-sat fee → 30 / 70.
# Partial-dispense to 50% gross → new_fee = 50.
# Original ratio (30/100 = 0.30) preserved.
new_platform, new_operator = self._recompute(100, 30, 50)
assert new_platform == 15
assert new_operator == 35
assert new_platform + new_operator == 50
def test_super_changed_rate_doesnt_affect_existing_settlement(self):
# Landed at super_fee_fraction=0.30 (fee 7965, platform 2390).
# Super then raises rate to 50% globally. Operator partial-dispenses
# to 50% gross → new_fee = 3982 (round(7965 * 0.5)).
# Original ratio (2390/7965 ≈ 0.30) MUST still apply, not 50%.
new_platform, new_operator = self._recompute(7965, 2390, 3982)
# Expected with original ratio: round(3982 * 0.30006...) = 1195
# With (broken) current rate of 50%: would be 1991 — much higher.
assert 1190 <= new_platform <= 1200
assert new_platform + new_operator == 3982
# Original platform share was ~30%; preserved within rounding.
assert abs(new_platform / 3982 - 2390 / 7965) < 0.001
def test_zero_original_fee_yields_zero_platform(self):
new_platform, new_operator = self._recompute(0, 0, 0)
assert new_platform == 0
assert new_operator == 0
def test_invariant_sum_equals_new_fee(self):
# Random-ish parameter sweep over realistic values.
cases = [
(100, 30, 50),
(100, 0, 50), # original platform_fee was 0 (super_fraction=0)
(100, 100, 50), # original platform_fee was 100 (super_fraction=100)
(7965, 2390, 3982),
(7965, 7965, 3982),
(1_000_000, 333_333, 250_000),
]
for orig_comm, orig_plat, new_comm in cases:
new_platform, new_operator = self._recompute(
orig_comm, orig_plat, new_comm
)
assert new_platform + new_operator == new_comm, (
f"sum invariant violated: {orig_comm=} {orig_plat=} "
f"{new_comm=} → {new_platform=} {new_operator=}"
)
assert 0 <= new_platform <= new_comm