4.RGB 协议开发实战指南

┌─────────────────────────────────────┐
│           应用层 (Python)            │  ← 业务逻辑、UI、自动化
├─────────────────────────────────────┤
│        RGB 协议层 (RGB CLI)          │  ← 状态管理、验证、承诺
├─────────────────────────────────────┤
│       承诺层 (Tapret/Opret)          │  ← 密码学承诺、锚定
├─────────────────────────────────────┤
│      比特币层 (Bitcoin Core)         │  ← UTXO、共识、安全性
└─────────────────────────────────────┘

def derive_wif_from_tprv(tprv: str, branch: int = 10, index: int = 1) -> str:
    """从主私钥派生指定路径的 WIF 私钥"""
    from bip32 import BIP32, HARDENED_INDEX
    import base58, hashlib
 
    # 构建 BIP32 对象
    bip32 = BIP32.from_xpriv(tprv)
 
    # RGB 使用的标准路径: m/86'/1'/0'/10/*
    derivation_path = [
        86 | HARDENED_INDEX,    # BIP86 (Taproot)
        1 | HARDENED_INDEX,     # Testnet
        0 | HARDENED_INDEX,     # Account 0
        branch,                 # Branch (通常为 10)
        index                   # Index
    ]
 
    # 派生私钥
    private_key = bip32.get_privkey_from_path(derivation_path)
 
    # 转换为 testnet WIF 格式
    payload = b"\\xEF" + private_key + b"\\x01"
    checksum = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4]
    return base58.b58encode(payload + checksum).decode()
 

def process_psbt_pipeline(psbt_b64: str, wif: str) -> dict:
    """完整的 PSBT 处理流程"""
 
    # 步骤 1: 创建 legacy 钱包 (兼容性考虑)
    ensure_legacy_wallet("alice_legacy")
 
    # 步骤 2: 导入私钥
    import_wif(wif, "alice_legacy", label="rgb-taproot-key")
 
    # 步骤 3: 处理 PSBT (添加签名信息)
    processed = bitcoin_cli([
        "walletprocesspsbt", psbt_b64, "true"
    ], wallet="alice_legacy")
 
    # 步骤 4: 完成 PSBT (提取最终交易)
    finalized = bitcoin_cli([
        "-named", "finalizepsbt",
        f"psbt={processed['psbt']}", "extract=true"
    ])
 
    return {
        "success": finalized.get("complete", False),
        "hex": finalized.get("hex"),
        "txid": None  # 需要广播后获得
    }
 

class RGBClient:
    def __init__(self, wallet_dir: str, network: str = "regtest"):
        self.wallet_dir = wallet_dir
        self.network = network
        self.esplora = "<http://localhost:3002>"
 
    def execute_command(self, args: list) -> dict:
        """执行 RGB 命令"""
        cmd = [
            "rgb", "-d", self.wallet_dir,
            "-n", self.network
        ] + args + [f"--esplora={self.esplora}"]
 
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=60
        )
 
        return {
            "success": result.returncode == 0,
            "output": result.stdout.strip(),
            "error": result.stderr if result.returncode != 0 else None
        }
 
    def create_invoice(self, contract_id: str, amount: int) -> str:
        """创建 RGB 发票"""
        result = self.execute_command([
            "invoice", contract_id, "--amount", str(amount)
        ])
        if result["success"]:
            return result["output"]
        raise Exception(f"Invoice creation failed: {result['error']}")
 
    def transfer(self, invoice: str, consignment_file: str, psbt_file: str):
        """创建 RGB 转账"""
        return self.execute_command([
            "transfer", invoice, consignment_file, psbt_file
        ])
 

def parse_rgb_state(output: str) -> dict:
    """解析 RGB 状态输出"""
    import re
 
    states = []
    lines = output.split('\\n')
 
    for line in lines:
        # 匹配状态行格式
        # "    99999997500    bc:tapret1st:...    bc:witness_tx (status)"
        match = re.match(
            r'\\s+(\\d+)\\s+(bc:tapret1st:[a-f0-9:]+)\\s+(bc:[a-f0-9]+)\\s*\\((.*?)\\)',
            line
        )
 
        if match:
            amount, seal, witness_tx, status = match.groups()
            states.append({
                "amount": int(amount),
                "seal": seal,
                "witness_tx": witness_tx.replace("bc:", ""),
                "status": "confirmed" if "bitcoin:" in status else "tentative",
                "block_height": extract_block_height(status)
            })
 
    return {
        "total_balance": sum(s["amount"] for s in states),
        "states": states,
        "active_states": [s for s in states if s["status"] == "tentative"]
    }
 

{
  "project_info": {
    "name": "RGB Transfer Toolkit",
    "version": "1.0.0",
    "compatible_clients": ["bitlight", "rgb-lib", "bitmask"]
  },
  "rgb_config": {
    "alice_dir": "/path/to/.alice",
    "bob_dir": "/path/to/.bob",
    "contract_id": "rgb:BppYGUUL-Qboz3UD-czwAaVV-!!Jkr1a-SE1!m1f-Cz$b0xs",
    "network": "regtest",
    "esplora_url": "<http://localhost:3002>"
  },
  "bitcoin_config": {
    "docker_project": "bitlight-local-env",
    "wallet_name": "alice_legacy",
    "rpc_url": "<http://localhost:18443>"
  }
}
 

class BitlightIntegration:
    """与 Bitlight 客户端的集成接口"""
 
    def sync_wallet_state(self, wallet_dir: str):
        """同步钱包状态到 Bitlight"""
        # 读取 RGB 状态
        rgb_client = RGBClient(wallet_dir)
        state = rgb_client.get_state(contract_id)
 
        # 转换为 Bitlight 格式
        bitlight_format = {
            "assets": [
                {
                    "contract_id": contract_id,
                    "balance": state["total_balance"],
                    "precision": "centiMicro",
                    "ticker": "TEST"
                }
            ],
            "transactions": self.format_transactions(state["states"])
        }
 
        return bitlight_format
 
    def export_consignment(self, consignment_file: str) -> dict:
        """导出 consignment 供 Bitlight 使用"""
        with open(consignment_file, 'rb') as f:
            data = f.read()
 
        return {
            "format": "rgb_consignment",
            "version": "0.10",
            "data": base64.b64encode(data).decode(),
            "size": len(data)
        }
 

class RGBTransferOrchestrator:
    """RGB 转账流程编排器"""
 
    def __init__(self, config_path: str):
        self.config = self.load_config(config_path)
        self.alice = RGBClient(self.config["rgb_config"]["alice_dir"])
        self.bob = RGBClient(self.config["rgb_config"]["bob_dir"])
 
    async def execute_transfer(self, amount: int) -> dict:
        """执行完整转账流程"""
 
        # 第 1 步: Bob 生成发票
        invoice = self.bob.create_invoice(
            self.config["rgb_config"]["contract_id"],
            amount
        )
 
        # 第 2 步: Alice 创建转账
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        consignment_file = f"transfer_{timestamp}.consignment"
        psbt_file = f"transfer_{timestamp}.psbt"
 
        transfer_result = self.alice.transfer(
            invoice, consignment_file, psbt_file
        )
 
        # 第 3 步: Bob 验证并接受
        validation = self.bob.validate(consignment_file)
        if not validation["success"]:
            raise Exception("Consignment validation failed")
 
        acceptance = self.bob.accept(consignment_file)
        if not acceptance["success"]:
            raise Exception("Transfer acceptance failed")
 
        # 第 4 步: 签名并广播 PSBT
        broadcast_result = await self.sign_and_broadcast_psbt(psbt_file)
 
        # 第 5 步: 智能 finalize (处理已知 bug)
        finalize_result = await self.smart_finalize(
            broadcast_result, consignment_file
        )
 
        # 第 6 步: 验证最终状态
        verification = self.verify_transfer_completion(amount)
 
        return {
            "success": verification["success"],
            "txid": broadcast_result.get("txid"),
            "amount": amount,
            "verification": verification
        }
 
    async def smart_finalize(self, broadcast_result: dict, consignment_file: str):
        """智能 finalize - 处理 RGB CLI bug"""
 
        # 尝试标准 finalize
        try:
            return self.alice.finalize_transfer(broadcast_result["signed_psbt_file"])
        except RGBFinalizeError:
            # 回退到状态验证
            return self.verify_state_without_finalize()
 

class RGBErrorHandler:
    """RGB 特定错误处理"""
 
    @staticmethod
    def handle_finalize_error(error_msg: str, context: dict) -> dict:
        """处理 RGB finalize 错误"""
 
        if "non-finalized input" in error_msg:
            # 已知的 RGB CLI bug
            return {
                "strategy": "bypass_finalize",
                "action": "verify_state_directly",
                "reason": "RGB CLI finalize bug detected"
            }
 
        elif "insufficient fee" in error_msg:
            return {
                "strategy": "fee_bump",
                "action": "increase_fee_and_retry",
                "suggested_fee": context.get("current_fee", 0) * 1.5
            }
 
        else:
            return {
                "strategy": "manual_intervention",
                "action": "require_user_input",
                "error": error_msg
            }
 

def batch_transfer(recipients: list, contract_id: str) -> dict:
    """批量转账优化 - 单个 PSBT 多个输出"""
 
    # 聚合所有接收方发票
    invoices = []
    for recipient in recipients:
        invoice = generate_invoice(recipient["address"], recipient["amount"])
        invoices.append(invoice)
 
    # 创建批量转账 PSBT
    batch_psbt = create_batch_psbt(invoices)
 
    # 单次签名和广播
    result = sign_and_broadcast(batch_psbt)
 
    return {
        "batch_size": len(recipients),
        "total_amount": sum(r["amount"] for r in recipients),
        "txid": result["txid"],
        "individual_results": parse_batch_results(result)
    }
 

class RGBStateMonitor:
    """RGB 状态变化监控"""
 
    def __init__(self, wallets: list):
        self.wallets = wallets
        self.last_states = {}
 
    async def monitor_changes(self, interval: int = 30):
        """监控状态变化"""
        while True:
            for wallet in self.wallets:
                current_state = get_wallet_state(wallet)
 
                if self.has_state_changed(wallet, current_state):
                    await self.handle_state_change(wallet, current_state)
                    self.last_states[wallet] = current_state
 
            await asyncio.sleep(interval)
 
    async def handle_state_change(self, wallet: str, new_state: dict):
        """处理状态变化"""
        changes = self.analyze_changes(wallet, new_state)
 
        for change in changes:
            if change["type"] == "incoming_transfer":
                await self.notify_incoming_transfer(wallet, change)
            elif change["type"] == "confirmation":
                await self.notify_confirmation(wallet, change)
 

# 与 Lightning Network 集成
class RGBLightningBridge:
    """RGB 与闪电网络桥接"""
 
    def create_lightning_invoice(self, rgb_amount: int, contract_id: str):
        """创建包含 RGB 信息的闪电发票"""
        lightning_invoice = lnd_client.add_invoice({
            "value": 1,  # 1 聪作为锚定
            "memo": f"RGB:{contract_id}:{rgb_amount}",
            "expiry": 3600
        })
 
        return {
            "lightning_invoice": lightning_invoice["payment_request"],
            "rgb_data": {
                "contract_id": contract_id,
                "amount": rgb_amount
            }
        }
 
# 与 DeFi 协议集成
class RGBDeFiAdapter:
    """RGB DeFi 协议适配器"""
 
    def create_swap_offer(self, offer_asset: str, want_asset: str, ratio: float):
        """创建资产交换提议"""
        swap_contract = deploy_swap_contract({
            "offer": {"asset": offer_asset, "amount": self.calculate_amount(ratio)},
            "want": {"asset": want_asset, "amount": 1000000}  # 1M 单位
        })
 
        return swap_contract
 

rgb-transfer-toolkit/
├── README.md
├── LICENSE
├── requirements.txt
├── complete_rgb_transfer.py
└── examples/
    ├── config_example.json
    └── docker-compose.example.yml

# 快速开发环境设置
setup_dev_env() {
    # 启动 Bitcoin regtest
    docker compose -p bitlight-local-env up -d
 
    # 初始化 RGB 钱包
    rgb -d .alice -n regtest init
    rgb -d .bob -n regtest init
 
    # 启动 Esplora
    esplora --network regtest --daemon
}
 
# 自动化测试
run_integration_tests() {
    python -m pytest tests/ -v --tb=short
}
 
# 部署到生产环境
deploy_production() {
    # 切换到主网配置
    sed -i 's/regtest/mainnet/g' config/rgb_config.json
 
    # 更新 Esplora 端点
    sed -i 's/localhost:3002/blockstream.info/g' config/rgb_config.json
}
 

import asyncio
from concurrent.futures import ThreadPoolExecutor
 
class OptimizedRGBTransfer:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=4)
 
    async def parallel_validation(self, consignments: list):
        """并行验证多个 consignment"""
        tasks = []
 
        for consignment in consignments:
            task = asyncio.get_event_loop().run_in_executor(
                self.executor,
                self.validate_consignment,
                consignment
            )
            tasks.append(task)
 
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if not isinstance(r, Exception)]
 

from functools import lru_cache
import redis
 
class RGBCache:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
 
    @lru_cache(maxsize=1000)
    def get_contract_info(self, contract_id: str) -> dict:
        """缓存合约信息"""
        cached = self.redis_client.get(f"contract:{contract_id}")
        if cached:
            return json.loads(cached)
 
        # 从 RGB 获取新数据
        info = self.fetch_contract_info(contract_id)
        self.redis_client.setex(
            f"contract:{contract_id}",
            3600,  # 1小时过期
            json.dumps(info)
        )
        return info
 

# Dockerfile
FROM python:3.11-slim
 
# 安装 RGB CLI
RUN curl -L <https://github.com/RGB-WG/rgb/releases/latest/download/rgb-x86_64-linux.tar.gz> | tar xz
RUN mv rgb /usr/local/bin/
 
# 安装 Python 依赖
COPY requirements.txt .
RUN pip install -r requirements.txt
 
# 复制应用代码
COPY . /app
WORKDIR /app
 
# 启动应用
CMD ["python", "main.py", "transfer"]
 

# docker-compose.yml
version: '3.8'
services:
  rgb-toolkit:
    build: .
    volumes:
      - ./config:/app/config
      - ./logs:/app/logs
    environment:
      - NETWORK=mainnet
      - ESPLORA_URL=https://blockstream.info/api
    depends_on:
      - bitcoin
      - esplora
 
  bitcoin:
    image: bitcoind/bitcoin
    volumes:
      - bitcoin_data:/bitcoin/.bitcoin
    ports:
      - "8332:8332"
 

# production_config.py
PRODUCTION_CONFIG = {
    "security": {
        "encrypt_private_keys": True,
        "use_hardware_wallet": True,
        "multi_sig_threshold": 2,
        "backup_seeds": True
    },
    "monitoring": {
        "enable_logging": True,
        "log_level": "INFO",
        "metrics_endpoint": "prometheus:9090",
        "alert_webhook": "<https://alerts.example.com/webhook>"
    },
    "performance": {
        "connection_pool_size": 10,
        "request_timeout": 30,
        "retry_attempts": 3,
        "cache_ttl": 3600
    }
}
 

# 安全的私钥管理
class SecureKeyManager:
    def __init__(self, hsm_config: dict):
        self.hsm = HSMClient(hsm_config)
 
    def sign_psbt(self, psbt: str, derivation_path: str) -> str:
        """使用 HSM 签名 PSBT"""
        # 永不暴露私钥到内存
        signature = self.hsm.sign_transaction(psbt, derivation_path)
        return signature
 
    def derive_address(self, path: str) -> str:
        """从 HSM 派生地址"""
        return self.hsm.get_address(path)
 

class RobustRGBClient:
    def __init__(self):
        self.max_retries = 3
        self.backoff_factor = 2
 
    @retry(max_attempts=3, backoff_factor=2)
    def resilient_transfer(self, *args, **kwargs):
        """具有重试机制的转账"""
        try:
            return self.execute_transfer(*args, **kwargs)
        except NetworkError as e:
            self.log_error(f"Network error: {e}, retrying...")
            raise
        except RGBError as e:
            if e.is_recoverable():
                self.log_warning(f"Recoverable RGB error: {e}")
                raise
            else:
                self.log_error(f"Fatal RGB error: {e}")
                return {"success": False, "error": str(e)}
 

import prometheus_client
from prometheus_client import Counter, Histogram, Gauge
 
# 定义指标
transfer_counter = Counter('rgb_transfers_total', 'Total RGB transfers')
transfer_duration = Histogram('rgb_transfer_duration_seconds', 'Transfer duration')
active_contracts = Gauge('rgb_active_contracts', 'Number of active contracts')
 
class RGBMetrics:
    @staticmethod
    def record_transfer(success: bool, duration: float):
        transfer_counter.labels(status='success' if success else 'failure').inc()
        transfer_duration.observe(duration)
 
    @staticmethod
    def update_contract_count(count: int):
        active_contracts.set(count)