Crypto payment platform development

Development of Crypto Payment Platform

The difference between "add a crypto payment button" and "build a payment platform" is fundamental. The first integrates a third-party service with its guarantees and limitations. The second builds your own infrastructure: address generation, transaction monitoring, key management, settlement, reconciliation. This page covers the second variant: when volume, regulatory requirements, or business model don't allow going through an intermediary.

Why build your own platform

Typical reasons:

• Volume — over $10M/month, processor commission of 0.5–1% becomes more expensive than own infrastructure
• Data control — financial data should not go to third parties
• Custom logic — conditional payments, escrow, recurring payments, payment splits
• White-label — platform for other businesses, own processing mandatory
• Regulatory requirements — some jurisdictions require licenses incompatible with foreign processors

Architecture: key components

Client → Payment API → Invoice Service → Address Generator
                                       → Blockchain Monitor
                                       → Settlement Engine
                                       → Merchant Webhook

HD wallets and address generation

Foundation of the platform — hierarchical deterministic wallets (BIP32/BIP44). One master seed generates unlimited child addresses — one unique address per invoice:

m / purpose' / coin_type' / account' / change / index
m / 44'      / 60'        / 0'       / 0      / invoice_id

from hdwallet import HDWallet
from hdwallet.symbols import ETH

def generate_payment_address(merchant_id: int, invoice_id: int) -> str:
    """
    Deterministic address generation: one invoice = one address
    merchant_id: account index (BIP44 level 3)
    invoice_id: address index (BIP44 level 5)
    """
    wallet = HDWallet(symbol=ETH)
    wallet.from_xprivate_key(MASTER_XPRIV)
    wallet.from_path(f"m/44'/60'/{merchant_id}'/0/{invoice_id}")
    return wallet.p2pkh_address()

Critically important: private key from master seed must be stored in HSM (Hardware Security Module) or at minimum in AWS KMS / HashiCorp Vault. No plaintext keys in config files or DB.

For public address generation (without private key access) — xpub is sufficient:

# Only xpub for watch-only addresses
wallet.from_xpublic_key(MERCHANT_XPUB)
wallet.from_path(f"m/0/{invoice_id}")  # relative from xpub
address = wallet.p2pkh_address()

Transaction monitoring

Two approaches to tracking incoming payments:

Polling via RPC — simple, works everywhere:

import asyncio
from web3 import Web3

async def monitor_invoice(
    address: str,
    expected_amount: int,  # in wei
    invoice_id: str,
    timeout_seconds: int = 3600
):
    w3 = Web3(Web3.HTTPProvider(ETH_RPC_URL))
    start_block = w3.eth.block_number
    deadline = time.time() + timeout_seconds

    while time.time() < deadline:
        # Check address balance
        balance = w3.eth.get_balance(address, "latest")
        if balance >= expected_amount:
            # Wait for confirmations
            await wait_for_confirmations(address, expected_amount, confirmations=6)
            await mark_invoice_paid(invoice_id, balance)
            return

        await asyncio.sleep(15)  # roughly per block

    await mark_invoice_expired(invoice_id)

WebSocket subscriptions — for minimal latency:

from web3 import AsyncWeb3
from web3.providers import WebsocketProviderV2

async def subscribe_to_address(address: str, callback):
    async with AsyncWeb3(WebsocketProviderV2("wss://eth-mainnet.g.alchemy.com/v2/KEY")) as w3:
        subscription_id = await w3.eth.subscribe(
            "logs",
            {"address": address}  # get all logs for address
        )
        async for response in w3.socket.process_subscriptions():
            if response["subscription"] == subscription_id:
                await callback(response["result"])

For ERC-20 payments monitoring is more complex: need to listen to Transfer events of token contract with filter by to address, not ETH balance.

Working with multiple networks and tokens

Production platform supports minimum: ETH/USDT/USDC on Ethereum, USDT/USDC on Polygon and Arbitrum (cheaper gas), BTC native. Multi-network architecture:

@dataclass
class NetworkConfig:
    chain_id: int
    rpc_url: str
    confirmations_required: int  # 1 for L2, 6 for ETH mainnet
    supported_tokens: dict[str, str]  # symbol -> contract address

NETWORKS = {
    "ethereum": NetworkConfig(
        chain_id=1,
        rpc_url=ETH_RPC,
        confirmations_required=6,
        supported_tokens={"USDT": "0xdAC17...", "USDC": "0xA0b86..."}
    ),
    "arbitrum": NetworkConfig(
        chain_id=42161,
        rpc_url=ARB_RPC,
        confirmations_required=1,
        supported_tokens={"USDT": "0xFd086...", "USDC": "0xFF970..."}
    ),
    # ...
}

Settlement: withdrawal

Collected payments must be aggregated and withdrawn to merchant. Two patterns:

Sweeping — periodic transfer from all addresses to hot wallet:

async def sweep_address(from_address: str, to_address: str, token: str):
    private_key = await kms.get_key(derive_key_path(from_address))

    if token == "ETH":
        balance = w3.eth.get_balance(from_address)
        gas_estimate = 21000
        gas_price = w3.eth.gas_price
        amount = balance - (gas_estimate * gas_price)
        if amount <= 0:
            return

        tx = {
            "to": to_address,
            "value": amount,
            "gas": gas_estimate,
            "gasPrice": gas_price,
            "nonce": w3.eth.get_transaction_count(from_address),
            "chainId": 1
        }
    else:
        # ERC-20 sweep: first ETH for gas, then transfer token
        await fund_gas(from_address)
        tx = build_erc20_transfer(token, from_address, to_address)

    signed = w3.eth.account.sign_transaction(tx, private_key)
    tx_hash = w3.eth.send_raw_transaction(signed.rawTransaction)
    return tx_hash.hex()

Batch payments — for payouts to merchants, use Multicall or specialized batch transfer contracts (e.g., Disperse.app pattern) — one gas fee for N transfers.

Working with ERC-20: the gas problem

ERC-20 payment weakness: receiving address is empty (no ETH for gas). Before sweep must send small amount of ETH to pay for token transfer gas. This creates dust problem: ETH remains after sweep.

Solution — EIP-2612 Permit: tokens with permit (USDC, DAI) allow signing transfer permission without on-chain approve transaction. Can sweep tokens in one transaction with signature, without pre-funding:

// USDC permit + transferFrom in one call
function sweepWithPermit(
    address token,
    address from,
    address to,
    uint256 amount,
    uint256 deadline,
    uint8 v, bytes32 r, bytes32 s
) external {
    IERC20Permit(token).permit(from, address(this), amount, deadline, v, r, s);
    IERC20(token).transferFrom(from, to, amount);
}

Exchange rates and pricing

Invoice created in fiat currency (USD/EUR), amount in crypto calculated dynamically. For stablecoins — 1:1 rate with small buffer. For volatile assets (ETH, BTC):

def calculate_crypto_amount(fiat_amount: Decimal, currency: str, crypto: str) -> dict:
    # Aggregate price from multiple sources
    prices = await asyncio.gather(
        fetch_binance_price(f"{crypto}USDT"),
        fetch_coinbase_price(f"{crypto}-USD"),
        fetch_chainlink_price(CHAINLINK_FEEDS[crypto])
    )
    median_price = statistics.median(prices)

    crypto_amount = fiat_amount / median_price
    # Add buffer for price movement (1-2%)
    crypto_amount_with_buffer = crypto_amount * Decimal("1.015")

    return {
        "amount": crypto_amount_with_buffer,
        "rate": median_price,
        "expires_at": datetime.now() + timedelta(minutes=15),
        "rate_locked": True
    }

Security and compliance

KYT (Know Your Transaction) — check addresses against sanction lists and connections to mixers/dark markets. Providers: Chainalysis, Elliptic, TRM Labs. API integrated into payment flow:

async def check_address_risk(address: str) -> RiskScore:
    response = await chainalysis_client.post("/v2/address/identifications", json={"address": address})
    risk = response.json()
    if risk["risk"] in ["HIGH", "SEVERE"]:
        await flag_for_review(address)
    return RiskScore(level=risk["risk"], categories=risk["categories"])

Transaction limits — automatic limits by volume for AML: daily limit per wallet, threshold for manual review, mandatory verification for large transactions.

Reconciliation — daily check: sum of merchant payouts + platform fees + balances on sweep addresses should match sum of all confirmed incoming transactions. Any discrepancy — alert.

Infrastructure and reliability

Node vs provider: for payment platform with reliability requirements — own node or Alchemy/Infura with fallback to another provider. Single point of failure is unacceptable.

from web3 import Web3
from web3.middleware import ExceptionRetryMiddleware

providers = [
    Web3.HTTPProvider(ALCHEMY_URL),
    Web3.HTTPProvider(INFURA_URL),
    Web3.HTTPProvider(QUICKNODE_URL),
]

def get_w3_with_fallback():
    for provider in providers:
        w3 = Web3(provider)
        w3.middleware_onion.add(ExceptionRetryMiddleware, retries=3)
        if w3.is_connected():
            return w3
    raise RuntimeError("All providers unavailable")

Idempotency: all operations must be idempotent — repeated webhook processing or payment check retry should not create duplicate payouts. Unique idempotency_key on each operation.

Timelines

Minimally viable platform with ETH/USDC on one network: 6–8 weeks. Full multi-network platform with compliance: 3–5 months.