Build a Real-Time Crypto Portfolio Tracker in Python

#python #automation

Build a Real-Time Crypto Portfolio Tracker in Python

Tags: python, cryptocurrency, portfolio, api

What You'll Build

A Python application that tracks your crypto portfolio across multiple wallets and exchanges, with real-time price updates and P&L calculations.

Architecture

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│  Price Feed      │────▶│  Portfolio Engine │────▶│  Dashboard      │
│  (CoinGecko API) │     │  (P&L Tracking)   │     │  (Web/Telegram) │
└─────────────────┘     └──────────────────┘     └─────────────────┘
         │                       │                         │
         ▼                       ▼                         ▼
┌─────────────────────────────────────────────────────────────────┐
│                   SQLite + JSON State                            │
│         Transactions • Holdings • Price History                  │
└─────────────────────────────────────────────────────────────────┘

Core Implementation

import json
import sqlite3
import requests
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass
from pathlib import Path

@dataclass
class Transaction:
    coin: str
    amount: float
    price_usd: float
    tx_type: str  # 'buy', 'sell', 'transfer'
    timestamp: datetime
    fee: float = 0.0
    notes: str = ""

class CryptoPortfolio:
    """Track crypto portfolio with real-time prices."""

    def __init__(self, db_path: str = "portfolio.db"):
        self.db = sqlite3.connect(db_path)
        self._init_db()
        self._price_cache: Dict[str, float] = {}
        self._cache_time: float = 0

    def _init_db(self):
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS transactions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                coin TEXT NOT NULL,
                amount REAL NOT NULL,
                price_usd REAL NOT NULL,
                tx_type TEXT NOT NULL,
                timestamp TEXT NOT NULL,
                fee REAL DEFAULT 0,
                notes TEXT
            );

            CREATE TABLE IF NOT EXISTS price_history (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                coin TEXT NOT NULL,
                price REAL NOT NULL,
                timestamp TEXT NOT NULL
            );
        """)
        self.db.commit()

    def add_transaction(self, tx: Transaction):
        """Record a transaction."""
        self.db.execute(
            """INSERT INTO transactions (coin, amount, price_usd, tx_type, timestamp, fee, notes)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (tx.coin, tx.amount, tx.price_usd, tx.tx_type,
             tx.timestamp.isoformat(), tx.fee, tx.notes)
        )
        self.db.commit()

    def get_holdings(self) -> Dict[str, float]:
        """Calculate current holdings per coin."""
        holdings = {}
        rows = self.db.execute("SELECT coin, amount, tx_type, fee FROM transactions").fetchall()

        for coin, amount, tx_type, fee in rows:
            if coin not in holdings:
                holdings[coin] = 0.0

            if tx_type == "buy":
                holdings[coin] += amount
            elif tx_type == "sell":
                holdings[coin] -= amount
            elif tx_type == "transfer_in":
                holdings[coin] += amount
            elif tx_type == "transfer_out":
                holdings[coin] -= amount

        # Subtract fees (paid in the same coin)
        return {c: max(0, a) for c, a in holdings.items()}

    def fetch_prices(self, coins: List[str]) -> Dict[str, float]:
        """Fetch current prices from CoinGecko."""
        try:
            ids = ",".join(coins)
            resp = requests.get(
                f"https://api.coingecko.com/api/v3/simple/price",
                params={"ids": ids, "vs_currencies": "usd"},
                timeout=10
            )
            if resp.status_code == 200:
                data = resp.json()
                return {coin: data.get(coin, {}).get("usd", 0) for coin in coins}
        except Exception:
            pass
        return {}

    def calculate_pnl(self) -> Dict[str, dict]:
        """Calculate P&L per coin and total."""
        holdings = self.get_holdings()
        prices = self.fetch_prices(list(holdings.keys()))

        results = {}
        total_invested = 0
        total_value = 0

        for coin, amount in holdings.items():
            # Get average buy price
            buys = self.db.execute(
                "SELECT SUM(amount * price_usd), SUM(amount) FROM transactions WHERE coin = ? AND tx_type = 'buy'",
                (coin,)
            ).fetchone()

            avg_price = (buys[0] / buys[1]) if buys[1] > 0 else 0
            invested = avg_price * amount
            current_price = prices.get(coin, 0)
            current_value = current_price * amount
            pnl = current_value - invested
            pnl_pct = (pnl / invested * 100) if invested > 0 else 0

            results[coin] = {
                "amount": amount,
                "avg_price": avg_price,
                "current_price": current_price,
                "invested": invested,
                "value": current_value,
                "pnl": pnl,
                "pnl_pct": pnl_pct,
            }

            total_invested += invested
            total_value += current_value

        results["_total"] = {
            "invested": total_invested,
            "value": total_value,
            "pnl": total_value - total_invested,
            "pnl_pct": ((total_value - total_invested) / total_invested * 100) if total_invested > 0 else 0,
        }

        return results

    def generate_report(self) -> str:
        """Generate human-readable portfolio report."""
        pnl = self.calculate_pnl()

        lines = ["📊 Crypto Portfolio Report\n"]
        lines.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n")

        for coin, data in pnl.items():
            if coin == "_total":
                continue
            emoji = "🟢" if data["pnl"] >= 0 else "🔴"
            lines.append(f"{emoji} {coin.upper()}")
            lines.append(f"  Holdings: {data['amount']:.8f}")
            lines.append(f"  Avg Price: ${data['avg_price']:,.2f}")
            lines.append(f"  Current: ${data['current_price']:,.2f}")
            lines.append(f"  Value: ${data['value']:,.2f}")
            lines.append(f"  P&L: ${data['pnl']:+,.2f} ({data['pnl_pct']:+.1f}%)")
            lines.append("")

        total = pnl["_total"]
        emoji = "🟢" if total["pnl"] >= 0 else "🔴"
        lines.append(f"{emoji} TOTAL")
        lines.append(f"  Invested: ${total['invested']:,.2f}")
        lines.append(f"  Value: ${total['value']:,.2f}")
        lines.append(f"  P&L: ${total['pnl']:+,.2f} ({total['pnl_pct']:+.1f}%)")

        return "\n".join(lines)


# Usage
if __name__ == "__main__":
    portfolio = CryptoPortfolio()

    # Add some transactions
    portfolio.add_transaction(Transaction("bitcoin", 0.05, 45000, "buy", datetime(2024, 1, 15)))
    portfolio.add_transaction(Transaction("ethereum", 2.0, 2500, "buy", datetime(2024, 2, 1)))
    portfolio.add_transaction(Transaction("monero", 10.0, 150, "buy", datetime(2024, 3, 1)))

    # Generate report
    print(portfolio.generate_report())