Розробка системи звітності крипто-фонду

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

У крипто-фондів специфічна проблема з відчітністю: активи розподілені по десяткам гаманців на кількох блокчейнах, частина знаходиться у DeFi протоколах (LP позиції, lending, stейкинг), частина на 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 та обчислювати поточну вартість через mathematical тиків:

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

Для податкової відчітності та інвесторської звітності потрібен історичний 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 звіт (Net Asset Value) — щоденний снапшот усіх позицій з ринковою вартістю
• P&L statement — реалізований та нереалізований P&L за період
• Transaction history — усі руху з хешами транзакцій для верифікації
• Tax звіт — реалізовані события за FIFO/LIFO з стоимостною базою

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

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