How to Build a Crypto Portfolio Tracker with Web Scraping
Tracking your crypto portfolio across multiple exchanges and wallets is tedious. Let's build an automated tracker that scrapes real-time prices, calculates your P&L, and sends you alerts.
What We're Building
- Real-time price scraping from CoinGecko and CoinMarketCap
- Portfolio value calculation with cost basis tracking
- Price alerts via email or Telegram
- Historical performance charts
Setup
pip install requests beautifulsoup4 pandas schedule
The Price Scraper
import requests
import time
from datetime import datetime
class CryptoScraper:
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
})
def get_prices_coingecko(self, coins):
"""Fetch prices from CoinGecko API (free tier)."""
ids = ",".join(coins)
url = f"https://api.coingecko.com/api/v3/simple/price"
params = {
"ids": ids,
"vs_currencies": "usd",
"include_24hr_change": "true",
"include_market_cap": "true"
}
response = self.session.get(url, params=params)
return response.json()
def scrape_coinmarketcap(self, limit=100):
"""Scrape top coins from CoinMarketCap."""
url = "https://coinmarketcap.com"
# Use ScraperAPI for JS-rendered content
proxy_url = f"http://api.scraperapi.com?api_key=YOUR_KEY&url={url}&render=true"
response = self.session.get(proxy_url)
from bs4 import BeautifulSoup
soup = BeautifulSoup(response.text, "html.parser")
coins = []
rows = soup.select("table tbody tr")[:limit]
for row in rows:
cells = row.find_all("td")
if len(cells) >= 4:
coins.append({
"rank": cells[1].text.strip(),
"name": cells[2].text.strip().split("\n")[0],
"price": cells[3].text.strip(),
"change_24h": cells[4].text.strip() if len(cells) > 4 else "N/A"
})
return coins
Portfolio Manager
import json
import pandas as pd
class Portfolio:
def __init__(self, holdings_file="holdings.json"):
self.holdings_file = holdings_file
self.holdings = self.load_holdings()
self.scraper = CryptoScraper()
def load_holdings(self):
try:
with open(self.holdings_file) as f:
return json.load(f)
except FileNotFoundError:
return {}
def save_holdings(self):
with open(self.holdings_file, "w") as f:
json.dump(self.holdings, f, indent=2)
def add_holding(self, coin_id, amount, cost_basis):
"""Add or update a holding."""
if coin_id in self.holdings:
existing = self.holdings[coin_id]
total_amount = existing["amount"] + amount
total_cost = (existing["amount"] * existing["cost_basis"]) + (amount * cost_basis)
existing["amount"] = total_amount
existing["cost_basis"] = total_cost / total_amount
else:
self.holdings[coin_id] = {
"amount": amount,
"cost_basis": cost_basis
}
self.save_holdings()
def get_portfolio_value(self):
"""Calculate current portfolio value and P&L."""
coin_ids = list(self.holdings.keys())
prices = self.scraper.get_prices_coingecko(coin_ids)
results = []
total_value = 0
total_cost = 0
for coin_id, holding in self.holdings.items():
current_price = prices.get(coin_id, {}).get("usd", 0)
value = holding["amount"] * current_price
cost = holding["amount"] * holding["cost_basis"]
pnl = value - cost
pnl_pct = (pnl / cost * 100) if cost > 0 else 0
results.append({
"coin": coin_id,
"amount": holding["amount"],
"price": current_price,
"value": round(value, 2),
"cost": round(cost, 2),
"pnl": round(pnl, 2),
"pnl_pct": round(pnl_pct, 2)
})
total_value += value
total_cost += cost
df = pd.DataFrame(results)
print(f"\nPortfolio Value: ${total_value:,.2f}")
print(f"Total Cost: ${total_cost:,.2f}")
print(f"Total P&L: ${total_value - total_cost:,.2f}")
print(f"\n{df.to_string(index=False)}")
return df
# Usage
portfolio = Portfolio()
portfolio.add_holding("bitcoin", 0.5, 42000)
portfolio.add_holding("ethereum", 10, 2200)
portfolio.add_holding("solana", 100, 95)
portfolio.get_portfolio_value()
Price Alerts
import schedule
import smtplib
from email.mime.text import MIMEText
class PriceAlert:
def __init__(self, portfolio):
self.portfolio = portfolio
self.alerts = []
def add_alert(self, coin_id, target_price, direction="above"):
self.alerts.append({
"coin_id": coin_id,
"target": target_price,
"direction": direction
})
def check_alerts(self):
prices = self.portfolio.scraper.get_prices_coingecko(
[a["coin_id"] for a in self.alerts]
)
for alert in self.alerts:
current = prices.get(alert["coin_id"], {}).get("usd", 0)
triggered = (
(alert["direction"] == "above" and current >= alert["target"]) or
(alert["direction"] == "below" and current <= alert["target"])
)
if triggered:
print(f"ALERT: {alert['coin_id']} is ${current} ({alert['direction']} ${alert['target']}!)")
self.send_notification(alert, current)
def send_notification(self, alert, current_price):
msg = f"{alert['coin_id']} hit ${current_price} (target: {alert['direction']} ${alert['target']})"
print(f"Notification: {msg}")
# Set up alerts
alert_system = PriceAlert(portfolio)
alert_system.add_alert("bitcoin", 100000, "above")
alert_system.add_alert("ethereum", 2000, "below")
# Check every 5 minutes
schedule.every(5).minutes.do(alert_system.check_alerts)
while True:
schedule.run_pending()
time.sleep(1)
Handling Rate Limits and Anti-Bot Protection
When scraping crypto sites at scale:
- ScraperAPI handles proxy rotation and CAPTCHA solving — essential for CoinMarketCap
- ThorData provides residential proxies for geo-specific price data
- ScrapeOps monitors your scrapers and alerts on failures
Conclusion
With this tracker, you've got real-time portfolio monitoring, P&L calculations, and price alerts — all powered by web scraping. Extend it with a web dashboard using Flask or Streamlit for a complete solution.
Follow for more Python automation tutorials!
Top comments (0)