Simple script to predict Cardano TX hash given inputs, outputs, change address and TTL. Requires Blockfrost key to fetch protocol params and local `cardano-cli`

predict-tx-hash.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = ["typer", "httpx"]
# ///
"""
Predict the hash of a Cardano transaction.

Usage:
  uv run scripts/predict-tx-hash.py \
    --input  "txhash#index:lovelace" \
    --input  "txhash#index:lovelace" \
    --output "address:lovelace" \
    --change-address "addr1..." \
    --cardano-cli cardano-cli \
    --ttl 186077892

BLOCKFROST_API_KEY must be set in the environment (or passed via --blockfrost-key).
"""

import json
import subprocess
import tempfile
from pathlib import Path
from typing import Annotated

import httpx
import typer

app = typer.Typer(add_completion=False)


def _bf_base(api_key: str) -> str:
    network = "mainnet" if api_key.startswith("mainnet") else "preprod"
    return f"https://cardano-{network}.blockfrost.io/api/v0"


def fetch_protocol_params(api_key: str, out: Path) -> None:
    r = httpx.get(f"{_bf_base(api_key)}/epochs/latest/parameters",
                  headers={"project_id": api_key}, timeout=30)
    r.raise_for_status()
    p = r.json()

    def n(field: str, fallback: int = 0) -> int | float:
        v = p.get(field)
        return (int(v) if isinstance(v, (int, str)) else float(v)) if v is not None else fallback

    params = {
        "txFeePerByte": p.get("min_fee_a"),
        "txFeeFixed": p.get("min_fee_b"),
        "maxTxSize": p.get("max_tx_size"),
        "maxBlockBodySize": p.get("max_block_size"),
        "maxBlockHeaderSize": p.get("max_block_header_size"),
        "stakeAddressDeposit": n("key_deposit"),
        "stakePoolDeposit": n("pool_deposit"),
        "poolRetireMaxEpoch": p.get("e_max"),
        "stakePoolTargetNum": p.get("n_opt"),
        "poolPledgeInfluence": p.get("a0"),
        "monetaryExpansion": p.get("rho"),
        "treasuryCut": p.get("tau"),
        "minPoolCost": n("min_pool_cost"),
        "utxoCostPerByte": n("coins_per_utxo_size"),
        "maxValueSize": n("max_val_size"),
        "collateralPercentage": p.get("collateral_percent"),
        "maxCollateralInputs": p.get("max_collateral_inputs"),
        "protocolVersion": {"major": p.get("protocol_major_ver"), "minor": p.get("protocol_minor_ver")},
        "executionUnitPrices": {"priceMemory": p.get("price_mem"), "priceSteps": p.get("price_step")},
        "maxTxExecutionUnits": {"memory": n("max_tx_ex_mem"), "steps": n("max_tx_ex_steps")},
        "maxBlockExecutionUnits": {"memory": n("max_block_ex_mem"), "steps": n("max_block_ex_steps")},
        "costModels": p.get("cost_models_raw"),
        "poolVotingThresholds": {
            "motionNoConfidence": n("pvt_motion_no_confidence"),
            "committeeNormal": n("pvt_committee_normal"),
            "committeeNoConfidence": n("pvt_committee_no_confidence"),
            "hardForkInitiation": n("pvt_hard_fork_initiation"),
            "ppSecurityGroup": n("pvt_pp_security_group"),
        },
        "dRepVotingThresholds": {
            "motionNoConfidence": n("dvt_motion_no_confidence"),
            "committeeNormal": n("dvt_committee_normal"),
            "committeeNoConfidence": n("dvt_committee_no_confidence"),
            "updateToConstitution": n("dvt_update_to_constitution"),
            "hardForkInitiation": n("dvt_hard_fork_initiation"),
            "ppNetworkGroup": n("dvt_pp_network_group"),
            "ppEconomicGroup": n("dvt_pp_economic_group"),
            "ppTechnicalGroup": n("dvt_pp_technical_group"),
            "ppGovGroup": n("dvt_pp_gov_group"),
            "treasuryWithdrawal": n("dvt_treasury_withdrawal"),
        },
        "committeeMinSize": n("committee_min_size"),
        "committeeMaxTermLength": n("committee_max_term_length"),
        "govActionLifetime": n("gov_action_lifetime"),
        "govActionDeposit": n("gov_action_deposit"),
        "dRepDeposit": n("drep_deposit"),
        "dRepActivity": n("drep_activity"),
        "minFeeRefScriptCostPerByte": n("min_fee_ref_script_cost_per_byte"),
        "decentralization": None,
        "extraPraosEntropy": None,
        "minUTxOValue": None,
    }
    out.write_text(json.dumps(params, indent=2))


def _run(cmd: str) -> str:
    result = subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
    return result.stdout.strip()


def build_raw(
    cli: str,
    inputs: list[dict],
    payment_outputs: list[dict],
    change_address: str,
    total_in: int,
    fee: int,
    ttl: int,
    tx_file: Path,
) -> None:
    change = total_in - sum(int(o["lovelace"]) for o in payment_outputs) - fee
    if change < 0:
        raise ValueError(f"Insufficient inputs: have {total_in}, need {total_in - change}")

    tx_ins  = " ".join(f'--tx-in "{i["txHash"]}#{i["outputIndex"]}"' for i in inputs)
    tx_outs = " ".join(f'--tx-out "{o["address"]}+{o["lovelace"]}"' for o in payment_outputs)
    tx_outs += f' --tx-out "{change_address}+{change}"'

    _run(f"{cli} babbage transaction build-raw "
         f"{tx_ins} {tx_outs} "
         f"--invalid-hereafter {ttl} --fee {fee} --out-file {tx_file}")


def calculate_min_fee(cli: str, tx_file: Path, protocol_file: Path, witness_count: int) -> int:
    out = _run(f"{cli} babbage transaction calculate-min-fee "
               f"--tx-body-file {tx_file} "
               f"--protocol-params-file {protocol_file} "
               f"--witness-count {witness_count}")
    try:
        return int(json.loads(out)["fee"])
    except (json.JSONDecodeError, KeyError):
        return int(out.split()[0])


def converge_fee(
    cli: str,
    inputs: list[dict],
    payment_outputs: list[dict],
    change_address: str,
    total_in: int,
    ttl: int,
    witness_count: int,
    tx_file: Path,
    protocol_file: Path,
) -> int:
    fee = 200_000
    for pass_num in range(1, 10):
        build_raw(cli, inputs, payment_outputs, change_address, total_in, fee, ttl, tx_file)
        new_fee = calculate_min_fee(cli, tx_file, protocol_file, witness_count)
        typer.echo(f"  pass {pass_num}: {new_fee} lovelace", err=True)
        if new_fee == fee:
            return fee
        fee = new_fee
    return fee


@app.command()
def predict(
    input: Annotated[list[str], typer.Option(
        "--input", "-i", help="UTxO input: txhash#index:lovelace (repeat for multiple)")],
    output: Annotated[list[str], typer.Option(
        "--output", "-o", help="Payment output: address:lovelace (repeat for multiple)")],
    change_address: Annotated[str, typer.Option("--change-address", "-c", help="Change address")],
    ttl: Annotated[int, typer.Option("--ttl", "-t", help="Absolute slot (invalid-hereafter)")],
    cardano_cli: Annotated[str, typer.Option("--cardano-cli", help="Path to cardano-cli binary")] = "cardano-cli",
    blockfrost_key: Annotated[str | None, typer.Option(
        "--blockfrost-key", "-k", envvar="BLOCKFROST_API_KEY",
        help="Blockfrost project ID (or set BLOCKFROST_API_KEY)")] = None,
    witness_count: Annotated[int, typer.Option("--witness-count", "-w", help="Number of signing keys")] = 1,
) -> None:
    if not blockfrost_key:
        typer.echo("Error: provide --blockfrost-key or set BLOCKFROST_API_KEY", err=True)
        raise typer.Exit(1)

    inputs: list[dict] = []
    total_in = 0
    for raw in input:
        ref, lovelace = raw.rsplit(":", 1)
        tx_hash, idx = ref.split("#")
        inputs.append({"txHash": tx_hash, "outputIndex": int(idx)})
        total_in += int(lovelace)

    payment_outputs = [{"address": a, "lovelace": v}
                       for a, v in (o.rsplit(":", 1) for o in output)]

    with tempfile.TemporaryDirectory() as tmpdir:
        tmp = Path(tmpdir)
        protocol_file = tmp / "protocol.json"
        tx_file = tmp / "tx.draft"

        typer.echo("Fetching protocol parameters...", err=True)
        fetch_protocol_params(blockfrost_key, protocol_file)

        typer.echo("Calculating fee...", err=True)
        fee = converge_fee(cardano_cli, inputs, payment_outputs, change_address,
                           total_in, ttl, witness_count, tx_file, protocol_file)

        tx_hash = _run(f"{cardano_cli} babbage transaction txid --tx-body-file {tx_file}")
        change = total_in - sum(int(o["lovelace"]) for o in payment_outputs) - fee

    all_outputs = [*payment_outputs, {"address": change_address, "lovelace": str(change)}]

    typer.echo(f"fee={fee} change={change} hash={tx_hash}")
    typer.echo("utxos:")
    for idx, out in enumerate(all_outputs):
        typer.echo(f"  {tx_hash}#{idx}:{out['lovelace']}  ({out['address']})")


if __name__ == "__main__":
    app()