Portfolio tracker

import os
import time
from decimal import Decimal

import requests

BASE = "https://api.0xinsider.com/api/v1"
H = {"Authorization": f"Bearer {os.environ['OXINSIDER_API_KEY']}"}
RETRYABLE = {408, 429, 502, 503, 504}


def call(method, path, **kwargs):
    for attempt in range(4):
        response = requests.request(method, f"{BASE}{path}", headers=H, timeout=35, **kwargs)
        if response.ok:
            return response.json(parse_float=Decimal)
        error = response.json()["error"]
        detail = f'{response.status_code} {error["code"]} {error.get("reason") or ""} {error["message"]}'
        if response.status_code not in RETRYABLE or error.get("reason") == "monthly_quota_exceeded":
            raise RuntimeError(detail)
        if attempt == 3:
            raise RuntimeError(f"{method} {path} still failing: {detail}")
        retry_after = response.headers.get("Retry-After")
        time.sleep(int(retry_after) if retry_after else 2**attempt)

wallets = [
    "0x204f72f35326db932158cba6adff0b9a1da95e14",
    # Add other complete wallet addresses here.
]

book = []
for start in range(0, len(wallets), 25):
    batch = call("POST", "/traders/batch", json={"traders": wallets[start:start + 25], "expand": ["trust"]})
    for item in batch["data"]:
        if item["status"] != "ok":
            book.append({"wallet": item["input"], "error": item["error"]["message"]})
            continue
        t = item["data"]
        pnl = t["pnl"]
        freshness = t["trust"]["total_pnl"]["freshness"]
        book.append({
            "wallet": t["address"],
            "name": t.get("username"),
            "grade": t.get("grade"),
            "sync_status": t.get("sync_status"),
            "total": pnl.get("total"),
            "realized": pnl.get("realized"),
            "unrealized": pnl.get("unrealized"),
            "pnl_freshness": freshness["status"],
            "pnl_as_of": freshness.get("as_of"),
        })

from datetime import datetime, timedelta, timezone

MAX_AGE = timedelta(hours=24)
now = datetime.now(timezone.utc)

known = [b for b in book if b.get("total") is not None]
missing = [b for b in book if b.get("total") is None]
stale = [
    b for b in known
    if b["pnl_as_of"] is None
    or now - datetime.fromisoformat(b["pnl_as_of"].replace("Z", "+00:00")) > MAX_AGE
]

subtotal = sum((b["total"] for b in known), Decimal("0"))
if missing:
    print(f"P&L for {len(known)} of {len(book)} wallets: ${subtotal:,.0f}. Book total unavailable.")
    for b in missing:
        print("  no P&L:", b["wallet"], b.get("error") or b.get("sync_status"))
else:
    print(f"Book total: ${subtotal:,.0f} across {len(book)} wallets.")
for b in stale:
    print("  older than 24h or unstamped:", b["wallet"], b["pnl_as_of"])

def positions_for_book(min_size=0):
    held = []
    for start in range(0, len(wallets), 25):
        cursor = None
        while True:
            params = {"wallet": wallets[start:start + 25], "limit": 100, "min_size": min_size}
            if cursor:
                params["cursor"] = cursor
            page = call("GET", "/positions", params=params)
            held += page["data"]
            if not page["has_more"]:
                break
            cursor = page["next_cursor"]
    return held


positions = positions_for_book()
if not positions:
    print(f"No open position in this book of {len(wallets)}.")
for p in positions:
    print(
        p["wallet"][:10], p["market"]["title"], p["side"],
        p["market"].get("outcome_label"), p["shares"], p.get("avg_price"),
        f'${p["current_value_usd"]:,.0f}', p.get("cash_pnl"), p["freshness"], p.get("last_reconciled_at"),
    )