Crypto fund reporting system development

Development of Crypto Fund Reporting System

Crypto funds have a specific problem with reporting: assets are distributed across dozens of wallets on multiple blockchains, some are in DeFi protocols (LP positions, lending, staking), some on CEX, and all this needs to be consolidated into a single P&L with correct cost basis accounting. No out-of-the-box solution covers this entire spectrum without significant compromises.

Architecture: what needs to be aggregated

Typical fund portfolio:

Key principle: data must be obtained from primary sources, not aggregators like Zapper or DeBank. Aggregators make mistakes, don't support all protocols, and data cannot be verified.

On-chain data: direct calls

For EVM wallets, balance of any 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)

For hundreds of wallets and tokens — use Multicall3 (0xcA11bde05977b3631167028862bE2a173976CA11, deployed on all major networks):

# One RPC call instead of 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 positions

LP in Uniswap V3 is an NFT with position ID. For correct accounting, need to read positions() from NonfungiblePositionManager and calculate current value via tick mathematics:

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}
    }

Cost basis and P&L calculation

For tax reporting and investor reporting, need historical P&L with correct cost basis method. Three approaches:

• FIFO (First In, First Out) — standard in most jurisdictions
• LIFO — allowed in some jurisdictions, optimizes taxes in rising markets
• ACB (Average Cost Basis) — simplified method, used in Canada and some other countries

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:
        """Returns realized 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  # positive = profit

Historical prices

For P&L need historical prices at the time of each transaction. Sources in order of reliability:

1. CoinGecko API (/coins/{id}/history?date=dd-mm-yyyy) — covers most tokens, free limit 30 req/min
2. Chainlink price feeds — for DeFi tokens with Chainlink integration, historical data via getHistoricalAnswer(roundId)
3. On-chain AMM TWAP — IUniswapV3Pool.observe() for accurate historical prices

Cache prices in local DB — don't request CoinGecko for each of thousands of transactions and hit rate limit.

Reports and export

Standard set for investor package:

• NAV report (Net Asset Value) — daily snapshot of all positions with market value
• P&L statement — realized and unrealized P&L by period
• Transaction history — all movements with transaction hashes for verification
• Tax report — realized events by FIFO/LIFO with cost basis

For auditors, reproducibility matters: any report should be generated deterministically from the same source data (transactions + prices).

System from zero with coverage of 5–10 addresses on 3–4 networks and basic DeFi protocols: 3–5 weeks.