Разработка системы отчетности крипто-фонда

Разработка системы отчётности крипто-фонда

У крипто-фондов специфическая проблема с отчётностью: активы распределены по десяткам кошельков на нескольких блокчейнах, часть находится в DeFi протоколах (LP позиции, lending, стейкинг), часть на CEX, и всё это нужно свести в единый P&L с правильным учётом стоимостной базы. Ни одно готовое решение не покрывает весь этот спектр без существенных компромиссов.

Архитектура: что нужно агрегировать

Типичный портфель фонда:

Ключевой принцип: данные нужно получать из первоисточника, не из агрегаторов вроде Zapper или DeBank. Агрегаторы ошибаются, не поддерживают все протоколы, и данные нельзя верифицировать.

On-chain данные: прямые вызовы

Для EVM-кошельков баланс любого ERC-20:

from web3 import Web3
from decimal import Decimal

w3 = Web3(Web3.HTTPProvider("https://eth-mainnet.g.alchemy.com/v2/KEY"))

ERC20_ABI = [{"inputs":[{"name":"account","type":"address"}],"name":"balanceOf","outputs":[{"name":"","type":"uint256"}],"type":"function"},
             {"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"type":"function"}]

def get_token_balance(wallet: str, token: str) -> Decimal:
    contract = w3.eth.contract(address=Web3.to_checksum_address(token), abi=ERC20_ABI)
    raw = contract.functions.balanceOf(wallet).call()
    decimals = contract.functions.decimals().call()
    return Decimal(raw) / Decimal(10 ** decimals)

Для сотен кошельков и токенов — Multicall3 (0xcA11bde05977b3631167028862bE2a173976CA11, деплоен на всех основных сетях):

# Один RPC вызов вместо N
def batch_balances(calls: list[tuple[str, str]]) -> list[Decimal]:
    # calls = [(wallet, token), ...]
    multicall = w3.eth.contract(address=MULTICALL3_ADDRESS, abi=MULTICALL3_ABI)
    encoded_calls = [(token, False, encode_balance_call(wallet)) for wallet, token in calls]
    results = multicall.functions.aggregate3(encoded_calls).call()
    return [decode_balance(r[1], get_decimals(token)) for r, (_, token) in zip(results, calls)]

Uniswap V3 LP позиции

LP в Uniswap V3 — это NFT с position ID. Для корректного учёта нужно читать positions() из NonfungiblePositionManager и вычислять текущую стоимость через квантовую математику тиков:

def get_uniswap_v3_position_value(token_id: int, block: int = None) -> dict:
    position_manager = w3.eth.contract(
        address="0xC36442b4a4522E871399CD717aBDD847Ab11FE88",
        abi=NPM_ABI
    )
    pos = position_manager.functions.positions(token_id).call(block_identifier=block or "latest")

    # pos: (nonce, operator, token0, token1, fee, tickLower, tickUpper,
    #        liquidity, feeGrowthInside0LastX128, feeGrowthInside1LastX128,
    #        tokensOwed0, tokensOwed1)

    liquidity = pos[7]
    tick_lower, tick_upper = pos[5], pos[6]

    pool = get_pool(pos[2], pos[3], pos[4])
    current_tick = pool.functions.slot0().call()[1]
    sqrt_price = pool.functions.slot0().call()[0]

    amount0, amount1 = calculate_amounts(liquidity, sqrt_price, tick_lower, tick_upper, current_tick)
    fees0, fees1 = calculate_uncollected_fees(pos, pool, tick_lower, tick_upper)

    return {
        "principal": {"token0": amount0, "token1": amount1},
        "fees": {"token0": fees0, "token1": fees1}
    }

Расчёт стоимостной базы и P&L

Для налоговой отчётности и investor reporting нужен исторический P&L с правильным методом стоимостной базы. Три подхода:

• FIFO (First In, First Out) — стандарт в большинстве юрисдикций
• LIFO — разрешён в некоторых юрисдикциях, оптимизирует налоги при растущем рынке
• ACB (Average Cost Basis) — упрощённый метод, используется в Канаде и ряде других стран

from dataclasses import dataclass
from collections import deque
from decimal import Decimal

@dataclass
class Lot:
    amount: Decimal
    cost_basis_usd: Decimal
    acquired_at: datetime

class FIFOLedger:
    def __init__(self):
        self.lots: dict[str, deque[Lot]] = {}  # symbol -> queue of lots

    def buy(self, symbol: str, amount: Decimal, price_usd: Decimal, ts: datetime):
        if symbol not in self.lots:
            self.lots[symbol] = deque()
        self.lots[symbol].append(Lot(amount, amount * price_usd, ts))

    def sell(self, symbol: str, amount: Decimal, price_usd: Decimal, ts: datetime) -> Decimal:
        """Возвращает реализованный P&L"""
        proceeds = amount * price_usd
        cost = Decimal(0)
        remaining = amount

        while remaining > 0 and self.lots[symbol]:
            lot = self.lots[symbol][0]
            if lot.amount <= remaining:
                cost += lot.cost_basis_usd
                remaining -= lot.amount
                self.lots[symbol].popleft()
            else:
                portion = remaining / lot.amount
                cost += lot.cost_basis_usd * portion
                lot.amount -= remaining
                lot.cost_basis_usd -= lot.cost_basis_usd * portion
                remaining = Decimal(0)

        return proceeds - cost  # положительный = прибыль

Исторические цены

Для P&L нужны исторические цены на момент каждой транзакции. Источники в порядке надёжности:

1. CoinGecko API (/coins/{id}/history?date=dd-mm-yyyy) — покрывает большинство токенов, бесплатный лимит 30 req/min
2. Chainlink price feeds — для DeFi токенов с Chainlink интеграцией, исторические данные через getHistoricalAnswer(roundId)
3. On-chain AMM TWAP — IUniswapV3Pool.observe() для точных исторических цен

Кешируйте цены в локальной БД — запрашивать CoinGecko для каждой из тысяч транзакций и получить rate limit.

Отчёты и экспорт

Стандартный набор для инвесторского пакета:

• NAV report (Net Asset Value) — ежедневный снапшот всех позиций с рыночной стоимостью
• P&L statement — реализованный и нереализованный P&L по периодам
• Transaction history — все движения с хэшами транзакций для верификации
• Tax report — реализованные события по FIFO/LIFO с стоимостной базой

Для аудиторов важна воспроизводимость: любой отчёт должен генерироваться детерминированно из одних и тех же исходных данных (транзакции + цены).

Система от нуля с покрытием 5–10 адресов на 3–4 сетях и базовыми DeFi протоколами: 3–5 недель.