Урок 5.1 — Создание переводов

Прежде чем использовать переводы, их нужно включить в настройках твоего приложения в @CryptoBot или @CryptoTestnetBot. По умолчанию переводы отключены для всех новых приложений.

Метод transfer() отправляет средства указанному пользователю @CryptoBot. Для получателя это выглядит как мгновенное зачисление на баланс в боте. Твоё приложение должно иметь достаточный баланс для отправки.

async def transfer(
    self,
    user_id: int,
    asset: str,
    amount: float | str,
    spend_id: str | None = None,
    comment: str | None = None,
    disable_send_notification: bool = False,
) -> Transfer

Только user_id, asset и amount обязательны. Остальные параметры опциональны.

from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# Простейший перевод — только обязательные параметры
transfer = await cp.transfer(
    user_id=123456789,
    asset="USDT",
    amount=10,
)

print(f"Перевод #{transfer.transfer_id}")
print(f"Получатель: {transfer.user_id}")
print(f"Сумма: {transfer.amount} {transfer.asset}")
print(f"Статус: {transfer.status}")
print(f"ID списания: {transfer.spend_id}")

Параметр spend_id — это механизм идемпотентности. Он гарантирует, что один и тот же перевод не будет выполнен дважды. Если по какой-то причине ты отправишь запрос с тем же spend_id, который уже был использован, API вернёт существующий объект Transfer вместо создания нового.

Это критически важно для финансовых операций: представь, что клиент отправил запрос на перевод, но из-за сетевой ошибки не получил ответ. Если он повторит запрос без spend_id — средства будут списаны повторно. С spend_id — API вернёт результат первого перевода.

import uuid
import secrets

# Вариант 1: UUID4 — стандартный способ
spend_id = str(uuid.uuid4())
print(f"spend_id: {spend_id}")
# → "a1b2c3d4-e5f6-7890-abcd-ef1234567890"

# Вариант 2: secrets — криптостойкая случайная строка
spend_id = secrets.token_hex(16)
print(f"spend_id: {spend_id}")
# → "d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8a9"

# Вариант 3: комбинация user_id + timestamp + random
import hashlib
import time

raw = f"user_{123456789}_time_{int(time.time())}_rand_{secrets.randbits(32)}"
spend_id = hashlib.sha256(raw.encode()).hexdigest()
print(f"spend_id: {spend_id}")
# → "a1b2c3d4..." (64 символа hex)

Требования к spend_id:

• Уникальная строка в кодировке UTF-8
• Генерируется на стороне клиента (ты сам отвечаешь за уникальность)
• После использования spend_id сохраняется в истории — повторная отправка с тем же ID вернёт существующий Transfer
• Рекомендуется использовать UUID4 или криптостойкие случайные строки

import uuid
from aiosend import CryptoPay

async def send_money(cp: CryptoPay, user_id: int, amount: float):
    # Генерируем уникальный spend_id
    spend_id = str(uuid.uuid4())

    transfer = await cp.transfer(
        user_id=user_id,
        asset="USDT",
        amount=amount,
        spend_id=spend_id,
        comment="Спасибо за покупку!",
    )
    print(f"Перевод выполнен: #{transfer.transfer_id}")
    return transfer

# Если запрос повторится с тем же spend_id —
# API вернёт существующий перевод, не списывая средства повторно
async def safe_send(cp: CryptoPay, user_id: int, amount: float):
    spend_id = "fixed-spend-id-for-order-12345"

    # Первый вызов — создаст перевод
    t1 = await cp.transfer(user_id=user_id, asset="USDT",
                          amount=amount, spend_id=spend_id)
    print(f"Первый раз: #{t1.transfer_id}")

    # Второй вызов с тем же spend_id — вернёт тот же перевод
    t2 = await cp.transfer(user_id=user_id, asset="USDT",
                          amount=amount, spend_id=spend_id)
    print(f"Второй раз: #{t2.transfer_id} (тот же!)")
    assert t1.transfer_id == t2.transfer_id  # True!

Метод transfer() возвращает объект Transfer (Pydantic-модель). Вот все его поля:

transfer = await cp.transfer(
    user_id=123456789,
    asset="USDT",
    amount=25.50,
    spend_id=str(uuid.uuid4()),
    comment="Бонус за регистрацию",
)

print(f"ID перевода: {transfer.transfer_id}")
print(f"Получатель: {transfer.user_id}")
print(f"Сумма: {transfer.amount} {transfer.asset}")
print(f"Статус: {transfer.status}")
print(f"Дата: {transfer.completed_at}")
print(f"Комментарий: {transfer.comment}")
print(f"spend_id: {transfer.spend_id}")

# Проверка на успешность
if transfer.status == "completed":
    print("✅ Перевод успешно выполнен")

Эти два параметра управляют тем, как получатель видит перевод.

# Перевод с комментарием — получатель увидит "Бонус!"
transfer = await cp.transfer(
    user_id=123456789,
    asset="USDT",
    amount=5,
    comment="Бонус за подписку! 🎉",
)

# Перевод без уведомления — тихая выплата
transfer = await cp.transfer(
    user_id=987654321,
    asset="TON",
    amount=10,
    disable_send_notification=True,
    comment="Автоматическая выплата",
)

# Перевод с комментарием и без уведомления
transfer = await cp.transfer(
    user_id=555555555,
    asset="USDT",
    amount=100,
    spend_id=str(uuid.uuid4()),
    comment="Возврат средств за заказ #12345",
    disable_send_notification=False,  # пусть знает
)

Каждый актив имеет ограничения по минимальной и максимальной сумме перевода. Точные значения можно получить через get_currencies(), но вот примерные ориентиры:

Максимальная сумма перевода ограничена ~25 000 USD в эквиваленте для всех активов. Если нужно отправить больше — отправляй частями.

from aiosend import CryptoPay

async def get_asset_limits(cp: CryptoPay, asset: str) -> tuple:
    """Возвращает (min_amount, max_amount) для актива."""
    currencies = await cp.get_currencies()
    for currency in currencies:
        if currency.code == asset:
            return (currency.min_amount, currency.max_amount)
    return (None, None)

async def safe_transfer(cp: CryptoPay, user_id: int, asset: str, amount: float):
    min_amt, max_amt = await get_asset_limits(cp, asset)

    if min_amt and float(amount) < float(min_amt):
        print(f"❌ Сумма {amount} меньше минимума {min_amt} для {asset}")
        return None

    if max_amt and float(amount) > float(max_amt):
        print(f"❌ Сумма {amount} превышает максимум {max_amt} для {asset}")
        return None

    # Все проверки пройдены — выполняем перевод
    transfer = await cp.transfer(
        user_id=user_id,
        asset=asset,
        amount=amount,
        spend_id=str(uuid.uuid4()),
    )
    return transfer

Важное условие: пользователь, которому ты отправляешь перевод, должен хотя бы один раз запустить @CryptoBot (или @CryptoTestnetBot для тестовой сети). Без этого API не сможет зачислить средства.

from aiosend import CryptoPay
from aiosend.exceptions import APIError

async def try_transfer(cp: CryptoPay, user_id: int, asset: str, amount: float):
    try:
        transfer = await cp.transfer(
            user_id=user_id,
            asset=asset,
            amount=amount,
            spend_id=str(uuid.uuid4()),
            comment="Проверка перевода",
        )
        print(f"✅ Перевод успешен: #{transfer.transfer_id}")
        return transfer

    except APIError as e:
        if "USER_NOT_FOUND" in str(e):
            print(f"❌ Пользователь {user_id} не найден в @CryptoBot.")
            print("   Убедитесь, что пользователь запустил @CryptoBot")
        elif "INSUFFICIENT_BALANCE" in str(e):
            print(f"❌ Недостаточно средств на балансе приложения")
        else:
            print(f"❌ Ошибка API: {e}")
        return None

Рассмотрим несколько реальных сценариев отправки переводов.

import uuid
from aiosend import CryptoPay

async def send_bonus(cp: CryptoPay, user_id: int, amount: float, reason: str):
    """Отправляет бонус пользователю."""
    spend_id = f"bonus_{user_id}_{uuid.uuid4().hex[:8]}"

    transfer = await cp.transfer(
        user_id=user_id,
        asset="USDT",
        amount=amount,
        spend_id=spend_id,
        comment=f"🎁 Бонус: {reason}",
    )
    print(f"✅ Бонус отправлен: #{transfer.transfer_id}")
    print(f"   Получатель: {transfer.user_id}")
    print(f"   Сумма: {transfer.amount} {transfer.asset}")
    print(f"   Комментарий: {transfer.comment}")
    return transfer

import uuid
import asyncio
from aiosend import CryptoPay

async def batch_payouts(cp: CryptoPay, payouts: list[dict]):
    """Массовые выплаты пользователям.
    payouts = [
        {"user_id": 123, "amount": 10, "asset": "USDT"},
        {"user_id": 456, "amount": 5, "asset": "USDT"},
    ]
    """
    results = []
    for payout in payouts:
        try:
            transfer = await cp.transfer(
                user_id=payout["user_id"],
                asset=payout.get("asset", "USDT"),
                amount=payout["amount"],
                spend_id=str(uuid.uuid4()),
                comment="Массовая выплата",
                disable_send_notification=True,
            )
            results.append(transfer)
            print(f"✅ {payout['user_id']}: #{transfer.transfer_id}")
            await asyncio.sleep(0.1)  # защита от rate limit
        except Exception as e:
            print(f"❌ {payout['user_id']}: {e}")
            results.append(None)
    return results

async def refund(cp: CryptoPay, user_id: int, amount: float, order_id: str):
    """Возврат средств за заказ."""
    spend_id = f"refund_{order_id}"

    transfer = await cp.transfer(
        user_id=user_id,
        asset="USDT",
        amount=amount,
        spend_id=spend_id,
        comment=f"Возврат за заказ #{order_id}",
    )
    print(f"✅ Возврат оформлен: #{transfer.transfer_id}")
    print(f"   ID заказа: {order_id}")
    print(f"   Сумма: {transfer.amount} {transfer.asset}")
    return transfer

# Идемпотентность: если refund вызвали дважды с тем же spend_id
# — второй раз вернёт тот же Transfer
# t1 = await refund(cp, 123, 10, "ORD-001")
# t2 = await refund(cp, 123, 10, "ORD-001")
# assert t1.transfer_id == t2.transfer_id  # OK!

Если ты работаешь в синхронном окружении (скрипты, Flask, простые утилиты), используй SyncCryptoPay. Он предоставляет те же методы, включая transfer() и get_transfers(), но без await.

from aiosend import SyncCryptoPay

cp = SyncCryptoPay(token="YOUR_TOKEN")

# Синхронный перевод — без await!
transfer = cp.transfer(
    user_id=123456789,
    asset="USDT",
    amount=10,
    comment="Синхронный перевод",
)

print(f"Перевод #{transfer.transfer_id}: {transfer.amount} {transfer.asset}")
print(f"Статус: {transfer.status}")

# Синхронное получение истории
transfers = cp.get_transfers(asset="USDT")
print(f"Найдено переводов: {len(transfers)}")

# Все параметры те же, что и в асинхронной версии
transfer = cp.transfer(
    user_id=987654321,
    asset="TON",
    amount=5,
    spend_id=str(uuid.uuid4()),
    comment="Бонус",
    disable_send_notification=True,
)

Важно понимать разницу между двумя способами отправки средств: чеки активируются получателем, переводы зачисляются принудительно. Рассмотрим ключевые отличия.