API Examples¶

https://api.chainsentinel.net

import requests
import json

API_KEY = "your_api_key_here"
BASE_URL = "https://api.chainsentinel.net"

headers = {
    "X-API-Key": API_KEY,
    "Content-Type": "application/json"
}

def analyze_token(token_address):
    """Analyze a Solana token for scam indicators"""

    # Check if token already analyzed
    response = requests.get(
        f"{BASE_URL}/tokens/{token_address}",
        headers=headers
    )

    if response.status_code == 200:
        data = response.json()
        return data['data']

    # Request new analysis
    response = requests.post(
        f"{BASE_URL}/tokens/analyze",
        headers=headers,
        json={"address": token_address}
    )

    if response.status_code == 200:
        job_data = response.json()['data']
        job_id = job_data['job_id']

        # Poll for results
        import time
        while True:
            response = requests.get(
                f"{BASE_URL}/tokens/analyze/{job_id}",
                headers=headers
            )

            result = response.json()['data']
            if result['status'] == 'completed':
                return result['token']
            elif result['status'] == 'failed':
                raise Exception(f"Analysis failed: {result['error']}")

            time.sleep(5)  # Wait 5 seconds before checking again

# Example usage
token = analyze_token("DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263")
print(f"Prediction: {token['prediction']}")
print(f"Confidence: {token['confidence']}%")
print(f"Risk Score: {token['risk_score']}")

def check_tokens(token_addresses):
    """Check multiple tokens and return scam predictions"""

    results = []

    for address in token_addresses:
        try:
            response = requests.get(
                f"{BASE_URL}/tokens/{address}",
                headers=headers
            )

            if response.status_code == 200:
                token = response.json()['data']
                results.append({
                    'address': address,
                    'name': token['name'],
                    'symbol': token['symbol'],
                    'prediction': token['prediction'],
                    'confidence': token['confidence'],
                    'risk_score': token['risk_score']
                })
            else:
                results.append({
                    'address': address,
                    'error': 'Not found'
                })
        except Exception as e:
            results.append({
                'address': address,
                'error': str(e)
            })

    return results

# Example usage
tokens = [
    "DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263",  # BONK
    "7GCihgDB8fe6KNjn2MYtkzZcRjQy3t9GHdC8uHYmW2hr",  # POPCAT
]

results = check_tokens(tokens)
for result in results:
    if 'error' not in result:
        print(f"{result['symbol']}: {result['prediction']} ({result['confidence']}%)")
    else:
        print(f"{result['address']}: {result['error']}")

def monitor_wallet(wallet_address):
    """Monitor a wallet for new token creations"""

    # Get wallet info
    response = requests.get(
        f"{BASE_URL}/wallets/{wallet_address}",
        headers=headers
    )

    if response.status_code != 200:
        raise Exception("Wallet not found")

    wallet = response.json()['data']

    print(f"Wallet: {wallet_address}")
    print(f"Risk Score: {wallet['risk_score']}")
    print(f"Tokens Created: {wallet['tokens_created']}")
    print(f"Scam Rate: {wallet['scam_rate'] * 100}%")

    # Get tokens created by this wallet
    response = requests.get(
        f"{BASE_URL}/wallets/{wallet_address}/tokens",
        headers=headers
    )

    tokens = response.json()['data']['tokens']

    print(f"\nRecent Tokens:")
    for token in tokens[:5]:  # Show last 5 tokens
        print(f"  {token['symbol']}: {token['prediction']} ({token['confidence']}%)")

    # Add to watchlist for alerts
    response = requests.post(
        f"{BASE_URL}/watchlist",
        headers=headers,
        json={
            "type": "wallet",
            "address": wallet_address,
            "alerts": ["new_token"],
            "sensitivity": "high"
        }
    )

    if response.status_code == 200:
        print(f"\n✅ Added to watchlist - you'll be alerted of new tokens")

# Example usage
monitor_wallet("ABC...123")

def find_scam_clusters(min_size=5):
    """Find large scam rings"""

    response = requests.get(
        f"{BASE_URL}/clusters",
        headers=headers,
        params={"min_size": min_size}
    )

    clusters = response.json()['data']['clusters']

    print(f"Found {len(clusters)} scam clusters:\n")

    for cluster in clusters:
        print(f"Cluster #{cluster['id']}")
        print(f"  Size: {cluster['size']} wallets")
        print(f"  Tokens: {cluster['tokens_created']}")
        print(f"  Scam Rate: {cluster['scam_rate'] * 100}%")
        print(f"  Total Stolen: ${cluster['total_stolen']:,.0f}")
        print(f"  Victims: {cluster['victims']:,}")
        print()

# Example usage
find_scam_clusters(min_size=5)

const API_KEY = 'your_api_key_here';
const BASE_URL = 'https://api.chainsentinel.net';

const headers = {
  'X-API-Key': API_KEY,
  'Content-Type': 'application/json'
};

async function analyzeToken(tokenAddress) {
  try {
    // Check if already analyzed
    let response = await fetch(`${BASE_URL}/tokens/${tokenAddress}`, {
      headers: headers
    });

    if (response.ok) {
      const data = await response.json();
      return data.data;
    }

    // Request new analysis
    response = await fetch(`${BASE_URL}/tokens/analyze`, {
      method: 'POST',
      headers: headers,
      body: JSON.stringify({ address: tokenAddress })
    });

    const jobData = await response.json();
    const jobId = jobData.data.job_id;

    // Poll for results
    while (true) {
      await new Promise(resolve => setTimeout(resolve, 5000)); // Wait 5s

      response = await fetch(`${BASE_URL}/tokens/analyze/${jobId}`, {
        headers: headers
      });

      const result = await response.json();

      if (result.data.status === 'completed') {
        return result.data.token;
      } else if (result.data.status === 'failed') {
        throw new Error(`Analysis failed: ${result.data.error}`);
      }
    }
  } catch (error) {
    console.error('Error analyzing token:', error);
    throw error;
  }
}

// Example usage
analyzeToken('DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263')
  .then(token => {
    console.log(`Prediction: ${token.prediction}`);
    console.log(`Confidence: ${token.confidence}%`);
    console.log(`Risk Score: ${token.risk_score}`);
  });

async function setupWatchlist(tokenAddresses) {
  // Add tokens to watchlist
  for (const address of tokenAddresses) {
    await fetch(`${BASE_URL}/watchlist`, {
      method: 'POST',
      headers: headers,
      body: JSON.stringify({
        type: 'token',
        address: address,
        alerts: ['rug_pull', 'price_drop'],
        sensitivity: 'medium'
      })
    });
  }

  // Poll for alerts every 30 seconds
  setInterval(async () => {
    const response = await fetch(`${BASE_URL}/alerts?unread=true`, {
      headers: headers
    });

    const data = await response.json();
    const alerts = data.data.alerts;

    for (const alert of alerts) {
      // Show notification
      if (Notification.permission === 'granted') {
        new Notification('Chain Sentinel Alert', {
          body: `${alert.type}: ${alert.token.symbol}`,
          icon: '/icon.png'
        });
      }

      // Mark as read
      await fetch(`${BASE_URL}/alerts/${alert.id}/read`, {
        method: 'POST',
        headers: headers
      });
    }
  }, 30000);
}

// Example usage
setupWatchlist([
  'DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263',
  '7GCihgDB8fe6KNjn2MYtkzZcRjQy3t9GHdC8uHYmW2hr'
]);

const axios = require('axios');

const API_KEY = 'your_api_key_here';
const BASE_URL = 'https://api.chainsentinel.net';

const client = axios.create({
  baseURL: BASE_URL,
  headers: {
    'X-API-Key': API_KEY,
    'Content-Type': 'application/json'
  }
});

async function batchAnalyze(tokenAddresses) {
  const results = await Promise.all(
    tokenAddresses.map(async (address) => {
      try {
        const response = await client.get(`/tokens/${address}`);
        return response.data.data;
      } catch (error) {
        return {
          address: address,
          error: error.message
        };
      }
    })
  );

  return results;
}

// Example usage
const tokens = [
  'DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263',
  '7GCihgDB8fe6KNjn2MYtkzZcRjQy3t9GHdC8uHYmW2hr'
];

batchAnalyze(tokens).then(results => {
  results.forEach(token => {
    if (!token.error) {
      console.log(`${token.symbol}: ${token.prediction} (${token.confidence}%)`);
    } else {
      console.log(`${token.address}: ${token.error}`);
    }
  });
});

const express = require('express');
const app = express();

app.use(express.json());

app.post('/webhook', (req, res) => {
  const alert = req.body;

  console.log(`Alert received: ${alert.alert_type}`);
  console.log(`Token: ${alert.token.symbol}`);
  console.log(`Severity: ${alert.severity}`);

  // Your custom logic
  if (alert.severity === 'critical') {
    // Send SMS, email, or other notification
    sendCriticalAlert(alert);
  }

  res.json({ status: 'ok' });
});

app.listen(3000, () => {
  console.log('Webhook server listening on port 3000');
});

function sendCriticalAlert(alert) {
  // Implement your notification logic
  console.log(`🚨 CRITICAL: ${alert.token.symbol} - ${alert.alert_type}`);
}

# Check if token exists
curl -H "X-API-Key: your_api_key" \
  "https://api.chainsentinel.net/tokens/DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263"

# Request new analysis
curl -X POST \
  -H "X-API-Key: your_api_key" \
  -H "Content-Type: application/json" \
  -d '{"address": "ABC...123"}' \
  "https://api.chainsentinel.net/tokens/analyze"

curl -H "X-API-Key: your_api_key" \
  "https://api.chainsentinel.net/tokens?prediction=SCAM&min_confidence=80&limit=10"

curl -X POST \
  -H "X-API-Key: your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "type": "token",
    "address": "ABC...123",
    "alerts": ["rug_pull", "price_drop"],
    "sensitivity": "high"
  }' \
  "https://api.chainsentinel.net/watchlist"

curl -H "X-API-Key: your_api_key" \
  "https://api.chainsentinel.net/alerts?severity=critical&unread=true"

import requests
import time

class ScamDetector:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.chainsentinel.net"
        self.headers = {"X-API-Key": api_key}

    def is_safe_to_buy(self, token_address, min_confidence=75):
        """Check if token is safe to buy"""

        response = requests.get(
            f"{self.base_url}/tokens/{token_address}",
            headers=self.headers
        )

        if response.status_code != 200:
            return False, "Token not found"

        token = response.json()['data']

        # Check prediction
        if token['prediction'] == 'SCAM':
            if token['confidence'] >= min_confidence:
                return False, f"SCAM detected ({token['confidence']}% confidence)"

        # Check risk score
        if token['risk_score'] > 70:
            return False, f"High risk score ({token['risk_score']})"

        # Check creator
        creator = token.get('creator', {})
        if creator.get('scam_rate', 0) > 0.5:
            return False, f"Creator has {creator['scam_rate']*100}% scam rate"

        return True, "Token appears safe"

    def before_trade(self, token_address):
        """Call this before executing a trade"""

        safe, reason = self.is_safe_to_buy(token_address)

        if not safe:
            print(f"❌ Trade blocked: {reason}")
            return False

        print(f"✅ Trade approved: {reason}")
        return True

# Example usage in trading bot
detector = ScamDetector("your_api_key")

def execute_trade(token_address, amount):
    # Check with Chain Sentinel first
    if not detector.before_trade(token_address):
        return False

    # Execute trade
    print(f"Buying {amount} of {token_address}")
    # ... your trading logic ...

    return True

import requests
import time
from datetime import datetime

class PortfolioMonitor:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.chainsentinel.net"
        self.headers = {"X-API-Key": api_key}
        self.portfolio = []

    def add_token(self, token_address, amount):
        """Add token to monitored portfolio"""

        # Get token info
        response = requests.get(
            f"{self.base_url}/tokens/{token_address}",
            headers=self.headers
        )

        if response.status_code == 200:
            token = response.json()['data']
            self.portfolio.append({
                'address': token_address,
                'symbol': token['symbol'],
                'amount': amount,
                'initial_prediction': token['prediction'],
                'initial_confidence': token['confidence'],
                'added_at': datetime.now()
            })

            # Add to watchlist
            requests.post(
                f"{self.base_url}/watchlist",
                headers=self.headers,
                json={
                    "type": "token",
                    "address": token_address,
                    "alerts": ["rug_pull", "price_drop"],
                    "sensitivity": "high"
                }
            )

    def check_portfolio(self):
        """Check all tokens in portfolio"""

        print(f"\n{'='*60}")
        print(f"Portfolio Check - {datetime.now()}")
        print(f"{'='*60}\n")

        for item in self.portfolio:
            response = requests.get(
                f"{self.base_url}/tokens/{item['address']}",
                headers=self.headers
            )

            if response.status_code == 200:
                token = response.json()['data']

                # Check if prediction changed
                if token['prediction'] != item['initial_prediction']:
                    print(f"⚠️  {item['symbol']}: Prediction changed!")
                    print(f"   Was: {item['initial_prediction']}")
                    print(f"   Now: {token['prediction']} ({token['confidence']}%)")

                # Check risk score
                if token['risk_score'] > 70:
                    print(f"🚨 {item['symbol']}: HIGH RISK ({token['risk_score']})")
                else:
                    print(f"✅ {item['symbol']}: OK (risk: {token['risk_score']})")

        print()

    def monitor_loop(self, interval=300):
        """Monitor portfolio continuously"""

        while True:
            self.check_portfolio()
            time.sleep(interval)  # Check every 5 minutes

# Example usage
monitor = PortfolioMonitor("your_api_key")

# Add your holdings
monitor.add_token("DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263", 1000000)  # BONK
monitor.add_token("7GCihgDB8fe6KNjn2MYtkzZcRjQy3t9GHdC8uHYmW2hr", 500)  # POPCAT

# Start monitoring
monitor.monitor_loop(interval=300)

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

class ChainSentinelClient:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.chainsentinel.net"

        # Setup session with retries
        self.session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount('https://', adapter)
        self.session.headers.update({'X-API-Key': api_key})

    def get_token(self, address):
        """Get token with error handling"""

        try:
            response = self.session.get(
                f"{self.base_url}/tokens/{address}",
                timeout=10
            )
            response.raise_for_status()
            return response.json()['data']

        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 404:
                print(f"Token not found: {address}")
            elif e.response.status_code == 429:
                print("Rate limit exceeded - please upgrade your plan")
            else:
                print(f"HTTP error: {e}")
            return None

        except requests.exceptions.Timeout:
            print("Request timed out")
            return None

        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
            return None

# Example usage
client = ChainSentinelClient("your_api_key")
token = client.get_token("DezXAZ8z7PnrnRJjz3wXBoRgixCa6xjnB7YaB1pPB263")
if token:
    print(f"Prediction: {token['prediction']}")