Module taker.config

Configuration for JoinMarket Taker.

Classes

class BroadcastPolicy (*values)
Expand source code 
class BroadcastPolicy(str, Enum):
    """
    Policy for how to broadcast the final CoinJoin transaction.

    Privacy implications:
    - SELF: Taker broadcasts via own node. Links taker's IP to the transaction.
    - RANDOM_PEER: Random selection from makers + self. Provides plausible deniability.
    - NOT_SELF: Only makers can broadcast. Maximum privacy - taker's node never touches tx.
                WARNING: No fallback if makers fail to broadcast!
    """

    SELF = "self"
    RANDOM_PEER = "random-peer"
    NOT_SELF = "not-self"

Policy for how to broadcast the final CoinJoin transaction.

Privacy implications: - SELF: Taker broadcasts via own node. Links taker's IP to the transaction. - RANDOM_PEER: Random selection from makers + self. Provides plausible deniability. - NOT_SELF: Only makers can broadcast. Maximum privacy - taker's node never touches tx. WARNING: No fallback if makers fail to broadcast!

Ancestors

  • builtins.str
  • enum.Enum

Class variables

var NOT_SELF

The type of the None singleton.

var RANDOM_PEER

The type of the None singleton.

var SELF

The type of the None singleton.

class MaxCjFee (**data: Any)
Expand source code 
class MaxCjFee(BaseModel):
    """Maximum CoinJoin fee limits."""

    abs_fee: int = Field(default=500, ge=0, description="Maximum absolute fee in sats")
    rel_fee: str = Field(default="0.001", description="Maximum relative fee (0.001 = 0.1%)")

Maximum CoinJoin fee limits.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

  • pydantic.main.BaseModel

Class variables

var abs_fee : int

The type of the None singleton.

var model_config

The type of the None singleton.

var rel_fee : str

The type of the None singleton.

class Schedule (**data: Any)
Expand source code 
class Schedule(BaseModel):
    """CoinJoin schedule for tumbler-style operations."""

    entries: list[ScheduleEntry] = Field(default_factory=list)
    current_index: int = Field(default=0, ge=0)

    def current_entry(self) -> ScheduleEntry | None:
        """Get current schedule entry."""
        if self.current_index >= len(self.entries):
            return None
        return self.entries[self.current_index]

    def advance(self) -> bool:
        """Advance to next entry. Returns True if more entries remain."""
        if self.current_index < len(self.entries):
            self.entries[self.current_index].completed = True
            self.current_index += 1
        return self.current_index < len(self.entries)

    def is_complete(self) -> bool:
        """Check if all entries are complete."""
        return self.current_index >= len(self.entries)

CoinJoin schedule for tumbler-style operations.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

  • pydantic.main.BaseModel

Class variables

var current_index : int

The type of the None singleton.

var entries : list[ScheduleEntry]

The type of the None singleton.

var model_config

The type of the None singleton.

Methods

def advance(self) ‑> bool
Expand source code 
def advance(self) -> bool:
    """Advance to next entry. Returns True if more entries remain."""
    if self.current_index < len(self.entries):
        self.entries[self.current_index].completed = True
        self.current_index += 1
    return self.current_index < len(self.entries)

Advance to next entry. Returns True if more entries remain.

def current_entry(self) ‑> ScheduleEntry | None
Expand source code 
def current_entry(self) -> ScheduleEntry | None:
    """Get current schedule entry."""
    if self.current_index >= len(self.entries):
        return None
    return self.entries[self.current_index]

Get current schedule entry.

def is_complete(self) ‑> bool
Expand source code 
def is_complete(self) -> bool:
    """Check if all entries are complete."""
    return self.current_index >= len(self.entries)

Check if all entries are complete.

class ScheduleEntry (**data: Any)
Expand source code 
class ScheduleEntry(BaseModel):
    """A single entry in a CoinJoin schedule."""

    mixdepth: int = Field(..., ge=0, le=9)
    amount: int | None = Field(
        default=None,
        ge=0,
        description="Amount in satoshis (mutually exclusive with amount_fraction)",
    )
    amount_fraction: float | None = Field(
        default=None,
        ge=0.0,
        le=1.0,
        description="Fraction of balance (0.0-1.0, mutually exclusive with amount)",
    )
    counterparty_count: int = Field(..., ge=1, le=20)
    destination: str = Field(..., description="Destination address or 'INTERNAL'")
    wait_time: float = Field(default=0.0, ge=0.0, description="Wait time after completion")
    rounding: int = Field(default=16, ge=1, description="Significant figures for rounding")
    completed: bool = False

    @model_validator(mode="after")
    def validate_amount_fields(self) -> ScheduleEntry:
        """Ensure exactly one of amount or amount_fraction is set."""
        if self.amount is None and self.amount_fraction is None:
            raise ValueError("Must specify either 'amount' or 'amount_fraction'")
        if self.amount is not None and self.amount_fraction is not None:
            raise ValueError("Cannot specify both 'amount' and 'amount_fraction'")
        return self

A single entry in a CoinJoin schedule.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

  • pydantic.main.BaseModel

Class variables

var amount : int | None

The type of the None singleton.

var amount_fraction : float | None

The type of the None singleton.

var completed : bool

The type of the None singleton.

var counterparty_count : int

The type of the None singleton.

var destination : str

The type of the None singleton.

var mixdepth : int

The type of the None singleton.

var model_config

The type of the None singleton.

var rounding : int

The type of the None singleton.

var wait_time : float

The type of the None singleton.

Methods

def validate_amount_fields(self) ‑> ScheduleEntry
Expand source code 
@model_validator(mode="after")
def validate_amount_fields(self) -> ScheduleEntry:
    """Ensure exactly one of amount or amount_fraction is set."""
    if self.amount is None and self.amount_fraction is None:
        raise ValueError("Must specify either 'amount' or 'amount_fraction'")
    if self.amount is not None and self.amount_fraction is not None:
        raise ValueError("Cannot specify both 'amount' and 'amount_fraction'")
    return self

Ensure exactly one of amount or amount_fraction is set.

class TakerConfig (**data: Any)
Expand source code 
class TakerConfig(WalletConfig):
    """
    Configuration for taker bot.

    Inherits base wallet configuration from jmcore.config.WalletConfig
    and adds taker-specific settings for CoinJoin execution, PoDLE,
    and broadcasting.
    """

    # CoinJoin settings
    destination_address: str = Field(
        default="", description="Target address for CJ output, empty = INTERNAL"
    )
    amount: int = Field(default=0, ge=0, description="Amount in sats (0 = sweep)")
    mixdepth: int = Field(default=0, ge=0, description="Source mixdepth")
    counterparty_count: int = Field(
        default=3, ge=1, le=20, description="Number of makers to select"
    )

    # Fee settings
    max_cj_fee: MaxCjFee = Field(
        default_factory=MaxCjFee, description="Maximum CoinJoin fee limits"
    )
    tx_fee_factor: float = Field(
        default=3.0, ge=1.0, description="Multiply estimated fee by this factor"
    )
    bondless_makers_allowance: float = Field(
        default=0.125,
        ge=0.0,
        le=1.0,
        description="Fraction of time to choose makers randomly (not by fidelity bond)",
    )

    # PoDLE settings
    taker_utxo_retries: int = Field(
        default=3,
        ge=1,
        le=10,
        description="Maximum PoDLE index retries per UTXO (reference: 3)",
    )
    taker_utxo_age: int = Field(default=5, ge=1, description="Minimum UTXO confirmations")
    taker_utxo_amtpercent: int = Field(
        default=20, ge=1, le=100, description="Min UTXO value as % of CJ amount"
    )

    # Timeouts
    maker_timeout_sec: int = Field(default=60, ge=10, description="Timeout for maker responses")
    order_wait_time: float = Field(
        default=10.0, ge=1.0, description="Seconds to wait for orderbook"
    )

    # Broadcast policy (privacy vs reliability tradeoff)
    tx_broadcast: BroadcastPolicy = Field(
        default=BroadcastPolicy.RANDOM_PEER,
        description="How to broadcast the final transaction: self, random-peer, or not-self",
    )
    broadcast_timeout_sec: int = Field(
        default=30,
        ge=5,
        description="Timeout waiting for maker to broadcast when delegating",
    )

    # Advanced options
    preferred_offer_type: OfferType = Field(
        default=OfferType.SW0_RELATIVE, description="Preferred offer type"
    )
    minimum_makers: int = Field(default=2, ge=1, description="Minimum number of makers required")

    # Wallet rescan configuration
    rescan_interval_sec: int = Field(
        default=600,
        ge=60,
        description="Interval in seconds for periodic wallet rescans (default: 10 minutes)",
    )

    @model_validator(mode="after")
    def set_bitcoin_network_default(self) -> TakerConfig:
        """If bitcoin_network is not set, default to the protocol network."""
        if self.bitcoin_network is None:
            object.__setattr__(self, "bitcoin_network", self.network)
        return self

Configuration for taker bot.

Inherits base wallet configuration from jmcore.config.WalletConfig and adds taker-specific settings for CoinJoin execution, PoDLE, and broadcasting.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Ancestors

Class variables

var amount : int

The type of the None singleton.

var bondless_makers_allowance : float

The type of the None singleton.

var broadcast_timeout_sec : int

The type of the None singleton.

var counterparty_count : int

The type of the None singleton.

var destination_address : str

The type of the None singleton.

var maker_timeout_sec : int

The type of the None singleton.

var max_cj_fee : MaxCjFee

The type of the None singleton.

var minimum_makers : int

The type of the None singleton.

var mixdepth : int

The type of the None singleton.

var order_wait_time : float

The type of the None singleton.

var preferred_offer_type : OfferType

The type of the None singleton.

var rescan_interval_sec : int

The type of the None singleton.

var taker_utxo_age : int

The type of the None singleton.

var taker_utxo_amtpercent : int

The type of the None singleton.

var taker_utxo_retries : int

The type of the None singleton.

var tx_broadcast : BroadcastPolicy

The type of the None singleton.

var tx_fee_factor : float

The type of the None singleton.

Inherited members