从0开始搭建自己的TRON钱包工具（批量生成地址/USDT转账/质押投票）

一些官方的虚拟货币钱包总是有各种限制，比如没法质押，没法投票，没法自由操作，有时候我们并不需要一个“功能很全但很重”的钱包应用，只需要把自己常用的 TRON 操作自动化并且可以自由扩展功能，完全有自己控制私钥,那么本文将会分享3个主要的相关脚本：

批量生成钱包地址，eth和tron网络
快速归集 USDT/TRX
质押、解质押、投票、委托能量

我这里用 3 个 Python 脚本，做了一个可直接跑的轻量工具链：

generate_wallet.py：多进程批量生成私钥/公钥/ETH地址/TRON地址
wallet_transfer.py：查询余额并转账 USDT + TRX
wallet_manipulate.py：质押、解质押、提取、领奖励、投票、能量委托

下面按可执行流程整理一遍。

零、区块链与 TRON 网络简介

区块链本质上是一个去中心化账本——每一笔转账、合约调用、质押投票都会打包成链上交易，全网节点验证并永久记录，任何人都可以公开查询。

TRON 网络有两个值得关注的特性：

确认速度快：出块间隔约 3 秒，链上交易通常数秒内完成确认。
资源费用模型：TRON 引入了 Energy 和 Bandwidth 两种资源来支付交易费用。TRX 直接转账消耗 Bandwidth，TRC20 合约调用（如 USDT 转账）消耗 Energy。资源不足时系统会燃烧 TRX 来补足。因此做钱包脚本时，除了"余额够不够"，还必须关注"资源够不够"，这也是本文把转账脚本和资源管理脚本拆开的原因。

TRON 地址生成原理：TRON 地址和以太坊地址共享同一套椭圆曲线（secp256k1）私钥/公钥体系，区别仅在于地址编码：以太坊取 keccak256(公钥) 后 20 字节加 0x 前缀，TRON 则在同样的 20 字节前加 0x41 再做 Base58Check 编码。因此同一个私钥可以同时对应一个 ETH 地址和一个 TRON 地址。

一、环境准备

1. Python 依赖

pip install tronpy coincurve base58 eth-hash

2. 脚本说明

你需要准备自己的 TRON 节点/TronGrid API Key。
脚本里涉及真实私钥，建议只在安全环境中运行。
不要把私钥、API Key 提交到公开仓库。

二、批量生成 TRON 地址（generate_wallet.py）

这个脚本是“地址工厂”：

用 os.urandom(32) 生成私钥
用 coincurve 推导公钥
用 keccak 取后20字节得到 ETH 地址
在 ETH 地址前拼接 0x41 并做 base58check 得到 TRON 地址

同时脚本用多进程跑满 CPU，速度很高，适合做批量地址池。

1. 运行示例

python generate_wallet.py 1000000

如果不传参数，默认生成 1000000 个地址。30s就可以生成100万的地址，其实eth和tron可以共用私钥。

2. 输出目录与结果

输出目录格式类似：

wallet-tool/dkg_batch/1713920000000_T16_N1000000_wallets/

每个 worker 一个 csv 文件，单行格式：

私钥hex,公钥hex,ETH地址,TRON地址

3. 注意点

当前脚本按 total // cpu 分配，余数会被丢掉（比如 1000003 在 16核下会少几个）。
如果你要求“严格生成 N 个”，可以补一个余数 worker。
生成后的 csv 一定要加密保存，这是高风险敏感数据。

4. 完整脚本（generate_wallet.py）

#!/usr/bin/env python3

import os
import time
import base58
from eth_hash.auto import keccak
import multiprocessing as mp
from coincurve import PrivateKey


BASE_DIR = "wallet-tool/dkg_batch"
DEFAULT_COUNT = 1000000


def generate_batch(args):

  count, worker_id, outdir = args

  filepath = os.path.join(outdir, f"wallets_{worker_id}.csv")

  with open(filepath, "w", buffering=1024*1024) as f:

    for _ in range(count):

      priv = os.urandom(32)
      pk = PrivateKey(priv)

      pub = pk.public_key.format(compressed=False)[1:]

      # ETH address
      eth_addr = keccak(pub)[-20:]
      eth = "0x" + eth_addr.hex()

      # TRON address
      tron_bytes = b'\x41' + eth_addr
      tron = base58.b58encode_check(tron_bytes).decode()

      f.write(
        f"{priv.hex()},{pub.hex()},{eth},{tron}\n"
      )


def create_output_dir(total):

  ts = int(time.time()*1000)

  folder = f"{ts}_T{mp.cpu_count()}_N{total}_wallets"

  path = os.path.join(BASE_DIR, folder)

  os.makedirs(path, exist_ok=True)

  return path


def main():

  import sys

  total = int(sys.argv[1]) if len(sys.argv) > 1 else DEFAULT_COUNT

  cpu = mp.cpu_count()

  outdir = create_output_dir(total)

  per_worker = total // cpu

  print(f"Generating {total} wallets")
  print(f"CPU cores: {cpu}")
  print(f"Output dir: {outdir}")
  print()

  start = time.time()

  args = [(per_worker, i, outdir) for i in range(cpu)]

  with mp.Pool(cpu) as p:
    p.map(generate_batch, args)

  elapsed = time.time() - start

  print("Done")
  print(f"Time: {elapsed:.2f} seconds")
  print(f"Speed: {int(total/elapsed):,} wallets/sec")


if __name__ == "__main__":
  main()

三、钱包归集与转账（wallet_transfer.py）

这个脚本做了三件事：

根据私钥推导发送地址
查询 USDT(TRC20) 余额 + TRX 余额 + 能量/带宽
交互式输入金额并发送 USDT、TRX

1. 先修改配置

脚本里需要确认以下参数：

TO_ADDRESS：归集地址（收款地址）
USDT_CONTRACT：主网 USDT 合约（地址是 TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t）
api_key：你自己的 TronGrid API Key

2. 运行示例

python wallet_transfer.py <PRIVATE_KEY_HEX>

运行后会先打印：

Sender
USDT Balance
TRX Balance
Energy Available
Bandwidth Available

然后按提示输入发送金额。

3. 手续费与资源说明

TRC20 USDT 转账需要能量，能量不足会消耗 TRX。
脚本里设置了 fee_limit(30_000_000)，即允许消耗一定上限的 TRX 作为手续费。
转账成功会输出 tx id 与 tronscan 链接，方便核验。

4. 完整脚本（wallet_transfer.py）

#!env python
from tronpy import Tron
from tronpy.keys import PrivateKey
from tronpy.providers import HTTPProvider
import sys

# ===== CONFIG =====
PRIVATE_KEY = sys.argv[1]
TO_ADDRESS = 'TF4YTndBMXGEB7EDz9G4L8WBPUi8PQjwtn' #替换成你的地址

CHAIN = "mainnet"

USDT_CONTRACT = "TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t"
DECIMALS = 6

# ===== CONNECT =====
client = Tron(provider=HTTPProvider(
  endpoint_uri="https://api.trongrid.io",
  api_key="765dd126-xxxx-48a7-a566-xxxxxxxxxx" #替换成你自己的api
))

private_key = PrivateKey(bytes.fromhex(PRIVATE_KEY))
sender = private_key.public_key.to_base58check_address()
print("Sender:", sender)

contract = client.get_contract(USDT_CONTRACT)

# ===== GET BALANCES =====
usdt_balance = contract.functions.balanceOf(sender)
human_usdt_balance = usdt_balance / (10 ** DECIMALS)
trx_balance = client.get_account_balance(sender)
print(f"USDT Balance: {human_usdt_balance}")
print(f"TRX Balance: {trx_balance}")

# Energy info
account_resource = client.get_account_resource(sender)
energy_available = account_resource.get("EnergyLimit", 0) - account_resource.get("EnergyUsed", 0)
bandwidth_available = account_resource.get("FreeNetLimit", 0) - account_resource.get("FreeNetUsed", 0)
print(f"Energy Available: {energy_available}")
print(f"Bandwidth Available: {bandwidth_available}")

# ===== SEND USDT =====
if usdt_balance > 0:
  try:
    usdt_input = float(input(f"Enter USDT amount to send (max {usdt_balance}): "))
  except ValueError:
    print("Invalid amount")
    usdt_input = 0

  if usdt_input > 0:
    if usdt_input > usdt_balance:
      print("Amount exceeds USDT balance")
    else:
      usdt_amount_in_units = int(usdt_input * (10 ** DECIMALS))
      try:
        txn = (
          contract.functions.transfer(TO_ADDRESS, usdt_amount_in_units)
          .with_owner(sender)
          .fee_limit(30_000_000)  # Use TRX to pay if no energy
          .build()
          .sign(private_key)
        )
        result = txn.broadcast().wait()
        tx_id = result["id"]
        print(f"Sent {usdt_input} USDT successfully")
        print("TX:", "https://tronscan.org/#/transaction/" + tx_id)
      except Exception as e:
        print("Error sending USDT:", e)
else:
  print("No USDT to send, skipping...")

# ===== SEND TRX =====
try:
  trx_input = float(input(f"Enter TRX amount to send (max {trx_balance}): "))
except ValueError:
  print("Invalid amount")
  trx_input = 0

if trx_input > 0:
  if trx_input > trx_balance:
    print("Amount exceeds TRX balance")
  else:
    try:
      trx_tx = (
        client.trx.transfer(sender, TO_ADDRESS, int(trx_input * 1_000_000))
        .build()
        .sign(private_key)
      )
      result = trx_tx.broadcast().wait()
      tx_id = result["id"]
      print(f"Sent {trx_input} TRX successfully")
      print("TX:", "https://tronscan.org/#/transaction/" + tx_id)
    except Exception as e:
      print("Error sending TRX:", e)
else:
  print("No TRX to send")

四、质押/投票/委托（wallet_manipulate.py）

这个脚本相当于一个 CLI 操作台，支持以下动作：

stake：质押 TRX（默认资源是 ENERGY，用于usdt转账）
unstake：解质押
withdraw：提取已解锁质押
claim：领取奖励
vote：投票给 SR(可以获得trx奖励)
delegate：委托资源（租借）
undelegate：撤销委托

1. 运行示例

# 质押 100 TRX
python wallet_manipulate.py <PRIVATE_KEY_HEX> stake 100

# 解质押 50 TRX
python wallet_manipulate.py <PRIVATE_KEY_HEX> unstake 50

# 提取解锁后余额
python wallet_manipulate.py <PRIVATE_KEY_HEX> withdraw

# 领取奖励
python wallet_manipulate.py <PRIVATE_KEY_HEX> claim

# 投票（示例：两个SR）
python wallet_manipulate.py <PRIVATE_KEY_HEX> vote TXXX...:10 TYYY...:20

# 委托能量
python wallet_manipulate.py <PRIVATE_KEY_HEX> delegate TReceiver...

# 撤销委托
python wallet_manipulate.py <PRIVATE_KEY_HEX> undelegate TReceiver...

2. 脚本里的校验逻辑

质押前会检查 TRX 余额是否足够。
投票前会读取 tron_power（或回退 frozenV2 计算）做额度校验。
投票参数必须是 <SR_ADDRESS>:<VOTES> 格式。
超额投票会直接报错并阻止广播。

3. 资源参数

脚本顶部的 RESOURCE 决定质押兑换哪种资源，根据需要调整：

ENERGY：用于 TRC20 合约调用（如 USDT 转账），一次约消耗 65,000 Energy。
BANDWIDTH：用于普通 TRX 转账，一次约消耗 300 Bandwidth。
注意：质押操作本身也消耗少量 Bandwidth，若账户 Bandwidth 不足会直接烧 TRX 支付。

RESOURCE = "ENERGY"   # 默认：用于 USDT 转账
# RESOURCE = "BANDWIDTH"  # 切换为带宽模式

4. 完整脚本（wallet_manipulate.py）

#!env python

from tronpy import Tron
from tronpy.keys import PrivateKey
from tronpy.providers import HTTPProvider
import sys

#https://tronscan.org/#/sr
# ===== CONFIG =====
TRONGRID_API_KEY = "765dd126-6d1d-xxxx-a566-xxxxxxxxxx"
CHAIN = "mainnet"
RESOURCE = "ENERGY"
#RESOURCE = "BANDWIDTH"

# ===== USAGE CHECK =====
if len(sys.argv) < 3:
  print("\nUsage:")
  print("  python tron_stake.py <PRIVATE_KEY> stake <amount>")
  print("  python tron_stake.py <PRIVATE_KEY> unstake <amount>")
  print("  python tron_stake.py <PRIVATE_KEY> withdraw")
  print("  python tron_stake.py <PRIVATE_KEY> claim")
  print("  python tron_stake.py <PRIVATE_KEY> vote <SR_ADDRESS>:<VOTES> [...]")
  sys.exit()

PRIVATE_KEY = sys.argv[1]
ACTION = sys.argv[2].lower()

# ===== CONNECT =====
provider = HTTPProvider(api_key=TRONGRID_API_KEY)
client = Tron(provider=provider)

private_key = PrivateKey(bytes.fromhex(PRIVATE_KEY))
sender = private_key.public_key.to_base58check_address()
print("\nSender:", sender)

# Show TRX balance
trx_balance = client.get_account_balance(sender)
print("TRX Balance:", trx_balance)

# =========================
# ===== STAKE TRX =====
# =========================
if ACTION == "stake":
  if len(sys.argv) < 4:
    raise Exception("Enter amount to stake")

  amount_trx = float(sys.argv[3])
  amount_sun = int(amount_trx * 1_000_000)

  if amount_trx > trx_balance:
    raise Exception("Not enough TRX")

  txn = (
    client.trx.freeze_balance(
      owner=sender,
      amount=amount_sun,
      resource=RESOURCE
    )
    .build()
    .sign(private_key)
  )

  result = txn.broadcast().wait()
  print("\nStaked", amount_trx, "TRX for", RESOURCE)
  print("TX ID:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])

# =========================
# ===== UNSTAKE TRX =====
# =========================
elif ACTION == "unstake":
  if len(sys.argv) < 4:
    raise Exception("Enter amount to unstake")

  amount_trx = float(sys.argv[3])
  amount_sun = int(amount_trx * 1_000_000)

  txn = (
    client.trx.unfreeze_balance(
      owner=sender,
      resource=RESOURCE,
      unfreeze_balance=amount_sun
    )
    .build()
    .sign(private_key)
  )

  result = txn.broadcast().wait()
  print("\nUnstaking initiated:", amount_trx, "TRX")
  print("Lock period: ~3 days (Stake 1.0)")
  print("TX ID:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])

# =========================
# ===== WITHDRAW TRX =====
# =========================
elif ACTION == "withdraw":

  txn = (
    client.trx.withdraw_stake_balance(sender)
    .build()
    .sign(private_key)
  )

  result = txn.broadcast().wait()
  print("\nWithdrawn unlocked TRX")
  print("TX ID:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])

# =========================
# ===== CLAIM REWARDS =====
# =========================
elif ACTION == "claim":

  txn = (
    client.trx.withdraw_rewards(sender)
    .build()
    .sign(private_key)
  )

  result = txn.broadcast().wait()
  print("\nRewards claimed!")
  print("TX ID:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])
# =========================
# ===== VOTE FOR SR =====
# =========================
elif ACTION == "vote":
  if len(sys.argv) < 4:
    raise Exception("Enter at least one SR vote in the format <SR_ADDRESS>:<VOTES>")

  # ===== GET ACCOUNT =====
  account_info = client.get_account(sender)

  print("\n=== ACCOUNT INFO ===")
  print(account_info)

  # ===== TRON POWER (Stake 2.0 safe) =====
  tron_power = account_info.get("tron_power", 0) / 1_000_000

  if tron_power == 0:
    tron_power = sum(
      item.get("amount", 0)
      for item in account_info.get("frozenV2", [])
      if isinstance(item, dict) and "amount" in item
    ) / 1_000_000

  print(f"\nTRON Power available: {tron_power}")

  if tron_power <= 0:
    raise Exception("No TRON Power available. Freeze TRX first to vote.")

  # ===== PARSE VOTES SAFELY =====
  votes_input = sys.argv[3:]
  votes = []
  total_votes = 0

  for v in votes_input:
    parts = v.split(":")

    if len(parts) != 2:
      raise Exception(f"Invalid format: {v} (use <SR_ADDRESS>:<VOTES>)")

    sr_address = parts[0].strip()

    try:
      vote_count = int(parts[1])
    except ValueError:
      raise Exception(f"Invalid vote number in: {v}")

    if vote_count <= 0:
      raise Exception(f"Vote count must be > 0: {v}")

    votes.append((sr_address, vote_count))
    total_votes += vote_count

  print("\nVotes parsed:", votes)
  print("Total votes requested:", total_votes)

  # ===== VALIDATION =====
  if total_votes > tron_power:
    raise Exception(
      f"Total votes ({total_votes}) exceed TRON Power ({tron_power})"
    )

  # ===== BUILD TX (tronpy expects tuple list) =====
  votes = [(str(a), int(b)) for a, b in votes]
  print("\nVotes final:", votes)
  txn = client.trx.vote_witness(sender, *votes).build().sign(private_key)

  # ===== BROADCAST =====
  result = txn.broadcast().wait()

  print("\nVote broadcasted successfully!")
  print("TX ID:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])

elif ACTION == "delegate":
  if len(sys.argv) < 3:
    raise Exception("Usage: delegate <receiver_address>")

  receiver = sys.argv[3]

  amount_trx = 5000
  amount_sun = int(amount_trx * 1_000_000)

  txn = (
    client.trx.delegate_resource(
      owner=sender,
      receiver=receiver,
      balance=amount_sun,
      resource=RESOURCE  # ENERGY
    )
    .build()
    .sign(private_key)
  )

  result = txn.broadcast().wait()

  print("\nEnergy delegated")
  print("To:", receiver)
  print("TRX staked:", amount_trx)
  print("TX:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])

elif ACTION == "undelegate":
  if len(sys.argv) < 3:
    raise Exception("Usage: undelegate <receiver_address>")

  receiver = sys.argv[3]

  amount_trx = 5000
  amount_sun = int(amount_trx * 1_000_000)

  txn = (
    client.trx.undelegate_resource(
      owner=sender,
      receiver=receiver,
      balance=amount_sun,
      resource=RESOURCE  # ENERGY
    )
    .build()
    .sign(private_key)
  )

  result = txn.broadcast().wait()

  print("\nEnergy undelegated")
  print("From:", receiver)
  print("TRX unstaked:", amount_trx)
  print("TX:", result["id"])
  print("https://tronscan.org/#/transaction/" + result["id"])

else:
  print("Invalid action. Use: stake | unstake | withdraw | vote")

五、建议增加的安全与工程化改造

如果你要长期使用，建议加这几项：

把私钥、API Key 改成环境变量读取。
转账前增加二次确认（地址、金额、手续费预估）。
生成地址文件加密（如 GPG 或磁盘加密）。
增加批量归集模式（读取 csv 自动归集尘埃资产）。
日志分级 + 失败重试（网络波动时更稳）。

六、常见问题

1) USDT 有余额但转账失败

常见原因：能量不足、fee_limit 太低、节点限流。

排查顺序：

先看 TRX 余额是否足够支付手续费。
查询账户资源（Energy/FreeNet）。
更换节点或提高 fee_limit。

2) 投票报“无 TRON Power”

说明账户还没质押，或者质押状态未生效。

先执行 stake，确认链上生效后再投票。

3) 私钥格式错误

脚本需要 64位 hex 私钥（不带 0x）,直接从生成的csv文件复制粘贴就行。

总结

这套脚本其实已经覆盖了一个 TRON 钱包的核心闭环：

地址生成（generate_wallet.py）
资产转账归集（wallet_transfer.py）
资源与治理操作（wallet_manipulate.py）

如果你和我一样，目标是“可控、可批量、可自动化”，这种轻量 Python 方案会比传统钱包 UI 更适合做生产脚本。

后续我会继续补一个“批量归集脚本 + 风险控制版”，把地址池资金一键归拢到主钱包。

从0开始搭建自己的TRON钱包工具（批量生成地址/USDT转账/质押投票）

零、区块链与 TRON 网络简介

一、环境准备

1. Python 依赖

2. 脚本说明

二、批量生成 TRON 地址（generate_wallet.py）

1. 运行示例

2. 输出目录与结果

3. 注意点

4. 完整脚本（generate_wallet.py）

三、钱包归集与转账（wallet_transfer.py）

1. 先修改配置

2. 运行示例

3. 手续费与资源说明

4. 完整脚本（wallet_transfer.py）

四、质押/投票/委托（wallet_manipulate.py）

1. 运行示例

2. 脚本里的校验逻辑

3. 资源参数

4. 完整脚本（wallet_manipulate.py）

五、建议增加的安全与工程化改造

六、常见问题

1) USDT 有余额但转账失败

2) 投票报“无 TRON Power”

3) 私钥格式错误

总结

相关文章

最新文章

分类

标签

友情链接

其它