aiolabs/satmachineadmin
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases 4 Packages Wiki Activity Actions
satmachineadmin/tests/test_calculations.py
padreug 6e86f53962 refactor: use Decimal instead of float for monetary calculations 
- calculations.py: Use Decimal for commission percentages, exchange rates,
  and client balances. Added to_decimal() helper for safe float conversion.
  Changed from banker's rounding to ROUND_HALF_UP.

- models.py: Changed all fiat amounts, percentages, and exchange rates to
  Decimal. Added json_encoders for API serialization.

- transaction_processor.py: Convert to Decimal at data ingestion boundary
  (CSV parsing). Updated all defaults and calculations to use Decimal.

- tests: Updated to work with Decimal return types.

This prevents floating-point precision issues in financial calculations.
All 23 tests pass.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
2026-01-11 14:47:56 +01:00

383 lines
15 KiB
Python
Raw Permalink Blame History

"""
Tests for DCA transaction calculations using empirical data.
These tests verify commission and distribution calculations against
real Lamassu transaction data to ensure the math is correct.
"""
import pytest
from decimal import Decimal
from typing import Dict, List, Tuple
# Import from the parent package (following lnurlp pattern)
from ..calculations import calculate_commission, calculate_distribution, calculate_exchange_rate, to_decimal
# =============================================================================
# COMMISSION CALCULATION TESTS
# =============================================================================
class TestCommissionCalculation:
"""Tests for commission calculation logic."""
# Empirical test cases: (crypto_atoms, commission%, discount%, expected_base, expected_commission)
# Formula: base = round(crypto_atoms / (1 + effective_commission))
# Where: effective_commission = commission_percentage * (100 - discount) / 100
EMPIRICAL_COMMISSION_CASES = [
# =============================================================
# REAL LAMASSU TRANSACTIONS (extracted from production database)
# =============================================================
# 8.75% commission, no discount - small transaction
# 15600 / 1.0875 = 14344.827... → 14345
(15600, 0.0875, 0.0, 14345, 1255),
# 8.75% commission, no discount - large transaction
# 309200 / 1.0875 = 284322.298... → 284322
(309200, 0.0875, 0.0, 284322, 24878),
# 5.5% commission, no discount
# 309500 / 1.055 = 293364.928... → 293365
(309500, 0.055, 0.0, 293365, 16135),
# 5.5% commission with 100% discount (no commission charged)
# effective = 0.055 * (100-100)/100 = 0
(292400, 0.055, 100.0, 292400, 0),
# 5.5% commission with 90% discount
# effective = 0.055 * (100-90)/100 = 0.0055
# 115000 / 1.0055 = 114370.96... → 114371
(115000, 0.055, 90.0, 114371, 629),
# 5.5% commission, no discount - 1300 GTQ transaction
# 205600 / 1.055 = 194881.516... → 194882
# Note: This tx showed 0.01 GTQ rounding discrepancy in per-client fiat
(205600, 0.055, 0.0, 194882, 10718),
# =============================================================
# SYNTHETIC TEST CASES (edge cases)
# =============================================================
# Zero commission - all goes to base
(100000, 0.0, 0.0, 100000, 0),
# Small amount edge case (1 sat minimum)
(100, 0.03, 0.0, 97, 3),
]
@pytest.mark.parametrize(
"crypto_atoms,commission_pct,discount,expected_base,expected_commission",
EMPIRICAL_COMMISSION_CASES,
ids=[
"lamassu_8.75pct_small",
"lamassu_8.75pct_large",
"lamassu_5.5pct_no_discount",
"lamassu_5.5pct_100pct_discount",
"lamassu_5.5pct_90pct_discount",
"lamassu_5.5pct_1300gtq",
"zero_commission",
"small_amount_100sats",
]
)
def test_commission_calculation(
self,
crypto_atoms: int,
commission_pct: float,
discount: float,
expected_base: int,
expected_commission: int
):
"""Test commission calculation against empirical data."""
base, commission, _ = calculate_commission(crypto_atoms, commission_pct, discount)
assert base == expected_base, f"Base amount mismatch: got {base}, expected {expected_base}"
assert commission == expected_commission, f"Commission mismatch: got {commission}, expected {expected_commission}"
# Invariant: base + commission must equal total
assert base + commission == crypto_atoms, "Base + commission must equal total crypto_atoms"
def test_commission_invariant_always_sums_to_total(self):
"""Commission + base must always equal the original amount."""
test_values = [1, 100, 1000, 10000, 100000, 266800, 1000000]
commission_rates = [0.0, 0.01, 0.03, 0.05, 0.10]
discounts = [0.0, 10.0, 25.0, 50.0]
for crypto_atoms in test_values:
for comm_rate in commission_rates:
for discount in discounts:
base, commission, _ = calculate_commission(crypto_atoms, comm_rate, discount)
assert base + commission == crypto_atoms, \
f"Invariant failed: {base} + {commission} != {crypto_atoms} " \
f"(rate={comm_rate}, discount={discount})"
# =============================================================================
# DISTRIBUTION CALCULATION TESTS
# =============================================================================
class TestDistributionCalculation:
"""Tests for proportional distribution logic."""
def test_single_client_gets_all(self):
"""Single client should receive entire distribution."""
distributions = calculate_distribution(
base_amount_sats=100000,
client_balances={"client_a": 500.00}
)
assert distributions == {"client_a": 100000}
def test_two_clients_equal_balance(self):
"""Two clients with equal balance should split evenly."""
distributions = calculate_distribution(
base_amount_sats=100000,
client_balances={
"client_a": 500.00,
"client_b": 500.00
}
)
assert distributions["client_a"] == 50000
assert distributions["client_b"] == 50000
assert sum(distributions.values()) == 100000
def test_two_clients_unequal_balance(self):
"""Two clients with 75/25 balance split."""
distributions = calculate_distribution(
base_amount_sats=100000,
client_balances={
"client_a": 750.00,
"client_b": 250.00
}
)
assert distributions["client_a"] == 75000
assert distributions["client_b"] == 25000
assert sum(distributions.values()) == 100000
def test_distribution_invariant_sums_to_total(self):
"""Total distributed sats must always equal base amount."""
# Test with various client configurations
test_cases = [
{"a": 100.0},
{"a": 100.0, "b": 100.0},
{"a": 100.0, "b": 200.0, "c": 300.0},
{"a": 33.33, "b": 33.33, "c": 33.34}, # Tricky rounding case
{"a": 1000.0, "b": 1.0}, # Large imbalance
]
for client_balances in test_cases:
for base_amount in [100, 1000, 10000, 100000, 258835]:
distributions = calculate_distribution(base_amount, client_balances)
total_distributed = sum(distributions.values())
assert total_distributed == base_amount, \
f"Distribution sum {total_distributed} != base {base_amount} " \
f"for balances {client_balances}"
def test_zero_balance_client_excluded(self):
"""Clients with zero balance should be excluded."""
distributions = calculate_distribution(
base_amount_sats=100000,
client_balances={
"client_a": 500.00,
"client_b": 0.0,
"client_c": 500.00
}
)
assert "client_b" not in distributions
assert distributions["client_a"] == 50000
assert distributions["client_c"] == 50000
def test_tiny_balance_excluded(self):
"""Clients with balance < 0.01 should be excluded."""
distributions = calculate_distribution(
base_amount_sats=100000,
client_balances={
"client_a": 500.00,
"client_b": 0.005, # Less than threshold
}
)
assert "client_b" not in distributions
assert distributions["client_a"] == 100000
def test_no_eligible_clients_returns_empty(self):
"""If no clients have balance, return empty distribution."""
distributions = calculate_distribution(
base_amount_sats=100000,
client_balances={
"client_a": 0.0,
"client_b": 0.0,
}
)
assert distributions == {}
def test_fiat_round_trip_invariant(self):
"""
Verify that distributed sats convert back to original fiat amount.
The sum of each client's fiat equivalent should equal the original
fiat amount (within rounding tolerance).
"""
# Use real Lamassu transaction data
test_cases = [
# (crypto_atoms, fiat_amount, commission_pct, discount, client_balances)
(309200, 2000.0, 0.0875, 0.0, {"a": 1000.0, "b": 1000.0}),
(309500, 2000.0, 0.055, 0.0, {"a": 500.0, "b": 1000.0, "c": 500.0}),
(292400, 2000.0, 0.055, 100.0, {"a": 800.0, "b": 1200.0}),
(115000, 800.0, 0.055, 90.0, {"a": 400.0, "b": 400.0}),
# Transaction that showed 0.01 GTQ rounding discrepancy with 4 clients
(205600, 1300.0, 0.055, 0.0, {"a": 1.0, "b": 986.0, "c": 14.0, "d": 4.0}),
]
for crypto_atoms, fiat_amount, comm_pct, discount, client_balances in test_cases:
# Calculate commission and base amount
base_sats, _, _ = calculate_commission(crypto_atoms, comm_pct, discount)
# Calculate exchange rate
exchange_rate = calculate_exchange_rate(base_sats, fiat_amount)
# Distribute sats to clients
distributions = calculate_distribution(base_sats, client_balances)
# Convert each client's sats back to fiat
total_fiat_distributed = sum(
Decimal(sats) / exchange_rate for sats in distributions.values()
)
# Should equal original fiat amount (within small rounding tolerance)
fiat_decimal = to_decimal(fiat_amount)
assert abs(total_fiat_distributed - fiat_decimal) < Decimal("0.01"), \
f"Fiat round-trip failed: {total_fiat_distributed:.2f} != {fiat_amount:.2f} " \
f"(crypto={crypto_atoms}, comm={comm_pct}, discount={discount})"
# =============================================================================
# EMPIRICAL END-TO-END TESTS
# =============================================================================
class TestEmpiricalTransactions:
"""
End-to-end tests using real Lamassu transaction data.
Add your empirical test cases here! Each case should include:
- Transaction details (crypto_atoms, fiat, commission, discount)
- Client balances at time of transaction
- Expected distribution outcome
"""
# TODO: Add your empirical data here
# Example structure:
EMPIRICAL_SCENARIOS = [
{
"name": "real_tx_266800sats_two_equal_clients",
"transaction": {
"crypto_atoms": 266800,
"fiat_amount": 2000,
"commission_percentage": 0.03,
"discount": 0.0,
},
"client_balances": {
"client_a": 1000.00, # 50% of total
"client_b": 1000.00, # 50% of total
},
# 266800 / 1.03 = 259029
"expected_base_sats": 259029,
"expected_commission_sats": 7771,
"expected_distributions": {
# 259029 / 2 = 129514.5 → both round to 129515 (ROUND_HALF_UP)
# Total = 259030, remainder = -1
# Both have same fractional (-0.5), client_a is first alphabetically
# So client_a gets -1 adjustment
"client_a": 129514,
"client_b": 129515,
},
},
# Add more scenarios from your real data!
]
@pytest.mark.parametrize(
"scenario",
EMPIRICAL_SCENARIOS,
ids=[s["name"] for s in EMPIRICAL_SCENARIOS]
)
def test_empirical_scenario(self, scenario):
"""Test full transaction flow against empirical data."""
tx = scenario["transaction"]
# Calculate commission
base, commission, _ = calculate_commission(
tx["crypto_atoms"],
tx["commission_percentage"],
tx["discount"]
)
assert base == scenario["expected_base_sats"], \
f"Base amount mismatch in {scenario['name']}"
assert commission == scenario["expected_commission_sats"], \
f"Commission mismatch in {scenario['name']}"
# Calculate distribution
distributions = calculate_distribution(
base,
scenario["client_balances"]
)
# Verify each client's allocation
for client_id, expected_sats in scenario["expected_distributions"].items():
actual_sats = distributions.get(client_id, 0)
assert actual_sats == expected_sats, \
f"Distribution mismatch for {client_id} in {scenario['name']}: " \
f"got {actual_sats}, expected {expected_sats}"
# Verify total distribution equals base
assert sum(distributions.values()) == base, \
f"Total distribution doesn't match base in {scenario['name']}"
# =============================================================================
# EDGE CASE TESTS
# =============================================================================
class TestEdgeCases:
"""Tests for edge cases and boundary conditions."""
def test_minimum_amount_1_sat(self):
"""Test with minimum possible amount (1 sat)."""
base, commission, _ = calculate_commission(1, 0.03, 0.0)
# With 3% commission on 1 sat, base rounds to 1, commission to 0
assert base + commission == 1
def test_large_transaction(self):
"""Test with large transaction (100 BTC worth of sats)."""
crypto_atoms = 10_000_000_000 # 100 BTC in sats
base, commission, _ = calculate_commission(crypto_atoms, 0.03, 0.0)
assert base + commission == crypto_atoms
assert commission > 0
def test_100_percent_discount(self):
"""100% discount should result in zero commission."""
base, commission, effective = calculate_commission(100000, 0.03, 100.0)
assert effective == 0.0
assert commission == 0
assert base == 100000
def test_many_clients_distribution(self):
"""Test distribution with many clients."""
# 10 clients with varying balances
client_balances = {f"client_{i}": float(i * 100) for i in range(1, 11)}
distributions = calculate_distribution(1000000, client_balances)
assert len(distributions) == 10
assert sum(distributions.values()) == 1000000
# Verify proportionality (client_10 should get ~18% with balance 1000)
# Total balance = 100+200+...+1000 = 5500
# client_10 proportion = 1000/5500 ≈ 18.18%
assert distributions["client_10"] > distributions["client_1"]
Reference in a new issue View git blame Copy permalink