jmwallet.history

jmwallet.history

Transaction history tracking for CoinJoin operations.

Stores a simple CSV log of all CoinJoin transactions with key metadata: - Role (maker/taker) - Fees (paid/received) - Peer count (only known by takers; None for makers) - Transaction details

Attributes

CLASSIFIED_ORIGINS = frozenset({ORIGIN_CJ_OUT, ORIGIN_CJ_CHANGE, ORIGIN_DEPOSIT, ORIGIN_NON_CJ_CHANGE}) module-attribute

HISTORY_FILENAME = 'history.csv' module-attribute

HistoryRole = Literal['maker', 'taker', 'send', 'deposit'] module-attribute

HistorySource = Literal['protocol', 'onchain'] module-attribute

LEGACY_HISTORY_FILENAME = 'coinjoin_history.csv' module-attribute

ORIGIN_CJ_CHANGE = 'cj_change' module-attribute

ORIGIN_CJ_OUT = 'cj_out' module-attribute

ORIGIN_DEPOSIT = 'deposit' module-attribute

ORIGIN_NON_CJ_CHANGE = 'non_cj_change' module-attribute

PENDING_CONFIRMATION_TRACKING_MAX = 6 module-attribute

VALID_HISTORY_SOURCES: frozenset[str] = frozenset({'protocol', 'onchain'}) module-attribute

YIELD_GENERATOR_REPORT_HEADER: list[str] = ['timestamp', 'cj amount/satoshi', 'my input count', 'my input value/satoshi', 'cjfee/satoshi', 'earned/satoshi', 'confirm time/min', 'notes'] module-attribute

Classes

HistoryWriteError

Bases: Exception

Raised when a history entry cannot be persisted to disk.

Source code in jmwallet/src/jmwallet/history.py

class HistoryWriteError(Exception):
    """Raised when a history entry cannot be persisted to disk."""

TransactionHistoryEntry

A single CoinJoin transaction record.

Source code in jmwallet/src/jmwallet/history.py

@dataclass
class TransactionHistoryEntry:
    """A single CoinJoin transaction record."""

    # Timestamps
    timestamp: str  # ISO format
    completed_at: str = ""  # ISO format

    # Role and outcome
    # "maker" / "taker" denote CoinJoin participation; "send" denotes a plain
    # (non-CoinJoin) wallet spend recorded so that the destination and change
    # addresses are persistently marked as used, even if Bitcoin Core's
    # transaction history later loses sight of them (for example, after an
    # interrupted full rescan or when the smart-scan window does not cover
    # the spend). "deposit" denotes an incoming payment; live flows never
    # write deposits (they are only reconstructed from chain data for
    # imported wallets).
    role: HistoryRole = "taker"
    success: bool = True
    failure_reason: str = ""

    # Confirmation tracking
    confirmations: int = 0  # Number of confirmations (0 = unconfirmed/pending)
    confirmed_at: str = ""  # ISO format - when first confirmation was seen

    # Core transaction data
    txid: str = ""
    cj_amount: int = 0  # satoshis

    # Peer information
    peer_count: int | None = None  # None for makers (unknown), count for takers
    counterparty_nicks: str = ""  # comma-separated

    # Fee information (in satoshis)
    fee_received: int = 0  # Only for makers - cjfee earned
    txfee_contribution: int = 0  # Mining fee contribution
    total_maker_fees_paid: int = 0  # Only for takers
    mining_fee_paid: int = 0  # Only for takers

    # Net profit/cost
    net_fee: int = 0  # Positive = profit, negative = cost

    # UTXO/address info
    source_mixdepth: int = 0
    destination_address: str = ""
    change_address: str = ""  # Change output address (must also be blacklisted!)
    utxos_used: str = ""  # comma-separated txid:vout

    # Broadcast method
    broadcast_method: str = ""  # "self", "maker:<nick>", etc.

    # Network
    network: str = "mainnet"

    # Wallet identity (issue #473): scopes the entry to a specific wallet so
    # that switching wallets from the same data directory does not surface
    # another wallet's history (and especially not its phantom pending
    # transactions). The value is the BIP32 master m/0 fingerprint hex
    # (8 chars) of the wallet that produced the entry. Defaults to an empty
    # string for backwards compatibility with pre-existing CSV files written
    # before this column existed; legacy rows are treated as belonging to no
    # known wallet and are therefore filtered out when a wallet filter is
    # active.
    wallet_fingerprint: str = ""

    # Comma-separated addresses corresponding to ``utxos_used`` (one address
    # per spent input, in the same order when known). Populated at entry
    # creation time so that ``get_used_addresses`` can blacklist input
    # addresses without re-querying the backend later. Defaults to "" for
    # backwards compatibility with rows written before this column existed;
    # such legacy rows can be backfilled out-of-band via ``gettransaction``.
    # Kept last in the field order so the existing CSV header migration
    # (which assigns trailing legacy cells positionally to appended columns)
    # continues to work for files written by older releases.
    source_addresses: str = ""

    # Provenance of this row (see ``HistorySource``): "protocol" for rows
    # written by live maker/taker/send flows (authoritative), "onchain" for
    # rows reconstructed from blockchain data after a seed import (best-effort
    # guesses: role and fees are inferred from the transaction structure).
    # Appended after ``source_addresses`` so the positional trailing-cell
    # migration for legacy headers keeps working.
    source: HistorySource = "protocol"

Attributes

broadcast_method: str = '' class-attribute instance-attribute

change_address: str = '' class-attribute instance-attribute

cj_amount: int = 0 class-attribute instance-attribute

completed_at: str = '' class-attribute instance-attribute

confirmations: int = 0 class-attribute instance-attribute

confirmed_at: str = '' class-attribute instance-attribute

counterparty_nicks: str = '' class-attribute instance-attribute

destination_address: str = '' class-attribute instance-attribute

failure_reason: str = '' class-attribute instance-attribute

fee_received: int = 0 class-attribute instance-attribute

mining_fee_paid: int = 0 class-attribute instance-attribute

net_fee: int = 0 class-attribute instance-attribute

network: str = 'mainnet' class-attribute instance-attribute

peer_count: int | None = None class-attribute instance-attribute

role: HistoryRole = 'taker' class-attribute instance-attribute

source: HistorySource = 'protocol' class-attribute instance-attribute

source_addresses: str = '' class-attribute instance-attribute

source_mixdepth: int = 0 class-attribute instance-attribute

success: bool = True class-attribute instance-attribute

timestamp: str instance-attribute

total_maker_fees_paid: int = 0 class-attribute instance-attribute

txfee_contribution: int = 0 class-attribute instance-attribute

txid: str = '' class-attribute instance-attribute

utxos_used: str = '' class-attribute instance-attribute

wallet_fingerprint: str = '' class-attribute instance-attribute

Functions:

abandon_transaction(txid: str, reason: str, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> bool

Mark a pending transaction as abandoned so the monitor stops checking it.

Sets completed_at and failure_reason so get_pending_transactions will no longer return the entry. success is left False and confirmations stays 0.

Args: txid: Transaction ID to abandon reason: Human-readable reason (stored in failure_reason) data_dir: Optional data directory wallet_fingerprint: If provided, only match entries belonging to this wallet.

Returns: True if transaction was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def abandon_transaction(
    txid: str,
    reason: str,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """Mark a pending transaction as abandoned so the monitor stops checking it.

    Sets ``completed_at`` and ``failure_reason`` so ``get_pending_transactions``
    will no longer return the entry.  ``success`` is left ``False`` and
    ``confirmations`` stays ``0``.

    Args:
        txid: Transaction ID to abandon
        reason: Human-readable reason (stored in ``failure_reason``)
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet.

    Returns:
        True if transaction was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    target = _select_entry_for_confirmation_update(entries, txid, wallet_fingerprint)
    if target is None:
        return False

    now = datetime.now().isoformat()
    target.completed_at = now
    target.failure_reason = reason
    logger.warning(f"Transaction {txid[:16]}... abandoned: {reason}")
    return _write_history_entries_atomic(entries, history_path)

append_history_entry(entry: TransactionHistoryEntry, data_dir: Path | None = None) -> None

Append a transaction history entry to the CSV file.

Args: entry: The transaction history entry to append data_dir: Optional data directory (defaults to get_default_data_dir())

Raises: HistoryWriteError: If the entry cannot be written to disk.

Source code in jmwallet/src/jmwallet/history.py

def append_history_entry(
    entry: TransactionHistoryEntry,
    data_dir: Path | None = None,
) -> None:
    """
    Append a transaction history entry to the CSV file.

    Args:
        entry: The transaction history entry to append
        data_dir: Optional data directory (defaults to get_default_data_dir())

    Raises:
        HistoryWriteError: If the entry cannot be written to disk.
    """
    history_path = _get_history_path(data_dir)
    fieldnames = _get_fieldnames()

    # Migrate legacy headers so new columns are not appended past the
    # on-disk header (which would otherwise silently lose data on read).
    _ensure_history_header_current(history_path)

    # Check if file exists to determine if we need to write header
    write_header = not history_path.exists()

    try:
        with open(history_path, "a", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            if write_header:
                writer.writeheader()

            # Convert entry to dict
            row = {f.name: getattr(entry, f.name) for f in fields(entry)}
            writer.writerow(row)

        logger.debug(f"Appended history entry: txid={entry.txid[:16]}... role={entry.role}")
    except Exception as e:
        raise HistoryWriteError(f"Failed to write history entry: {e}") from e

classify_imported_output(analysis: CoinjoinAnalysis, value: int, is_external: bool) -> str

Classify a wallet output from its creating transaction's structure.

Returns one of the address-origin tags (:data:ORIGIN_CJ_OUT, :data:ORIGIN_CJ_CHANGE, :data:ORIGIN_DEPOSIT, :data:ORIGIN_NON_CJ_CHANGE) following the same rules legacy joinmarket-clientserver uses to label coins on display:

• an equal-amount CoinJoin output -> cj_out
• our other output inside a CoinJoin (the change) -> cj_change
• a non-CoinJoin coin on the external branch -> deposit
• a non-CoinJoin coin on the internal branch -> non_cj_change

The CoinJoin cases are decided purely by output value (parity with the reference), so an equal-amount output is labeled cj_out regardless of which branch it sits on.

Source code in jmwallet/src/jmwallet/history.py

def classify_imported_output(
    analysis: CoinjoinAnalysis,
    value: int,
    is_external: bool,
) -> str:
    """Classify a wallet output from its creating transaction's structure.

    Returns one of the address-origin tags (:data:`ORIGIN_CJ_OUT`,
    :data:`ORIGIN_CJ_CHANGE`, :data:`ORIGIN_DEPOSIT`,
    :data:`ORIGIN_NON_CJ_CHANGE`) following the same rules legacy
    joinmarket-clientserver uses to label coins on display:

    - an equal-amount CoinJoin output -> ``cj_out``
    - our other output inside a CoinJoin (the change) -> ``cj_change``
    - a non-CoinJoin coin on the external branch -> ``deposit``
    - a non-CoinJoin coin on the internal branch -> ``non_cj_change``

    The CoinJoin cases are decided purely by output value (parity with the
    reference), so an equal-amount output is labeled ``cj_out`` regardless of
    which branch it sits on.
    """
    if analysis.is_coinjoin and value == analysis.cj_amount:
        return ORIGIN_CJ_OUT
    if analysis.is_coinjoin:
        return ORIGIN_CJ_CHANGE
    return ORIGIN_DEPOSIT if is_external else ORIGIN_NON_CJ_CHANGE

cleanup_stale_pending_transactions(max_age_minutes: int = 60, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> int

Mark all stale pending transactions as failed.

This is a cleanup function for entries that got stuck in pending state (e.g., from before the timeout feature was implemented, or due to bugs).

Args: max_age_minutes: Mark entries older than this as failed (default: 60) data_dir: Optional data directory wallet_fingerprint: If provided, only clean up stale pending entries for the given wallet (issue #473). Other wallets' pending entries in the same shared CSV are left untouched.

Returns: Number of entries marked as failed

Source code in jmwallet/src/jmwallet/history.py

def cleanup_stale_pending_transactions(
    max_age_minutes: int = 60,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> int:
    """
    Mark all stale pending transactions as failed.

    This is a cleanup function for entries that got stuck in pending state
    (e.g., from before the timeout feature was implemented, or due to bugs).

    Args:
        max_age_minutes: Mark entries older than this as failed (default: 60)
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only clean up stale pending entries
            for the given wallet (issue #473). Other wallets' pending entries
            in the same shared CSV are left untouched.

    Returns:
        Number of entries marked as failed
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return 0

    # Read full history (no wallet filter) so the rewrite preserves all entries
    entries = read_history(data_dir)
    count = 0
    now = datetime.now()

    for entry in entries:
        # Only process pending entries (success=False, confirmations=0, no completed_at)
        if not entry.success and entry.confirmations == 0 and not entry.completed_at:
            # Skip entries that don't belong to the active wallet when a
            # filter is set so we don't mutate other wallets' pending state.
            if wallet_fingerprint is not None and entry.wallet_fingerprint != wallet_fingerprint:
                continue
            try:
                timestamp = datetime.fromisoformat(entry.timestamp)
                age_minutes = (now - timestamp).total_seconds() / 60

                if age_minutes >= max_age_minutes:
                    entry.completed_at = now.isoformat()
                    entry.failure_reason = (
                        f"Cleaned up: pending for {int(age_minutes)} minutes without confirmation"
                    )
                    txid_str = f" (txid: {entry.txid[:16]}...)" if entry.txid else ""
                    logger.info(
                        f"Marked stale pending entry{txid_str} as failed "
                        f"(age: {int(age_minutes)} minutes)"
                    )
                    count += 1
            except (ValueError, TypeError) as e:
                logger.debug(f"Error parsing timestamp for entry: {e}")
                continue

    if count == 0:
        return 0

    if _write_history_entries_atomic(entries, history_path):
        return count
    return 0

count_other_wallet_entries(data_dir: Path | None = None, wallet_fingerprint: str | None = None, role_filter: HistoryRole | None = None) -> int

Count history rows that a per-wallet view hides for wallet_fingerprint.

Returns the number of rows whose wallet_fingerprint differs from the active wallet (including legacy rows with an empty fingerprint). Used by jm-wallet history to tell the user how many entries are excluded so the per-wallet scoping is never silent (they can pass --all-wallets).

When wallet_fingerprint is None (no scoping) this returns 0.

Source code in jmwallet/src/jmwallet/history.py

def count_other_wallet_entries(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
    role_filter: HistoryRole | None = None,
) -> int:
    """Count history rows that a per-wallet view hides for ``wallet_fingerprint``.

    Returns the number of rows whose ``wallet_fingerprint`` differs from the
    active wallet (including legacy rows with an empty fingerprint). Used by
    ``jm-wallet history`` to tell the user how many entries are excluded so
    the per-wallet scoping is never silent (they can pass ``--all-wallets``).

    When ``wallet_fingerprint`` is ``None`` (no scoping) this returns ``0``.
    """
    if wallet_fingerprint is None:
        return 0
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return 0

    hidden = 0
    try:
        with open(history_path, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                entry = _row_to_entry(row)
                if entry is None:
                    continue
                if role_filter and entry.role != role_filter:
                    continue
                if entry.wallet_fingerprint != wallet_fingerprint:
                    hidden += 1
    except Exception as e:  # pragma: no cover - defensive
        logger.warning(f"Failed to count other-wallet history entries: {e}")
        return 0
    return hidden

create_maker_history_entry(taker_nick: str, cj_amount: int, fee_received: int, txfee_contribution: int, cj_address: str, change_address: str, our_utxos: list[tuple[str, int]], txid: str | None = None, network: str = 'mainnet', wallet_fingerprint: str = '', source_addresses: list[str] | None = None) -> TransactionHistoryEntry

Create a history entry for a maker CoinJoin (initially marked as pending).

The transaction is created with success=False and confirmations=0 to indicate it's pending confirmation. A background task should later update this entry once the transaction is confirmed on-chain.

Args: taker_nick: The taker's nick cj_amount: CoinJoin amount in sats fee_received: CoinJoin fee received txfee_contribution: Mining fee contribution cj_address: Our CoinJoin output address change_address: Our change output address our_utxos: List of (txid, vout) tuples for our inputs txid: Transaction ID (may not be known by maker) network: Network name

Returns: TransactionHistoryEntry ready to be appended (marked as pending)

Source code in jmwallet/src/jmwallet/history.py

def create_maker_history_entry(
    taker_nick: str,
    cj_amount: int,
    fee_received: int,
    txfee_contribution: int,
    cj_address: str,
    change_address: str,
    our_utxos: list[tuple[str, int]],
    txid: str | None = None,
    network: str = "mainnet",
    wallet_fingerprint: str = "",
    source_addresses: list[str] | None = None,
) -> TransactionHistoryEntry:
    """
    Create a history entry for a maker CoinJoin (initially marked as pending).

    The transaction is created with success=False and confirmations=0 to indicate
    it's pending confirmation. A background task should later update this entry
    once the transaction is confirmed on-chain.

    Args:
        taker_nick: The taker's nick
        cj_amount: CoinJoin amount in sats
        fee_received: CoinJoin fee received
        txfee_contribution: Mining fee contribution
        cj_address: Our CoinJoin output address
        change_address: Our change output address
        our_utxos: List of (txid, vout) tuples for our inputs
        txid: Transaction ID (may not be known by maker)
        network: Network name

    Returns:
        TransactionHistoryEntry ready to be appended (marked as pending)
    """
    now = datetime.now().isoformat()
    net_fee = fee_received - txfee_contribution

    return TransactionHistoryEntry(
        timestamp=now,
        completed_at="",  # Not completed until confirmed
        role="maker",
        success=False,  # Pending confirmation
        failure_reason="Pending confirmation",
        confirmations=0,
        confirmed_at="",
        txid=txid or "",
        cj_amount=cj_amount,
        peer_count=None,  # Makers don't know total peer count
        counterparty_nicks=taker_nick,
        fee_received=fee_received,
        txfee_contribution=txfee_contribution,
        net_fee=net_fee,
        source_mixdepth=0,  # Would need to determine from UTXOs
        destination_address=cj_address,
        change_address=change_address,
        utxos_used=",".join(f"{txid}:{vout}" for txid, vout in our_utxos),
        source_addresses=",".join(source_addresses) if source_addresses else "",
        network=network,
        wallet_fingerprint=wallet_fingerprint,
    )

create_send_history_entry(destination: str, change_address: str, amount: int, mining_fee: int, source_mixdepth: int, selected_utxos: list[tuple[str, int]], txid: str = '', success: bool = True, failure_reason: str = '', network: str = 'mainnet', wallet_fingerprint: str = '', source_addresses: list[str] | None = None) -> TransactionHistoryEntry

Create a history entry for a plain (non-CoinJoin) wallet send.

The point of recording these entries is privacy/correctness of the next-unused-address pointer: once the wallet has signed a transaction that exposes destination and/or change_address, both must be treated as used regardless of broadcast outcome (the signed bytes are already out of the wallet's control) and regardless of whether Bitcoin Core's transaction history still surfaces the transaction (an interrupted background rescan or a smart-scan window that drops the spend would otherwise leave the addresses looking fresh).

get_used_addresses() consumes every row regardless of role, so persisting a row with role="send" is enough to keep WalletService.get_next_address_index() from handing the same address out twice.

Args: destination: Destination address (recorded so we never propose it as a fresh deposit address if it happens to be one of ours). change_address: Change output address (always ours; empty if the send had no change, e.g., a sweep). amount: Amount sent to destination in sats. mining_fee: Mining fee paid in sats. source_mixdepth: Source mixdepth the spend was funded from. selected_utxos: List of (txid, vout) tuples for our inputs. txid: Transaction ID (empty string if not yet known / not broadcast). success: Whether the transaction was successfully broadcast. failure_reason: Reason for failure if any. network: Network name. wallet_fingerprint: Wallet fingerprint for issue #473 scoping.

Returns: A TransactionHistoryEntry ready to be appended via :func:append_history_entry.

Source code in jmwallet/src/jmwallet/history.py

def create_send_history_entry(
    destination: str,
    change_address: str,
    amount: int,
    mining_fee: int,
    source_mixdepth: int,
    selected_utxos: list[tuple[str, int]],
    txid: str = "",
    success: bool = True,
    failure_reason: str = "",
    network: str = "mainnet",
    wallet_fingerprint: str = "",
    source_addresses: list[str] | None = None,
) -> TransactionHistoryEntry:
    """Create a history entry for a plain (non-CoinJoin) wallet send.

    The point of recording these entries is privacy/correctness of the
    next-unused-address pointer: once the wallet has signed a transaction
    that exposes ``destination`` and/or ``change_address``, both must be
    treated as used regardless of broadcast outcome (the signed bytes are
    already out of the wallet's control) and regardless of whether Bitcoin
    Core's transaction history still surfaces the transaction (an
    interrupted background rescan or a smart-scan window that drops the
    spend would otherwise leave the addresses looking fresh).

    ``get_used_addresses()`` consumes every row regardless of ``role``, so
    persisting a row with ``role="send"`` is enough to keep
    ``WalletService.get_next_address_index()`` from handing the same
    address out twice.

    Args:
        destination: Destination address (recorded so we never propose it
            as a fresh deposit address if it happens to be one of ours).
        change_address: Change output address (always ours; empty if the
            send had no change, e.g., a sweep).
        amount: Amount sent to ``destination`` in sats.
        mining_fee: Mining fee paid in sats.
        source_mixdepth: Source mixdepth the spend was funded from.
        selected_utxos: List of ``(txid, vout)`` tuples for our inputs.
        txid: Transaction ID (empty string if not yet known / not broadcast).
        success: Whether the transaction was successfully broadcast.
        failure_reason: Reason for failure if any.
        network: Network name.
        wallet_fingerprint: Wallet fingerprint for issue #473 scoping.

    Returns:
        A ``TransactionHistoryEntry`` ready to be appended via
        :func:`append_history_entry`.
    """
    now = datetime.now().isoformat()
    return TransactionHistoryEntry(
        timestamp=now,
        completed_at=now,
        role="send",
        success=success,
        failure_reason=failure_reason,
        confirmations=0,
        confirmed_at="",
        txid=txid,
        cj_amount=amount,
        peer_count=None,
        counterparty_nicks="",
        mining_fee_paid=mining_fee,
        net_fee=-mining_fee,
        source_mixdepth=source_mixdepth,
        destination_address=destination,
        change_address=change_address,
        utxos_used=",".join(f"{t}:{v}" for t, v in selected_utxos),
        source_addresses=",".join(source_addresses) if source_addresses else "",
        broadcast_method="self",
        network=network,
        wallet_fingerprint=wallet_fingerprint,
    )

create_taker_history_entry(maker_nicks: list[str], cj_amount: int, total_maker_fees: int, mining_fee: int, destination: str, change_address: str, source_mixdepth: int, selected_utxos: list[tuple[str, int]], txid: str = '', broadcast_method: str = 'self', network: str = 'mainnet', success: bool = False, failure_reason: str = 'Awaiting transaction', wallet_fingerprint: str = '', source_addresses: list[str] | None = None) -> TransactionHistoryEntry

Create a history entry for a taker CoinJoin.

This should be called BEFORE sending !tx to makers, to ensure addresses are recorded before they're revealed. Initially created with failure_reason="Awaiting transaction", then updated after broadcast.

The transaction is created with success=False and confirmations=0 by default to indicate it's pending confirmation. A background task should later update this entry once the transaction is confirmed on-chain.

Args: maker_nicks: List of maker nicks cj_amount: CoinJoin amount in sats total_maker_fees: Total maker fees paid mining_fee: Mining fee paid (may be 0 initially, updated after signing) destination: Destination address (CoinJoin output) change_address: Change output address (must be recorded for privacy!) source_mixdepth: Source mixdepth selected_utxos: List of (txid, vout) tuples for our inputs txid: Transaction ID (empty string if not yet known) broadcast_method: How the tx was/will be broadcast network: Network name success: Whether the CoinJoin succeeded (default False for pending) failure_reason: Reason for failure if any (default "Awaiting transaction")

Returns: TransactionHistoryEntry ready to be appended

Source code in jmwallet/src/jmwallet/history.py

def create_taker_history_entry(
    maker_nicks: list[str],
    cj_amount: int,
    total_maker_fees: int,
    mining_fee: int,
    destination: str,
    change_address: str,
    source_mixdepth: int,
    selected_utxos: list[tuple[str, int]],
    txid: str = "",
    broadcast_method: str = "self",
    network: str = "mainnet",
    success: bool = False,  # Default to pending
    failure_reason: str = "Awaiting transaction",
    wallet_fingerprint: str = "",
    source_addresses: list[str] | None = None,
) -> TransactionHistoryEntry:
    """
    Create a history entry for a taker CoinJoin.

    This should be called BEFORE sending !tx to makers, to ensure addresses
    are recorded before they're revealed. Initially created with
    failure_reason="Awaiting transaction", then updated after broadcast.

    The transaction is created with success=False and confirmations=0 by default
    to indicate it's pending confirmation. A background task should later update
    this entry once the transaction is confirmed on-chain.

    Args:
        maker_nicks: List of maker nicks
        cj_amount: CoinJoin amount in sats
        total_maker_fees: Total maker fees paid
        mining_fee: Mining fee paid (may be 0 initially, updated after signing)
        destination: Destination address (CoinJoin output)
        change_address: Change output address (must be recorded for privacy!)
        source_mixdepth: Source mixdepth
        selected_utxos: List of (txid, vout) tuples for our inputs
        txid: Transaction ID (empty string if not yet known)
        broadcast_method: How the tx was/will be broadcast
        network: Network name
        success: Whether the CoinJoin succeeded (default False for pending)
        failure_reason: Reason for failure if any (default "Awaiting transaction")

    Returns:
        TransactionHistoryEntry ready to be appended
    """
    now = datetime.now().isoformat()
    net_fee = -(total_maker_fees + mining_fee)  # Negative = cost

    return TransactionHistoryEntry(
        timestamp=now,
        completed_at="" if not success else now,
        role="taker",
        success=success,
        failure_reason=failure_reason,
        confirmations=0,
        confirmed_at="",
        txid=txid,
        cj_amount=cj_amount,
        peer_count=len(maker_nicks),
        counterparty_nicks=",".join(maker_nicks),
        total_maker_fees_paid=total_maker_fees,
        mining_fee_paid=mining_fee,
        net_fee=net_fee,
        source_mixdepth=source_mixdepth,
        destination_address=destination,
        change_address=change_address,
        utxos_used=",".join(f"{txid}:{vout}" for txid, vout in selected_utxos),
        source_addresses=",".join(source_addresses) if source_addresses else "",
        broadcast_method=broadcast_method,
        network=network,
        wallet_fingerprint=wallet_fingerprint,
    )

detect_coinjoin_peer_count(backend: BlockchainBackend | Any, txid: str, cj_amount: int) -> int | None async

Detect the number of CoinJoin participants by counting equal-amount outputs.

When makers participate in a CoinJoin, they don't know the total number of participants. However, once the transaction confirms, we can analyze it to count outputs with the CoinJoin amount.

Args: backend: Blockchain backend to fetch transaction data txid: Transaction ID to analyze cj_amount: The CoinJoin amount in satoshis

Returns: Number of equal-amount outputs (peer count), or None if detection fails

Source code in jmwallet/src/jmwallet/history.py

async def detect_coinjoin_peer_count(
    backend: BlockchainBackend | Any,
    txid: str,
    cj_amount: int,
) -> int | None:
    """
    Detect the number of CoinJoin participants by counting equal-amount outputs.

    When makers participate in a CoinJoin, they don't know the total number of
    participants. However, once the transaction confirms, we can analyze it to
    count outputs with the CoinJoin amount.

    Args:
        backend: Blockchain backend to fetch transaction data
        txid: Transaction ID to analyze
        cj_amount: The CoinJoin amount in satoshis

    Returns:
        Number of equal-amount outputs (peer count), or None if detection fails
    """
    try:
        from jmcore.bitcoin import parse_transaction

        # Fetch the transaction
        tx = await backend.get_transaction(txid)
        if not tx:
            logger.warning(f"Could not fetch transaction {txid} for peer count detection")
            return None

        # Parse the raw transaction to get outputs
        parsed_tx = parse_transaction(tx.raw)

        # Count outputs with the CoinJoin amount
        equal_amount_count = sum(1 for output in parsed_tx.outputs if output["value"] == cj_amount)

        if equal_amount_count == 0:
            logger.warning(
                f"No outputs matching CoinJoin amount {cj_amount} sats in tx {txid[:16]}..."
            )
            return None

        logger.debug(
            f"Detected {equal_amount_count} equal-amount outputs "
            f"({cj_amount:,} sats each) in tx {txid[:16]}..."
        )
        return equal_amount_count

    except Exception as e:
        logger.warning(f"Failed to detect peer count for tx {txid[:16]}...: {e}")
        return None

format_yield_generator_report(data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> list[str]

Build the yield-generator earnings report as reference-format CSV rows.

Returns a list of comma-separated strings: a header row, an initial Connected marker row (kept for parity with the reference, which the Earn report UI discards), and one row per successful maker CoinJoin from history.csv (the row's fee_received is the cjfee earned, and net_fee is the amount earned after the mining-fee contribution). Reconstructed on-chain guesses are excluded because their maker role and gross cjfee cannot be established from public transaction data, while this compatibility report has no provenance field and promises exact earnings.

The my input value/satoshi column is reported as 0 because the per-input values are not retained in the history log; the earnings columns (cjfee/earned) are exact.

Args: data_dir: Data directory holding history.csv. wallet_fingerprint: When set, restrict to one wallet's maker rows.

Source code in jmwallet/src/jmwallet/history.py

def format_yield_generator_report(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> list[str]:
    """Build the yield-generator earnings report as reference-format CSV rows.

    Returns a list of comma-separated strings: a header row, an initial
    ``Connected`` marker row (kept for parity with the reference, which the
    Earn report UI discards), and one row per *successful* maker CoinJoin from
    ``history.csv`` (the row's ``fee_received`` is the cjfee earned, and
    ``net_fee`` is the amount earned after the mining-fee contribution).
    Reconstructed on-chain guesses are excluded because their maker role and
    gross cjfee cannot be established from public transaction data, while this
    compatibility report has no provenance field and promises exact earnings.

    The ``my input value/satoshi`` column is reported as ``0`` because the
    per-input values are not retained in the history log; the earnings columns
    (``cjfee``/``earned``) are exact.

    Args:
        data_dir: Data directory holding ``history.csv``.
        wallet_fingerprint: When set, restrict to one wallet's maker rows.
    """
    rows: list[str] = [",".join(YIELD_GENERATOR_REPORT_HEADER)]
    # A startup marker keeps the response shape identical to the reference,
    # whose clients filter out rows whose notes contain "Connected".
    marker_ts = datetime.now().strftime(_YIELD_GENERATOR_TIMESTAMP_FORMAT)
    rows.append(f"{marker_ts},,,,,,,Connected")

    entries = read_history(
        data_dir,
        role_filter="maker",
        wallet_fingerprint=wallet_fingerprint,
    )
    # read_history returns most-recent-first; the statement reads chronologically.
    for entry in sorted(entries, key=lambda e: e.timestamp):
        if not entry.success or entry.source != "protocol":
            continue
        input_count = len(_parse_utxos(entry.utxos_used))
        row = [
            _format_yield_generator_timestamp(entry.timestamp),
            str(entry.cj_amount),
            str(input_count),
            "0",  # per-input values are not retained in history.csv
            str(entry.fee_received),
            str(entry.net_fee),
            _yield_generator_confirm_minutes(entry),
            "",  # notes
        ]
        rows.append(",".join(row))

    return rows

get_address_history_types(data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> dict[str, str]

Get the history type for each address used in CoinJoin history.

This maps addresses to their role in CoinJoin transactions: - "cj_out": CoinJoin output address (destination) - from successful CJ - "change": Change address - from successful CJ - "flagged": Address was shared but ALL transactions using it failed

Plain wallet spends (role="send") are intentionally excluded: their destination and change addresses are ordinary deposits/change, not CoinJoin outputs, so classifying them here would mislabel them (issue #517).

Priority: successful transactions take precedence over failed ones. Once an address is used in a successful CoinJoin, it remains cj_out/change even if later transactions using the same address failed.

Args: data_dir: Optional data directory (defaults to get_default_data_dir()) wallet_fingerprint: If provided, only consider entries belonging to the given wallet (issue #473).

Returns: Dict mapping address -> type string

Source code in jmwallet/src/jmwallet/history.py

def get_address_history_types(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> dict[str, str]:
    """
    Get the history type for each address used in CoinJoin history.

    This maps addresses to their role in CoinJoin transactions:
    - "cj_out": CoinJoin output address (destination) - from successful CJ
    - "change": Change address - from successful CJ
    - "flagged": Address was shared but ALL transactions using it failed

    Plain wallet spends (``role="send"``) are intentionally excluded: their
    destination and change addresses are ordinary deposits/change, not CoinJoin
    outputs, so classifying them here would mislabel them (issue #517).

    Priority: successful transactions take precedence over failed ones.
    Once an address is used in a successful CoinJoin, it remains cj_out/change
    even if later transactions using the same address failed.

    Args:
        data_dir: Optional data directory (defaults to get_default_data_dir())
        wallet_fingerprint: If provided, only consider entries belonging to
            the given wallet (issue #473).

    Returns:
        Dict mapping address -> type string
    """
    entries = read_history(data_dir, wallet_fingerprint=wallet_fingerprint)
    address_types: dict[str, str] = {}

    for entry in entries:
        # Only CoinJoin participation (maker/taker) produces CoinJoin output and
        # change addresses. Plain wallet spends (role="send", e.g. internal
        # mixdepth-to-mixdepth transfers used to reset PoDLE retry counters) are
        # recorded only to keep their addresses marked as used; classifying their
        # destination as "cj_out" or change as "change" would mislabel them as
        # CoinJoin outputs and create false privacy expectations (issue #517).
        # Reconstructed deposits are likewise ordinary receives, not CoinJoin
        # outputs.
        if entry.role in ("send", "deposit"):
            continue

        if entry.destination_address:
            # CoinJoin output address
            if entry.success:
                # Successful transaction - mark as cj_out (overrides any previous flagged)
                address_types[entry.destination_address] = "cj_out"
            else:
                # Transaction failed - only mark as flagged if not already used successfully
                if entry.destination_address not in address_types:
                    address_types[entry.destination_address] = "flagged"

        if entry.change_address:
            # Change address
            if entry.success:
                # Successful transaction - mark as change (overrides any previous flagged)
                address_types[entry.change_address] = "change"
            else:
                # Transaction failed - only mark as flagged if not already used successfully
                if entry.change_address not in address_types:
                    address_types[entry.change_address] = "flagged"

    return address_types

get_history_stats(data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> dict[str, int | float]

Get aggregate statistics from transaction history.

Args: data_dir: Optional data directory. wallet_fingerprint: If provided, restrict statistics to entries belonging to the given wallet (issue #473).

Returns: Dict with statistics (see _compute_stats for full list).

Source code in jmwallet/src/jmwallet/history.py

def get_history_stats(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> dict[str, int | float]:
    """
    Get aggregate statistics from transaction history.

    Args:
        data_dir: Optional data directory.
        wallet_fingerprint: If provided, restrict statistics to entries
            belonging to the given wallet (issue #473).

    Returns:
        Dict with statistics (see _compute_stats for full list).
    """
    entries = read_history(data_dir, wallet_fingerprint=wallet_fingerprint)
    return _compute_stats(entries)

get_history_stats_for_period(hours: float, role_filter: HistoryRole | None = None, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> dict[str, int | float]

Get aggregate statistics for a specific time period.

Filters history entries to only include those within the last hours hours, then computes the same aggregate statistics as get_history_stats().

This is used by the periodic summary notification to report daily/weekly stats.

Args: hours: Number of hours to look back (e.g., 24 for daily, 168 for weekly) role_filter: Optional filter by role ("maker" or "taker") data_dir: Optional data directory wallet_fingerprint: If provided, restrict statistics to entries belonging to the given wallet (issue #473).

Returns: Dict with statistics (see _compute_stats for full list).

Source code in jmwallet/src/jmwallet/history.py

def get_history_stats_for_period(
    hours: float,
    role_filter: HistoryRole | None = None,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> dict[str, int | float]:
    """
    Get aggregate statistics for a specific time period.

    Filters history entries to only include those within the last `hours` hours,
    then computes the same aggregate statistics as get_history_stats().

    This is used by the periodic summary notification to report daily/weekly stats.

    Args:
        hours: Number of hours to look back (e.g., 24 for daily, 168 for weekly)
        role_filter: Optional filter by role ("maker" or "taker")
        data_dir: Optional data directory
        wallet_fingerprint: If provided, restrict statistics to entries
            belonging to the given wallet (issue #473).

    Returns:
        Dict with statistics (see _compute_stats for full list).
    """
    entries = read_history(data_dir, role_filter=role_filter, wallet_fingerprint=wallet_fingerprint)

    if not entries:
        return _compute_stats([])

    cutoff = datetime.now() - timedelta(hours=hours)

    filtered: list[TransactionHistoryEntry] = []
    for entry in entries:
        try:
            entry_time = datetime.fromisoformat(entry.timestamp)
            if entry_time >= cutoff:
                filtered.append(entry)
        except (ValueError, TypeError):
            continue

    return _compute_stats(filtered)

get_pending_transactions(data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> list[TransactionHistoryEntry]

Get all pending (unconfirmed) transactions from history.

Returns entries that are: - Not yet confirmed (success=False, confirmations=0) - Not yet completed (completed_at is empty) - excludes failed transactions - Either have a txid waiting for confirmation, or no txid yet (needs discovery) - Without a sibling row (same txid + wallet_fingerprint) that is already marked successful. Such ghost duplicates would otherwise cause the background monitors to poll the same already-confirmed txid forever (the duplicated successful row keeps absorbing the update via update_transaction_confirmation*, leaving this stale pending row untouched). - With confirmations < PENDING_CONFIRMATION_TRACKING_MAX. This is a safety net: under normal flow confirmations flips above 0 only via update_transaction_confirmation* which simultaneously sets success=True (so the row is no longer pending). If a row somehow ends up with confirmations >= the cap while still pending, we simply stop polling it.

Args: data_dir: Optional data directory. wallet_fingerprint: If provided, only return pending entries for the given wallet (issue #473). This is what prevents another wallet's phantom pending transactions from showing up under a freshly generated wallet.

Returns: List of pending entries (includes entries without txid)

Source code in jmwallet/src/jmwallet/history.py

def get_pending_transactions(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> list[TransactionHistoryEntry]:
    """
    Get all pending (unconfirmed) transactions from history.

    Returns entries that are:
    - Not yet confirmed (success=False, confirmations=0)
    - Not yet completed (completed_at is empty) - excludes failed transactions
    - Either have a txid waiting for confirmation, or no txid yet (needs discovery)
    - Without a sibling row (same txid + wallet_fingerprint) that is already
      marked successful. Such ghost duplicates would otherwise cause the
      background monitors to poll the same already-confirmed txid forever
      (the duplicated successful row keeps absorbing the update via
      ``update_transaction_confirmation*``, leaving this stale pending row
      untouched).
    - With ``confirmations < PENDING_CONFIRMATION_TRACKING_MAX``. This is a
      safety net: under normal flow ``confirmations`` flips above 0 only via
      ``update_transaction_confirmation*`` which simultaneously sets
      ``success=True`` (so the row is no longer pending). If a row somehow
      ends up with confirmations >= the cap while still pending, we simply
      stop polling it.

    Args:
        data_dir: Optional data directory.
        wallet_fingerprint: If provided, only return pending entries for the
            given wallet (issue #473). This is what prevents another wallet's
            phantom pending transactions from showing up under a freshly
            generated wallet.

    Returns:
        List of pending entries (includes entries without txid)
    """
    entries = read_history(data_dir, wallet_fingerprint=wallet_fingerprint)

    # Index already-successful txids per wallet so we can drop any pending
    # row that is shadowed by a successful sibling. Keying on
    # ``(wallet_fingerprint, txid)`` keeps the check correct when several
    # wallets share the same data directory.
    successful_txids: set[tuple[str, str]] = {
        (e.wallet_fingerprint, e.txid) for e in entries if e.success and e.txid
    }

    return [
        e
        for e in entries
        if not e.success
        and e.confirmations < PENDING_CONFIRMATION_TRACKING_MAX
        and not e.completed_at
        and (not e.txid or (e.wallet_fingerprint, e.txid) not in successful_txids)
    ]

get_used_addresses(data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> set[str]

Get all addresses that have been used in CoinJoin history.

Returns destination addresses (CoinJoin outputs), change addresses, and source (input) addresses from all history entries, regardless of success or confirmation status.

This is critical for privacy: once an address has been shared with peers (even if the transaction failed or wasn't confirmed), it should never be reused. Input addresses are included so that a spent deposit address is never proposed again, even if the backend later loses sight of the spending transaction (e.g., interrupted rescan, smart-scan window miss).

Args: data_dir: Optional data directory wallet_fingerprint: If provided, only consider entries belonging to the given wallet (issue #473).

Returns: Set of addresses that should not be reused

Source code in jmwallet/src/jmwallet/history.py

def get_used_addresses(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> set[str]:
    """
    Get all addresses that have been used in CoinJoin history.

    Returns destination addresses (CoinJoin outputs), change addresses, and
    source (input) addresses from all history entries, regardless of success
    or confirmation status.

    This is critical for privacy: once an address has been shared with peers
    (even if the transaction failed or wasn't confirmed), it should never be
    reused. Input addresses are included so that a spent deposit address is
    never proposed again, even if the backend later loses sight of the
    spending transaction (e.g., interrupted rescan, smart-scan window miss).

    Args:
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only consider entries belonging to
            the given wallet (issue #473).

    Returns:
        Set of addresses that should not be reused
    """
    entries = read_history(data_dir, wallet_fingerprint=wallet_fingerprint)
    used_addresses = set()

    for entry in entries:
        if entry.destination_address:
            used_addresses.add(entry.destination_address)
        if entry.change_address:
            used_addresses.add(entry.change_address)
        if entry.source_addresses:
            for addr in entry.source_addresses.split(","):
                addr = addr.strip()
                if addr:
                    used_addresses.add(addr)

    return used_addresses

get_utxo_label(address: str, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> str

Get a human-readable label for a UTXO based on its address history.

Labels are derived from CoinJoin history: - "cj-out": CoinJoin output (equal-amount output from successful CJ) - "cj-change": CoinJoin change output (change from successful CJ) - "deposit": External deposit (not from CoinJoin) - "flagged": Address was shared but transaction failed

Args: address: The address to get a label for data_dir: Optional data directory (defaults to get_default_data_dir()) wallet_fingerprint: If provided, only consider entries belonging to the given wallet (issue #473).

Returns: Human-readable label for the UTXO

Source code in jmwallet/src/jmwallet/history.py

def get_utxo_label(
    address: str,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> str:
    """
    Get a human-readable label for a UTXO based on its address history.

    Labels are derived from CoinJoin history:
    - "cj-out": CoinJoin output (equal-amount output from successful CJ)
    - "cj-change": CoinJoin change output (change from successful CJ)
    - "deposit": External deposit (not from CoinJoin)
    - "flagged": Address was shared but transaction failed

    Args:
        address: The address to get a label for
        data_dir: Optional data directory (defaults to get_default_data_dir())
        wallet_fingerprint: If provided, only consider entries belonging to
            the given wallet (issue #473).

    Returns:
        Human-readable label for the UTXO
    """
    history_types = get_address_history_types(data_dir, wallet_fingerprint=wallet_fingerprint)

    if address in history_types:
        history_type = history_types[address]
        if history_type == "cj_out":
            return "cj-out"
        elif history_type == "change":
            return "cj-change"
        elif history_type == "flagged":
            return "flagged"

    # If not in history, it's a deposit (external receive)
    return "deposit"

list_history_fingerprints(data_dir: Path | None = None) -> list[str]

List distinct wallet fingerprints recorded in the history CSV.

Used by jm-wallet history to auto-select the active wallet when only one is present in the data directory, and to print a helpful error listing the available choices when several are. Empty fingerprints (legacy rows written before issue #473 added the column) are excluded; callers can fall back to --all-wallets if they want to see those rows.

Args: data_dir: Optional data directory (defaults to get_default_data_dir()).

Returns: Sorted list of distinct non-empty wallet fingerprints found in history.csv. Returns an empty list when the file is missing or unreadable.

Source code in jmwallet/src/jmwallet/history.py

def list_history_fingerprints(data_dir: Path | None = None) -> list[str]:
    """List distinct wallet fingerprints recorded in the history CSV.

    Used by ``jm-wallet history`` to auto-select the active wallet when
    only one is present in the data directory, and to print a helpful
    error listing the available choices when several are. Empty
    fingerprints (legacy rows written before issue #473 added the
    column) are excluded; callers can fall back to ``--all-wallets``
    if they want to see those rows.

    Args:
        data_dir: Optional data directory (defaults to
            ``get_default_data_dir()``).

    Returns:
        Sorted list of distinct non-empty wallet fingerprints found in
        ``history.csv``. Returns an empty list when the file is missing
        or unreadable.
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return []

    fingerprints: set[str] = set()
    try:
        with open(history_path, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                fp = (row.get("wallet_fingerprint") or "").strip()
                if fp:
                    fingerprints.add(fp)
    except Exception as e:  # pragma: no cover - defensive
        logger.warning(f"Failed to scan history for wallet fingerprints: {e}")
        return []
    return sorted(fingerprints)

mark_pending_transaction_failed(destination_address: str, failure_reason: str, data_dir: Path | None = None, txid: str | None = None, wallet_fingerprint: str | None = None) -> bool

Mark a pending transaction as failed by matching the destination address and optionally txid.

This is used when a pending CoinJoin times out - the taker never broadcast the transaction, so we mark it as failed rather than leaving it pending indefinitely.

Args: destination_address: The CoinJoin destination address to match failure_reason: Reason for marking as failed (e.g., "Timed out after 60 minutes") data_dir: Optional data directory txid: Optional transaction ID for more precise matching (when multiple entries share the same destination address) wallet_fingerprint: If provided, only match entries belonging to this wallet (issue #473).

Returns: True if a matching entry was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def mark_pending_transaction_failed(
    destination_address: str,
    failure_reason: str,
    data_dir: Path | None = None,
    txid: str | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """
    Mark a pending transaction as failed by matching the destination address and optionally txid.

    This is used when a pending CoinJoin times out - the taker never broadcast
    the transaction, so we mark it as failed rather than leaving it pending
    indefinitely.

    Args:
        destination_address: The CoinJoin destination address to match
        failure_reason: Reason for marking as failed (e.g., "Timed out after 60 minutes")
        data_dir: Optional data directory
        txid: Optional transaction ID for more precise matching (when multiple entries
              share the same destination address)
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet (issue #473).

    Returns:
        True if a matching entry was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    updated = False

    for entry in entries:
        # Match by destination address and pending status
        # (success=False, confirmations=0, no completed_at)
        if (
            entry.destination_address == destination_address
            and not entry.success
            and entry.confirmations == 0
            and not entry.completed_at
        ):
            # If txid is provided, also match by txid
            if txid is not None and entry.txid != txid:
                continue

            if wallet_fingerprint is not None and entry.wallet_fingerprint != wallet_fingerprint:
                continue

            entry.success = False
            entry.failure_reason = failure_reason
            entry.completed_at = datetime.now().isoformat()
            # Keep confirmations at 0 to distinguish from confirmed then reorged
            txid_str = f" (txid: {entry.txid[:16]}...)" if entry.txid else ""
            logger.info(
                f"Marked pending transaction for {destination_address[:20]}...{txid_str} "
                f"as failed: {failure_reason}"
            )
            updated = True
            break

    if not updated:
        return False

    return _write_history_entries_atomic(entries, history_path)

purge_reconstructed_entries(data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> int

Remove on-chain-reconstructed rows (source="onchain") from history.

Used by jm-wallet reconstruct-history to rebuild the guessed portion of a wallet's history from scratch without touching authoritative protocol-time rows. When wallet_fingerprint is given, only that wallet's reconstructed rows are removed.

Returns: The number of rows removed.

Raises: HistoryWriteError: If the pruned file cannot be written back.

Source code in jmwallet/src/jmwallet/history.py

def purge_reconstructed_entries(
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> int:
    """Remove on-chain-reconstructed rows (``source="onchain"``) from history.

    Used by ``jm-wallet reconstruct-history`` to rebuild the guessed portion
    of a wallet's history from scratch without touching authoritative
    protocol-time rows. When ``wallet_fingerprint`` is given, only that
    wallet's reconstructed rows are removed.

    Returns:
        The number of rows removed.

    Raises:
        HistoryWriteError: If the pruned file cannot be written back.
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return 0

    entries = read_history(data_dir)
    kept = [
        e
        for e in entries
        if not (
            e.source == "onchain"
            and (wallet_fingerprint is None or e.wallet_fingerprint == wallet_fingerprint)
        )
    ]
    removed = len(entries) - len(kept)
    if removed == 0:
        return 0
    if not _write_history_entries_atomic(kept, history_path):
        raise HistoryWriteError(f"Failed to rewrite {history_path} after purging entries")
    logger.info(f"Purged {removed} reconstructed history entries")
    return removed

read_history(data_dir: Path | None = None, limit: int | None = None, role_filter: HistoryRole | None = None, wallet_fingerprint: str | None = None) -> list[TransactionHistoryEntry]

Read transaction history from the CSV file.

Args: data_dir: Optional data directory (defaults to get_default_data_dir()) limit: Maximum number of entries to return (most recent first) role_filter: Filter by role (maker/taker) wallet_fingerprint: If provided, only return entries belonging to this wallet (matched by their wallet_fingerprint column). Entries without a recorded fingerprint (legacy rows from before issue #473) are excluded from the filtered view to prevent another wallet's entries from leaking into the active wallet's history.

Returns: List of TransactionHistoryEntry objects

Source code in jmwallet/src/jmwallet/history.py

def read_history(
    data_dir: Path | None = None,
    limit: int | None = None,
    role_filter: HistoryRole | None = None,
    wallet_fingerprint: str | None = None,
) -> list[TransactionHistoryEntry]:
    """
    Read transaction history from the CSV file.

    Args:
        data_dir: Optional data directory (defaults to get_default_data_dir())
        limit: Maximum number of entries to return (most recent first)
        role_filter: Filter by role (maker/taker)
        wallet_fingerprint: If provided, only return entries belonging to this
            wallet (matched by their ``wallet_fingerprint`` column). Entries
            without a recorded fingerprint (legacy rows from before issue #473)
            are excluded from the filtered view to prevent another wallet's
            entries from leaking into the active wallet's history.

    Returns:
        List of TransactionHistoryEntry objects
    """
    history_path = _get_history_path(data_dir)

    if not history_path.exists():
        return []

    # Recover fingerprint cells appended past a stale legacy header so
    # per-wallet filters and updates work for pre-existing files even
    # when no new write has occurred yet.
    try:
        _ensure_history_header_current(history_path)
    except HistoryWriteError as e:
        # Migration is best-effort on read; keep going so the user can
        # still inspect history even if the file is read-only.
        logger.warning(f"History header migration failed during read: {e}")

    entries: list[TransactionHistoryEntry] = []

    try:
        with open(history_path, newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for row in reader:
                entry = _row_to_entry(row)
                if entry is None:
                    continue

                # Apply role filter
                if role_filter and entry.role != role_filter:
                    continue

                # Apply wallet filter (issue #473): legacy rows without a
                # recorded fingerprint do not match any specific wallet
                # and are therefore hidden from the per-wallet view.
                if wallet_fingerprint is not None:
                    if entry.wallet_fingerprint != wallet_fingerprint:
                        continue

                entries.append(entry)

    except Exception as e:
        logger.error(f"Failed to read history: {e}")
        return []

    # Sort by timestamp (most recent first) and apply limit
    entries.sort(key=lambda e: e.timestamp, reverse=True)
    if limit:
        entries = entries[:limit]

    return entries

update_all_pending_transactions(backend: BlockchainBackend | Any, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> int async

Update the status of all pending transactions using the blockchain backend.

This function is called when displaying wallet info or history to ensure pending transactions are updated with their current confirmation status. Particularly important for one-shot coinjoin commands that exit before the background monitor can update the status.

Args: backend: Blockchain backend to query transaction status data_dir: Optional data directory wallet_fingerprint: If provided, only update pending entries for the given wallet (issue #473).

Returns: Number of transactions that were updated

Source code in jmwallet/src/jmwallet/history.py

async def update_all_pending_transactions(
    backend: BlockchainBackend | Any,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> int:
    """
    Update the status of all pending transactions using the blockchain backend.

    This function is called when displaying wallet info or history to ensure
    pending transactions are updated with their current confirmation status.
    Particularly important for one-shot coinjoin commands that exit before
    the background monitor can update the status.

    Args:
        backend: Blockchain backend to query transaction status
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only update pending entries for the
            given wallet (issue #473).

    Returns:
        Number of transactions that were updated
    """
    pending = get_pending_transactions(data_dir, wallet_fingerprint=wallet_fingerprint)
    if not pending:
        return 0

    updated_count = 0
    # Full node / mempool-API backends report confirmation depth via
    # get_transaction(). Light clients (Neutrino) are mempool-only there
    # (get_transaction reports confirmations=0 and returns None once a tx
    # confirms), so they must confirm via verify_tx_output() against the output
    # address. has_mempool_access() is not a reliable proxy: neutrino with the
    # watched mempool tracker has mempool access yet still cannot report
    # confirmations by txid.
    can_lookup_by_txid = backend.can_get_confirmations_by_txid()

    for entry in pending:
        if not entry.txid:
            # Can't check without txid
            continue

        try:
            if can_lookup_by_txid:
                # Full node: can check confirmation depth directly by txid
                tx_info = await backend.get_transaction(entry.txid)
                if tx_info is not None:
                    # Only mark as success after first block confirmation.
                    if tx_info.confirmations > 0:
                        update_transaction_confirmation(
                            txid=entry.txid,
                            confirmations=tx_info.confirmations,
                            data_dir=data_dir,
                        )
                        updated_count += 1
                        logger.debug(
                            f"Updated pending tx {entry.txid[:16]}... "
                            f"({tx_info.confirmations} confs)"
                        )
            else:
                # Neutrino: get_transaction cannot report confirmations, so
                # confirm via a compact-filter match on the output address.
                if not entry.destination_address:
                    continue

                try:
                    current_height = await backend.get_block_height()
                except Exception:
                    current_height = None

                verified = await backend.verify_tx_output(
                    txid=entry.txid,
                    vout=0,  # CJ outputs are typically first
                    address=entry.destination_address,
                    start_height=current_height,
                )

                if verified:
                    update_transaction_confirmation(
                        txid=entry.txid,
                        confirmations=1,
                        data_dir=data_dir,
                    )
                    updated_count += 1
                    logger.debug(f"Updated pending tx {entry.txid[:16]}... via Neutrino")

        except Exception as e:
            logger.debug(f"Could not update pending tx {entry.txid[:16]}...: {e}")

    if updated_count > 0:
        logger.info(f"Updated {updated_count} pending transaction(s)")

    return updated_count

update_awaiting_transaction_signed(destination_address: str, txid: str, fee_received: int, txfee_contribution: int, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> bool

Update a pending "Awaiting transaction" entry when the maker signs the tx.

This is called after the maker successfully signs a transaction. The entry was created earlier (during !ioauth) with failure_reason="Awaiting transaction" to ensure the addresses were recorded before revealing them.

Args: destination_address: The CoinJoin destination address to match txid: The transaction ID fee_received: CoinJoin fee earned txfee_contribution: Mining fee contribution data_dir: Optional data directory wallet_fingerprint: If provided, only match entries belonging to this wallet (issue #473).

Returns: True if a matching entry was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def update_awaiting_transaction_signed(
    destination_address: str,
    txid: str,
    fee_received: int,
    txfee_contribution: int,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """
    Update a pending "Awaiting transaction" entry when the maker signs the tx.

    This is called after the maker successfully signs a transaction. The entry
    was created earlier (during !ioauth) with failure_reason="Awaiting transaction"
    to ensure the addresses were recorded before revealing them.

    Args:
        destination_address: The CoinJoin destination address to match
        txid: The transaction ID
        fee_received: CoinJoin fee earned
        txfee_contribution: Mining fee contribution
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet (issue #473).

    Returns:
        True if a matching entry was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    updated = False

    for entry in entries:
        # Match by destination address and "Awaiting transaction" status
        if (
            entry.destination_address == destination_address
            and entry.failure_reason == "Awaiting transaction"
            and not entry.txid  # Should not have txid yet
        ):
            if wallet_fingerprint is not None and entry.wallet_fingerprint != wallet_fingerprint:
                continue
            entry.txid = txid
            entry.fee_received = fee_received
            entry.txfee_contribution = txfee_contribution
            entry.net_fee = fee_received - txfee_contribution
            entry.failure_reason = "Pending confirmation"  # Now awaiting confirmation
            logger.info(
                f"Updated awaiting transaction for {destination_address[:20]}... "
                f"with txid {txid[:16]}..., fee={fee_received} sats"
            )
            updated = True
            break

    if not updated:
        return False

    return _write_history_entries_atomic(entries, history_path)

update_pending_transaction_txid(destination_address: str, txid: str, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> bool

Update a pending transaction's txid by matching the destination address.

This is used when a maker doesn't initially know the txid (didn't receive !push), but can discover it later by finding which transaction paid to the CoinJoin address.

Args: destination_address: The CoinJoin destination address to match txid: The discovered transaction ID data_dir: Optional data directory wallet_fingerprint: If provided, only match entries belonging to this wallet (issue #473).

Returns: True if a matching entry was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def update_pending_transaction_txid(
    destination_address: str,
    txid: str,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """
    Update a pending transaction's txid by matching the destination address.

    This is used when a maker doesn't initially know the txid (didn't receive !push),
    but can discover it later by finding which transaction paid to the CoinJoin address.

    Args:
        destination_address: The CoinJoin destination address to match
        txid: The discovered transaction ID
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet (issue #473).

    Returns:
        True if a matching entry was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    updated = False

    for entry in entries:
        # Match by destination address and empty txid (pending without txid)
        if entry.destination_address == destination_address and not entry.txid:
            if wallet_fingerprint is not None and entry.wallet_fingerprint != wallet_fingerprint:
                continue
            entry.txid = txid
            logger.info(
                f"Updated pending transaction for {destination_address[:20]}... "
                f"with txid {txid[:16]}..."
            )
            updated = True
            break

    if not updated:
        return False

    return _write_history_entries_atomic(entries, history_path)

update_send_awaiting_broadcast(pending_entry: TransactionHistoryEntry, *, txid: str, success: bool, failure_reason: str, data_dir: Path | None = None) -> bool

Finalize a "send" history row created with failure_reason="awaiting broadcast".

The send CLI calls :func:create_send_history_entry and immediately appends the row so the destination/change addresses are persisted to disk before the broadcast attempt. After broadcast resolves (success or failure) the same row is updated in place with the final outcome.

Args: pending_entry: The in-memory entry that was just appended. Its timestamp and destination_address identify the row. txid: Final transaction ID (empty string if broadcast failed). success: True if the transaction was broadcast successfully. failure_reason: Final failure reason (empty string on success). data_dir: Optional data directory.

Returns: True if a matching row was found and rewritten, False otherwise.

Source code in jmwallet/src/jmwallet/history.py

def update_send_awaiting_broadcast(
    pending_entry: TransactionHistoryEntry,
    *,
    txid: str,
    success: bool,
    failure_reason: str,
    data_dir: Path | None = None,
) -> bool:
    """Finalize a "send" history row created with ``failure_reason="awaiting broadcast"``.

    The send CLI calls :func:`create_send_history_entry` and immediately
    appends the row so the destination/change addresses are persisted to
    disk *before* the broadcast attempt. After broadcast resolves (success
    or failure) the same row is updated in place with the final outcome.

    Args:
        pending_entry: The in-memory entry that was just appended. Its
            ``timestamp`` and ``destination_address`` identify the row.
        txid: Final transaction ID (empty string if broadcast failed).
        success: True if the transaction was broadcast successfully.
        failure_reason: Final failure reason (empty string on success).
        data_dir: Optional data directory.

    Returns:
        True if a matching row was found and rewritten, False otherwise.
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    updated = False
    for entry in entries:
        if (
            entry.role == "send"
            and entry.timestamp == pending_entry.timestamp
            and entry.destination_address == pending_entry.destination_address
            and entry.failure_reason == "awaiting broadcast"
            and not entry.txid
        ):
            entry.txid = txid
            entry.success = success
            entry.failure_reason = failure_reason
            entry.completed_at = entry.timestamp
            updated = True
            break

    if not updated:
        return False

    return _write_history_entries_atomic(entries, history_path)

update_taker_awaiting_transaction_broadcast(destination_address: str, change_address: str, txid: str, mining_fee: int, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> bool

Update a pending "Awaiting transaction" entry when the taker broadcasts the tx.

This is called after the taker successfully broadcasts a transaction. The entry was created earlier (before sending !tx) with failure_reason="Awaiting transaction" to ensure the addresses were recorded before revealing them.

Args: destination_address: The CoinJoin destination address to match change_address: The change address to match (for extra precision) txid: The transaction ID mining_fee: Actual mining fee paid (may differ from estimate) data_dir: Optional data directory wallet_fingerprint: If provided, only match entries belonging to this wallet (issue #473).

Returns: True if a matching entry was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def update_taker_awaiting_transaction_broadcast(
    destination_address: str,
    change_address: str,
    txid: str,
    mining_fee: int,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """
    Update a pending "Awaiting transaction" entry when the taker broadcasts the tx.

    This is called after the taker successfully broadcasts a transaction. The entry
    was created earlier (before sending !tx) with failure_reason="Awaiting transaction"
    to ensure the addresses were recorded before revealing them.

    Args:
        destination_address: The CoinJoin destination address to match
        change_address: The change address to match (for extra precision)
        txid: The transaction ID
        mining_fee: Actual mining fee paid (may differ from estimate)
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet (issue #473).

    Returns:
        True if a matching entry was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    updated = False

    for entry in entries:
        # Match by destination + change address and "Awaiting transaction" status
        # Both addresses must match exactly (including empty string for no change)
        if (
            entry.destination_address == destination_address
            and entry.change_address == change_address
            and entry.failure_reason == "Awaiting transaction"
            and not entry.txid  # Should not have txid yet
        ):
            if wallet_fingerprint is not None and entry.wallet_fingerprint != wallet_fingerprint:
                continue
            entry.txid = txid
            entry.mining_fee_paid = mining_fee
            entry.net_fee = -(entry.total_maker_fees_paid + mining_fee)
            entry.failure_reason = "Pending confirmation"  # Now awaiting confirmation
            logger.info(
                f"Updated awaiting transaction for {destination_address[:20]}... "
                f"with txid {txid[:16]}..., mining_fee={mining_fee} sats"
            )
            updated = True
            break

    if not updated:
        return False

    return _write_history_entries_atomic(entries, history_path)

update_transaction_confirmation(txid: str, confirmations: int, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> bool

Update a transaction's confirmation status in the history file.

This function rewrites the entire CSV file with the updated entry. If confirmations > 0, marks the transaction as successful.

Note: This is the synchronous version. For makers who want automatic peer count detection, use update_transaction_confirmation_with_detection().

Args: txid: Transaction ID to update confirmations: Current number of confirmations data_dir: Optional data directory wallet_fingerprint: If provided, only match entries belonging to this wallet (issue #473). Prevents updating another wallet's entry when multiple wallets share the same data directory.

Returns: True if transaction was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def update_transaction_confirmation(
    txid: str,
    confirmations: int,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """
    Update a transaction's confirmation status in the history file.

    This function rewrites the entire CSV file with the updated entry.
    If confirmations > 0, marks the transaction as successful.

    Note: This is the synchronous version. For makers who want automatic
    peer count detection, use update_transaction_confirmation_with_detection().

    Args:
        txid: Transaction ID to update
        confirmations: Current number of confirmations
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet (issue #473). Prevents updating another wallet's entry when
            multiple wallets share the same data directory.

    Returns:
        True if transaction was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    target = _select_entry_for_confirmation_update(entries, txid, wallet_fingerprint)
    if target is None:
        return False

    target.confirmations = confirmations
    if confirmations > 0 and not target.success:
        # Mark as successful on first confirmation
        target.success = True
        target.failure_reason = ""
        target.confirmed_at = datetime.now().isoformat()
        target.completed_at = target.confirmed_at
        logger.info(f"Transaction {txid[:16]}... confirmed with {confirmations} confirmations")
    elif confirmations > 0:
        # Already marked as successful, just update confirmation count
        logger.debug(f"Updated confirmations for {txid[:16]}...: {confirmations}")

    return _write_history_entries_atomic(entries, history_path)

update_transaction_confirmation_with_detection(txid: str, confirmations: int, backend: BlockchainBackend | Any | None = None, data_dir: Path | None = None, wallet_fingerprint: str | None = None) -> bool async

Update transaction confirmation and detect peer count for makers.

This async version can detect the CoinJoin peer count by analyzing the transaction outputs when it confirms. This is useful for makers who don't know the peer count during the CoinJoin.

Args: txid: Transaction ID to update confirmations: Current number of confirmations backend: Blockchain backend for fetching transaction (optional, for peer detection) data_dir: Optional data directory wallet_fingerprint: If provided, only match entries belonging to this wallet (issue #473).

Returns: True if transaction was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

async def update_transaction_confirmation_with_detection(
    txid: str,
    confirmations: int,
    backend: BlockchainBackend | Any | None = None,
    data_dir: Path | None = None,
    wallet_fingerprint: str | None = None,
) -> bool:
    """
    Update transaction confirmation and detect peer count for makers.

    This async version can detect the CoinJoin peer count by analyzing the
    transaction outputs when it confirms. This is useful for makers who don't
    know the peer count during the CoinJoin.

    Args:
        txid: Transaction ID to update
        confirmations: Current number of confirmations
        backend: Blockchain backend for fetching transaction (optional, for peer detection)
        data_dir: Optional data directory
        wallet_fingerprint: If provided, only match entries belonging to this
            wallet (issue #473).

    Returns:
        True if transaction was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    target = _select_entry_for_confirmation_update(entries, txid, wallet_fingerprint)
    if target is None:
        return False

    target.confirmations = confirmations
    if confirmations > 0 and not target.success:
        # Mark as successful on first confirmation
        target.success = True
        target.failure_reason = ""
        target.confirmed_at = datetime.now().isoformat()
        target.completed_at = target.confirmed_at
        logger.info(f"Transaction {txid[:16]}... confirmed with {confirmations} confirmations")

        # Detect peer count for makers
        if (
            target.role == "maker"
            and target.peer_count is None
            and backend is not None
            and target.cj_amount > 0
        ):
            detected_count = await detect_coinjoin_peer_count(backend, txid, target.cj_amount)
            if detected_count is not None:
                target.peer_count = detected_count
                logger.info(f"Detected {detected_count} participants in CoinJoin {txid[:16]}...")

    elif confirmations > 0:
        # Already marked as successful, just update confirmation count
        logger.debug(f"Updated confirmations for {txid[:16]}...: {confirmations}")

    return _write_history_entries_atomic(entries, history_path)

update_transaction_peer_count(txid: str, peer_count: int, data_dir: Path | None = None) -> bool

Update a transaction's peer count in the history file.

This is used for makers to update the peer count after detecting it from the confirmed transaction's equal-amount outputs.

Args: txid: Transaction ID to update peer_count: Detected peer count data_dir: Optional data directory

Returns: True if transaction was found and updated, False otherwise

Source code in jmwallet/src/jmwallet/history.py

def update_transaction_peer_count(
    txid: str,
    peer_count: int,
    data_dir: Path | None = None,
) -> bool:
    """
    Update a transaction's peer count in the history file.

    This is used for makers to update the peer count after detecting it
    from the confirmed transaction's equal-amount outputs.

    Args:
        txid: Transaction ID to update
        peer_count: Detected peer count
        data_dir: Optional data directory

    Returns:
        True if transaction was found and updated, False otherwise
    """
    history_path = _get_history_path(data_dir)
    if not history_path.exists():
        return False

    entries = read_history(data_dir)
    updated = False

    for entry in entries:
        if entry.txid == txid and entry.peer_count is None:
            entry.peer_count = peer_count
            logger.info(f"Updated peer count for tx {txid[:16]}... to {peer_count}")
            updated = True
            break

    if not updated:
        return False

    return _write_history_entries_atomic(entries, history_path)