Skip to content

jmcore.fee_policy

jmcore.fee_policy

Shared minimum miner-fee policy for CoinJoin participants.

Classes

MinimumFeeRateExceedsCapError

Bases: ValueError

Raised when a required minimum fee rate exceeds the configured cap.

Source code in jmcore/src/jmcore/fee_policy.py 
43
44
45
46
47
48
49
class MinimumFeeRateExceedsCapError(ValueError):
    """Raised when a required minimum fee rate exceeds the configured cap."""

    def __init__(self, *, source: str, fee_rate: float) -> None:
        self.source = source
        self.fee_rate = fee_rate
        super().__init__(f"Minimum fee rate from {source} exceeds the configured maximum fee rate")
Attributes
fee_rate = fee_rate instance-attribute
source = source instance-attribute
Methods:
__init__(*, source: str, fee_rate: float) -> None
Source code in jmcore/src/jmcore/fee_policy.py 
46
47
48
49
def __init__(self, *, source: str, fee_rate: float) -> None:
    self.source = source
    self.fee_rate = fee_rate
    super().__init__(f"Minimum fee rate from {source} exceeds the configured maximum fee rate")

Functions:

estimate_p2wpkh_vsize(num_inputs: int, num_outputs: int) -> int

Return the conservative P2WPKH virtual-size estimate used by CoinJoin.

Source code in jmcore/src/jmcore/fee_policy.py 
52
53
54
def estimate_p2wpkh_vsize(num_inputs: int, num_outputs: int) -> int:
    """Return the conservative P2WPKH virtual-size estimate used by CoinJoin."""
    return num_inputs * 68 + num_outputs * 31 + 11

fee_rate_meets_minimum(fee: int, vsize: int, minimum_fee_rate: float) -> bool

Check whether an integer fee meets the exact minimum fee rate.

Source code in jmcore/src/jmcore/fee_policy.py 
57
58
59
def fee_rate_meets_minimum(fee: int, vsize: int, minimum_fee_rate: float) -> bool:
    """Check whether an integer fee meets the exact minimum fee rate."""
    return fee >= 0 and vsize > 0 and fee >= minimum_fee_rate * vsize

format_low_fee_error(fee_rate: float, minimum_fee_rate: float) -> str

Return a human-readable !error with only the two fee-policy values.

Source code in jmcore/src/jmcore/fee_policy.py 
18
19
20
21
22
23
def format_low_fee_error(fee_rate: float, minimum_fee_rate: float) -> str:
    """Return a human-readable !error with only the two fee-policy values."""
    return (
        f"CoinJoin miner fee rate {fee_rate:.4f} sat/vB is below required "
        f"{minimum_fee_rate:.4f} sat/vB"
    )

parse_low_fee_error(message: object) -> tuple[float, float] | None

Extract bounded numeric diagnostics without logging arbitrary peer text.

Accept the current and earlier decimal precision. Full matching excludes identifiers, terminal escapes, and extra lines from ordinary INFO logs.

Source code in jmcore/src/jmcore/fee_policy.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def parse_low_fee_error(message: object) -> tuple[float, float] | None:
    """Extract bounded numeric diagnostics without logging arbitrary peer text.

    Accept the current and earlier decimal precision. Full matching excludes
    identifiers, terminal escapes, and extra lines from ordinary INFO logs.
    """
    if not isinstance(message, str):
        return None
    match = _LOW_FEE_ERROR.fullmatch(message)
    if match is None:
        return None
    fee_rate, minimum_fee_rate = map(float, match.groups())
    if minimum_fee_rate <= 0 or fee_rate > minimum_fee_rate:
        return None
    return fee_rate, minimum_fee_rate

resolve_min_fee_rate(backend: Any, *, static_floor: float, block_target: int, max_fee_rate: float) -> float async

Resolve the maximum of static, mempool, and estimate fee floors within the cap.

Source code in jmcore/src/jmcore/fee_policy.py 
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def resolve_min_fee_rate(
    backend: Any,
    *,
    static_floor: float,
    block_target: int,
    max_fee_rate: float,
) -> float:
    """Resolve the maximum of static, mempool, and estimate fee floors within the cap."""
    if not _is_finite_positive(static_floor):
        raise ValueError("Minimum fee static floor must be a finite positive sat/vB value")
    if not _is_finite_positive(max_fee_rate):
        raise ValueError("Maximum fee rate must be a finite positive sat/vB value")
    sources: list[tuple[str, float]] = [("static floor", static_floor)]

    try:
        mempool_minimum = await backend.get_mempool_min_fee()
        if mempool_minimum is not None:
            if _is_finite_positive(mempool_minimum):
                sources.append(("mempool minimum", mempool_minimum))
            else:
                logger.warning("Ignoring invalid local mempool minimum fee rate")
    except Exception as exc:
        logger.warning(f"Could not resolve local mempool minimum fee rate: {exc}")

    try:
        can_estimate = backend.can_estimate_fee()
        if isawaitable(can_estimate):
            can_estimate = await can_estimate
        if can_estimate:
            estimate = await backend.estimate_fee(block_target)
            if _is_finite_positive(estimate):
                sources.append(("backend estimate", estimate))
            else:
                logger.warning("Ignoring invalid fee estimate for minimum miner-fee policy")
    except Exception as exc:
        logger.warning(f"Could not resolve fee estimate for minimum miner-fee policy: {exc}")

    source, minimum_fee_rate = max(sources, key=lambda source_and_rate: source_and_rate[1])
    if minimum_fee_rate > max_fee_rate:
        raise MinimumFeeRateExceedsCapError(source=source, fee_rate=minimum_fee_rate)
    return minimum_fee_rate