Python Integration Guide for Smart Money API

This comprehensive guide covers integrating Smart Money API into Python applications. Learn how to set up the environment, create a reusable client library, handle authentication, manage pagination, implement caching, and deploy to production with proper error handling and monitoring.

Table of Contents

Environment Setup
Installing Dependencies
Creating a Reusable Client Class
Authentication Implementation
Pagination and Loops
Caching Strategy
Error Handling and Retries
Logging and Monitoring
Production Deployment

Environment Setup

Start by setting up your Python development environment. Smart Money API requires Python 3.7+ for modern async support and type hints. We recommend using virtual environments to isolate dependencies.

        BASH
      

# Create project directory
mkdir smart-money-project
cd smart-money-project

# Create virtual environment
python3 -m venv venv

# Activate virtual environment
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Verify Python version
python --version  # Should be 3.7+
      

Installing Dependencies

Create a requirements.txt file with essential dependencies for Smart Money API integration:

        TEXT
      

requests==2.31.0
python-dotenv==1.0.0
aiohttp==3.9.0
websockets==12.0
redis==5.0.0
pydantic==2.5.0
tenacity==8.2.3
      

        BASH
      
pip install -r requirements.txt

Dependencies explained:

requests: HTTP client for REST API calls
python-dotenv: Load environment variables from .env files
aiohttp: Async HTTP client for concurrent requests
websockets: WebSocket support for real-time data
redis: Caching layer for improved performance
pydantic: Data validation and type checking
tenacity: Retry logic with exponential backoff

Creating a Reusable Client Class

Build a reusable client class that encapsulates API interaction logic. This approach keeps code DRY and provides a clean interface for your application.

        PYTHON
      

import os
import requests
import hmac
import hashlib
import time
from typing import Dict, Optional, Any
from datetime import datetime
from dotenv import load_dotenv

load_dotenv()

class SmartMoneyClient:
    """
    Smart Money API client with authentication,
    error handling, and rate limit management
    """

    def __init__(
        self,
        public_key: Optional[str] = None,
        secret_key: Optional[str] = None,
        api_version: str = 'v1'
    ):
        self.public_key = public_key or os.getenv('SMARTMONEY_PUBLIC_KEY')
        self.secret_key = secret_key or os.getenv('SMARTMONEY_SECRET_KEY')
        self.base_url = 'https://api.smartmoneyapi.com'
        self.api_version = api_version
        self.session = requests.Session()

        if not self.public_key:
            raise ValueError('API key required. Set SMARTMONEY_PUBLIC_KEY environment variable.')

    def _create_signature(
        self,
        method: str,
        path: str,
        timestamp: str,
        params: Optional[Dict] = None
    ) -> str:
        """Generate HMAC-SHA256 signature for request"""
        message = f'{method}{path}{timestamp}'
        if params:
            query_string = '&'.join([f'{k}={v}' for k, v in sorted(params.items())])
            message += f'?{query_string}'

        signature = hmac.new(
            self.secret_key.encode() if self.secret_key else b'',
            message.encode(),
            hashlib.sha256
        ).hexdigest()

        return signature

    def _prepare_headers(
        self,
        method: str,
        path: str,
        params: Optional[Dict] = None
    ) -> Dict[str, str]:
        """Prepare request headers with authentication"""
        headers = {
            'X-API-Key': self.public_key,
            'Content-Type': 'application/json'
        }

        if self.secret_key:
            timestamp = str(int(time.time() * 1000))
            signature = self._create_signature(method, path, timestamp, params)
            headers['X-Timestamp'] = timestamp
            headers['X-Signature'] = signature

        return headers

    def request(
        self,
        method: str,
        endpoint: str,
        params: Optional[Dict] = None,
        json_data: Optional[Dict] = None,
        timeout: int = 30
    ) -> Dict[str, Any]:
        """Make API request with error handling"""
        path = f'/v1{endpoint}'
        url = f'{self.base_url}{path}'

        headers = self._prepare_headers(method, path, params)

        try:
            response = self.session.request(
                method,
                url,
                params=params,
                json=json_data,
                headers=headers,
                timeout=timeout
            )

            # Handle rate limiting
            remaining = response.headers.get('X-RateLimit-Remaining')
            if remaining:
                self.rate_limit_remaining = int(remaining)

            response.raise_for_status()
            return response.json()

        except requests.exceptions.HTTPError as e:
            status = e.response.status_code
            if status == 429:
                raise Exception('Rate limit exceeded')
            elif status == 401:
                raise Exception('Invalid API key')
            elif status == 403:
                raise Exception('Insufficient permissions')
            else:
                raise

    def get(
        self,
        endpoint: str,
        params: Optional[Dict] = None
    ) -> Dict[str, Any]:
        """GET request"""
        return self.request('GET', endpoint, params=params)

    def post(
        self,
        endpoint: str,
        json_data: Optional[Dict] = None,
        params: Optional[Dict] = None
    ) -> Dict[str, Any]:
        """POST request"""
        return self.request('POST', endpoint, params=params, json_data=json_data)

    def get_whales(self, limit: int = 50) -> Dict[str, Any]:
        """Get top whale positions"""
        return self.get('/whales/top-positions', params={'limit': limit})

    def get_derivatives(self, symbol: str) -> Dict[str, Any]:
        """Get derivatives data for symbol"""
        return self.get('/derivatives/funding-rates', params={'symbol': symbol})

    def get_onchain_flows(self) -> Dict[str, Any]:
        """Get on-chain exchange flows"""
        return self.get('/onchain/exchange-flows')
      

Authentication Implementation

Smart Money API supports multiple authentication methods. For production applications, use signed requests with HMAC-SHA256 to prove key ownership.

        PYTHON
      

# .env file - NEVER commit this to version control
SMARTMONEY_PUBLIC_KEY=pk_live_abc123xyz
SMARTMONEY_SECRET_KEY=sk_live_secret789xyz

# Initialize client
client = SmartMoneyClient()

# Simple request (public key only)
data = client.get_whales(limit=10)

# For signed requests, both keys are used automatically
derivatives = client.get_derivatives('BTC')
      

Security Best Practice: Never hardcode API keys in your source code. Always use environment variables or secure vaults like AWS Secrets Manager or HashiCorp Vault for production systems.

Pagination and Loops

Many Smart Money API endpoints return paginated results. Implement pagination efficiently to fetch all data without unnecessary requests.

        PYTHON
      

from typing import Generator, Dict, Any

class PaginatedSmartMoneyClient(SmartMoneyClient):
    """Extended client with pagination support"""

    def get_paginated(
        self,
        endpoint: str,
        params: Optional[Dict] = None,
        page_size: int = 50
    ) -> Generator[Dict[str, Any], None, None]:
        """Fetch paginated data, yielding items as you go"""
        params = params or {}
        page = 1

        while True:
            params['page'] = page
            params['limit'] = page_size

            response = self.get(endpoint, params=params)
            items = response.get('data', [])

            if not items:
                break

            for item in items:
                yield item

            # Check if there are more pages
            meta = response.get('meta', {})
            if not meta.get('has_more', False):
                break

            page += 1

    def get_all_whale_positions(self) -> Generator[Dict, None, None]:
        """Get all whale positions with pagination"""
        yield from self.get_paginated('/whales/positions', page_size=100)


# Usage
client = PaginatedSmartMoneyClient()

print("Fetching all whale positions...")
count = 0
for position in client.get_all_whale_positions():
    print(f"Whale: {position['wallet_address']}")
    print(f"Position: {position['asset']} - {position['size_usd']}")
    count += 1
    if count >= 10:  # Stop after first 10 for demo
        break

print(f"Processed {count} positions")
      

Caching Strategy

Reduce API calls and improve performance by caching frequently accessed data. Use Redis for distributed caching in production systems.

        PYTHON
      

import json
import redis
from functools import wraps
from typing import Optional

class CachedSmartMoneyClient(SmartMoneyClient):
    """Client with Redis caching support"""

    def __init__(
        self,
        public_key: Optional[str] = None,
        secret_key: Optional[str] = None,
        redis_url: str = 'redis://localhost:6379/0',
        cache_ttl: int = 300  # 5 minutes
    ):
        super().__init__(public_key, secret_key)
        try:
            self.redis = redis.from_url(redis_url)
            self.cache_ttl = cache_ttl
        except Exception as e:
            print(f"Redis connection failed: {e}")
            self.redis = None

    def _cache_key(self, endpoint: str, params: Optional[Dict]) -> str:
        """Generate cache key from endpoint and parameters"""
        param_str = json.dumps(params or {}, sort_keys=True)
        return f'smartmoney:{endpoint}:{param_str}'

    def get_cached(
        self,
        endpoint: str,
        params: Optional[Dict] = None,
        ttl: Optional[int] = None
    ) -> Dict[str, Any]:
        """GET with caching"""
        if not self.redis:
            return self.get(endpoint, params)

        cache_key = self._cache_key(endpoint, params)

        # Try to get from cache
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # Fetch from API
        data = self.get(endpoint, params)

        # Cache result
        ttl = ttl or self.cache_ttl
        self.redis.setex(
            cache_key,
            ttl,
            json.dumps(data)
        )

        return data


# Usage
client = CachedSmartMoneyClient(redis_url='redis://localhost:6379')

# First call hits API
whales = client.get_cached('/whales/top-positions', {'limit': 50})

# Subsequent calls within 5 minutes come from cache
whales2 = client.get_cached('/whales/top-positions', {'limit': 50})

print(f"Fetched {len(whales['data'])} whale positions")
      

Error Handling and Retries

Implement robust error handling with exponential backoff for transient failures. The tenacity library simplifies retry logic.

        PYTHON
      

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type
)
import requests

class RobustSmartMoneyClient(CachedSmartMoneyClient):
    """Client with automatic retries for transient failures"""

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(
            (requests.exceptions.ConnectionError,
             requests.exceptions.Timeout)
        )
    )
    def request(self, method, endpoint, **kwargs):
        """Request with automatic retries"""
        return super().request(method, endpoint, **kwargs)

    def safe_get(
        self,
        endpoint: str,
        params: Optional[Dict] = None,
        on_error: Optional[str] = 'return_none'
    ) -> Optional[Dict[str, Any]]:
        """GET with error handling"""
        try:
            return self.get_cached(endpoint, params)
        except requests.exceptions.Timeout:
            print(f"Request to {endpoint} timed out after retries")
            return None if on_error == 'return_none' else {}
        except Exception as e:
            print(f"Error fetching {endpoint}: {e}")
            return None if on_error == 'return_none' else {}


# Usage
client = RobustSmartMoneyClient()

whales = client.safe_get('/whales/top-positions')
if whales:
    print(f"Successfully fetched {len(whales['data'])} positions")
else:
    print("Failed to fetch whale positions after retries")
      

Logging and Monitoring

Implement comprehensive logging to track API calls, errors, and performance metrics. This is essential for debugging and production monitoring.

        PYTHON
      

import logging
from datetime import datetime
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger('SmartMoneyAPI')

class MonitoredSmartMoneyClient(RobustSmartMoneyClient):
    """Client with logging and monitoring"""

    def request(self, method, endpoint, params=None, **kwargs):
        """Request with logging"""
        start_time = time.time()

        try:
            logger.info(f'{method} {endpoint}')
            response = super().request(method, endpoint, params, **kwargs)
            elapsed = time.time() - start_time

            logger.info(
                f'{endpoint} completed in {elapsed:.2f}s - '
                f'Status: Success, Items: {len(response.get("data", []))}'
            )

            return response

        except Exception as e:
            elapsed = time.time() - start_time
            logger.error(
                f'{endpoint} failed after {elapsed:.2f}s - '
                f'Error: {str(e)}'
            )
            raise

    def log_metrics(self):
        """Log current metrics"""
        logger.info(f'Rate limit remaining: {self.rate_limit_remaining}')


# Usage
client = MonitoredSmartMoneyClient()
whales = client.safe_get('/whales/top-positions')
client.log_metrics()
      

Production Deployment

Deploy your Python application to production with proper configuration management, health checks, and scaling considerations.

        PYTHON
      

# config.py - Configuration management
import os
from typing import Optional

class Config:
    """Base configuration"""
    DEBUG = False
    TESTING = False
    SMARTMONEY_PUBLIC_KEY = os.getenv('SMARTMONEY_PUBLIC_KEY')
    SMARTMONEY_SECRET_KEY = os.getenv('SMARTMONEY_SECRET_KEY')
    REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
    CACHE_TTL = int(os.getenv('CACHE_TTL', '300'))
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')

class DevelopmentConfig(Config):
    """Development configuration"""
    DEBUG = True
    CACHE_TTL = 60

class ProductionConfig(Config):
    """Production configuration"""
    DEBUG = False
    CACHE_TTL = 600

def get_config(env: Optional[str] = None) -> Config:
    """Get configuration for environment"""
    env = env or os.getenv('APP_ENV', 'development')
    if env == 'production':
        return ProductionConfig()
    return DevelopmentConfig()

# main.py - Application entry point
from config import get_config

config = get_config()
client = MonitoredSmartMoneyClient(
    public_key=config.SMARTMONEY_PUBLIC_KEY,
    secret_key=config.SMARTMONEY_SECRET_KEY,
    redis_url=config.REDIS_URL,
    cache_ttl=config.CACHE_TTL
)

# Health check
def check_health():
    """Check API connectivity"""
    try:
        response = client.safe_get('/market/status')
        return response is not None
    except Exception as e:
        logger.error(f'Health check failed: {e}')
        return False

if __name__ == '__main__':
    if check_health():
        print('API health check passed')
        # Start your application
    else:
        print('API health check failed')
        exit(1)
      

Deployment Checklist: Use environment variables for all secrets, implement health checks, configure logging to stdout, use Redis for caching in distributed systems, monitor rate limits closely, and set up alerts for API errors.

For more advanced topics, check out our guides on WebSocket streaming, building trading bots, and rate limiting best practices.

Ready to Build with Python?

Start integrating Smart Money API into your Python applications. Access whale tracking, derivatives intelligence, and on-chain analysis with our comprehensive Python client.

View Python Plans