Portfolio SDK

pip install requests cryptography

import os
import time
import base64
import requests
from cryptography.hazmat.primitives.asymmetric import ed25519

class PolymarketPortfolioSDK:
    """SDK for Polymarket US Portfolio API"""

    def __init__(self, api_key_id, private_key_base64):
        """
        Initialize the SDK

        Args:
            api_key_id: Your API key ID from developer portal
            private_key_base64: Your base64-encoded Ed25519 private key (64 bytes)
        """
        self.api_key_id = api_key_id
        self.base_url = "https://api.polymarket.us"

        # Parse Ed25519 private key
        private_key_bytes = base64.b64decode(private_key_base64)
        self.private_key = ed25519.Ed25519PrivateKey.from_private_bytes(
            private_key_bytes[:32]
        )

    def _sign_request(self, method, path):
        """Generate Ed25519 signature for request"""
        timestamp = str(int(time.time() * 1000))
        message = f"{timestamp}{method}{path}"
        signature = self.private_key.sign(message.encode('utf-8'))
        signature_base64 = base64.b64encode(signature).decode('utf-8')

        return {
            "X-PM-Access-Key": self.api_key_id,
            "X-PM-Timestamp": timestamp,
            "X-PM-Signature": signature_base64
        }

    def get_positions(self, market_slug=None, limit=None, cursor=None):
        """
        Get user's trading positions

        Args:
            market_slug: Optional market slug to filter by
            limit: Maximum number of positions to return
            cursor: Pagination cursor from previous response

        Returns:
            dict: List of positions with pagination cursor
        """
        path = "/v1/portfolio/positions"
        headers = self._sign_request("GET", path)

        params = {}
        if market_slug:
            params["market"] = market_slug
        if limit:
            params["limit"] = limit
        if cursor:
            params["cursor"] = cursor

        response = requests.get(
            f"{self.base_url}{path}",
            headers=headers,
            params=params
        )
        response.raise_for_status()
        return response.json()

    def get_all_positions(self):
        """
        Get all positions across all pages

        Returns:
            list: Complete list of all positions
        """
        all_positions = []
        cursor = None

        while True:
            response = self.get_positions(cursor=cursor)
            all_positions.extend(response.get("positions", []))

            cursor = response.get("nextCursor")
            if not cursor:
                break

        return all_positions

    def get_activities(self, limit=None, cursor=None, market_slug=None,
                      activity_types=None, sort_order="SORT_ORDER_DESCENDING"):
        """
        Get trading activities and account changes

        Args:
            limit: Maximum number of activities to return
            cursor: Pagination cursor from previous response
            market_slug: Optional market slug to filter by
            activity_types: List of activity types to filter
                (ACTIVITY_TYPE_TRADE, ACTIVITY_TYPE_POSITION_RESOLUTION,
                 ACTIVITY_TYPE_ACCOUNT_DEPOSIT, ACTIVITY_TYPE_ACCOUNT_WITHDRAWAL,
                 ACTIVITY_TYPE_REFERRAL_BONUS, ACTIVITY_TYPE_TRANSFER)
            sort_order: SORT_ORDER_DESCENDING or SORT_ORDER_ASCENDING

        Returns:
            dict: List of activities with pagination cursor
        """
        path = "/v1/portfolio/activities"
        headers = self._sign_request("GET", path)

        params = {"sortOrder": sort_order}
        if limit:
            params["limit"] = limit
        if cursor:
            params["cursor"] = cursor
        if market_slug:
            params["marketSlug"] = market_slug
        if activity_types:
            params["types"] = activity_types

        response = requests.get(
            f"{self.base_url}{path}",
            headers=headers,
            params=params
        )
        response.raise_for_status()
        return response.json()

    def get_trades_only(self, market_slug=None, limit=50):
        """
        Get only trade activities

        Args:
            market_slug: Optional market slug to filter by
            limit: Maximum number of trades to return

        Returns:
            dict: List of trade activities
        """
        return self.get_activities(
            limit=limit,
            market_slug=market_slug,
            activity_types=["ACTIVITY_TYPE_TRADE"]
        )

    def calculate_portfolio_value(self):
        """
        Calculate total portfolio value from positions

        Returns:
            dict: Portfolio summary with total value and P/L
        """
        positions = self.get_all_positions()

        total_cost = 0
        total_value = 0

        for position in positions:
            quantity = float(position.get("quantity", 0))
            avg_price = float(position.get("avgPrice", {}).get("value", 0))
            last_price = float(position.get("lastPrice", {}).get("value", 0))

            cost = quantity * avg_price
            value = quantity * last_price

            total_cost += cost
            total_value += value

        return {
            "totalCost": round(total_cost, 2),
            "totalValue": round(total_value, 2),
            "unrealizedPnL": round(total_value - total_cost, 2),
            "positionCount": len(positions)
        }


# Example usage
if __name__ == "__main__":
    # Load credentials from environment
    api_key_id = os.environ.get("POLYMARKET_API_KEY")
    private_key = os.environ.get("POLYMARKET_PRIVATE_KEY")

    # Initialize SDK
    sdk = PolymarketPortfolioSDK(api_key_id, private_key)

    # Example 1: Get all positions
    positions = sdk.get_positions()
    print(f"Total positions: {len(positions.get('positions', []))}")

    for position in positions.get("positions", [])[:5]:  # Show first 5
        market = position.get("marketMetadata", {})
        print(f"Market: {market.get('title')}")
        print(f"  Quantity: {position.get('quantity')}")
        print(f"  Avg Price: \${position.get('avgPrice', {}).get('value')}")
        print(f"  Last Price: \${position.get('lastPrice', {}).get('value')}")

    # Example 2: Get positions for specific market
    market_positions = sdk.get_positions(market_slug="will-trump-win-2024")
    print(f"\nPositions in market: {len(market_positions.get('positions', []))}")

    # Example 3: Get all positions (paginated)
    all_positions = sdk.get_all_positions()
    print(f"\nTotal positions (all pages): {len(all_positions)}")

    # Example 4: Get recent trading activity
    activities = sdk.get_activities(limit=10)
    print(f"\nRecent activities: {len(activities.get('activities', []))}")

    for activity in activities.get("activities", [])[:3]:  # Show first 3
        print(f"Type: {activity.get('type')}")
        print(f"Time: {activity.get('timestamp')}")

    # Example 5: Get only trades
    trades = sdk.get_trades_only(limit=20)
    print(f"\nRecent trades: {len(trades.get('activities', []))}")

    # Example 6: Calculate portfolio value
    portfolio = sdk.calculate_portfolio_value()
    print(f"\nPortfolio Summary:")
    print(f"  Total Cost: \${portfolio['totalCost']}")
    print(f"  Total Value: \${portfolio['totalValue']}")
    print(f"  Unrealized P/L: \${portfolio['unrealizedPnL']}")
    print(f"  Position Count: {portfolio['positionCount']}")

# Get all positions across multiple pages
all_positions = []
cursor = None

while True:
    response = sdk.get_positions(cursor=cursor, limit=100)
    all_positions.extend(response.get("positions", []))

    cursor = response.get("nextCursor")
    if not cursor:
        break

print(f"Total positions: {len(all_positions)}")

try:
    positions = sdk.get_positions()
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print("Authentication failed - check your API key")
    else:
        print(f"Error: {e.response.text}")