| name | trader-analysis |
| description | Analyze Polymarket traders, identify profitable traders to follow, and track their performance. Use when building copy trading features or trader discovery. |
Trader Analysis Skill
Tracking Trader Activity
On-Chain Data
from web3 import Web3
import httpx
from typing import AsyncIterator
CTF_EXCHANGE = "0x4bFb41d5B3570DeFd03C39a9A4D8dE6Bd8B8982E"
class TraderTracker:
def __init__(self, polygon_rpc: str):
self.w3 = Web3(Web3.HTTPProvider(polygon_rpc))
self.exchange = self.w3.eth.contract(
address=CTF_EXCHANGE,
abi=CTF_EXCHANGE_ABI
)
async def get_trader_trades(
self,
address: str,
from_block: int = None
) -> list[dict]:
"""Fetch all trades for an address."""
events = self.exchange.events.OrderFilled.get_logs(
fromBlock=from_block or "earliest",
argument_filters={"maker": address}
)
return [self._parse_trade_event(e) for e in events]
def _parse_trade_event(self, event: dict) -> dict:
"""Parse OrderFilled event into trade dict."""
return {
"tx_hash": event.transactionHash.hex(),
"block_number": event.blockNumber,
"maker": event.args.maker,
"taker": event.args.taker,
"token_id": str(event.args.tokenId),
"amount": event.args.amount / 1e6,
"price": event.args.price / 1e18,
"side": "BUY" if event.args.side == 0 else "SELL",
"timestamp": self._get_block_timestamp(event.blockNumber)
}
Polymarket Data API
class PolymarketDataClient:
BASE_URL = "https://data-api.polymarket.com"
def __init__(self):
self.client = httpx.AsyncClient(
base_url=self.BASE_URL,
timeout=30.0
)
async def get_trader_profile(self, address: str) -> dict:
"""Fetch trader profile and stats."""
response = await self.client.get(f"/users/{address}")
response.raise_for_status()
return response.json()
async def get_trader_positions(self, address: str) -> list[dict]:
"""Get all positions for a trader."""
response = await self.client.get(
"/positions",
params={"user": address}
)
response.raise_for_status()
return response.json()
async def get_trader_activity(
self,
address: str,
limit: int = 100,
offset: int = 0
) -> list[dict]:
"""Get recent trading activity."""
response = .client.get(
,
params={
: address,
: limit,
: offset
}
)
response.raise_for_status()
response.json()
() -> []:
response = .client.get(
,
params={: period, : limit}
)
response.raise_for_status()
response.json()
Trader Scoring System
from dataclasses import dataclass
from datetime import datetime, timedelta
import numpy as np
from typing import Optional
@dataclass
class TraderMetrics:
address: str
total_pnl: float
realized_pnl: float
unrealized_pnl: float
win_rate: float
avg_return_per_trade: float
sharpe_ratio: float
total_trades: int
unique_markets: int
avg_position_size: float
avg_hold_time: timedelta
consistency_score: float
recency_score: float
largest_win: float
largest_loss: float
profit_factor: float
class TraderAnalyzer:
def __init__(self, data_client: PolymarketDataClient):
self.client = data_client
async def analyze_trader(
self,
address: str,
days: int = 90
) -> TraderMetrics:
"""Comprehensive trader analysis."""
activity = await self.client.get_trader_activity(
address, limit=1000
)
positions = .client.get_trader_positions(address)
cutoff = datetime.utcnow() - timedelta(days=days)
recent_trades = [
t t activity
datetime.fromisoformat(t[]) > cutoff
]
TraderMetrics(
address=address,
total_pnl=._calculate_total_pnl(positions, recent_trades),
realized_pnl=._calculate_realized_pnl(recent_trades),
unrealized_pnl=._calculate_unrealized_pnl(positions),
win_rate=._calculate_win_rate(recent_trades),
avg_return_per_trade=._calculate_avg_return(recent_trades),
sharpe_ratio=._calculate_sharpe(recent_trades),
total_trades=(recent_trades),
unique_markets=((t[] t recent_trades)),
avg_position_size=._calculate_avg_size(recent_trades),
avg_hold_time=._calculate_avg_hold_time(recent_trades),
consistency_score=._calculate_consistency(recent_trades),
recency_score=._calculate_recency_score(recent_trades),
largest_win=((t.get(, ) t recent_trades), default=),
largest_loss=((t.get(, ) t recent_trades), default=),
profit_factor=._calculate_profit_factor(recent_trades)
)
() -> :
trades:
winning = ( t trades t.get(, ) > )
winning / (trades)
() -> :
returns = [t.get(, ) t trades t]
(returns) < :
mean_return = np.mean(returns)
std_return = np.std(returns)
std_return == :
(mean_return * **) / std_return
() -> :
(trades) < :
weekly_pnl = {}
trade trades:
week = datetime.fromisoformat(trade[]).isocalendar()[:]
weekly_pnl[week] = weekly_pnl.get(week, ) + trade.get(, )
(weekly_pnl) < :
profitable_weeks = ( pnl weekly_pnl.values() pnl > )
profitable_weeks / (weekly_pnl)
() -> :
trades:
latest = (
datetime.fromisoformat(t[]) t trades
)
days_since = (datetime.utcnow() - latest).days
(, - (days_since / ))
() -> :
gross_profit = (t.get(, ) t trades t.get(, ) > )
gross_loss = ((t.get(, ) t trades t.get(, ) < ))
gross_loss == :
() gross_profit >
gross_profit / gross_loss
:
():
.weights = weights {
: ,
: ,
: ,
: ,
: ,
: ,
: ,
:
}
() -> :
scores = {
: ._normalize_pnl(metrics.total_pnl),
: metrics.win_rate * ,
: ._normalize_sharpe(metrics.sharpe_ratio),
: metrics.consistency_score * ,
: metrics.recency_score * ,
: ._normalize_profit_factor(metrics.profit_factor),
: ._normalize_trades(metrics.total_trades),
: ._normalize_markets(metrics.unique_markets)
}
(scores[k] * .weights[k] k .weights)
() -> :
pnl <= :
(, + pnl / )
(, + np.log1p(pnl) * )
() -> :
(, (, sharpe * ))
() -> :
pf == ():
(, pf * )
() -> :
(, trades)
() -> :
(, markets * )
Finding Traders to Follow
class TraderDiscovery:
def __init__(
self,
data_client: PolymarketDataClient,
analyzer: TraderAnalyzer
):
self.client = data_client
self.analyzer = analyzer
self.scorer = TraderScorer()
async def find_top_traders(
self,
min_trades: int = 50,
min_pnl: float = 1000,
min_win_rate: float = 0.5,
days: int = 30
) -> list[tuple[str, float, TraderMetrics]]:
"""Discover top performing traders."""
leaderboard = await self.client.get_leaderboard(limit=500)
candidates = []
for trader in leaderboard:
try:
metrics = await self.analyzer.analyze_trader(
trader["address"],
days=days
)
if (metrics.total_trades >= min_trades and
metrics.total_pnl >= min_pnl and
metrics.win_rate >= min_win_rate):
score = self.scorer.calculate_score(metrics)
candidates.append((trader["address"], score, metrics))
Exception e:
(candidates, key= x: x[], reverse=)
() -> []:
() -> []:
leaderboard = .client.get_leaderboard(limit=)
original_traders = []
trader leaderboard:
activity = .client.get_trader_activity(
trader[],
limit=
)
originality = ._calculate_originality(activity)
originality >= min_originality_score:
original_traders.append(trader[])
original_traders
() -> :
:
():
.client = data_client
.trading = trading_service
.tracked_traders: [, ] = {}
.copy_delay = config.get(, )
.size_multiplier = config.get(, )
.max_position_pct = config.get(, )
():
.tracked_traders[address] = {
: multiplier .size_multiplier,
: markets,
:
}
():
.tracked_traders.pop(address, )
():
address = trade[]
address .tracked_traders:
config = .tracked_traders[address]
config[] trade[] config[]:
asyncio.sleep(.copy_delay)
size = trade[] * config[]
portfolio = .trading.get_portfolio()
max_size = portfolio[] * .max_position_pct / trade[]
size = (size, max_size)
.trading.place_order(
token_id=trade[],
side=trade[],
price=trade[],
size=size,
metadata={: address}
)
Real-Time Monitoring
import asyncio
from collections import defaultdict
from typing import Callable, Awaitable
class LiveTraderMonitor:
def __init__(self, tracked_addresses: list[str]):
self.tracked = set(tracked_addresses)
self.callbacks: dict[str, list[Callable]] = defaultdict(list)
self._running = False
def on_trade(self, callback: Callable[[dict], Awaitable[None]]):
"""Register callback for trade events."""
self.callbacks["trade"].append(callback)
return callback
def on_position_change(self, callback: Callable[[dict], Awaitable[None]]):
"""Register callback for position changes."""
self.callbacks["position"].append(callback)
return callback
async def start(self):
"""Start monitoring tracked traders."""
._running =
event ._watch_events():
._running:
event.get() .tracked:
event_type = event.get(, )
callback .callbacks[event_type]:
:
callback(event)
Exception e:
()
():
._running =
():
.tracked.add(address)
():
.tracked.discard(address)
():
._running:
asyncio.sleep()
{}
