trader-analysis

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,  # Assuming 6 decimals
            "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  # gross profit / gross loss

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
                )
                
                # Apply filters
                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()
             {}