Skip to main content

Python SDK

The Prysmatic Python SDK is the fastest way to build bots, scanners, notebooks, dashboards, and backend jobs on top of the Prysmatic API.

It wraps the REST API and the live WebSocket feed in typed Python clients, while keeping the same product contract as the raw API:

  • Wallets are represented by aliases such as W12.
  • REST calls use Authorization: Bearer <API_KEY>.
  • Live trades use the same /ws endpoint as raw WebSocket clients. The SDK handles the connection details for you.
  • Credits are spent the same way they are spent when calling the API directly.
  • The SDK exposes data primitives; it does not ship a black-box trading signal.

Install

pip install prysmatic-sdk

Until the package is published, install it from GitHub:

pip install git+https://github.com/eserya77/prysmatic-sdk.git

For local development from a cloned checkout:

git clone https://github.com/eserya77/prysmatic-sdk.git
cd prysmatic-sdk
pip install -e ".[dev]"

Create a client

from prysmatic_sdk import PrysmaticClient

client = PrysmaticClient(api_key="prys_...")

You can also use a context manager so the underlying HTTP client is closed automatically:

from prysmatic_sdk import PrysmaticClient

with PrysmaticClient(api_key="prys_...") as client:
wallets = client.wallets.list(page=1)

The default API base is:

https://api.prysmatic-sol.xyz

Override it only when testing against a local or staging deployment:

client = PrysmaticClient(
api_key="prys_...",
base_url="http://127.0.0.1:8090",
)

Wallets

Use the wallet resource to list tracked wallets, inspect a wallet's aggregate performance, and read its current derived holdings.

from prysmatic_sdk import PrysmaticClient

with PrysmaticClient(api_key="prys_...") as client:
page = client.wallets.list(page=1)

for wallet in page.items:
alias = wallet.identity.wallet
score = wallet.identity.score
winrate = wallet.behavior.winrate
avg_buy = wallet.behavior.avg_buy_amount

print(alias, score, winrate, avg_buy)

Fetch one wallet by alias:

metrics = client.wallets.metrics("W12")
print(metrics.pnl.pnl_sol)
print(metrics.behavior.avg_holding_time)

Fetch its current holdings:

holdings = client.wallets.holdings("W12")

for item in holdings.tokens:
print(
item.token.address,
item.position.balance,
item.position.sol_invested,
)

To page through every wallet:

for wallet in client.wallets.iter_all():
print(wallet.identity.wallet, wallet.identity.score)

Tokens

Use the token resource to discover tokens co-held by multiple tracked wallets and inspect tracked-wallet flow around a specific mint.

coheld = client.tokens.held(min_wallets=3, page=1)

for item in coheld.items:
print(
item.token.address,
item.holders.wallet_count,
item.aggregate.sol_invested,
)

Read raw tracked-wallet swaps for one mint:

swaps = client.tokens.swaps(
"FSA7iqBeeENna2LUmZyqCYce7op2jBc9ztXLKdM6bonk",
)

for swap in swaps.items:
print(swap.block_time, swap.wallet, swap.side, swap.quote_amount)

Read one row per tracked wallet instead:

flow = client.tokens.swaps(
"FSA7iqBeeENna2LUmZyqCYce7op2jBc9ztXLKdM6bonk",
aggregate=True,
)

for wallet in flow.wallets:
print(wallet.wallet, wallet.buys, wallet.sells, wallet.pct_held)

Async client

Use AsyncPrysmaticClient when you are already inside an async application.

import asyncio
from prysmatic_sdk import AsyncPrysmaticClient

async def main() -> None:
async with AsyncPrysmaticClient(api_key="prys_...") as client:
wallets = await client.wallets.list(page=1)
print(wallets.total)

asyncio.run(main())

Live trades

The async client also exposes the WebSocket stream. It handles the Prysmatic Bearer authentication, ignores free heartbeat pings, resubscribes after network reconnects, and stops on account-level control messages such as balance_exhausted or superseded.

import asyncio
from prysmatic_sdk import AsyncPrysmaticClient

async def main() -> None:
async with AsyncPrysmaticClient(api_key="prys_...") as client:
async for trade in client.stream.trades(wallets=["W12", "W37"]):
print(
trade.data.block_time,
trade.data.wallet,
trade.data.side,
trade.data.mint,
trade.data.sol_amount,
)

asyncio.run(main())

Omit the wallet filter to receive all live tracked-wallet trades:

async for trade in client.stream.trades():
print(trade.data.wallet, trade.data.mint)

Only delivered trade payloads spend credits. Heartbeats do not spend credits, and trades outside your wallet filter are not delivered.

Error handling

The SDK maps common API responses to typed exceptions.

from prysmatic_sdk import (
InsufficientCreditsError,
PrysmaticClient,
RateLimitedError,
UnauthorizedError,
WalletNotFoundError,
)

client = PrysmaticClient(api_key="prys_...")

try:
wallet = client.wallets.metrics("W12")
except UnauthorizedError:
print("The API key is missing or invalid.")
except InsufficientCreditsError:
print("The account needs more credits.")
except WalletNotFoundError:
print("That wallet alias does not exist.")
except RateLimitedError:
print("Back off before retrying.")

For the live stream, balance_exhausted is raised as InsufficientCreditsError, and superseded is raised as StreamSupersededError.

Build patterns

The SDK is intentionally small. It gives you reliable access to facts, not a prebuilt conclusion. Useful applications usually combine several primitives:

  • Rank wallets by score, winrate, average buy size, or holding time.
  • Watch only a filtered set of wallet aliases over WebSocket.
  • Alert when multiple high-quality wallets buy the same mint inside a time window.
  • Combine tokens.held() with tokens.swaps(..., aggregate=True) to inspect whether smart-wallet exposure is increasing or fading.
  • Persist live trades into your own database and apply your own scoring logic.

Source

The SDK source lives at:

https://github.com/eserya77/prysmatic-sdk

The package is named prysmatic-sdk, and the import name is prysmatic_sdk.