Marius Mutu
51c9202556
- Bitcoin and Ethereum address tracking - Identifies first purchase from exchanges - Interactive CLI mode with historical price lookup links - Test suite with public addresses - Documentation for Claude Code 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
289 lines
12 KiB
Python
289 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Crypto Address Tracker - Finds Primary Purchases
|
||
Tracks Bitcoin and Ethereum addresses back to first exchange acquisition
|
||
"""
|
||
|
||
import requests
|
||
import time
|
||
from datetime import datetime
|
||
from typing import List, Dict, Optional
|
||
|
||
class CryptoTracker:
|
||
def __init__(self):
|
||
# Known exchange addresses (Bitcoin)
|
||
self.known_exchanges = {
|
||
'binance': ['1NDyJtNTjmwk5xPNhjgAMu4HDHigtobu1s', '34xp4vRoCGJym3xR7yCVPFHoCNxv4Twseo'],
|
||
'coinbase': ['3D2oetdNuZUqQHPJmcMDDHYoqkyNVsFk9r', '3Nxwenay9Z8Lc9JBiywExpnEFiLp6Afp8v'],
|
||
'kraken': ['3BMEXVx3hL3qFAahnWgQ8buSr89oHWqBHX'],
|
||
'bitstamp': ['1Kr6QSydW9bFQG1mXiPNNu6WpJGmUa9i1g']
|
||
}
|
||
|
||
# Known exchange markers (for Ethereum - common patterns)
|
||
self.eth_exchange_markers = [
|
||
'binance', 'coinbase', 'kraken', 'bitstamp', 'bitfinex',
|
||
'okex', 'huobi', 'gemini', 'ftx', 'kucoin'
|
||
]
|
||
|
||
def track_bitcoin_address(self, address: str, max_depth: int = 10) -> Dict:
|
||
"""
|
||
Track a Bitcoin address back to find primary purchase
|
||
Uses blockchain.com API (free, no API key needed)
|
||
"""
|
||
print(f"\n🔍 Tracking Bitcoin address: {address}")
|
||
print("=" * 70)
|
||
|
||
results = {
|
||
'address': address,
|
||
'type': 'Bitcoin',
|
||
'transactions': [],
|
||
'first_purchase': None,
|
||
'exchange_found': None
|
||
}
|
||
|
||
try:
|
||
# Get address info
|
||
url = f"https://blockchain.info/rawaddr/{address}?limit=50"
|
||
response = requests.get(url, timeout=10)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
if 'txs' not in data or len(data['txs']) == 0:
|
||
print("❌ No transactions found for this address")
|
||
return results
|
||
|
||
transactions = data['txs']
|
||
print(f"✅ Found {len(transactions)} transactions\n")
|
||
|
||
# Find first incoming transaction
|
||
incoming_txs = []
|
||
for tx in reversed(transactions): # Start from oldest
|
||
# Check if this transaction sent BTC to our address
|
||
for output in tx['out']:
|
||
if output.get('addr') == address:
|
||
tx_date = datetime.fromtimestamp(tx['time'])
|
||
tx_info = {
|
||
'hash': tx['hash'],
|
||
'date': tx_date,
|
||
'value': output['value'] / 100000000, # Convert satoshi to BTC
|
||
'from_addresses': []
|
||
}
|
||
|
||
# Get source addresses
|
||
for inp in tx['inputs']:
|
||
if 'prev_out' in inp and 'addr' in inp['prev_out']:
|
||
from_addr = inp['prev_out']['addr']
|
||
tx_info['from_addresses'].append(from_addr)
|
||
|
||
# Check if it's a known exchange
|
||
exchange = self._check_exchange(from_addr)
|
||
if exchange:
|
||
tx_info['exchange'] = exchange
|
||
if not results['first_purchase']:
|
||
results['first_purchase'] = tx_info
|
||
results['exchange_found'] = exchange
|
||
|
||
incoming_txs.append(tx_info)
|
||
break
|
||
|
||
results['transactions'] = incoming_txs
|
||
|
||
# Display results
|
||
self._display_bitcoin_results(results)
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Error fetching data: {e}")
|
||
except Exception as e:
|
||
print(f"❌ Unexpected error: {e}")
|
||
|
||
return results
|
||
|
||
def track_ethereum_address(self, address: str, etherscan_api_key: Optional[str] = None) -> Dict:
|
||
"""
|
||
Track an Ethereum address back to find primary purchase
|
||
Free tier: 5 requests/second, no API key needed for basic use
|
||
"""
|
||
print(f"\n🔍 Tracking Ethereum address: {address}")
|
||
print("=" * 70)
|
||
|
||
results = {
|
||
'address': address,
|
||
'type': 'Ethereum',
|
||
'transactions': [],
|
||
'first_purchase': None,
|
||
'exchange_found': None
|
||
}
|
||
|
||
try:
|
||
# Build API URL
|
||
base_url = "https://api.etherscan.io/api"
|
||
params = {
|
||
'module': 'account',
|
||
'action': 'txlist',
|
||
'address': address,
|
||
'startblock': 0,
|
||
'endblock': 99999999,
|
||
'page': 1,
|
||
'offset': 100,
|
||
'sort': 'asc' # Oldest first
|
||
}
|
||
|
||
if etherscan_api_key:
|
||
params['apikey'] = etherscan_api_key
|
||
|
||
response = requests.get(base_url, params=params, timeout=10)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
if data['status'] != '1' or 'result' not in data:
|
||
print("❌ No transactions found or API error")
|
||
return results
|
||
|
||
transactions = data['result']
|
||
print(f"✅ Found {len(transactions)} transactions\n")
|
||
|
||
# Find incoming transactions
|
||
incoming_txs = []
|
||
for tx in transactions:
|
||
# Check if ETH was sent TO our address
|
||
if tx['to'].lower() == address.lower() and tx['value'] != '0':
|
||
tx_date = datetime.fromtimestamp(int(tx['timeStamp']))
|
||
value_eth = int(tx['value']) / 1e18 # Convert wei to ETH
|
||
|
||
tx_info = {
|
||
'hash': tx['hash'],
|
||
'date': tx_date,
|
||
'value': value_eth,
|
||
'from_address': tx['from']
|
||
}
|
||
|
||
# Check for exchange markers in address tags
|
||
# Note: In production, you'd check against Etherscan's address tags
|
||
# For now, we'll flag it for manual review
|
||
incoming_txs.append(tx_info)
|
||
|
||
if not results['first_purchase']:
|
||
results['first_purchase'] = tx_info
|
||
|
||
results['transactions'] = incoming_txs[:10] # First 10 incoming
|
||
|
||
# Display results
|
||
self._display_ethereum_results(results)
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ Error fetching data: {e}")
|
||
except Exception as e:
|
||
print(f"❌ Unexpected error: {e}")
|
||
|
||
return results
|
||
|
||
def _check_exchange(self, address: str) -> Optional[str]:
|
||
"""Check if address belongs to a known exchange"""
|
||
for exchange, addresses in self.known_exchanges.items():
|
||
if address in addresses:
|
||
return exchange
|
||
return None
|
||
|
||
def _display_bitcoin_results(self, results: Dict):
|
||
"""Display Bitcoin tracking results"""
|
||
print("\n📊 REZULTATE:")
|
||
print("=" * 70)
|
||
|
||
if results['first_purchase']:
|
||
fp = results['first_purchase']
|
||
print(f"\n🎯 PRIMA ACHIZIȚIE GĂSITĂ!")
|
||
print(f" Exchange: {fp.get('exchange', 'Unknown').upper()}")
|
||
print(f" Data: {fp['date'].strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Sumă: {fp['value']:.8f} BTC")
|
||
print(f" Hash: {fp['hash']}")
|
||
print(f"\n 🔗 Vezi tranzacția: https://blockchain.com/btc/tx/{fp['hash']}")
|
||
|
||
# Get historical price
|
||
print(f"\n 💰 Pentru preț istoric, verifică:")
|
||
print(f" https://coinmarketcap.com/currencies/bitcoin/historical-data/")
|
||
print(f" Data: {fp['date'].strftime('%Y-%m-%d')}")
|
||
else:
|
||
print("\n⚠️ Nu am găsit exchange-uri cunoscute")
|
||
print(" Primele tranzacții incoming:")
|
||
for i, tx in enumerate(results['transactions'][:3], 1):
|
||
print(f"\n {i}. Data: {tx['date'].strftime('%Y-%m-%d')}")
|
||
print(f" Sumă: {tx['value']:.8f} BTC")
|
||
print(f" De la: {tx['from_addresses'][0] if tx['from_addresses'] else 'Unknown'}")
|
||
print(f" Hash: {tx['hash'][:16]}...")
|
||
|
||
def _display_ethereum_results(self, results: Dict):
|
||
"""Display Ethereum tracking results"""
|
||
print("\n📊 REZULTATE:")
|
||
print("=" * 70)
|
||
|
||
if results['first_purchase']:
|
||
fp = results['first_purchase']
|
||
print(f"\n🎯 PRIMA TRANZACȚIE INCOMING GĂSITĂ!")
|
||
print(f" Data: {fp['date'].strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Sumă: {fp['value']:.6f} ETH")
|
||
print(f" De la: {fp['from_address']}")
|
||
print(f" Hash: {fp['hash']}")
|
||
print(f"\n 🔗 Vezi tranzacția: https://etherscan.io/tx/{fp['hash']}")
|
||
|
||
print(f"\n ℹ️ Verifică manual pe Etherscan dacă adresa de mai sus")
|
||
print(f" este un exchange cunoscut (va avea tag)")
|
||
|
||
print(f"\n 💰 Pentru preț istoric ETH:")
|
||
print(f" https://coinmarketcap.com/currencies/ethereum/historical-data/")
|
||
print(f" Data: {fp['date'].strftime('%Y-%m-%d')}")
|
||
|
||
print("\n 📋 Primele 5 tranzacții incoming:")
|
||
for i, tx in enumerate(results['transactions'][:5], 1):
|
||
print(f"\n {i}. Data: {tx['date'].strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f" Sumă: {tx['value']:.6f} ETH")
|
||
print(f" De la: {tx['from_address'][:16]}...{tx['from_address'][-4:]}")
|
||
|
||
def main():
|
||
"""Main function - interactive mode"""
|
||
print("=" * 70)
|
||
print("🔍 CRYPTO ADDRESS TRACKER - Găsește Prima Achiziție")
|
||
print("=" * 70)
|
||
|
||
tracker = CryptoTracker()
|
||
|
||
print("\nSelectează tipul de crypto:")
|
||
print("1. Bitcoin (BTC)")
|
||
print("2. Ethereum (ETH)")
|
||
|
||
choice = input("\nAlege (1/2): ").strip()
|
||
|
||
if choice == '1':
|
||
address = input("\nIntroduCe adresa Bitcoin: ").strip()
|
||
if not address:
|
||
print("❌ Adresă invalidă!")
|
||
return
|
||
|
||
tracker.track_bitcoin_address(address)
|
||
|
||
elif choice == '2':
|
||
address = input("\nIntroduCe adresa Ethereum (0x...): ").strip()
|
||
if not address.startswith('0x'):
|
||
print("❌ Adresă Ethereum invalidă! Trebuie să înceapă cu 0x")
|
||
return
|
||
|
||
print("\n📝 Optional: Etherscan API key (ENTER pentru a sări):")
|
||
print(" Obține gratuit de la: https://etherscan.io/apis")
|
||
api_key = input(" API Key: ").strip()
|
||
|
||
tracker.track_ethereum_address(address, api_key if api_key else None)
|
||
|
||
else:
|
||
print("❌ Opțiune invalidă!")
|
||
return
|
||
|
||
print("\n" + "=" * 70)
|
||
print("✅ Analiză completă!")
|
||
print("\n💡 Următorii pași:")
|
||
print(" 1. Notează data primei achiziții")
|
||
print(" 2. Verifică prețul istoric pentru acea dată pe CoinMarketCap")
|
||
print(" 3. Acela este prețul tău de achiziție pentru calcul impozit")
|
||
print("=" * 70)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|