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 = await self.client.get(
            "/activity",
            params={
                "user": address,
                "limit": limit,
                "offset": offset
            }
        )
        response.raise_for_status()
        return response.json()
    
    async def get_leaderboard(
        self,
        period: str = "all",
        limit: int = 100
    ) -> list[dict]:
        """Get top traders by P&L."""
        response = await self.client.get(
            "/leaderboard",
            params={"period": period, "limit": limit}
        )
        response.raise_for_status()
        return 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 =