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>
This commit is contained in:
padreug 2026-01-11 14:47:56 +01:00
parent 397fd4b002
commit 6e86f53962
4 changed files with 180 additions and 101 deletions

View file

@ -3,16 +3,34 @@ Pure calculation functions for DCA transaction processing.
These functions have no external dependencies (no lnbits, no database)
and can be easily tested in isolation.
All monetary calculations use Decimal for precision. Satoshi amounts
remain as int since they are the smallest indivisible unit.
"""
from typing import Dict, Tuple
from decimal import Decimal, ROUND_HALF_UP
from typing import Dict, Tuple, Union
# Type alias for values that can be Decimal or numeric types that will be converted
DecimalLike = Union[Decimal, float, int, str]
def to_decimal(value: DecimalLike) -> Decimal:
"""Convert a value to Decimal, handling floats carefully."""
if isinstance(value, Decimal):
return value
# Convert floats via string to avoid binary float precision issues
# e.g., Decimal(0.055) gives 0.054999999... but Decimal("0.055") is exact
if isinstance(value, float):
return Decimal(str(value))
return Decimal(value)
def calculate_commission(
crypto_atoms: int,
commission_percentage: float,
discount: float = 0.0
) -> Tuple[int, int, float]:
commission_percentage: DecimalLike,
discount: DecimalLike = Decimal("0")
) -> Tuple[int, int, Decimal]:
"""
Calculate commission split from a Lamassu transaction.
@ -34,15 +52,25 @@ def calculate_commission(
Tuple of (base_amount_sats, commission_amount_sats, effective_commission_rate)
Example:
>>> calculate_commission(266800, 0.03, 0.0)
(259029, 7771, 0.03)
>>> calculate_commission(266800, Decimal("0.03"), Decimal("0"))
(259029, 7771, Decimal('0.03'))
"""
if commission_percentage > 0:
effective_commission = commission_percentage * (100 - discount) / 100
base_crypto_atoms = round(crypto_atoms / (1 + effective_commission))
# Convert inputs to Decimal for precise calculations
comm_pct = to_decimal(commission_percentage)
disc = to_decimal(discount)
if comm_pct > 0:
# effective = commission_percentage * (100 - discount) / 100
effective_commission = comm_pct * (Decimal("100") - disc) / Decimal("100")
# base = crypto_atoms / (1 + effective_commission), rounded to nearest int
divisor = Decimal("1") + effective_commission
exact_base = Decimal(crypto_atoms) / divisor
# Use ROUND_HALF_UP (standard rounding: 0.5 rounds up)
base_crypto_atoms = int(exact_base.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
commission_amount_sats = crypto_atoms - base_crypto_atoms
else:
effective_commission = 0.0
effective_commission = Decimal("0")
base_crypto_atoms = crypto_atoms
commission_amount_sats = 0
@ -51,8 +79,8 @@ def calculate_commission(
def calculate_distribution(
base_amount_sats: int,
client_balances: Dict[str, float],
min_balance_threshold: float = 0.01
client_balances: Dict[str, DecimalLike],
min_balance_threshold: DecimalLike = Decimal("0.01")
) -> Dict[str, int]:
"""
Calculate proportional distribution of sats to clients based on their fiat balances.
@ -69,15 +97,18 @@ def calculate_distribution(
Dict of {client_id: allocated_sats}
Example:
>>> calculate_distribution(100000, {"a": 500.0, "b": 500.0})
>>> calculate_distribution(100000, {"a": Decimal("500"), "b": Decimal("500")})
{"a": 50000, "b": 50000}
"""
# Filter out clients with balance below threshold
active_balances = {
client_id: balance
for client_id, balance in client_balances.items()
if balance >= min_balance_threshold
}
# Convert threshold to Decimal
threshold = to_decimal(min_balance_threshold)
# Filter out clients with balance below threshold, converting to Decimal
active_balances: Dict[str, Decimal] = {}
for client_id, balance in client_balances.items():
bal = to_decimal(balance)
if bal >= threshold:
active_balances[client_id] = bal
if not active_balances:
return {}
@ -90,11 +121,13 @@ def calculate_distribution(
# First pass: calculate base allocations and track for remainder distribution
client_calculations = []
distributed_sats = 0
base_sats_decimal = Decimal(base_amount_sats)
for client_id, balance in active_balances.items():
proportion = balance / total_balance
exact_share = base_amount_sats * proportion
allocated_sats = round(exact_share)
exact_share = base_sats_decimal * proportion
# Round to nearest integer using ROUND_HALF_UP
allocated_sats = int(exact_share.quantize(Decimal("1"), rounding=ROUND_HALF_UP))
client_calculations.append({
'client_id': client_id,
@ -109,8 +142,9 @@ def calculate_distribution(
if remainder != 0:
# Sort by largest fractional remainder to distribute fairly
# The fractional part is exact_share - allocated_sats
client_calculations.sort(
key=lambda x: x['exact_share'] - x['allocated_sats'],
key=lambda x: x['exact_share'] - Decimal(x['allocated_sats']),
reverse=True
)
@ -131,7 +165,7 @@ def calculate_distribution(
return distributions
def calculate_exchange_rate(base_crypto_atoms: int, fiat_amount: float) -> float:
def calculate_exchange_rate(base_crypto_atoms: int, fiat_amount: DecimalLike) -> Decimal:
"""
Calculate exchange rate in sats per fiat unit.
@ -140,8 +174,9 @@ def calculate_exchange_rate(base_crypto_atoms: int, fiat_amount: float) -> float
fiat_amount: Fiat amount dispensed
Returns:
Exchange rate as sats per fiat unit
Exchange rate as sats per fiat unit (Decimal for precision)
"""
if fiat_amount <= 0:
return 0.0
return base_crypto_atoms / fiat_amount
fiat = to_decimal(fiat_amount)
if fiat <= 0:
return Decimal("0")
return Decimal(base_crypto_atoms) / fiat