Smart-money co-buys
A small, self-contained example that ties the REST and WebSocket APIs together into something useful: it watches the tracked wallets with the biggest average buys and flags any token that 2 or more of them buy within 15 minutes — an early sign of smart money converging on a mint.
Everything runs in memory (no database). The result is written to a co_buys.json
file that updates live as clusters form.
What you'll use
- REST
GET /walletsto rank tracked wallets by average buy size. - WebSocket
/wsto stream those wallets' trades in real time.
Prerequisites
pip install websockets certifi
You also need an API key with some credits (see Authentication).
certifi is only there so TLS verification works on machines with a stale certificate
store (common on Windows); on most systems the standard library is enough.
Set up the shared bits — your key, the endpoints, and a TLS context:
import asyncio, json, random, ssl, time, urllib.request
from pathlib import Path
import websockets
API_KEY = "<your-api-key>"
API_BASE = "https://api.prysmatic-sol.xyz"
WS_URL = "wss://api.prysmatic-sol.xyz/ws"
WINDOW_SECONDS = 15 * 60 # buys must land within this span to count as a cluster
MIN_WALLETS = 2 # how many distinct wallets make it a co-buy
MIN_AVG_BUY_SOL = 2.0 # only watch wallets whose average buy is at least this
MAX_WATCH = 100 # the WebSocket filter accepts up to 100 wallets
try:
import certifi
SSL_CONTEXT = ssl.create_default_context(cafile=certifi.where())
except ImportError:
SSL_CONTEXT = ssl.create_default_context()
Step 1 — Pick the biggest buyers
Page through GET /wallets, read each wallet's behavior.avg_buy_amount, and keep the
ones that buy big. The response tells you when to stop with has_more.
def get(path: str) -> dict:
req = urllib.request.Request(
f"{API_BASE}{path}", headers={"Authorization": f"Bearer {API_KEY}"})
with urllib.request.urlopen(req, timeout=30, context=SSL_CONTEXT) as resp:
return json.loads(resp.read())
def top_buyers() -> list[str]:
wallets = []
page = 1
while True:
data = get(f"/wallets?page={page}")
for item in data["items"]:
alias = item["identity"]["wallet"]
avg_buy = item.get("behavior", {}).get("avg_buy_amount") or 0.0
wallets.append((alias, float(avg_buy)))
if not data.get("has_more"):
break
page += 1
wallets.sort(key=lambda w: w[1], reverse=True)
picked = [a for a, avg in wallets if avg >= MIN_AVG_BUY_SOL][:MAX_WATCH]
print(f"Watching {len(picked)} wallets")
return picked
Step 2 — Detect co-buys in a sliding window
For each mint, keep the buys from the last 15 minutes. When two or more distinct wallets show up in that window, you have a cluster.
class CoBuyTracker:
def __init__(self, watched: list[str]):
self.watched = watched
self.buys: dict[str, list[dict]] = {} # mint -> [{wallet, block_time}]
self.clusters: dict[str, dict] = {} # mint -> cluster record
def add_buy(self, mint: str, wallet: str, block_time: int) -> bool:
events = self.buys.setdefault(mint, [])
cutoff = block_time - WINDOW_SECONDS
events[:] = [e for e in events if e["block_time"] >= cutoff] # drop old
if not any(e["wallet"] == wallet for e in events):
events.append({"wallet": wallet, "block_time": block_time})
distinct = {e["wallet"] for e in events}
if len(distinct) < MIN_WALLETS:
return False
self.clusters[mint] = {
"mint": mint,
"wallets": sorted(distinct),
"first_buy": min(e["block_time"] for e in events),
"last_buy": max(e["block_time"] for e in events),
"buys": sorted(events, key=lambda e: e["block_time"]),
}
return True
def dump(self) -> None:
Path("co_buys.json").write_text(json.dumps({
"generated_at": int(time.time()),
"window_minutes": WINDOW_SECONDS // 60,
"watched_wallets": self.watched,
"co_buys": list(self.clusters.values()),
}, indent=2))
Step 3 — Stream the live feed
Open the WebSocket, subscribe to trades filtered to your watchlist, and feed every
buy into the tracker. Note the small details that make it production-shaped:
- Ignore the
{"type": "ping"}heartbeat and stop onbalance_exhausted. - Reconnect with exponential backoff and jitter instead of a tight loop.
- Re-subscribe on every reconnect — subscriptions do not survive a dropped connection.
async def stream(watched: list[str], tracker: CoBuyTracker):
delay = 1.0
while True:
try:
async with websockets.connect(
WS_URL,
extra_headers={"Authorization": f"Bearer {API_KEY}"},
ssl=SSL_CONTEXT,
) as ws:
await ws.send(json.dumps({
"action": "subscribe", "channels": ["trades"], "wallets": watched,
}))
delay = 1.0
async for raw in ws:
msg = json.loads(raw)
if msg.get("type") == "ping":
continue
if msg.get("type") == "balance_exhausted":
return
if msg.get("channel") != "trades":
continue
t = msg["data"]
if t.get("side") != "buy":
continue
if tracker.add_buy(t["mint"], t["wallet"], int(t["block_time"])):
c = tracker.clusters[t["mint"]]
print(f"CO-BUY {c['mint']} by {c['wallets']}")
tracker.dump()
except websockets.ConnectionClosed:
pass
await asyncio.sleep(delay + random.uniform(0, delay)) # backoff + jitter
delay = min(delay * 2, 30)
Step 4 — Run it
async def main():
watched = top_buyers()
tracker = CoBuyTracker(watched)
tracker.dump()
await stream(watched, tracker)
if __name__ == "__main__":
asyncio.run(main())
Save the snippets above as co_buys.py and run:
python co_buys.py
You'll see the watchlist print, then live CO-BUY lines as clusters form, and a
co_buys.json that looks like this:
{
"generated_at": 1781450000,
"window_minutes": 15,
"watched_wallets": ["W145", "W342", "W57"],
"co_buys": [
{
"mint": "4Sm4YsjRrvSRQ4x1P86ncqumV3ZVgXxtFMKjiL8vpump",
"wallets": ["W57", "W145"],
"first_buy": 1781449880,
"last_buy": 1781449990,
"buys": [
{ "wallet": "W145", "block_time": 1781449880 },
{ "wallet": "W57", "block_time": 1781449990 }
]
}
]
}
Each delivered payload costs 1 credit, but the wallets filter means you only pay for
the wallets you actually watch. Co-buys depend on real on-chain activity, so leave it
running for a while.
Ideas to extend it
- Tighten the window or raise
MIN_WALLETSto surface only stronger signals. - Weight clusters by the wallets'
score(from the Wallets API). - Push clusters to a webhook or a Telegram bot instead of a JSON file.