#!/usr/bin/env python3
"""
Blockchain Transaction Tracer for Cost Basis Reconstruction
Searches multiple blockchain APIs for specific transfer amounts on specific dates.
Target transactions (from 1099-DA forms):
- STX: 314.147779 transferred ~5/12/2025 (Stacks blockchain)
- AVAX: 5.15124903 transferred ~5/12/2025 (Avalanche C-Chain)
- ZIL: 4678.83799196 transferred various dates (Zilliqa)
- XRP: 72.98391 transferred various dates (XRP Ledger)
- ADA: 196.134034 transferred various dates (Cardano)
"""
import requests
import json
import time
import sys
from datetime import datetime, timezone, timedelta
from decimal import Decimal
# ── Configuration ──────────────────────────────────────────────────────────────
TARGETS = {
"STX": {
"amount": Decimal("314.147779"),
"micro_amount": 314147779, # STX uses 6 decimal places (micro-STX)
"transfer_date": "2025-05-12",
"search_window_days": 3,
"chain": "stacks",
},
"STX_small": {
"amount": Decimal("0.5"),
"micro_amount": 500000,
"transfer_date": "2025-05-11",
"search_window_days": 3,
"chain": "stacks",
},
"AVAX": {
"amount": Decimal("5.15124903"),
"wei_amount": 5151249030000000000, # 18 decimals
"transfer_date": "2025-05-12",
"search_window_days": 3,
"chain": "avalanche",
},
"XRP": {
"amount": Decimal("72.983910"),
"drops_amount": 72983910, # XRP uses 6 decimal places (drops)
"transfer_date": "2025-12-08", # sold this date, transferred in earlier
"search_window_days": 90, # wider window since "VARIOUS" date
"chain": "xrp",
},
"ZIL": {
"amount": Decimal("4678.83799196"),
"transfer_date": "2025-05-12",
"search_window_days": 30,
"chain": "zilliqa",
},
"ADA": {
"amount": Decimal("196.134034"),
"lovelace_amount": 196134034, # ADA uses 6 decimal places
"transfer_date": "2025-11-28",
"search_window_days": 30,
"chain": "cardano",
},
}
HIRO_API = "https://api.hiro.so"
RESULTS_FILE = "trace_results.json"
# ── Stacks (STX) Scanner ──────────────────────────────────────────────────────
def stx_find_block_at_time(target_ts):
"""Binary search for a Stacks block near a given Unix timestamp."""
print(f" [STX] Binary searching for block at {datetime.fromtimestamp(target_ts, tz=timezone.utc).isoformat()}...")
# Get the latest block for upper bound
resp = requests.get(f"{HIRO_API}/extended/v2/blocks?limit=1&offset=0")
resp.raise_for_status()
data = resp.json()
high = data["results"][0]["height"]
low = 1
best_height = high
best_diff = float("inf")
for _ in range(40): # binary search iterations
mid = (low + high) // 2
try:
resp = requests.get(f"{HIRO_API}/extended/v2/blocks/{mid}")
if resp.status_code == 404:
high = mid - 1
continue
resp.raise_for_status()
block = resp.json()
block_ts = block["burn_block_time"]
diff = abs(block_ts - target_ts)
if diff < best_diff:
best_diff = diff
best_height = mid
if block_ts < target_ts:
low = mid + 1
elif block_ts > target_ts:
high = mid - 1
else:
break
if high - low < 10:
break
except Exception as e:
print(f" Block {mid}: error {e}")
high = mid - 1
continue
time.sleep(0.1) # rate limiting
print(f" [STX] Found block {best_height} (off by {best_diff}s / {best_diff/3600:.1f}h)")
return best_height
def stx_scan_blocks(target_amount_micro, start_block, end_block):
"""Scan Stacks blocks for STX transfers matching a specific amount."""
matches = []
total_blocks = end_block - start_block
checked = 0
transfers_found = 0
print(f" [STX] Scanning blocks {start_block} to {end_block} ({total_blocks} blocks)")
print(f" [STX] Looking for transfers of {target_amount_micro} micro-STX ({target_amount_micro / 1e6:.6f} STX)")
block = start_block
while block <= end_block:
try:
# Get transactions for this block
resp = requests.get(
f"{HIRO_API}/extended/v2/blocks/{block}/transactions",
params={"limit": 50, "offset": 0},
)
if resp.status_code == 404:
block += 1
continue
resp.raise_for_status()
tx_data = resp.json()
total_txs = tx_data.get("total", 0)
offset = 0
while offset < total_txs:
if offset > 0:
resp = requests.get(
f"{HIRO_API}/extended/v2/blocks/{block}/transactions",
params={"limit": 50, "offset": offset},
)
resp.raise_for_status()
tx_data = resp.json()
for tx in tx_data.get("results", []):
tx_id = tx.get("tx_id", "")
# Check token_transfer type transactions
if tx.get("tx_type") == "token_transfer":
transfer = tx.get("token_transfer", {})
amount = int(transfer.get("amount", 0))
transfers_found += 1
# Check for exact match or close match (within 1%)
if amount == target_amount_micro:
match = {
"type": "EXACT_MATCH",
"tx_id": tx_id,
"sender": tx.get("sender_address", ""),
"recipient": transfer.get("recipient_address", ""),
"amount_micro": amount,
"amount_stx": amount / 1e6,
"block_height": block,
"timestamp": tx.get("burn_block_time_iso", ""),
"memo": transfer.get("memo", ""),
}
matches.append(match)
print(f"\n *** EXACT MATCH FOUND ***")
print(f" TX: {tx_id}")
print(f" From: {match['sender']}")
print(f" To: {match['recipient']}")
print(f" Amount: {match['amount_stx']} STX")
print(f" Date: {match['timestamp']}")
elif abs(amount - target_amount_micro) / target_amount_micro < 0.02:
match = {
"type": "CLOSE_MATCH",
"tx_id": tx_id,
"sender": tx.get("sender_address", ""),
"recipient": transfer.get("recipient_address", ""),
"amount_micro": amount,
"amount_stx": amount / 1e6,
"block_height": block,
"timestamp": tx.get("burn_block_time_iso", ""),
"memo": transfer.get("memo", ""),
"diff_pct": abs(amount - target_amount_micro) / target_amount_micro * 100,
}
matches.append(match)
print(f"\n ~ Close match ({match['diff_pct']:.2f}% off): {match['amount_stx']} STX")
print(f" TX: {tx_id}")
# Also check STX transfers embedded in contract calls
events = tx.get("events", [])
for event in events:
if event.get("event_type") == "stx_asset":
asset = event.get("asset", {})
amount = int(asset.get("amount", 0))
if amount == target_amount_micro or (
target_amount_micro > 0
and abs(amount - target_amount_micro) / target_amount_micro < 0.02
):
match = {
"type": "EVENT_MATCH",
"tx_id": tx_id,
"sender": asset.get("sender", ""),
"recipient": asset.get("recipient", ""),
"amount_micro": amount,
"amount_stx": amount / 1e6,
"block_height": block,
"timestamp": tx.get("burn_block_time_iso", ""),
}
matches.append(match)
print(f"\n *** EVENT MATCH FOUND ***")
print(f" TX: {tx_id} | {match['amount_stx']} STX")
offset += 50
time.sleep(0.05)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
print(" [STX] Rate limited, waiting 10s...")
time.sleep(10)
continue
else:
print(f" [STX] Error at block {block}: {e}")
except Exception as e:
print(f" [STX] Error at block {block}: {e}")
block += 1
checked += 1
if checked % 500 == 0:
print(f" [STX] Progress: {checked}/{total_blocks} blocks scanned, {transfers_found} transfers checked, {len(matches)} matches")
time.sleep(0.05) # rate limiting
print(f"\n [STX] Scan complete: {checked} blocks, {transfers_found} transfers checked, {len(matches)} matches found")
return matches
def scan_stacks(target_key):
"""Main entry point for Stacks scanning."""
target = TARGETS[target_key]
target_date = datetime.strptime(target["transfer_date"], "%Y-%m-%d").replace(tzinfo=timezone.utc)
window = target["search_window_days"]
start_ts = int((target_date - timedelta(days=window // 2)).timestamp())
end_ts = int((target_date + timedelta(days=window // 2)).timestamp())
print(f"\n{'='*70}")
print(f"SCANNING STACKS BLOCKCHAIN FOR {target_key}")
print(f"Target amount: {target['amount']} STX ({target['micro_amount']} micro-STX)")
print(f"Date range: {datetime.fromtimestamp(start_ts, tz=timezone.utc).date()} to {datetime.fromtimestamp(end_ts, tz=timezone.utc).date()}")
print(f"{'='*70}")
start_block = stx_find_block_at_time(start_ts)
end_block = stx_find_block_at_time(end_ts)
matches = stx_scan_blocks(target["micro_amount"], start_block, end_block)
return matches
# ── Avalanche (AVAX) Scanner ──────────────────────────────────────────────────
def scan_avalanche(target_key):
"""
Scan Avalanche C-Chain for AVAX transfers.
Uses the Snowtrace/Routescan API (Etherscan-compatible).
"""
target = TARGETS[target_key]
target_date = datetime.strptime(target["transfer_date"], "%Y-%m-%d").replace(tzinfo=timezone.utc)
window = target["search_window_days"]
start_ts = int((target_date - timedelta(days=window // 2)).timestamp())
end_ts = int((target_date + timedelta(days=window // 2)).timestamp())
print(f"\n{'='*70}")
print(f"SCANNING AVALANCHE C-CHAIN FOR {target_key}")
print(f"Target: {target['amount']} AVAX")
print(f"Date range: {datetime.fromtimestamp(start_ts, tz=timezone.utc).date()} to {datetime.fromtimestamp(end_ts, tz=timezone.utc).date()}")
print(f"{'='*70}")
# Avalanche C-Chain has high tx volume, so we need a different approach.
# We'll try the Routescan API which supports searching.
# First, try to find the block range for our date window.
target_wei = target["wei_amount"]
target_avax = float(target["amount"])
matches = []
# Try Routescan API (the current Snowtrace backend)
api_url = "https://api.routescan.io/v2/network/mainnet/evm/43114/etherscan/api"
# Get block number for start timestamp
resp = requests.get(api_url, params={
"module": "block",
"action": "getblocknobytime",
"timestamp": start_ts,
"closest": "after",
})
if resp.status_code == 200:
data = resp.json()
start_block = int(data.get("result", 0))
print(f" [AVAX] Start block: {start_block}")
else:
print(f" [AVAX] Could not get start block: {resp.status_code}")
return matches
resp = requests.get(api_url, params={
"module": "block",
"action": "getblocknobytime",
"timestamp": end_ts,
"closest": "before",
})
if resp.status_code == 200:
data = resp.json()
end_block = int(data.get("result", 0))
print(f" [AVAX] End block: {end_block}")
else:
print(f" [AVAX] Could not get end block: {resp.status_code}")
return matches
total_blocks = end_block - start_block
print(f" [AVAX] Block range: {start_block} to {end_block} ({total_blocks} blocks)")
print(f" [AVAX] Note: Avalanche has ~2 sec blocks, so {total_blocks} blocks is expected for {window} days")
# For Avalanche, we can't scan every block (too many).
# Instead, we'll try to search known OKX addresses.
# Known OKX hot wallet on EVM chains: 0x4b4e14a3773ee558b6597070797fd51eb48606e5
okx_addresses = [
"0x4b4e14a3773ee558b6597070797fd51eb48606e5", # OKX Hot Wallet (Ethereum, may be same on AVAX)
"0x6cc5f688a315f3dc28a7781717a9a798a59fda7b", # OKX (Ethereum, may be same on AVAX)
]
for addr in okx_addresses:
print(f"\n [AVAX] Checking OKX address: {addr}")
try:
resp = requests.get(api_url, params={
"module": "account",
"action": "txlist",
"address": addr,
"startblock": start_block,
"endblock": end_block,
"sort": "asc",
})
time.sleep(0.3)
if resp.status_code != 200:
print(f" [AVAX] API returned {resp.status_code}")
continue
data = resp.json()
txs = data.get("result", [])
if isinstance(txs, str):
print(f" [AVAX] API message: {txs}")
continue
print(f" [AVAX] Found {len(txs)} transactions for {addr}")
for tx in txs:
value = int(tx.get("value", 0))
if value == 0:
continue
avax_value = value / 1e18
# Check for match within 1%
if abs(avax_value - target_avax) / target_avax < 0.02:
ts = int(tx.get("timeStamp", 0))
match = {
"type": "MATCH",
"tx_hash": tx.get("hash", ""),
"from": tx.get("from", ""),
"to": tx.get("to", ""),
"value_wei": value,
"value_avax": avax_value,
"block": tx.get("blockNumber", ""),
"timestamp": datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() if ts else "",
}
matches.append(match)
print(f"\n *** AVAX MATCH FOUND ***")
print(f" TX: {match['tx_hash']}")
print(f" From: {match['from']}")
print(f" To: {match['to']}")
print(f" Amount: {match['value_avax']} AVAX")
print(f" Date: {match['timestamp']}")
except Exception as e:
print(f" [AVAX] Error checking {addr}: {e}")
if not matches:
print(f" [AVAX] No matches found via known OKX addresses.")
print(f" [AVAX] The transfer may have been internal (within OKX) or used a different address.")
return matches
# ── XRP Ledger Scanner ────────────────────────────────────────────────────────
def scan_xrp(target_key):
"""
Search XRP Ledger for specific payment amounts.
Uses the XRP Ledger JSON-RPC API.
"""
target = TARGETS[target_key]
target_amount = float(target["amount"])
print(f"\n{'='*70}")
print(f"SCANNING XRP LEDGER FOR {target_key}")
print(f"Target: {target['amount']} XRP")
print(f"{'='*70}")
matches = []
# Known exchange XRP addresses (exchanges use shared addresses with destination tags)
known_addresses = {
"rLNaPoKeeBjZe2qs6x52yVPZpZ8td4dc6w": "Possible Coinbase",
"rLHzPsX6oXkzU2qL12kHCH8G8cnZv1rBJh": "Kraken",
"rEy8TFcrAPvhpKrwyrscNYyqBGUkE9hKaJ": "Coinbase (reported)",
"rhub8VRN55s94qWKDv6jmDy1pUykJzF3wq": "GateHub",
}
# Try XRP Ledger public API
xrpl_api = "https://s1.ripple.com:51234/"
for addr, label in known_addresses.items():
print(f"\n [XRP] Checking {label}: {addr}")
try:
# Get account transactions
payload = {
"method": "account_tx",
"params": [{
"account": addr,
"ledger_index_min": -1,
"ledger_index_max": -1,
"limit": 200,
"forward": False,
}]
}
resp = requests.post(xrpl_api, json=payload, timeout=30)
if resp.status_code != 200:
print(f" [XRP] API returned {resp.status_code}")
continue
data = resp.json()
result = data.get("result", {})
txs = result.get("transactions", [])
print(f" [XRP] Got {len(txs)} recent transactions")
for tx_entry in txs:
tx = tx_entry.get("tx", tx_entry.get("tx_blob", {}))
if isinstance(tx, str):
continue
if tx.get("TransactionType") != "Payment":
continue
# Amount can be a string (XRP drops) or object (issued currency)
amount = tx.get("Amount", "0")
if isinstance(amount, dict):
continue # Skip non-XRP payments
xrp_amount = int(amount) / 1e6
if abs(xrp_amount - target_amount) / target_amount < 0.02:
close_time = tx_entry.get("close_time_iso", "")
match = {
"type": "MATCH",
"tx_hash": tx.get("hash", ""),
"from": tx.get("Account", ""),
"to": tx.get("Destination", ""),
"destination_tag": tx.get("DestinationTag", ""),
"amount_xrp": xrp_amount,
"timestamp": close_time,
"label": label,
}
matches.append(match)
print(f"\n *** XRP MATCH FOUND ***")
print(f" TX: {match['tx_hash']}")
print(f" From: {match['from']}")
print(f" To: {match['to']} ({label})")
print(f" Tag: {match['destination_tag']}")
print(f" Amount: {match['amount_xrp']} XRP")
print(f" Date: {match['timestamp']}")
time.sleep(0.5)
except Exception as e:
print(f" [XRP] Error checking {addr}: {e}")
return matches
# ── Historical Price Lookup ───────────────────────────────────────────────────
def lookup_historical_prices():
"""Fetch historical crypto prices for known acquisition dates."""
print(f"\n{'='*70}")
print("LOOKING UP HISTORICAL PRICES FOR COST BASIS ESTIMATION")
print(f"{'='*70}")
prices = {}
# Use CoinGecko (may need API key) or alternatives
price_lookups = [
("avalanche-2", "2021-03-19", "AVAX"),
("blockstack", "2021-02-23", "STX"),
("cardano", "2021-12-04", "ADA"),
("zilliqa", "2021-03-19", "ZIL"),
("ripple", "2021-06-15", "XRP (mid-2021 estimate)"),
]
for coin_id, date_str, label in price_lookups:
# Try CoinGecko free API
parts = date_str.split("-")
cg_date = f"{parts[2]}-{parts[1]}-{parts[0]}"
url = f"https://api.coingecko.com/api/v3/coins/{coin_id}/history?date={cg_date}"
print(f"\n Fetching {label} price on {date_str}...")
try:
resp = requests.get(url, timeout=15)
if resp.status_code == 200:
data = resp.json()
price = data.get("market_data", {}).get("current_price", {}).get("usd")
if price:
prices[label] = {"date": date_str, "price_usd": price}
print(f" {label} on {date_str}: ${price:.6f}")
else:
print(f" {label}: price data not found in response")
elif resp.status_code == 401:
print(f" {label}: CoinGecko requires API key, trying alternative...")
# Try CoinCap as alternative
alt_url = f"https://api.coincap.io/v2/assets/{coin_id.replace('-2', '')}/history?interval=d1&start={int(datetime.strptime(date_str, '%Y-%m-%d').timestamp()) * 1000}&end={int((datetime.strptime(date_str, '%Y-%m-%d') + timedelta(days=1)).timestamp()) * 1000}"
alt_resp = requests.get(alt_url, timeout=15)
if alt_resp.status_code == 200:
alt_data = alt_resp.json()
data_points = alt_data.get("data", [])
if data_points:
price = float(data_points[0].get("priceUsd", 0))
prices[label] = {"date": date_str, "price_usd": price}
print(f" {label} on {date_str}: ${price:.6f} (from CoinCap)")
else:
print(f" {label}: no data from CoinCap either")
else:
print(f" {label}: CoinCap returned {alt_resp.status_code}")
else:
print(f" {label}: HTTP {resp.status_code}")
time.sleep(1) # rate limiting
except Exception as e:
print(f" {label}: Error - {e}")
return prices
# ── Main ──────────────────────────────────────────────────────────────────────
def main():
all_results = {}
print("\n" + "=" * 70)
print("CRYPTOCURRENCY TRANSACTION TRACER")
print("Cost Basis Reconstruction for Alexis Mattos - Tax Year 2025")
print("=" * 70)
# Parse command-line args
chains_to_scan = sys.argv[1:] if len(sys.argv) > 1 else ["stx", "avax", "xrp", "prices"]
# 1. Scan Stacks blockchain for STX transfers
if "stx" in chains_to_scan or "all" in chains_to_scan:
try:
stx_matches = scan_stacks("STX")
all_results["STX_314"] = stx_matches
except Exception as e:
print(f" [STX] Scanner failed: {e}")
all_results["STX_314"] = {"error": str(e)}
# 2. Scan Avalanche for AVAX transfers
if "avax" in chains_to_scan or "all" in chains_to_scan:
try:
avax_matches = scan_avalanche("AVAX")
all_results["AVAX"] = avax_matches
except Exception as e:
print(f" [AVAX] Scanner failed: {e}")
all_results["AVAX"] = {"error": str(e)}
# 3. Scan XRP Ledger
if "xrp" in chains_to_scan or "all" in chains_to_scan:
try:
xrp_matches = scan_xrp("XRP")
all_results["XRP"] = xrp_matches
except Exception as e:
print(f" [XRP] Scanner failed: {e}")
all_results["XRP"] = {"error": str(e)}
# 4. Historical price lookup
if "prices" in chains_to_scan or "all" in chains_to_scan:
try:
prices = lookup_historical_prices()
all_results["historical_prices"] = prices
except Exception as e:
print(f" [PRICES] Lookup failed: {e}")
all_results["historical_prices"] = {"error": str(e)}
# Save results
with open(RESULTS_FILE, "w") as f:
json.dump(all_results, f, indent=2, default=str)
print(f"\nResults saved to {RESULTS_FILE}")
# Summary
print(f"\n{'='*70}")
print("SUMMARY")
print(f"{'='*70}")
for key, result in all_results.items():
if isinstance(result, list):
print(f" {key}: {len(result)} matches found")
for m in result:
if isinstance(m, dict):
print(f" - {m.get('type', '?')}: {m.get('amount_stx', m.get('value_avax', m.get('amount_xrp', '?')))} "
f"| {m.get('timestamp', '?')}")
print(f" From: {m.get('sender', m.get('from', '?'))}")
print(f" To: {m.get('recipient', m.get('to', '?'))}")
elif isinstance(result, dict) and "error" not in result:
print(f" {key}:")
for label, info in result.items():
if isinstance(info, dict):
print(f" {label}: ${info.get('price_usd', '?'):.4f} on {info.get('date', '?')}")
print(f"\n{'='*70}")
print("COST BASIS ESTIMATION (from known per-unit costs)")
print(f"{'='*70}")
print(f" AVAX: 5.15124903 units × $38.75/unit (from Account 1) = ${5.15124903 * 38.75:.2f}")
print(f" STX: 314.147779 units × $0.74/unit (from Account 1) = ${314.147779 * 0.74:.2f}")
print(f" ADA: 196.134034 units × $1.41/unit (from Binance known lot) = ${196.134034 * 1.41:.2f}")
print(f" ")
print(f" These estimates are based on known per-unit costs from the same")
print(f" exchange accounts where the assets were originally purchased.")
if __name__ == "__main__":
main()