jmwallet.wallet.spend

jmwallet.wallet.spend

Reusable direct-send (non-CoinJoin) transaction building, signing, and broadcasting.

This module contains the core spending logic extracted from the CLI so that both the CLI and the jmwalletd HTTP daemon can share it without duplication.

Attributes

DUST_THRESHOLD = 546 module-attribute

Classes

DirectSendResult dataclass

Result returned by :func:direct_send.

Source code in jmwallet/src/jmwallet/wallet/spend.py

@dataclass
class DirectSendResult:
    """Result returned by :func:`direct_send`."""

    txid: str
    tx_hex: str
    fee: int
    fee_rate: float
    send_amount: int
    change_amount: int
    num_inputs: int
    num_outputs: int
    inputs: list[dict[str, object]] = field(default_factory=list)
    outputs: list[dict[str, object]] = field(default_factory=list)

Attributes

change_amount: int instance-attribute

fee: int instance-attribute

fee_rate: float instance-attribute

inputs: list[dict[str, object]] = field(default_factory=list) class-attribute instance-attribute

num_inputs: int instance-attribute

num_outputs: int instance-attribute

outputs: list[dict[str, object]] = field(default_factory=list) class-attribute instance-attribute

send_amount: int instance-attribute

tx_hex: str instance-attribute

txid: str instance-attribute

Functions

direct_send(*, wallet: WalletService, backend: BlockchainBackend, mixdepth: int, amount_sats: int, destination: str, fee_rate: float | None = None, fee_target_blocks: int = 6) -> DirectSendResult async

Build, sign, and broadcast a direct (non-CoinJoin) transaction.

Parameters:

Name	Type	Description	Default
wallet	WalletService	An initialised and synced :class:WalletService.	required
backend	BlockchainBackend	The blockchain backend for fee estimation and broadcasting.	required
mixdepth	int	The mixdepth (account) to spend from.	required
amount_sats	int	Amount in satoshis to send. 0 means sweep the entire mixdepth.	required
destination	str	Destination Bitcoin address (bech32 only).	required
fee_rate	float | None	Explicit fee rate in sat/vB. When None, the rate is estimated from the backend using fee_target_blocks.	None
fee_target_blocks	int	Number of blocks for fee estimation (ignored when fee_rate is set).	6

Returns:

Type	Description
DirectSendResult

Source code in jmwallet/src/jmwallet/wallet/spend.py

async def direct_send(
    *,
    wallet: WalletService,
    backend: BlockchainBackend,
    mixdepth: int,
    amount_sats: int,
    destination: str,
    fee_rate: float | None = None,
    fee_target_blocks: int = 6,
) -> DirectSendResult:
    """Build, sign, and broadcast a direct (non-CoinJoin) transaction.

    Parameters
    ----------
    wallet:
        An initialised and synced :class:`WalletService`.
    backend:
        The blockchain backend for fee estimation and broadcasting.
    mixdepth:
        The mixdepth (account) to spend from.
    amount_sats:
        Amount in satoshis to send.  ``0`` means sweep the entire mixdepth.
    destination:
        Destination Bitcoin address (bech32 only).
    fee_rate:
        Explicit fee rate in sat/vB.  When *None*, the rate is estimated
        from the backend using *fee_target_blocks*.
    fee_target_blocks:
        Number of blocks for fee estimation (ignored when *fee_rate* is set).

    Returns
    -------
    DirectSendResult
    """
    if not destination.startswith(("bc1", "tb1", "bcrt1")):
        msg = "Only bech32 addresses are currently supported"
        raise ValueError(msg)

    # --- Fee rate resolution ---
    if fee_rate is None:
        fee_rate = await backend.estimate_fee(target_blocks=fee_target_blocks)
        logger.debug("Estimated fee rate: {:.2f} sat/vB ({} blocks)", fee_rate, fee_target_blocks)

    # --- UTXO selection ---
    utxos: list[UTXOInfo]
    if amount_sats == 0:
        # Sweep: use all spendable UTXOs.
        raw_utxos = await wallet.get_utxos(mixdepth)
        utxos = select_spendable_utxos(raw_utxos)
    else:
        # Non-sweep: use greedy coin selection to pick the minimum UTXOs needed.
        # This avoids building oversized transactions when the wallet has many UTXOs.
        # Add a generous fee buffer (5× estimated fee) to ensure enough inputs.
        fee_buffer = max(10_000, int(amount_sats * 0.05))
        try:
            utxos = wallet.select_utxos(mixdepth, amount_sats + fee_buffer)
        except ValueError:
            # Fallback: use all spendable UTXOs if coin selection fails
            # (e.g. many dust UTXOs where the sum exceeds target but individually small).
            raw_utxos = await wallet.get_utxos(mixdepth)
            utxos = select_spendable_utxos(raw_utxos)

    if not utxos:
        msg = f"No spendable UTXOs in mixdepth {mixdepth}"
        raise ValueError(msg)

    total_input = sum(u.value for u in utxos)
    is_sweep = amount_sats == 0

    # --- Fee estimation ---
    has_change = not is_sweep
    fee, _vsize = estimate_fee(utxos, destination, fee_rate, has_change=has_change)

    if is_sweep:
        send_amount = total_input - fee
        if send_amount <= 0:
            msg = "Insufficient funds after fee deduction for sweep"
            raise ValueError(msg)
        change_amount = 0
    else:
        send_amount = amount_sats
        change_amount = total_input - send_amount - fee
        if change_amount < 0:
            msg = f"Insufficient funds: need {send_amount + fee}, have {total_input}"
            raise ValueError(msg)
        if change_amount < DUST_THRESHOLD:
            fee += change_amount
            change_amount = 0
            # Re-estimate without change output
            fee, _vsize = estimate_fee(utxos, destination, fee_rate, has_change=False)

    # --- Destination scriptPubKey ---
    dest_script = _decode_bech32_scriptpubkey(destination)

    # --- Change output ---
    change_script: bytes | None = None
    if change_amount > 0:
        change_index = wallet.get_next_address_index(mixdepth, 1)
        change_addr = wallet.get_change_address(mixdepth, change_index)
        change_key = wallet.get_key_for_address(change_addr)
        if change_key is None:
            msg = f"Cannot derive key for change address {change_addr}"
            raise ValueError(msg)
        change_script = pubkey_to_p2wpkh_script(
            change_key.get_public_key_bytes(compressed=True).hex()
        )

    # --- Build unsigned tx ---
    unsigned_tx, version, inputs_data, outputs_data, num_outputs = _build_unsigned_tx(
        utxos, dest_script, send_amount, change_script, change_amount
    )

    # --- Sign ---
    witnesses = _sign_inputs(unsigned_tx, utxos, wallet)

    # --- Assemble signed tx ---
    locktime_bytes = unsigned_tx[-4:]
    signed_tx = _assemble_signed_tx(
        version, inputs_data, num_outputs, outputs_data, locktime_bytes, witnesses, len(utxos)
    )
    tx_hex = signed_tx.hex()

    # --- Broadcast ---
    logger.info("Broadcasting direct-send transaction ({} bytes)", len(signed_tx))
    broadcast_txid = await backend.broadcast_transaction(tx_hex)
    txid = broadcast_txid or ""

    logger.info("Broadcast OK: {}", txid)
    return DirectSendResult(
        txid=txid,
        tx_hex=tx_hex,
        fee=fee,
        fee_rate=fee_rate,
        send_amount=send_amount,
        change_amount=change_amount,
        num_inputs=len(utxos),
        num_outputs=num_outputs,
        inputs=[
            {
                "outpoint": f"{u.txid}:{u.vout}",
                "scriptSig": "",
                "nSequence": 0xFFFFFFFE
                if any(ut.is_timelocked and ut.locktime is not None for ut in utxos)
                else 0xFFFFFFFF,
                "witness": "",
            }
            for u in utxos
        ],
        outputs=[
            {"value_sats": send_amount, "scriptPubKey": dest_script.hex(), "address": destination},
        ],
    )

estimate_fee(utxos: list[UTXOInfo], destination: str, fee_rate: float, *, has_change: bool) -> tuple[int, int]

Estimate the transaction fee and vsize.

Returns (fee, vsize).

Source code in jmwallet/src/jmwallet/wallet/spend.py

def estimate_fee(
    utxos: list[UTXOInfo],
    destination: str,
    fee_rate: float,
    *,
    has_change: bool,
) -> tuple[int, int]:
    """Estimate the transaction fee and vsize.

    Returns ``(fee, vsize)``.
    """
    input_types = ["p2wpkh"] * len(utxos)
    try:
        dest_type = get_address_type(destination)
    except ValueError:
        dest_type = "p2wpkh"

    output_types = [dest_type]
    if has_change:
        output_types.append("p2wpkh")

    vsize = estimate_vsize(input_types, output_types)
    return math.ceil(vsize * fee_rate), vsize

select_spendable_utxos(utxos: list[UTXOInfo], *, include_frozen: bool = False, include_fidelity_bonds: bool = False) -> list[UTXOInfo]

Filter UTXOs to only those safe for auto-spending.

Source code in jmwallet/src/jmwallet/wallet/spend.py

def select_spendable_utxos(
    utxos: list[UTXOInfo],
    *,
    include_frozen: bool = False,
    include_fidelity_bonds: bool = False,
) -> list[UTXOInfo]:
    """Filter UTXOs to only those safe for auto-spending."""
    result = []
    for u in utxos:
        if not include_frozen and u.frozen:
            continue
        if not include_fidelity_bonds and u.is_fidelity_bond:
            continue
        result.append(u)
    return result