Module taker.cli
Command-line interface for JoinMarket Taker.
Functions
def coinjoin(amount: "Annotated[int, typer.Option('--amount', '-a', help='Amount in sats (0 for sweep)')]",
destination: Annotated[str, typer.Option(\'--destination\', \'-d\', help="Destination address (or \'INTERNAL\' for next mixdepth)")] = 'INTERNAL',
mixdepth: "Annotated[int, typer.Option('--mixdepth', '-m', help='Source mixdepth')]" = 0,
counterparties: "Annotated[int, typer.Option('--counterparties', '-n', help='Number of makers')]" = 3,
mnemonic: "Annotated[str | None, typer.Option('--mnemonic', envvar='MNEMONIC', help='Wallet mnemonic phrase')]" = None,
mnemonic_file: "Annotated[Path | None, typer.Option('--mnemonic-file', '-f', help='Path to mnemonic file')]" = None,
password: "Annotated[str | None, typer.Option('--password', '-p', help='Password for encrypted mnemonic file')]" = None,
network: "Annotated[str, typer.Option('--network', help='Protocol network for handshakes')]" = 'mainnet',
bitcoin_network: "Annotated[str | None, typer.Option('--bitcoin-network', help='Bitcoin network for addresses (defaults to --network)')]" = None,
backend_type: "Annotated[str, typer.Option('--backend', '-b', help='Backend type: full_node | neutrino')]" = 'full_node',
rpc_url: "Annotated[str, typer.Option('--rpc-url', envvar='BITCOIN_RPC_URL', help='Bitcoin full node RPC URL')]" = 'http://127.0.0.1:8332',
rpc_user: "Annotated[str, typer.Option('--rpc-user', envvar='BITCOIN_RPC_USER', help='Bitcoin full node RPC user')]" = '',
rpc_password: "Annotated[str, typer.Option('--rpc-password', envvar='BITCOIN_RPC_PASSWORD', help='Bitcoin full node RPC password')]" = '',
neutrino_url: "Annotated[str, typer.Option('--neutrino-url', envvar='NEUTRINO_URL', help='Neutrino REST API URL')]" = 'http://127.0.0.1:8334',
directory_servers: "Annotated[str | None, typer.Option('--directory', '-D', envvar='DIRECTORY_SERVERS', help='Directory servers (comma-separated). Defaults to mainnet directory nodes.')]" = None,
tor_socks_host: "Annotated[str, typer.Option(envvar='TOR_SOCKS_HOST', help='Tor SOCKS proxy host')]" = '127.0.0.1',
tor_socks_port: "Annotated[int, typer.Option(envvar='TOR_SOCKS_PORT', help='Tor SOCKS proxy port')]" = 9050,
max_abs_fee: "Annotated[int, typer.Option('--max-abs-fee', help='Max absolute fee in sats')]" = 500,
max_rel_fee: "Annotated[str, typer.Option('--max-rel-fee', help='Max relative fee (0.001=0.1%)')]" = '0.001',
bondless_makers_allowance: "Annotated[float, typer.Option('--bondless-allowance', help='Fraction of time to choose makers randomly (0.0-1.0)')]" = 0.125,
yes: "Annotated[bool, typer.Option('--yes', '-y', help='Skip confirmation prompt')]" = False,
log_level: "Annotated[str, typer.Option('--log-level', '-l', help='Log level')]" = 'INFO') ‑> None-
Expand source code
@app.command() def coinjoin( amount: Annotated[int, typer.Option("--amount", "-a", help="Amount in sats (0 for sweep)")], destination: Annotated[ str, typer.Option( "--destination", "-d", help="Destination address (or 'INTERNAL' for next mixdepth)", ), ] = "INTERNAL", mixdepth: Annotated[int, typer.Option("--mixdepth", "-m", help="Source mixdepth")] = 0, counterparties: Annotated[ int, typer.Option("--counterparties", "-n", help="Number of makers") ] = 3, mnemonic: Annotated[ str | None, typer.Option("--mnemonic", envvar="MNEMONIC", help="Wallet mnemonic phrase") ] = None, mnemonic_file: Annotated[ Path | None, typer.Option("--mnemonic-file", "-f", help="Path to mnemonic file") ] = None, password: Annotated[ str | None, typer.Option("--password", "-p", help="Password for encrypted mnemonic file") ] = None, network: Annotated[ str, typer.Option("--network", help="Protocol network for handshakes") ] = "mainnet", bitcoin_network: Annotated[ str | None, typer.Option( "--bitcoin-network", help="Bitcoin network for addresses (defaults to --network)" ), ] = None, backend_type: Annotated[ str, typer.Option("--backend", "-b", help="Backend type: full_node | neutrino") ] = "full_node", rpc_url: Annotated[ str, typer.Option( "--rpc-url", envvar="BITCOIN_RPC_URL", help="Bitcoin full node RPC URL", ), ] = "http://127.0.0.1:8332", rpc_user: Annotated[ str, typer.Option("--rpc-user", envvar="BITCOIN_RPC_USER", help="Bitcoin full node RPC user"), ] = "", rpc_password: Annotated[ str, typer.Option( "--rpc-password", envvar="BITCOIN_RPC_PASSWORD", help="Bitcoin full node RPC password" ), ] = "", neutrino_url: Annotated[ str, typer.Option( "--neutrino-url", envvar="NEUTRINO_URL", help="Neutrino REST API URL", ), ] = "http://127.0.0.1:8334", directory_servers: Annotated[ str | None, typer.Option( "--directory", "-D", envvar="DIRECTORY_SERVERS", help="Directory servers (comma-separated). Defaults to mainnet directory nodes.", ), ] = None, tor_socks_host: Annotated[ str, typer.Option(envvar="TOR_SOCKS_HOST", help="Tor SOCKS proxy host") ] = "127.0.0.1", tor_socks_port: Annotated[ int, typer.Option(envvar="TOR_SOCKS_PORT", help="Tor SOCKS proxy port") ] = 9050, max_abs_fee: Annotated[ int, typer.Option("--max-abs-fee", help="Max absolute fee in sats") ] = 500, max_rel_fee: Annotated[ str, typer.Option("--max-rel-fee", help="Max relative fee (0.001=0.1%)") ] = "0.001", bondless_makers_allowance: Annotated[ float, typer.Option( "--bondless-allowance", help="Fraction of time to choose makers randomly (0.0-1.0)" ), ] = 0.125, yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt")] = False, log_level: Annotated[str, typer.Option("--log-level", "-l", help="Log level")] = "INFO", ) -> None: """Execute a single CoinJoin transaction.""" setup_logging(log_level) # Load mnemonic try: resolved_mnemonic = load_mnemonic(mnemonic, mnemonic_file, password) except ValueError as e: logger.error(str(e)) raise typer.Exit(1) # Parse network try: network_type = NetworkType(network) except ValueError: logger.error(f"Invalid network: {network}") raise typer.Exit(1) # Parse bitcoin network (defaults to protocol network) actual_bitcoin_network = bitcoin_network or network try: bitcoin_network_type = NetworkType(actual_bitcoin_network) except ValueError: logger.error(f"Invalid bitcoin network: {actual_bitcoin_network}") raise typer.Exit(1) # Parse directory servers: use provided list or default for network if directory_servers: dir_servers = [s.strip() for s in directory_servers.split(",")] else: dir_servers = get_default_directory_nodes(network_type) # Build backend config based on type if backend_type == "neutrino": backend_config = { "neutrino_url": neutrino_url, "network": actual_bitcoin_network, } else: backend_config = { "rpc_url": rpc_url, "rpc_user": rpc_user, "rpc_password": rpc_password, } # Build config config = TakerConfig( mnemonic=resolved_mnemonic, network=network_type, bitcoin_network=bitcoin_network_type, backend_type=backend_type, backend_config=backend_config, directory_servers=dir_servers, socks_host=tor_socks_host, socks_port=tor_socks_port, destination_address=destination, amount=amount, mixdepth=mixdepth, counterparty_count=counterparties, max_cj_fee=MaxCjFee(abs_fee=max_abs_fee, rel_fee=max_rel_fee), bondless_makers_allowance=bondless_makers_allowance, ) asyncio.run(_run_coinjoin(config, amount, destination, mixdepth, counterparties, yes))Execute a single CoinJoin transaction.
def load_mnemonic(mnemonic: str | None, mnemonic_file: Path | None, password: str | None) ‑> str-
Expand source code
def load_mnemonic( mnemonic: str | None, mnemonic_file: Path | None, password: str | None, ) -> str: """ Load mnemonic from argument, file, or environment variable. Priority: 1. --mnemonic argument 2. --mnemonic-file argument 3. MNEMONIC_FILE environment variable (path to mnemonic file) 4. MNEMONIC environment variable Args: mnemonic: Direct mnemonic string mnemonic_file: Path to mnemonic file password: Password for encrypted file Returns: The mnemonic phrase Raises: ValueError: If no mnemonic source is available """ if mnemonic: return mnemonic # Check for mnemonic file (from argument or environment) actual_mnemonic_file = mnemonic_file if not actual_mnemonic_file: env_mnemonic_file = os.environ.get("MNEMONIC_FILE") if env_mnemonic_file: actual_mnemonic_file = Path(env_mnemonic_file) if actual_mnemonic_file: if not actual_mnemonic_file.exists(): raise ValueError(f"Mnemonic file not found: {actual_mnemonic_file}") # Import the mnemonic loading utilities from jmwallet from jmwallet.cli import load_mnemonic_file try: return load_mnemonic_file(actual_mnemonic_file, password) except ValueError: # File is encrypted, need password if password is None: password = typer.prompt("Enter mnemonic file password", hide_input=True) return load_mnemonic_file(actual_mnemonic_file, password) env_mnemonic = os.environ.get("MNEMONIC") if env_mnemonic: return env_mnemonic raise ValueError( "Mnemonic required. Use --mnemonic, --mnemonic-file, MNEMONIC_FILE, or MNEMONIC env var" )Load mnemonic from argument, file, or environment variable.
Priority: 1. –mnemonic argument 2. –mnemonic-file argument 3. MNEMONIC_FILE environment variable (path to mnemonic file) 4. MNEMONIC environment variable
Args
mnemonic- Direct mnemonic string
mnemonic_file- Path to mnemonic file
password- Password for encrypted file
Returns
The mnemonic phrase
Raises
ValueError- If no mnemonic source is available
def main() ‑> None-
Expand source code
def main() -> None: """Entry point.""" app()Entry point.
def setup_logging(level: str) ‑> None-
Expand source code
def setup_logging(level: str) -> None: """Configure loguru logging.""" logger.remove() logger.add( sys.stderr, level=level.upper(), format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | {message}", )Configure loguru logging.
def tumble(schedule_file: "Annotated[Path, typer.Argument(help='Path to schedule JSON file')]",
mnemonic: "Annotated[str | None, typer.Option('--mnemonic', envvar='MNEMONIC', help='Wallet mnemonic phrase')]" = None,
mnemonic_file: "Annotated[Path | None, typer.Option('--mnemonic-file', '-f', help='Path to mnemonic file')]" = None,
password: "Annotated[str | None, typer.Option('--password', '-p', help='Password for encrypted mnemonic file')]" = None,
network: "Annotated[str, typer.Option('--network', help='Bitcoin network')]" = 'mainnet',
backend_type: "Annotated[str, typer.Option('--backend', '-b', help='Backend type: full_node | neutrino')]" = 'full_node',
rpc_url: "Annotated[str, typer.Option('--rpc-url', envvar='BITCOIN_RPC_URL', help='Bitcoin full node RPC URL')]" = 'http://127.0.0.1:8332',
rpc_user: "Annotated[str, typer.Option('--rpc-user', envvar='BITCOIN_RPC_USER', help='Bitcoin full node RPC user')]" = '',
rpc_password: "Annotated[str, typer.Option('--rpc-password', envvar='BITCOIN_RPC_PASSWORD', help='Bitcoin full node RPC password')]" = '',
neutrino_url: "Annotated[str, typer.Option('--neutrino-url', envvar='NEUTRINO_URL', help='Neutrino REST API URL')]" = 'http://127.0.0.1:8334',
directory_servers: "Annotated[str | None, typer.Option('--directory', '-D', envvar='DIRECTORY_SERVERS', help='Directory servers (comma-separated). Defaults to mainnet directory nodes.')]" = None,
tor_socks_host: "Annotated[str, typer.Option(envvar='TOR_SOCKS_HOST', help='Tor SOCKS proxy host')]" = '127.0.0.1',
tor_socks_port: "Annotated[int, typer.Option(envvar='TOR_SOCKS_PORT', help='Tor SOCKS proxy port')]" = 9050,
log_level: "Annotated[str, typer.Option('--log-level', '-l', help='Log level')]" = 'INFO') ‑> None-
Expand source code
@app.command() def tumble( schedule_file: Annotated[Path, typer.Argument(help="Path to schedule JSON file")], mnemonic: Annotated[ str | None, typer.Option("--mnemonic", envvar="MNEMONIC", help="Wallet mnemonic phrase") ] = None, mnemonic_file: Annotated[ Path | None, typer.Option("--mnemonic-file", "-f", help="Path to mnemonic file") ] = None, password: Annotated[ str | None, typer.Option("--password", "-p", help="Password for encrypted mnemonic file") ] = None, network: Annotated[str, typer.Option("--network", help="Bitcoin network")] = "mainnet", backend_type: Annotated[ str, typer.Option("--backend", "-b", help="Backend type: full_node | neutrino") ] = "full_node", rpc_url: Annotated[ str, typer.Option( "--rpc-url", envvar="BITCOIN_RPC_URL", help="Bitcoin full node RPC URL", ), ] = "http://127.0.0.1:8332", rpc_user: Annotated[ str, typer.Option("--rpc-user", envvar="BITCOIN_RPC_USER", help="Bitcoin full node RPC user"), ] = "", rpc_password: Annotated[ str, typer.Option( "--rpc-password", envvar="BITCOIN_RPC_PASSWORD", help="Bitcoin full node RPC password" ), ] = "", neutrino_url: Annotated[ str, typer.Option( "--neutrino-url", envvar="NEUTRINO_URL", help="Neutrino REST API URL", ), ] = "http://127.0.0.1:8334", directory_servers: Annotated[ str | None, typer.Option( "--directory", "-D", envvar="DIRECTORY_SERVERS", help="Directory servers (comma-separated). Defaults to mainnet directory nodes.", ), ] = None, tor_socks_host: Annotated[ str, typer.Option(envvar="TOR_SOCKS_HOST", help="Tor SOCKS proxy host") ] = "127.0.0.1", tor_socks_port: Annotated[ int, typer.Option(envvar="TOR_SOCKS_PORT", help="Tor SOCKS proxy port") ] = 9050, log_level: Annotated[str, typer.Option("--log-level", "-l", help="Log level")] = "INFO", ) -> None: """Run a tumbler schedule of CoinJoins.""" setup_logging(log_level) # Load mnemonic try: resolved_mnemonic = load_mnemonic(mnemonic, mnemonic_file, password) except ValueError as e: logger.error(str(e)) raise typer.Exit(1) if not schedule_file.exists(): logger.error(f"Schedule file not found: {schedule_file}") raise typer.Exit(1) # Load schedule import json try: with open(schedule_file) as f: schedule_data = json.load(f) entries = [ScheduleEntry(**entry) for entry in schedule_data["entries"]] schedule = Schedule(entries=entries) except Exception as e: logger.error(f"Failed to load schedule: {e}") raise typer.Exit(1) # Parse network try: network_type = NetworkType(network) except ValueError: logger.error(f"Invalid network: {network}") raise typer.Exit(1) # Parse directory servers: use provided list or default for network if directory_servers: dir_servers = [s.strip() for s in directory_servers.split(",")] else: dir_servers = get_default_directory_nodes(network_type) # Build backend config based on type if backend_type == "neutrino": backend_config = { "neutrino_url": neutrino_url, "network": network, } else: backend_config = { "rpc_url": rpc_url, "rpc_user": rpc_user, "rpc_password": rpc_password, } # Build config config = TakerConfig( mnemonic=resolved_mnemonic, network=network_type, backend_type=backend_type, backend_config=backend_config, directory_servers=dir_servers, socks_host=tor_socks_host, socks_port=tor_socks_port, ) asyncio.run(_run_tumble(config, schedule))Run a tumbler schedule of CoinJoins.