cryptoport.py

#!/usr/bin/env python3
"""
CryptoPort v2.0 - CLI Crypto Portfolio Tracker
Tracks ETH/BTC/USDT balances across multiple chains.
Features: Price Alerts, Historical Tracking, Multi-chain portfolio.
No API keys needed. Uses free public RPCs.

GitHub: https://github.com/Byaigo/cryptoport
Donate: ETH 0x18da907cb9d981bc798acb87ac27b03a2dc3cbb7
"""
import urllib.request, json, sys, time, os
from datetime import datetime

# ===== Configuration =====
CHAINS = {
    "ETH": "https://eth.llamarpc.com",
    "ARB": "https://arb1.arbitrum.io/rpc",
    "BASE": "https://mainnet.base.org",
    "OP": "https://mainnet.optimism.io",
    "POLYGON": "https://polygon-rpc.com",
}

TOKENS = {
    "ETH": {"USDT": (6, "0xdAC17F958D2ee523a2206206994597C13D831ec7"),
            "USDC": (6, "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")},
    "ARB": {"USDT": (6, "0xFd086bC7CD5C481DCC9C85ebE478A1C0b69FCbb9"),
            "USDC": (6, "0xaf88d065e77c8cC2239327C5EDb3A432268e5831")},
    "BASE": {"USDC": (6, "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913")},
}

HISTORY_FILE = os.path.expanduser("~/.cryptoport_history.json")
ALERTS_FILE = os.path.expanduser("~/.cryptoport_alerts.json")

def rpc_call(chain, method, params):
    url = CHAINS[chain]
    data = json.dumps({"jsonrpc":"2.0","method":method,"params":params,"id":1}).encode()
    req = urllib.request.Request(url, data=data,
        headers={"Content-Type":"application/json"})
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return json.loads(resp.read())
    except Exception as e:
        return {"error": str(e)}

def get_balance(chain, address):
    result = rpc_call(chain, "eth_getBalance", [address, "latest"])
    if "error" in result:
        return None
    return int(result.get("result", "0x0"), 16) / 1e18

def get_token_balance(chain, address, token_addr, decimals):
    data = "0x70a08231" + address[2:].zfill(64)
    result = rpc_call(chain, "eth_call", [
        {"to": token_addr, "data": data}, "latest"
    ])
    if "error" in result:
        return None
    raw = result.get("result", "0x0")
    return int(raw, 16) / (10 ** decimals) if raw != "0x" else 0

def get_prices():
    try:
        url = "https://api.coingecko.com/api/v3/simple/price?ids=ethereum,bitcoin,tether,usd-coin&vs_currencies=usd,cny"
        req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
        with urllib.request.urlopen(req, timeout=10) as resp:
            return json.loads(resp.read())
    except:
        return None

def load_history():
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE) as f:
            return json.load(f)
    return []

def save_snapshot(portfolio_data):
    history = load_history()
    snapshot = {
        "timestamp": time.time(),
        "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "assets": portfolio_data
    }
    history.append(snapshot)
    if len(history) > 1000:
        history = history[-1000:]
    with open(HISTORY_FILE, "w") as f:
        json.dump(history, f, indent=2)
    return history

def show_history(days=7):
    history = load_history()
    if not history:
        print("No historical data yet. Run a portfolio check first!")
        return
    cutoff = time.time() - (days * 86400)
    recent = [h for h in history if h["timestamp"] >= cutoff]
    if not recent:
        print(f"No data in the last {days} days. Try --history 30")
        return
    print(f"\n{'='*70}")
    print(f"  Portfolio History (Last {days} days)")
    print(f"  {recent[0]['date']} -> {recent[-1]['date']}")
    print(f"{'='*70}")
    daily = {}
    for snap in recent:
        day = snap["date"][:10]
        if day not in daily:
            daily[day] = []
        daily[day].append(snap["assets"].get("total_usd", 0))
    prev_val = None
    for day in sorted(daily.keys()):
        val = daily[day][-1]
        change = ""
        if prev_val:
            diff = val - prev_val
            pct = (diff / prev_val) * 100 if prev_val > 0 else 0
            arrow = "UP" if diff > 0 else "DOWN" if diff < 0 else "--"
            change = f"{arrow} {pct:+.2f}%"
        prev_val = val
        print(f"  {day:<12} ${val:>13,.2f} {change}")
    first_total = daily[list(daily.keys())[0]][-1] if daily else 0
    last_total = daily[list(daily.keys())[-1]][-1] if daily else 0
    if first_total and last_total:
        total_change = last_total - first_total
        total_pct = (total_change / first_total) * 100 if first_total > 0 else 0
        print(f"\n  Total Change: ${total_change:+,.2f} ({total_pct:+.2f}%)")
    print(f"{'='*70}\n")

def export_csv():
    history = load_history()
    if not history:
        print("No history to export.")
        return None
    csv_path = os.path.expanduser("~/.cryptoport_export.csv")
    with open(csv_path, "w") as f:
        f.write("date,timestamp,total_usd,eth_balance,usdt_balance,usdc_balance\n")
        for snap in history:
            a = snap["assets"]
            f.write(f"{snap['date']},{snap['timestamp']},{a.get('total_usd',0)},"
                    f"{a.get('eth_total',0)},{a.get('usdt_total',0)},{a.get('usdc_total',0)}\n")
    print(f"Exported {len(history)} snapshots to {csv_path}")
    return csv_path

def load_alerts():
    if os.path.exists(ALERTS_FILE):
        with open(ALERTS_FILE) as f:
            return json.load(f)
    return []

def save_alerts(alerts):
    with open(ALERTS_FILE, "w") as f:
        json.dump(alerts, f, indent=2)

def add_alert(asset, direction, price):
    alerts = load_alerts()
    alert = {
        "id": len(alerts) + 1,
        "asset": asset.lower(),
        "direction": direction.lower(),
        "price": float(price),
        "created": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "triggered": False
    }
    alerts.append(alert)
    save_alerts(alerts)
    print(f"Alert #{alert['id']} set: {asset.upper()} {direction} ${price:,.2f}")
    return alert

def list_alerts():
    alerts = load_alerts()
    if not alerts:
        print("No alerts set. Use --alert-add to create one.")
        return
    print(f"\n{'='*50}")
    print(f"  Price Alerts ({len(alerts)} active)")
    print(f"{'='*50}")
    for a in alerts:
        status = "TRIGGERED" if a.get("triggered") else "Waiting"
        print(f"  #{a['id']:02d} | {a['asset'].upper():6s} {a['direction']:5s} ${a['price']:>10,.2f} | {status}")
    print(f"{'='*50}\n")

def remove_alert(alert_id):
    alerts = load_alerts()
    alerts = [a for a in alerts if a["id"] != alert_id]
    save_alerts(alerts)
    print(f"Alert #{alert_id} removed.")

def check_alerts(prices):
    if not prices:
        return
    alerts = load_alerts()
    price_map = {
        "eth": prices.get("ethereum", {}).get("usd"),
        "btc": prices.get("bitcoin", {}).get("usd"),
        "usdt": prices.get("tether", {}).get("usd"),
        "usdc": prices.get("usd-coin", {}).get("usd"),
    }
    updated = False
    for alert in alerts:
        if alert.get("triggered"):
            continue
        current = price_map.get(alert["asset"])
        if current is None:
            continue
        should_trigger = False
        if alert["direction"] == "above" and current > alert["price"]:
            should_trigger = True
        elif alert["direction"] == "below" and current < alert["price"]:
            should_trigger = True
        if should_trigger:
            alert["triggered"] = True
            alert["triggered_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            alert["triggered_price"] = current
            updated = True
            print(f"\n*** ALERT TRIGGERED! ***")
            print(f"   {alert['asset'].upper()} is now ${current:,.2f} ({alert['direction']} ${alert['price']:,.2f})")
    if updated:
        save_alerts(alerts)

def get_full_portfolio(addresses, prices):
    portfolio = {"total_usd": 0, "eth_total": 0, "usdt_total": 0, "usdc_total": 0, "wallets": []}
    if not prices:
        prices = {"ethereum":{"usd":2000}, "tether":{"usd":1}, "usd-coin":{"usd":1}}
    for name, addr in addresses.items():
        wallet = {"name": name, "address": addr, "chains": {}}
        for chain in ["ETH", "ARB", "BASE", "OP", "POLYGON"]:
            eth = get_balance(chain, addr)
            chain_data = {}
            if eth is not None and eth > 0.000001:
                eth_usd = eth * prices.get("ethereum",{}).get("usd",2000)
                portfolio["total_usd"] += eth_usd
                portfolio["eth_total"] += eth
                chain_data["ETH"] = {"balance": eth, "usd": eth_usd}
            for token, (dec, taddr) in TOKENS.get(chain, {}).items():
                bal = get_token_balance(chain, addr, taddr, dec)
                if bal and bal > 0.01:
                    price_key = {"USDT":"tether","USDC":"usd-coin"}.get(token,"")
                    token_price = prices.get(price_key,{}).get("usd",1) if price_key else 1
                    token_usd = bal * token_price
                    portfolio["total_usd"] += token_usd
                    if token == "USDT":
                        portfolio["usdt_total"] += bal
                    elif token == "USDC":
                        portfolio["usdc_total"] += bal
                    chain_data[token] = {"balance": bal, "usd": token_usd}
            if chain_data:
                wallet["chains"][chain] = chain_data
        if wallet["chains"]:
            portfolio["wallets"].append(wallet)
    return portfolio

def print_portfolio(addresses, prices):
    portfolio = get_full_portfolio(addresses, prices)
    print(f"\n{'='*70}")
    print(f"  CryptoPort v2.0 - Portfolio Summary")
    print(f"  {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'='*70}")
    for wallet in portfolio["wallets"]:
        print(f"\n  {wallet['name']} ({wallet['address'][:8]}...{wallet['address'][-6:]})")
        print(f"  {'-'*50}")
        for chain, assets in wallet["chains"].items():
            for asset, data in assets.items():
                cny = data["usd"] * 7.2
                print(f"    {chain:8s} {asset:6s}: {data['balance']:12.6f}  (${data['usd']:8.2f} / Y{cny:8.2f})")
    print(f"\n  {'-'*50}")
    print(f"  Total Value: ${portfolio['total_usd']:,.2f} USD / Y{portfolio['total_usd']*7.2:,.2f} CNY")
    print(f"  ETH: {portfolio['eth_total']:.4f} | USDT: {portfolio['usdt_total']:.2f} | USDC: {portfolio['usdc_total']:.2f}")
    print(f"{'='*70}\n")
    check_alerts(prices)
    save_snapshot(portfolio)
    return portfolio

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(
        description="CryptoPort v2.0 - CLI Portfolio Tracker with Alerts & History",
        epilog="Donate: ETH 0x18da907cb9d981bc798acb87ac27b03a2dc3cbb7"
    )
    parser.add_argument("addresses", nargs="*", help="ETH addresses to track")
    parser.add_argument("--watch", "-w", type=int, default=0, help="Watch mode (seconds)")
    parser.add_argument("--config", "-c", help="Config file with addresses")
    parser.add_argument("--history", type=int, nargs="?", const=7, metavar="DAYS",
                        help="Show portfolio history (default: 7 days)")
    parser.add_argument("--export-csv", action="store_true", help="Export history to CSV")
    parser.add_argument("--alert-add", nargs=3, metavar=("ASSET", "DIRECTION", "PRICE"),
                        help="Add price alert (e.g., --alert-add ETH above 3000)")
    parser.add_argument("--alert-list", action="store_true", help="List all price alerts")
    parser.add_argument("--alert-remove", type=int, metavar="ID", help="Remove an alert by ID")
    args = parser.parse_args()

    if args.alert_list:
        list_alerts()
        sys.exit(0)
    if args.alert_remove:
        remove_alert(args.alert_remove)
        sys.exit(0)
    if args.alert_add:
        asset, direction, price = args.alert_add
        add_alert(asset, direction, price)
        sys.exit(0)
    if args.history is not None:
        show_history(args.history if isinstance(args.history, int) else 7)
        sys.exit(0)
    if args.export_csv:
        export_csv()
        sys.exit(0)

    addrs = {}
    if args.config and os.path.exists(args.config):
        import configparser
        cfg = configparser.ConfigParser()
        cfg.read(args.config)
        for section in cfg.sections():
            addrs[section] = cfg[section].get("address","")
    for i, a in enumerate(args.addresses):
        addrs[f"Wallet{i+1}"] = a

    if not addrs:
        print("""
CryptoPort v2.0 - CLI Portfolio Tracker

  Portfolio:    cryptoport.py ADDR1 ADDR2 ...
  Watch mode:   cryptoport.py ADDR --watch 60
  History:      cryptoport.py --history 30
  Export CSV:   cryptoport.py --export-csv
  Add Alert:    cryptoport.py --alert-add ETH above 3000
  List Alerts:  cryptoport.py --alert-list
  Remove Alert: cryptoport.py --alert-remove 1

  Star: github.com/Byaigo/cryptoport
  Donate: 0x18da907cb9d981bc798acb87ac27b03a2dc3cbb7
""")
        sys.exit(0)

    if args.watch > 0:
        print(f"Watching portfolio every {args.watch}s... (Ctrl+C to stop)")
        try:
            while True:
                os.system('clear' if os.name == 'posix' else 'cls')
                prices = get_prices()
                print_portfolio(addrs, prices)
                time.sleep(args.watch)
        except KeyboardInterrupt:
            print("\nGoodbye!")
    else:
        prices = get_prices()
        print_portfolio(addrs, prices)