القائمة
Track crypto wallets across multiple chains with real-time balance and...
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/tools-only/X-Skills --skill track-walletيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
Index of Build Systems Skills
Coordination patterns for distributed dataflow systems including barriers, epochs, and distributed snapshots
Windowing, sessionization, time-series aggregation, and late data handling for streaming systems
استنادا إلى تصنيف SOC المهني
عرض SKILL.md
| name | track-wallet |
| description | Track crypto wallets across multiple chains with real-time balance and... |
| shortcut | tw |
Monitor cryptocurrency wallets across multiple blockchain networks with comprehensive portfolio analytics, DeFi position tracking, NFT holdings, transaction history, profit/loss analysis, and tax reporting integration. This command provides institutional-grade portfolio tracking capabilities for individual investors, traders, and portfolio managers.
The Wallet Portfolio Tracker provides real-time visibility into cryptocurrency holdings across Ethereum, Bitcoin, Binance Smart Chain, Solana, Polygon, and other major blockchain networks. It aggregates token balances, NFT collections, DeFi positions (staking, lending, liquidity pools), and transaction history into a unified portfolio view with automated valuation using real-time price feeds from CoinGecko and other market data providers.
/track-wallet [address] [options]
address (required): Wallet address or ENS name to track--chains: Comma-separated list of chains (default: all)--include-nfts: Include NFT valuations (default: true)--include-defi: Include DeFi positions (default: true)--fiat: Fiat currency for valuation (default: USD)--historical-days: Days of historical data (default: 30)--export: Export format (csv, excel, json)--tax-year: Generate tax report for specific year--label: Save address with custom label/track-wallet 0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb1
Portfolio Summary for 0x742d...bEb1
===============================================
Total Value: $125,432.18 USD
Last Updated: 2025-10-11 14:32:18 UTC
Chain Distribution:
- Ethereum: $89,234.50 (71.1%)
- Polygon: $21,432.18 (17.1%)
- BSC: $10,234.90 (8.2%)
- Arbitrum: $4,530.60 (3.6%)
Top Holdings:
1. ETH: 15.234 ($48,936.00) - 39.0%
2. USDC: $25,000.00 - 19.9%
3. MATIC: 8,432 ($6,745.60) - 5.4%
4. LINK: 423 ($6,345.00) - 5.1%
5. AAVE: 67 ($5,226.00) - 4.2%
DeFi Positions:
- Aave Lending: $15,432 supplied
- Uniswap V3 LP: $8,234 in ETH/USDC pool
- Lido Staking: 2.5 stETH ($8,025)
NFT Holdings:
- 3 Collections, 12 NFTs
- Estimated Value: $18,430
- Floor Value: $14,230
/track-wallet vitalik.eth --chains ethereum,polygon,optimism --historical-days 90
Multi-Chain Portfolio Analysis
===============================================
Address: vitalik.eth (0xd8dA...9045)
Analysis Period: 90 days
Current Value: $2,145,432.18 USD
90-Day Change: +$245,432.18 (+12.9%)
Historical Performance:
- 7-day: +$12,432 (+0.6%)
- 30-day: +$89,234 (+4.3%)
- 90-day: +$245,432 (+12.9%)
ROI Metrics:
- Total ROI: +234.5%
- Annualized: +89.3%
- Max Drawdown: -18.4%
- Sharpe Ratio: 1.87
Chain Performance:
Ethereum:
- Current: $1,834,234
- 90d Change: +8.9%
- Transaction Count: 1,234
Polygon:
- Current: $234,123
- 90d Change: +18.4%
- Transaction Count: 567
Optimism:
- Current: $77,075
- 90d Change: +45.2%
- Transaction Count: 89
/track-wallet 0x742d...bEb1 --include-defi
DeFi Portfolio Analysis
===============================================
Total DeFi Value: $45,234.50 (36.0% of portfolio)
Protocol Count: 8
Active Positions: 15
Lending Positions:
Aave V3 (Ethereum):
- Supplied: 10,000 USDC ($10,000)
- Borrowed: 5 ETH ($16,060) at 2.1% APY
- Health Factor: 2.45
- Net APY: +1.8%
Compound V3:
- Supplied: 3 ETH ($9,639)
- Borrowed: None
- Supply APY: 2.4%
Liquidity Pools:
Uniswap V3 (ETH/USDC 0.3%):
- Position: $8,234.50
- Price Range: $2,800 - $3,400
- Fees Earned (24h): $12.45
- Impermanent Loss: -2.3%
- APY (7d avg): 18.4%
Curve Finance (3pool):
- Position: $5,000 in USDC/USDT/DAI
- Gauge Staked: Yes
- CRV Rewards: 12.5 CRV ($15.23/day)
- APY: 8.7% + 4.2% CRV
Staking:
Lido (stETH):
- Staked: 2.5 ETH ($8,025)
- APY: 3.8%
- Rewards: 0.0234 ETH ($75.14)
Rocket Pool (rETH):
- Staked: 1.2 ETH ($3,854)
- APY: 4.1%
- Exchange Rate: 1.0234 ETH
Yield Aggregators:
Yearn Finance (yvUSDC):
- Deposited: $5,000
- Current Value: $5,234.50
- APY: 6.8%
- Auto-compounding: Enabled
/track-wallet 0x742d...bEb1 --include-nfts
NFT Portfolio Summary
===============================================
Total NFT Value: $18,430 (14.7% of portfolio)
Collections: 3
Total NFTs: 12
Bored Ape Yacht Club (BAYC):
- Owned: 1 NFT (#8234)
- Floor Price: $45.5 ETH ($146,160)
- Estimated Value: $146,160
- Rarity Rank: #2,341/10,000
- Last Sale: 52 ETH (6 months ago)
- Unrealized P&L: -$18,900 (-11.5%)
CryptoPunks:
- Owned: 1 NFT (#4523)
- Floor Price: $89 ETH ($285,840)
- Estimated Value: $285,840
- Type: Ape, Hoodie
- Last Sale: 75 ETH (18 months ago)
- Unrealized P&L: +$45,840 (+19.1%)
Art Blocks Curated:
- Owned: 10 NFTs
- Total Floor Value: $34,230
- Estimated Portfolio Value: $38,450
- Projects: Fidenza (2), Ringers (3), Others (5)
- Avg Purchase Price: $2,100/NFT
- Unrealized P&L: +$17,450 (+83.1%)
Collection Performance (30d):
- BAYC: -8.4%
- CryptoPunks: +2.1%
- Art Blocks: +12.8%
/track-wallet 0x742d...bEb1 --export csv --tax-year 2025
Transaction History Export
===============================================
Total Transactions: 1,234
Date Range: 2025-01-01 to 2025-10-11
Export Format: CSV
Transaction Breakdown:
- Transfers: 456 (37.0%)
- Swaps: 342 (27.7%)
- DeFi Interactions: 289 (23.4%)
- NFT Trades: 89 (7.2%)
- Contract Calls: 58 (4.7%)
Gas Costs:
- Total Gas Paid: 2.34 ETH ($7,515.60)
- Average per Transaction: 0.0019 ETH ($6.09)
- Highest Gas Transaction: 0.234 ETH ($751.56)
- Optimization Potential: $1,234.50 (16.4%)
Tax Summary (2025):
Capital Gains:
- Short-term: $12,345.67 (456 trades)
- Long-term: $34,567.89 (123 trades)
- Total Realized: $46,913.56
Income:
- Staking Rewards: $2,345.67
- LP Fees: $1,234.50
- Airdrops: $890.23
- Total Income: $4,470.40
Total Taxable: $51,383.96
Exported Files:
- transactions_2025.csv (1,234 rows)
- tax_summary_2025.pdf
- irs_form_8949_draft.pdf
- cost_basis_report.xlsx
Below is a comprehensive Python implementation of the wallet portfolio tracker with multi-chain support, real-time pricing, DeFi position tracking, and tax reporting capabilities.
#!/usr/bin/env python3
"""
Wallet Portfolio Tracker - Production Implementation
Track crypto wallets across multiple chains with comprehensive analytics
"""
import os
import json
import asyncio
import aiohttp
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
from collections import defaultdict
import pandas as pd
from web3 import Web3
from bitcoinlib.wallets import HDWallet
from solana.rpc.async_api import AsyncClient as SolanaClient
import sqlite3
from functools import lru_cache
# Configuration
COINGECKO_API_KEY = os.getenv('COINGECKO_API_KEY', '')
ETHERSCAN_API_KEY = os.getenv('ETHERSCAN_API_KEY', '')
BSCSCAN_API_KEY = os.getenv('BSCSCAN_API_KEY', '')
POLYGONSCAN_API_KEY = os.getenv('POLYGONSCAN_API_KEY', '')
SOLSCAN_API_KEY = os.getenv('SOLSCAN_API_KEY', '')
BLOCKCHAIN_INFO_KEY = os.getenv('BLOCKCHAIN_INFO_KEY', '')
# RPC Endpoints
RPC_ENDPOINTS = {
'ethereum': os.getenv('ETH_RPC_URL', 'https://eth.llamarpc.com'),
'bsc': os.getenv('BSC_RPC_URL', 'https://bsc-dataseed.binance.org'),
'polygon': os.getenv('POLYGON_RPC_URL', 'https://polygon-rpc.com'),
'arbitrum': os.getenv('ARB_RPC_URL', 'https://arb1.arbitrum.io/rpc'),
'optimism': os.getenv('OP_RPC_URL', 'https://mainnet.optimism.io'),
'avalanche': os.getenv('AVAX_RPC_URL', 'https://api.avax.network/ext/bc/C/rpc'),
'solana': os.getenv('SOLANA_RPC_URL', 'https://api.mainnet-beta.solana.com')
}
# Chain configurations
CHAIN_CONFIGS = {
'ethereum': {
'chain_id': 1,
'native_token': 'ETH',
'decimals': 18,
'explorer_api': 'https://api.etherscan.io/api',
'api_key': ETHERSCAN_API_KEY,
'coingecko_id': 'ethereum'
},
'bsc': {
'chain_id': 56,
'native_token': 'BNB',
'decimals': 18,
'explorer_api': 'https://api.bscscan.com/api',
'api_key': BSCSCAN_API_KEY,
'coingecko_id': 'binancecoin'
},
'polygon': {
'chain_id': 137,
'native_token': 'MATIC',
'decimals': 18,
'explorer_api': 'https://api.polygonscan.com/api',
'api_key': POLYGONSCAN_API_KEY,
'coingecko_id': 'matic-network'
},
'arbitrum': {
'chain_id': 42161,
'native_token': 'ETH',
'decimals': 18,
'explorer_api': 'https://api.arbiscan.io/api',
'api_key': os.getenv('ARBISCAN_API_KEY', ''),
'coingecko_id': 'ethereum'
},
'optimism': {
'chain_id': 10,
'native_token': 'ETH',
'decimals': 18,
'explorer_api': 'https://api-optimistic.etherscan.io/api',
'api_key': os.getenv('OPSCAN_API_KEY', ''),
'coingecko_id': 'ethereum'
},
'avalanche': {
'chain_id': 43114,
'native_token': 'AVAX',
'decimals': 18,
'explorer_api': 'https://api.snowtrace.io/api',
'api_key': os.getenv('SNOWTRACE_API_KEY', ''),
'coingecko_id': 'avalanche-2'
}
}
# ERC-20 ABI (minimal)
ERC20_ABI = [
{
"constant": True,
"inputs": [{"name": "_owner", "type": "address"}],
"name": "balanceOf",
"outputs": [{"name": "balance", "type": "uint256"}],
"type": "function"
},
{
"constant": True,
"inputs": [],
"name": "decimals",
"outputs": [{"name": "", "type": "uint8"}],
"type": "function"
},
{
"constant": True,
"inputs": [],
"name": "symbol",
"outputs": [{"name": "", "type": "string"}],
"type": "function"
}
]
# Data models
@dataclass
class TokenBalance:
"""Token balance information"""
chain: str
address: str
symbol: str
name: str
balance: Decimal
decimals: int
price_usd: Decimal
value_usd: Decimal
contract_address: Optional[str] = None
logo_url: Optional[str] = None
@dataclass
class NFTAsset:
"""NFT asset information"""
chain: str
collection: str
token_id: str
name: str
description: Optional[str]
image_url: Optional[str]
floor_price: Decimal
estimated_value: Decimal
rarity_rank: Optional[int]
attributes: Dict
last_sale_price: Optional[Decimal] = None
@dataclass
class DeFiPosition:
"""DeFi position information"""
protocol: str
chain: str
position_type: str # lending, staking, liquidity_pool, yield_farm
supplied_tokens: List[TokenBalance]
borrowed_tokens: List[TokenBalance]
rewards_tokens: List[TokenBalance]
total_value_usd: Decimal
apy: Optional[Decimal]
health_factor: Optional[Decimal]
metadata: Dict
@dataclass
class Transaction:
"""Transaction information"""
chain: str
hash: str
timestamp: datetime
from_address: str
to_address: str
value: Decimal
gas_used: Decimal
gas_price: Decimal
transaction_type: str # transfer, swap, defi, nft, contract
tokens_transferred: List[Dict]
usd_value: Decimal
category: str
@dataclass
class PortfolioSnapshot:
"""Complete portfolio snapshot"""
address: str
timestamp: datetime
total_value_usd: Decimal
token_balances: List[TokenBalance]
nft_assets: List[NFTAsset]
defi_positions: List[DeFiPosition]
recent_transactions: List[Transaction]
chain_distribution: Dict[str, Decimal]
asset_allocation: Dict[str, Decimal]
analytics: Dict
class PriceOracle:
"""Handle price fetching from multiple sources"""
def __init__(self):
self.cache = {}
self.cache_duration = 300 # 5 minutes
async def get_token_price(
self,
token_address: str,
chain: str,
session: aiohttp.ClientSession
) -> Decimal:
"""Get token price in USD from CoinGecko"""
cache_key = f"{chain}:{token_address}"
if cache_key in self.cache:
cached_time, cached_price = self.cache[cache_key]
if (datetime.now() - cached_time).seconds < self.cache_duration:
return cached_price
try:
# Map chain to CoinGecko platform ID
platform_map = {
'ethereum': 'ethereum',
'bsc': 'binance-smart-chain',
'polygon': 'polygon-pos',
'arbitrum': 'arbitrum-one',
'optimism': 'optimistic-ethereum',
'avalanche': 'avalanche'
}
platform = platform_map.get(chain, 'ethereum')
url = f"https://api.coingecko.com/api/v3/simple/token_price/{platform}"
params = {
'contract_addresses': token_address,
'vs_currencies': 'usd',
'x_cg_demo_api_key': COINGECKO_API_KEY
}
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
price = Decimal(str(data.get(token_address.lower(), {}).get('usd', 0)))
self.cache[cache_key] = (datetime.now(), price)
return price
except Exception as e:
print(f"Error fetching price for {token_address}: {e}")
return Decimal('0')
async def get_native_token_price(
self,
chain: str,
session: aiohttp.ClientSession
) -> Decimal:
"""Get native token price (ETH, BNB, MATIC, etc.)"""
coingecko_id = CHAIN_CONFIGS[chain]['coingecko_id']
cache_key = f"native:{coingecko_id}"
if cache_key in self.cache:
cached_time, cached_price = self.cache[cache_key]
if (datetime.now() - cached_time).seconds < self.cache_duration:
return cached_price
try:
url = f"https://api.coingecko.com/api/v3/simple/price"
params = {
'ids': coingecko_id,
'vs_currencies': 'usd',
'x_cg_demo_api_key': COINGECKO_API_KEY
}
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
price = Decimal(str(data.get(coingecko_id, {}).get('usd', 0)))
self.cache[cache_key] = (datetime.now(), price)
return price
except Exception as e:
print(f"Error fetching native token price for {chain}: {e}")
return Decimal('0')
class ChainScanner:
"""Scan blockchain for wallet data"""
def __init__(self, chain: str):
self.chain = chain
self.config = CHAIN_CONFIGS[chain]
self.w3 = Web3(Web3.HTTPProvider(RPC_ENDPOINTS[chain]))
self.price_oracle = PriceOracle()
async def get_native_balance(
self,
address: str,
session: aiohttp.ClientSession
) -> TokenBalance:
"""Get native token balance (ETH, BNB, etc.)"""
try:
balance_wei = self.w3.eth.get_balance(Web3.to_checksum_address(address))
balance = Decimal(balance_wei) / Decimal(10 ** self.config['decimals'])
price = await self.price_oracle.get_native_token_price(self.chain, session)
value_usd = balance * price
return TokenBalance(
chain=self.chain,
address=address,
symbol=self.config['native_token'],
name=self.config['native_token'],
balance=balance,
decimals=self.config['decimals'],
price_usd=price,
value_usd=value_usd,
contract_address=None
)
except Exception as e:
print(f"Error getting native balance on {self.chain}: {e}")
return None
async def get_erc20_balances(
self,
address: str,
session: aiohttp.ClientSession
) -> List[TokenBalance]:
"""Get all ERC-20 token balances"""
balances = []
try:
# Get token list from block explorer API
url = self.config['explorer_api']
params = {
'module': 'account',
'action': 'tokentx',
'address': address,
'startblock': 0,
'endblock': 99999999,
'sort': 'desc',
'apikey': self.config['api_key']
}
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
if data.get('status') == '1':
# Extract unique token contracts
token_contracts = set()
for tx in data.get('result', [])[:100]: # Limit to recent
token_contracts.add(tx['contractAddress'])
# Get balance for each token
for contract_address in token_contracts:
balance = await self._get_token_balance(
address,
contract_address,
session
)
if balance and balance.balance > 0:
balances.append(balance)
except Exception as e:
print(f"Error getting ERC-20 balances on {self.chain}: {e}")
return balances
async def _get_token_balance(
self,
wallet_address: str,
contract_address: str,
session: aiohttp.ClientSession
) -> Optional[TokenBalance]:
"""Get balance for a specific ERC-20 token"""
try:
contract = self.w3.eth.contract(
address=Web3.to_checksum_address(contract_address),
abi=ERC20_ABI
)
balance_raw = contract.functions.balanceOf(
Web3.to_checksum_address(wallet_address)
).call()
if balance_raw == 0:
return None
decimals = contract.functions.decimals().call()
symbol = contract.functions.symbol().call()
balance = Decimal(balance_raw) / Decimal(10 ** decimals)
price = await self.price_oracle.get_token_price(
contract_address,
self.chain,
session
)
value_usd = balance * price
return TokenBalance(
chain=self.chain,
address=wallet_address,
symbol=symbol,
name=symbol,
balance=balance,
decimals=decimals,
price_usd=price,
value_usd=value_usd,
contract_address=contract_address
)
except Exception as e:
print(f"Error getting token balance for {contract_address}: {e}")
return None
async def get_transactions(
self,
address: str,
session: aiohttp.ClientSession,
limit: int = 50
) -> List[Transaction]:
"""Get recent transactions"""
transactions = []
try:
url = self.config['explorer_api']
params = {
'module': 'account',
'action': 'txlist',
'address': address,
'startblock': 0,
'endblock': 99999999,
'sort': 'desc',
'page': 1,
'offset': limit,
'apikey': self.config['api_key']
}
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
if data.get('status') == '1':
for tx in data.get('result', []):
transaction = Transaction(
chain=self.chain,
hash=tx['hash'],
timestamp=datetime.fromtimestamp(int(tx['timeStamp'])),
from_address=tx['from'],
to_address=tx['to'],
value=Decimal(tx['value']) / Decimal(10 ** 18),
gas_used=Decimal(tx['gasUsed']),
gas_price=Decimal(tx['gasPrice']) / Decimal(10 ** 9),
transaction_type=self._categorize_transaction(tx),
tokens_transferred=[],
usd_value=Decimal('0'),
category='transfer'
)
transactions.append(transaction)
except Exception as e:
print(f"Error getting transactions on {self.chain}: {e}")
return transactions
def _categorize_transaction(self, tx: Dict) -> str:
"""Categorize transaction type"""
if tx.get('functionName'):
func_name = tx['functionName'].lower()
if 'swap' in func_name:
return 'swap'
elif 'stake' in func_name or 'deposit' in func_name:
return 'defi'
elif 'mint' in func_name or 'safetransfer' in func_name:
return 'nft'
else:
return 'contract'
return 'transfer'
class DeFiScanner:
"""Scan DeFi protocols for positions"""
def __init__(self, chain: str):
self.chain = chain
self.w3 = Web3(Web3.HTTPProvider(RPC_ENDPOINTS[chain]))
async def scan_lending_protocols(
self,
address: str,
session: aiohttp.ClientSession
) -> List[DeFiPosition]:
"""Scan Aave, Compound, etc. for lending positions"""
positions = []
# Aave V3 scanning
aave_positions = await self._scan_aave_v3(address, session)
positions.extend(aave_positions)
# Compound V3 scanning
compound_positions = await self._scan_compound_v3(address, session)
positions.extend(compound_positions)
return positions
async def _scan_aave_v3(
self,
address: str,
session: aiohttp.ClientSession
) -> List[DeFiPosition]:
"""Scan Aave V3 positions"""
# Simplified - would need full Aave ABI and contract addresses
return []
async def _scan_compound_v3(
self,
address: str,
session: aiohttp.ClientSession
) -> List[DeFiPosition]:
"""Scan Compound V3 positions"""
# Simplified - would need full Compound ABI
return []
async def scan_liquidity_pools(
self,
address: str,
session: aiohttp.ClientSession
) -> List[DeFiPosition]:
"""Scan Uniswap, Curve, etc. for LP positions"""
positions = []
# Uniswap V3 NFT positions
uniswap_positions = await self._scan_uniswap_v3(address, session)
positions.extend(uniswap_positions)
return positions
async def _scan_uniswap_v3(
self,
address: str,
session: aiohttp.ClientSession
) -> List[DeFiPosition]:
"""Scan Uniswap V3 NFT LP positions"""
# Simplified - would need Uniswap V3 position manager contract
return []
class PortfolioTracker:
"""Main portfolio tracking orchestrator"""
def __init__(self):
self.price_oracle = PriceOracle()
self.db_path = os.path.expanduser('~/.crypto-portfolio/portfolio.db')
self._init_database()
def _init_database(self):
"""Initialize SQLite database for historical tracking"""
os.makedirs(os.path.dirname(self.db_path), exist_ok=True)
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# Portfolio snapshots table
cursor.execute('''
CREATE TABLE IF NOT EXISTS portfolio_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
address TEXT NOT NULL,
timestamp INTEGER NOT NULL,
total_value_usd REAL NOT NULL,
chain_data TEXT NOT NULL,
UNIQUE(address, timestamp)
)
''')
# Address labels table
cursor.execute('''
CREATE TABLE IF NOT EXISTS address_labels (
address TEXT PRIMARY KEY,
label TEXT NOT NULL,
category TEXT,
notes TEXT
)
''')
conn.commit()
conn.close()
async def track_wallet(
self,
address: str,
chains: Optional[List[str]] = None,
include_nfts: bool = True,
include_defi: bool = True
) -> PortfolioSnapshot:
"""Track wallet across all chains"""
if chains is None:
chains = ['ethereum', 'bsc', 'polygon', 'arbitrum', 'optimism', 'avalanche']
print(f"\nScanning wallet: {address}")
print("=" * 60)
all_balances = []
all_nfts = []
all_defi = []
all_transactions = []
async with aiohttp.ClientSession() as session:
# Scan each chain
for chain in chains:
print(f"\nScanning {chain}...")
scanner = ChainScanner(chain)
# Get native balance
native_balance = await scanner.get_native_balance(address, session)
if native_balance:
all_balances.append(native_balance)
print(f" {native_balance.symbol}: {native_balance.balance:.4f} "
f"(${native_balance.value_usd:.2f})")
# Get ERC-20 balances
token_balances = await scanner.get_erc20_balances(address, session)
all_balances.extend(token_balances)
for balance in token_balances[:5]: # Show top 5
print(f" {balance.symbol}: {balance.balance:.4f} "
f"(${balance.value_usd:.2f})")
# Get transactions
transactions = await scanner.get_transactions(address, session)
all_transactions.extend(transactions)
# DeFi positions
if include_defi:
defi_scanner = DeFiScanner(chain)
lending_positions = await defi_scanner.scan_lending_protocols(
address, session
)
all_defi.extend(lending_positions)
lp_positions = await defi_scanner.scan_liquidity_pools(
address, session
)
all_defi.extend(lp_positions)
# Calculate totals
total_value = sum(b.value_usd for b in all_balances)
# Chain distribution
chain_distribution = defaultdict(Decimal)
for balance in all_balances:
chain_distribution[balance.chain] += balance.value_usd
# Asset allocation
asset_allocation = {}
sorted_balances = sorted(all_balances, key=lambda x: x.value_usd, reverse=True)
for balance in sorted_balances[:10]:
percentage = (balance.value_usd / total_value * 100) if total_value > 0 else 0
asset_allocation[balance.symbol] = percentage
# Create snapshot
snapshot = PortfolioSnapshot(
address=address,
timestamp=datetime.now(),
total_value_usd=total_value,
token_balances=sorted_balances,
nft_assets=all_nfts,
defi_positions=all_defi,
recent_transactions=sorted(
all_transactions,
key=lambda x: x.timestamp,
reverse=True
)[:50],
chain_distribution=dict(chain_distribution),
asset_allocation=asset_allocation,
analytics=self._calculate_analytics(
sorted_balances,
all_defi,
all_transactions
)
)
# Save to database
self._save_snapshot(snapshot)
return snapshot
def _calculate_analytics(
self,
balances: List[TokenBalance],
defi_positions: List[DeFiPosition],
transactions: List[Transaction]
) -> Dict:
"""Calculate portfolio analytics"""
total_value = sum(b.value_usd for b in balances)
defi_value = sum(p.total_value_usd for p in defi_positions)
# Gas analysis
total_gas = sum(
tx.gas_used * tx.gas_price / Decimal(10 ** 9)
for tx in transactions
)
return {
'token_count': len(balances),
'defi_exposure_pct': (defi_value / total_value * 100) if total_value > 0 else 0,
'total_gas_spent_eth': total_gas,
'avg_gas_per_tx': total_gas / len(transactions) if transactions else 0,
'transaction_count': len(transactions),
'diversification_score': self._calculate_diversification(balances)
}
def _calculate_diversification(self, balances: List[TokenBalance]) -> float:
"""Calculate Herfindahl-Hirschman Index for diversification"""
total_value = sum(b.value_usd for b in balances)
if total_value == 0:
return 0
hhi = sum(
(float(b.value_usd) / float(total_value)) ** 2
for b in balances
)
# Normalize to 0-100 scale (0 = concentrated, 100 = diversified)
return (1 - hhi) * 100
def _save_snapshot(self, snapshot: PortfolioSnapshot):
"""Save portfolio snapshot to database"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO portfolio_snapshots
(address, timestamp, total_value_usd, chain_data)
VALUES (?, ?, ?, ?)
''', (
snapshot.address,
int(snapshot.timestamp.timestamp()),
float(snapshot.total_value_usd),
json.dumps({
'chain_distribution': {
k: float(v) for k, v in snapshot.chain_distribution.items()
},
'asset_allocation': snapshot.asset_allocation,
'analytics': snapshot.analytics
})
))
conn.commit()
conn.close()
def get_historical_performance(
self,
address: str,
days: int = 30
) -> pd.DataFrame:
"""Get historical portfolio performance"""
conn = sqlite3.connect(self.db_path)
cutoff_timestamp = int((datetime.now() - timedelta(days=days)).timestamp())
query = '''
SELECT timestamp, total_value_usd, chain_data
FROM portfolio_snapshots
WHERE address = ? AND timestamp >= ?
ORDER BY timestamp ASC
'''
df = pd.read_sql_query(
query,
conn,
params=(address, cutoff_timestamp)
)
conn.close()
if not df.empty:
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
df.set_index('timestamp', inplace=True)
return df
def export_to_csv(
self,
snapshot: PortfolioSnapshot,
output_path: str
):
"""Export portfolio to CSV"""
# Token balances
balances_df = pd.DataFrame([
{
'Chain': b.chain,
'Symbol': b.symbol,
'Balance': float(b.balance),
'Price_USD': float(b.price_usd),
'Value_USD': float(b.value_usd),
'Contract_Address': b.contract_address or 'Native'
}
for b in snapshot.token_balances
])
# Transactions
transactions_df = pd.DataFrame([
{
'Chain': tx.chain,
'Hash': tx.hash,
'Timestamp': tx.timestamp.isoformat(),
'From': tx.from_address,
'To': tx.to_address,
'Value': float(tx.value),
'Gas_Used': float(tx.gas_used),
'Type': tx.transaction_type,
'Category': tx.category
}
for tx in snapshot.recent_transactions
])
# Export
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
balances_df.to_excel(writer, sheet_name='Balances', index=False)
transactions_df.to_excel(writer, sheet_name='Transactions', index=False)
# Summary
summary_df = pd.DataFrame([
{'Metric': 'Total Value USD', 'Value': float(snapshot.total_value_usd)},
{'Metric': 'Token Count', 'Value': len(snapshot.token_balances)},
{'Metric': 'Transaction Count', 'Value': len(snapshot.recent_transactions)},
{'Metric': 'Timestamp', 'Value': snapshot.timestamp.isoformat()}
])
summary_df.to_excel(writer, sheet_name='Summary', index=False)
print(f"\nPortfolio exported to: {output_path}")
# CLI interface
async def main():
"""Main CLI entry point"""
import argparse
parser = argparse.ArgumentParser(description='Track crypto wallet portfolio')
parser.add_argument('address', help='Wallet address to track')
parser.add_argument('--chains', help='Comma-separated list of chains')
parser.add_argument('--export', help='Export to CSV/Excel file')
parser.add_argument('--historical-days', type=int, default=30,
help='Days of historical data to fetch')
args = parser.parse_args()
tracker = PortfolioTracker()
chains = args.chains.split(',') if args.chains else None
snapshot = await tracker.track_wallet(
address=args.address,
chains=chains
)
# Print summary
print("\n" + "=" * 60)
print(f"Portfolio Summary for {snapshot.address}")
print("=" * 60)
print(f"Total Value: ${snapshot.total_value_usd:,.2f} USD")
print(f"Token Count: {len(snapshot.token_balances)}")
print(f"DeFi Positions: {len(snapshot.defi_positions)}")
print(f"Recent Transactions: {len(snapshot.recent_transactions)}")
print("\nChain Distribution:")
for chain, value in sorted(
snapshot.chain_distribution.items(),
key=lambda x: x[1],
reverse=True
):
pct = (value / snapshot.total_value_usd * 100) if snapshot.total_value_usd > 0 else 0
print(f" {chain}: ${value:,.2f} ({pct:.1f}%)")
print("\nTop Holdings:")
for i, balance in enumerate(snapshot.token_balances[:10], 1):
pct = (balance.value_usd / snapshot.total_value_usd * 100) if snapshot.total_value_usd > 0 else 0
print(f" {i}. {balance.symbol}: {balance.balance:.4f} "
f"(${balance.value_usd:,.2f}) - {pct:.1f}%")
# Export if requested
if args.export:
tracker.export_to_csv(snapshot, args.export)
# Historical performance
if args.historical_days > 0:
historical_df = tracker.get_historical_performance(
args.address,
args.historical_days
)
if not historical_df.empty and len(historical_df) > 1:
first_value = historical_df['total_value_usd'].iloc[0]
last_value = historical_df['total_value_usd'].iloc[-1]
change = last_value - first_value
change_pct = (change / first_value * 100) if first_value > 0 else 0
print(f"\n{args.historical_days}-Day Performance:")
print(f" Starting Value: ${first_value:,.2f}")
print(f" Current Value: ${last_value:,.2f}")
print(f" Change: ${change:,.2f} ({change_pct:+.2f}%)")
if __name__ == '__main__':
asyncio.run(main())
Create a .env file with your API keys:
# Price data
COINGECKO_API_KEY=your_coingecko_api_key
# Block explorers
ETHERSCAN_API_KEY=your_etherscan_api_key
BSCSCAN_API_KEY=your_bscscan_api_key
POLYGONSCAN_API_KEY=your_polygonscan_api_key
ARBISCAN_API_KEY=your_arbiscan_api_key
OPSCAN_API_KEY=your_optimism_scan_key
SNOWTRACE_API_KEY=your_snowtrace_api_key
# RPC endpoints (optional - uses public RPCs by default)
ETH_RPC_URL=https://eth.llamarpc.com
BSC_RPC_URL=https://bsc-dataseed.binance.org
POLYGON_RPC_URL=https://polygon-rpc.com
ARB_RPC_URL=https://arb1.arbitrum.io/rpc
OP_RPC_URL=https://mainnet.optimism.io
AVAX_RPC_URL=https://api.avax.network/ext/bc/C/rpc
SOLANA_RPC_URL=https://api.mainnet-beta.solana.com
pip install web3 aiohttp pandas openpyxl python-bitcoinlib solana
The tracker automatically categorizes transactions for tax purposes:
This comprehensive wallet portfolio tracker provides institutional-grade analysis capabilities for crypto investors and traders across multiple blockchain networks.