Portfolio tracker

import os
import time

import requests

BASE = "https://api.0xinsider.com/api/v1"
H = {"Authorization": f"Bearer {os.environ['OXINSIDER_API_KEY']}"}
PERMANENT = {
    "invalid_api_key",
    "subscription_required",
    "forbidden",
    "insufficient_scope",
    "account_locked",
    "bad_request",
    "not_found",
}


def call(method, path, **kwargs):
    """Returns the JSON body of a 2xx. Raises RuntimeError on a permanent error or after 4 tries."""
    for attempt in range(4):
        r = requests.request(method, f"{BASE}{path}", headers=H, timeout=35, **kwargs)
        if r.ok:
            return r.json()
        err = r.json()["error"]
        detail = f'{r.status_code} {err["code"]} {err.get("reason") or ""} {err["message"]}'
        if err["code"] in PERMANENT or err.get("reason") == "monthly_quota_exceeded":
            raise RuntimeError(detail)
        retry_after = r.headers.get("Retry-After")
        time.sleep(int(retry_after) if retry_after else 2**attempt)
    raise RuntimeError(f"{method} {path} still failing: {detail}")

wallets = [
    "0x204f72f35326db932158cba6adff0b9a1da95e14",
    "0x863...",
    "0x885...",
]

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)
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"),
    )