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 available fee floors, with zero disabling the static floor.

A block target of -1 skips estimation but still honors the local mempool minimum. With no positive source, return zero rather than inventing a floor.

Source code in jmcore/src/jmcore/fee_policy.py 
 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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def resolve_min_fee_rate(
    backend: Any,
    *,
    static_floor: float,
    block_target: int,
    max_fee_rate: float,
) -> float:
    """Resolve available fee floors, with zero disabling the static floor.

    A block target of -1 skips estimation but still honors the local mempool
    minimum. With no positive source, return zero rather than inventing a floor.
    """
    if not _is_valid_fee_rate(static_floor, allow_zero=True):
        raise ValueError("Minimum fee static floor must be a finite nonnegative sat/vB value")
    if not _is_valid_fee_rate(max_fee_rate):
        raise ValueError("Maximum fee rate must be a finite positive sat/vB value")
    validate_min_fee_block_target(block_target)
    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_valid_fee_rate(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}")

    if block_target != -1:
        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_valid_fee_rate(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)
    logger.debug(
        "Minimum CoinJoin miner fee rate: {:.4f} sat/vB from {} (block target {})",
        minimum_fee_rate,
        source,
        block_target,
    )
    return minimum_fee_rate

validate_min_fee_block_target(value: int) -> int

Accept a confirmation target or -1 to disable the estimator floor.

Source code in jmcore/src/jmcore/fee_policy.py 
72
73
74
75
76
77
78
79
80
def validate_min_fee_block_target(value: int) -> int:
    """Accept a confirmation target or -1 to disable the estimator floor."""
    if (
        isinstance(value, bool)
        or not isinstance(value, int)
        or not (value == -1 or 1 <= value <= 1008)
    ):
        raise ValueError("Minimum fee block target must be -1 or between 1 and 1008")
    return value