Skip to content

taker.tx_builder

taker.tx_builder

Transaction builder for CoinJoin transactions.

Builds the unsigned CoinJoin transaction from: - Taker's UTXOs and change address - Maker UTXOs, CJ addresses, and change addresses - CoinJoin amount and fees

Attributes

logger = logging.getLogger(__name__) module-attribute

varint = encode_varint module-attribute

Classes

CoinJoinTxBuilder

Builds CoinJoin transactions.

The transaction structure: - Inputs: Taker inputs + Maker inputs (shuffled) - Outputs: Equal CJ outputs + Change outputs (shuffled)

Source code in taker/src/taker/tx_builder.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class CoinJoinTxBuilder:
    """
    Builds CoinJoin transactions.

    The transaction structure:
    - Inputs: Taker inputs + Maker inputs (shuffled)
    - Outputs: Equal CJ outputs + Change outputs (shuffled)
    """

    def __init__(self, network: str = "mainnet"):
        self.network = network

    def build_unsigned_tx(self, tx_data: CoinJoinTxData) -> tuple[bytes, dict[str, Any]]:
        """
        Build an unsigned CoinJoin transaction.

        Args:
            tx_data: Transaction data with all inputs and outputs

        Returns:
            (tx_bytes, metadata) where metadata maps inputs/outputs to owners
        """
        import random

        # Collect all inputs with owner info
        all_inputs: list[tuple[TxInput, str]] = []

        for inp in tx_data.taker_inputs:
            all_inputs.append((inp, "taker"))

        for nick, inputs in tx_data.maker_inputs.items():
            for inp in inputs:
                all_inputs.append((inp, nick))

        # Collect all outputs with owner info
        all_outputs: list[tuple[TxOutput, str, str]] = []  # (output, owner, type)

        # CJ outputs (equal amounts)
        all_outputs.append((tx_data.taker_cj_output, "taker", "cj"))
        for nick, out in tx_data.maker_cj_outputs.items():
            all_outputs.append((out, nick, "cj"))

        # Change outputs
        if tx_data.taker_change_output:
            all_outputs.append((tx_data.taker_change_output, "taker", "change"))
        for nick, out in tx_data.maker_change_outputs.items():
            all_outputs.append((out, nick, "change"))

        # Shuffle for privacy
        random.shuffle(all_inputs)
        random.shuffle(all_outputs)

        # Build metadata
        metadata = {
            "input_owners": [owner for _, owner in all_inputs],
            "output_owners": [(owner, out_type) for _, owner, out_type in all_outputs],
            "input_values": [inp.value for inp, _ in all_inputs],
            "fee": tx_data.tx_fee,
        }

        # Serialize transaction
        tx_bytes = self._serialize_tx(
            inputs=[inp for inp, _ in all_inputs],
            outputs=[out for out, _, _ in all_outputs],
        )

        return tx_bytes, metadata

    def _serialize_tx(self, inputs: list[TxInput], outputs: list[TxOutput]) -> bytes:
        """Serialize transaction to bytes.

        For unsigned transactions, we use non-SegWit format (no marker/flag/witness).
        The SegWit marker (0x00, 0x01) is only added when witnesses are present.
        """
        return serialize_transaction(
            version=2,
            inputs=inputs,
            outputs=outputs,
            locktime=0,
            witnesses=None,
        )

    def add_signatures(
        self,
        tx_bytes: bytes,
        signatures: dict[str, list[dict[str, Any]]],
        metadata: dict[str, Any],
    ) -> bytes:
        """
        Add signatures to transaction.

        Every input must have a matching signature. A CoinJoin transaction with
        any unsigned input is invalid and must never be broadcast.

        Args:
            tx_bytes: Unsigned transaction
            signatures: Dict of nick -> list of signature info
            metadata: Transaction metadata with input owners

        Returns:
            Signed transaction bytes

        Raises:
            ValueError: If any input is missing a signature
        """
        from loguru import logger as log

        # Parse unsigned tx using jmcore
        parsed = parse_transaction_bytes(tx_bytes)

        log.debug(f"add_signatures: {len(parsed.inputs)} inputs, {len(parsed.outputs)} outputs")
        log.debug(f"input_owners: {metadata.get('input_owners', [])}")
        log.debug(f"signatures keys: {list(signatures.keys())}")

        # Build witness data
        new_witnesses: list[list[bytes]] = []
        input_owners = metadata["input_owners"]
        unsigned_inputs: list[str] = []

        for i, owner in enumerate(input_owners):
            inp = parsed.inputs[i]
            log.debug(f"Input {i}: owner={owner}, txid={inp.txid[:16]}..., vout={inp.vout}")

            if owner in signatures:
                # Find matching signature
                for sig_info in signatures[owner]:
                    if sig_info.get("txid") == inp.txid and sig_info.get("vout") == inp.vout:
                        witness = sig_info.get("witness", [])
                        new_witnesses.append([bytes.fromhex(w) for w in witness])
                        log.debug(f"  -> Found matching signature, witness len={len(witness)}")
                        break
                else:
                    unsigned_inputs.append(
                        f"input {i} (owner={owner}, txid={inp.txid[:16]}...:{inp.vout})"
                    )
                    new_witnesses.append([])
            else:
                unsigned_inputs.append(
                    f"input {i} (owner={owner}, txid={inp.txid[:16]}...:{inp.vout})"
                )
                new_witnesses.append([])

        if unsigned_inputs:
            raise ValueError(
                f"Cannot assemble transaction: {len(unsigned_inputs)} input(s) missing "
                f"signatures: {', '.join(unsigned_inputs)}. "
                f"All inputs must be signed for a valid transaction."
            )

        # Reserialize with witnesses using jmcore
        return serialize_transaction(
            version=parsed.version,
            inputs=parsed.inputs,
            outputs=parsed.outputs,
            locktime=parsed.locktime,
            witnesses=new_witnesses,
        )

    def get_txid(self, tx_bytes: bytes) -> str:
        """Calculate txid (double SHA256 of non-witness data)."""
        parsed = parse_transaction_bytes(tx_bytes)

        # Serialize without witness for txid calculation
        data = serialize_transaction(
            version=parsed.version,
            inputs=parsed.inputs,
            outputs=parsed.outputs,
            locktime=parsed.locktime,
            witnesses=None,
        )

        return hash256(data)[::-1].hex()
Attributes
network = network instance-attribute
Functions
__init__(network: str = 'mainnet')
Source code in taker/src/taker/tx_builder.py 
61
62
def __init__(self, network: str = "mainnet"):
    self.network = network
add_signatures(tx_bytes: bytes, signatures: dict[str, list[dict[str, Any]]], metadata: dict[str, Any]) -> bytes

Add signatures to transaction.

Every input must have a matching signature. A CoinJoin transaction with any unsigned input is invalid and must never be broadcast.

Args: tx_bytes: Unsigned transaction signatures: Dict of nick -> list of signature info metadata: Transaction metadata with input owners

Returns: Signed transaction bytes

Raises: ValueError: If any input is missing a signature

Source code in taker/src/taker/tx_builder.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def add_signatures(
    self,
    tx_bytes: bytes,
    signatures: dict[str, list[dict[str, Any]]],
    metadata: dict[str, Any],
) -> bytes:
    """
    Add signatures to transaction.

    Every input must have a matching signature. A CoinJoin transaction with
    any unsigned input is invalid and must never be broadcast.

    Args:
        tx_bytes: Unsigned transaction
        signatures: Dict of nick -> list of signature info
        metadata: Transaction metadata with input owners

    Returns:
        Signed transaction bytes

    Raises:
        ValueError: If any input is missing a signature
    """
    from loguru import logger as log

    # Parse unsigned tx using jmcore
    parsed = parse_transaction_bytes(tx_bytes)

    log.debug(f"add_signatures: {len(parsed.inputs)} inputs, {len(parsed.outputs)} outputs")
    log.debug(f"input_owners: {metadata.get('input_owners', [])}")
    log.debug(f"signatures keys: {list(signatures.keys())}")

    # Build witness data
    new_witnesses: list[list[bytes]] = []
    input_owners = metadata["input_owners"]
    unsigned_inputs: list[str] = []

    for i, owner in enumerate(input_owners):
        inp = parsed.inputs[i]
        log.debug(f"Input {i}: owner={owner}, txid={inp.txid[:16]}..., vout={inp.vout}")

        if owner in signatures:
            # Find matching signature
            for sig_info in signatures[owner]:
                if sig_info.get("txid") == inp.txid and sig_info.get("vout") == inp.vout:
                    witness = sig_info.get("witness", [])
                    new_witnesses.append([bytes.fromhex(w) for w in witness])
                    log.debug(f"  -> Found matching signature, witness len={len(witness)}")
                    break
            else:
                unsigned_inputs.append(
                    f"input {i} (owner={owner}, txid={inp.txid[:16]}...:{inp.vout})"
                )
                new_witnesses.append([])
        else:
            unsigned_inputs.append(
                f"input {i} (owner={owner}, txid={inp.txid[:16]}...:{inp.vout})"
            )
            new_witnesses.append([])

    if unsigned_inputs:
        raise ValueError(
            f"Cannot assemble transaction: {len(unsigned_inputs)} input(s) missing "
            f"signatures: {', '.join(unsigned_inputs)}. "
            f"All inputs must be signed for a valid transaction."
        )

    # Reserialize with witnesses using jmcore
    return serialize_transaction(
        version=parsed.version,
        inputs=parsed.inputs,
        outputs=parsed.outputs,
        locktime=parsed.locktime,
        witnesses=new_witnesses,
    )
build_unsigned_tx(tx_data: CoinJoinTxData) -> tuple[bytes, dict[str, Any]]

Build an unsigned CoinJoin transaction.

Args: tx_data: Transaction data with all inputs and outputs

Returns: (tx_bytes, metadata) where metadata maps inputs/outputs to owners

Source code in taker/src/taker/tx_builder.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def build_unsigned_tx(self, tx_data: CoinJoinTxData) -> tuple[bytes, dict[str, Any]]:
    """
    Build an unsigned CoinJoin transaction.

    Args:
        tx_data: Transaction data with all inputs and outputs

    Returns:
        (tx_bytes, metadata) where metadata maps inputs/outputs to owners
    """
    import random

    # Collect all inputs with owner info
    all_inputs: list[tuple[TxInput, str]] = []

    for inp in tx_data.taker_inputs:
        all_inputs.append((inp, "taker"))

    for nick, inputs in tx_data.maker_inputs.items():
        for inp in inputs:
            all_inputs.append((inp, nick))

    # Collect all outputs with owner info
    all_outputs: list[tuple[TxOutput, str, str]] = []  # (output, owner, type)

    # CJ outputs (equal amounts)
    all_outputs.append((tx_data.taker_cj_output, "taker", "cj"))
    for nick, out in tx_data.maker_cj_outputs.items():
        all_outputs.append((out, nick, "cj"))

    # Change outputs
    if tx_data.taker_change_output:
        all_outputs.append((tx_data.taker_change_output, "taker", "change"))
    for nick, out in tx_data.maker_change_outputs.items():
        all_outputs.append((out, nick, "change"))

    # Shuffle for privacy
    random.shuffle(all_inputs)
    random.shuffle(all_outputs)

    # Build metadata
    metadata = {
        "input_owners": [owner for _, owner in all_inputs],
        "output_owners": [(owner, out_type) for _, owner, out_type in all_outputs],
        "input_values": [inp.value for inp, _ in all_inputs],
        "fee": tx_data.tx_fee,
    }

    # Serialize transaction
    tx_bytes = self._serialize_tx(
        inputs=[inp for inp, _ in all_inputs],
        outputs=[out for out, _, _ in all_outputs],
    )

    return tx_bytes, metadata
get_txid(tx_bytes: bytes) -> str

Calculate txid (double SHA256 of non-witness data).

Source code in taker/src/taker/tx_builder.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def get_txid(self, tx_bytes: bytes) -> str:
    """Calculate txid (double SHA256 of non-witness data)."""
    parsed = parse_transaction_bytes(tx_bytes)

    # Serialize without witness for txid calculation
    data = serialize_transaction(
        version=parsed.version,
        inputs=parsed.inputs,
        outputs=parsed.outputs,
        locktime=parsed.locktime,
        witnesses=None,
    )

    return hash256(data)[::-1].hex()

CoinJoinTxData

Data for building a CoinJoin transaction.

Source code in taker/src/taker/tx_builder.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@dataclass
class CoinJoinTxData:
    """Data for building a CoinJoin transaction."""

    # Taker data
    taker_inputs: list[TxInput]
    taker_cj_output: TxOutput
    taker_change_output: TxOutput | None

    # Maker data (by nick)
    maker_inputs: dict[str, list[TxInput]]
    maker_cj_outputs: dict[str, TxOutput]
    maker_change_outputs: dict[str, TxOutput]

    # Amounts
    cj_amount: int
    total_maker_fee: int
    tx_fee: int
Attributes
cj_amount: int instance-attribute
maker_change_outputs: dict[str, TxOutput] instance-attribute
maker_cj_outputs: dict[str, TxOutput] instance-attribute
maker_inputs: dict[str, list[TxInput]] instance-attribute
taker_change_output: TxOutput | None instance-attribute
taker_cj_output: TxOutput instance-attribute
taker_inputs: list[TxInput] instance-attribute
total_maker_fee: int instance-attribute
tx_fee: int instance-attribute

Functions

build_coinjoin_tx(taker_utxos: list[dict[str, Any]], taker_cj_address: str, taker_change_address: str, taker_total_input: int, maker_data: dict[str, dict[str, Any]], cj_amount: int, tx_fee: int, network: str = 'mainnet', dust_threshold: int = 27300) -> tuple[bytes, dict[str, Any]]

Build a complete CoinJoin transaction.

Args: taker_utxos: List of taker's UTXOs taker_cj_address: Taker's CJ output address taker_change_address: Taker's change address (empty string if no change needed) taker_total_input: Total value of taker's inputs maker_data: Dict of maker nick -> {utxos, cj_addr, change_addr, cjfee, txfee} cj_amount: Equal CoinJoin output amount tx_fee: Total transaction fee network: Network name dust_threshold: Minimum output value in satoshis (default: 27300)

Returns: (tx_bytes, metadata)

Source code in taker/src/taker/tx_builder.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
def build_coinjoin_tx(
    # Taker data
    taker_utxos: list[dict[str, Any]],
    taker_cj_address: str,
    taker_change_address: str,
    taker_total_input: int,
    # Maker data
    maker_data: dict[str, dict[str, Any]],  # nick -> {utxos, cj_addr, change_addr, cjfee, txfee}
    # Amounts
    cj_amount: int,
    tx_fee: int,
    network: str = "mainnet",
    dust_threshold: int = 27300,  # Default to DUST_THRESHOLD from jmcore.constants
) -> tuple[bytes, dict[str, Any]]:
    """
    Build a complete CoinJoin transaction.

    Args:
        taker_utxos: List of taker's UTXOs
        taker_cj_address: Taker's CJ output address
        taker_change_address: Taker's change address (empty string if no change needed)
        taker_total_input: Total value of taker's inputs
        maker_data: Dict of maker nick -> {utxos, cj_addr, change_addr, cjfee, txfee}
        cj_amount: Equal CoinJoin output amount
        tx_fee: Total transaction fee
        network: Network name
        dust_threshold: Minimum output value in satoshis (default: 27300)

    Returns:
        (tx_bytes, metadata)
    """
    builder = CoinJoinTxBuilder(network)

    # Build taker inputs
    taker_inputs = [
        TxInput.from_hex(
            txid=u["txid"],
            vout=u["vout"],
            value=u["value"],
            scriptpubkey=u.get("scriptpubkey", ""),
        )
        for u in taker_utxos
    ]

    # Calculate taker's fees paid to makers
    total_maker_fee = sum(m["cjfee"] for m in maker_data.values())

    # Taker's change = total_input - cj_amount - maker_fees - tx_fee
    taker_change = taker_total_input - cj_amount - total_maker_fee - tx_fee

    # Taker CJ output
    taker_cj_output = TxOutput.from_address(taker_cj_address, cj_amount)

    # Taker change output (if any)
    taker_change_output = None
    if taker_change > dust_threshold and taker_change_address:
        taker_change_output = TxOutput.from_address(taker_change_address, taker_change)
    elif taker_change > 0:
        logger.warning(
            f"Taker change {taker_change} sats "
            + (
                "has no address (sweep mode)"
                if not taker_change_address
                else f"is below dust threshold ({dust_threshold})"
            )
            + ", no change output will be created"
        )

    # Build maker data
    maker_inputs: dict[str, list[TxInput]] = {}
    maker_cj_outputs: dict[str, TxOutput] = {}
    maker_change_outputs: dict[str, TxOutput] = {}

    for nick, data in maker_data.items():
        # Maker inputs
        maker_inputs[nick] = [
            TxInput.from_hex(
                txid=u["txid"],
                vout=u["vout"],
                value=u["value"],
                scriptpubkey=u.get("scriptpubkey", ""),
            )
            for u in data["utxos"]
        ]

        # Maker CJ output (cj_amount)
        maker_cj_outputs[nick] = TxOutput.from_address(data["cj_addr"], cj_amount)

        # Maker change output
        # Formula: change = inputs - cj_amount - txfee + cjfee
        # (Maker pays txfee, receives cjfee from taker)
        maker_total_input = sum(u["value"] for u in data["utxos"])
        maker_txfee = data.get("txfee", 0)
        maker_change = maker_total_input - cj_amount - maker_txfee + data["cjfee"]

        logger.debug(
            f"Maker {nick} change calculation: "
            f"inputs={maker_total_input}, cj_amount={cj_amount}, "
            f"cjfee={data['cjfee']}, txfee={maker_txfee}, change={maker_change}, "
            f"dust_threshold={dust_threshold}"
        )

        if maker_change < 0:
            # Negative change means maker's UTXOs are insufficient
            # This can happen if UTXO verification failed (value=0) or if UTXOs were spent
            raise ValueError(
                f"Maker {nick} has insufficient funds: inputs={maker_total_input} sats, "
                f"required={cj_amount + maker_txfee - data['cjfee']} sats, "
                f"change={maker_change} sats. Maker's UTXOs may have been spent."
            )
        elif maker_change > dust_threshold:
            maker_change_outputs[nick] = TxOutput.from_address(data["change_addr"], maker_change)
        else:
            logger.warning(
                f"Maker {nick} change {maker_change} sats is below dust threshold "
                f"({dust_threshold}), "
                "no change output will be created"
            )

    tx_data = CoinJoinTxData(
        taker_inputs=taker_inputs,
        taker_cj_output=taker_cj_output,
        taker_change_output=taker_change_output,
        maker_inputs=maker_inputs,
        maker_cj_outputs=maker_cj_outputs,
        maker_change_outputs=maker_change_outputs,
        cj_amount=cj_amount,
        total_maker_fee=total_maker_fee,
        tx_fee=tx_fee,
    )

    return builder.build_unsigned_tx(tx_data)

calculate_tx_fee(num_taker_inputs: int, num_maker_inputs: int, num_outputs: int, fee_rate: float) -> int

Calculate transaction fee based on estimated vsize.

SegWit P2WPKH inputs: ~68 vbytes each P2WPKH outputs: 31 vbytes each Overhead: ~11 vbytes

Args: fee_rate: Fee rate in sat/vB (can be fractional, e.g. 0.5)

Returns: Fee in satoshis (rounded up to ensure minimum relay fee)

Source code in taker/src/taker/tx_builder.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def calculate_tx_fee(
    num_taker_inputs: int,
    num_maker_inputs: int,
    num_outputs: int,
    fee_rate: float,
) -> int:
    """
    Calculate transaction fee based on estimated vsize.

    SegWit P2WPKH inputs: ~68 vbytes each
    P2WPKH outputs: 31 vbytes each
    Overhead: ~11 vbytes

    Args:
        fee_rate: Fee rate in sat/vB (can be fractional, e.g. 0.5)

    Returns:
        Fee in satoshis (rounded up to ensure minimum relay fee)
    """
    # Estimate virtual size
    input_vsize = (num_taker_inputs + num_maker_inputs) * 68
    output_vsize = num_outputs * 31
    overhead = 11

    vsize = input_vsize + output_vsize + overhead

    # Round up to ensure we pay at least the minimum
    import math

    return math.ceil(vsize * fee_rate)