Python开发环境展示Pangolin API调用代码和亚马逊数据采集实时结果,Python调用Pangolin API

When you first attempt to scrape product data from Amazon, you might encounter this frustrating scenario: spending hours writing scraper code, debugging CSS selectors and XPath expressions, finally getting your script to work, only to discover the next day that Amazon updated their page structure and all your code needs to be rewritten. Even more frustrating is that even when your code runs successfully, it frequently gets blocked due to anti-scraping mechanisms or account restrictions from excessive request rates. This “works today, breaks tomorrow” state causes many developers to invest enormous time in data collection projects with minimal results, dragging out project timelines and accumulating technical debt.

The root problem is that traditional web scraping development requires developers to handle too many low-level details simultaneously: HTTP request construction, Cookie and Session management, anti-scraping countermeasures, page parsing, data cleaning, exception handling, concurrency control, and more. A seemingly simple “get product price” requirement might actually require over 500 lines of code, with most of it handling technical issues unrelated to business logic. More critically, this approach has extremely high maintenance costs—whenever the target website updates its page structure, you need to re-analyze the DOM tree, adjust selectors, and test validation, a process that could take days or even a week.

The Pangolin API completely transforms this situation. By providing a stable RESTful interface and structured JSON data returns, it simplifies complex data collection work into just a few lines of Python API calls. You don’t need to worry about how Amazon’s page structure changes, don’t need to handle anti-scraping mechanisms, don’t need to maintain proxy IP pools—you only need to focus on your business logic: getting data, analyzing data, applying data. This article will guide you through 7 complete code examples, from environment setup to production projects, teaching you to master all Python Pangolin API skills and build a functional data collection system in just 30 minutes.

Step 1: Python Environment Setup and Dependency Installation

Before writing code, we need to set up an appropriate Python development environment. Pangolin API has very flexible Python version requirements, supporting Python 3.7 and above, but I strongly recommend using Python 3.9 or 3.10, as these versions have significant improvements in performance and stability, and have the best compatibility with mainstream data analysis libraries. If Python isn’t installed on your system yet, you can download the installer for your operating system from the official website. Windows users should check the “Add Python to PATH” option to use the python command directly in the command line.

Once the environment is ready, we need to install necessary Python libraries. Pangolin API doesn’t require a dedicated SDK—you only need the requests module from Python’s standard library to complete all operations, greatly reducing project dependency complexity. Open your command line tool and execute the following commands to install dependencies:

# Install core dependencies
pip install requests

# Install data processing libraries (optional, for advanced data analysis)
pip install pandas numpy

# Install JSON processing enhancement library (optional, for beautified output)
pip install json5

After installation, we can create a project directory to organize our code. Here’s the recommended project structure:

pangolin-api-project/
├── config.py          # Configuration file (stores API keys, etc.)
├── api_client.py      # API client wrapper
├── examples/          # Example code directory
│   ├── basic_call.py      # Basic call example
│   ├── batch_fetch.py     # Batch collection example
│   └── monitor.py         # Monitoring project example
├── utils/             # Utility functions
│   ├── error_handler.py   # Error handling
│   └── data_parser.py     # Data parsing
└── data/              # Data storage directory
    ├── raw/               # Raw data
    └── processed/         # Processed data

This structured organization not only makes code easier to maintain but also reserves space for future feature expansion. Next, we’ll configure API authentication information in config.py. Note that you should never hardcode API keys in your code or commit them to Git repositories. The correct approach is to use environment variables or configuration files, and add configuration files to .gitignore.

Step 2: API Authentication and Token Acquisition

Pangolin API uses Bearer Token authentication, a simple yet secure authentication mechanism. The entire authentication process has two steps: first, use your email and password to obtain a Token, then carry this Token in all subsequent API requests. The Token is long-lived, meaning you only need to obtain it once at program startup and can reuse it throughout the session without re-authenticating for each request.

Let’s write our first complete Python code example to implement API authentication:

import requests
import json
from typing import Optional

class PangolinAPIClient:
    """Pangolin API client wrapper class"""
    
    def __init__(self, email: str, password: str):
        """
        Initialize API client
        
        Args:
            email: Registered email
            password: Account password
        """
        self.base_url = "https://scrapeapi.pangolinfo.com"
        self.email = email
        self.password = password
        self.token: Optional[str] = None
        
    def authenticate(self) -> bool:
        """
        Perform API authentication to obtain access Token
        
        Returns:
            bool: Whether authentication was successful
        """
        auth_url = f"{self.base_url}/api/v1/auth"
        
        payload = {
            "email": self.email,
            "password": self.password
        }
        
        headers = {
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                auth_url,
                json=payload,
                headers=headers,
                timeout=10
            )
            
            # Check HTTP status code
            response.raise_for_status()
            
            # Parse response
            result = response.json()
            
            if result.get("code") == 0:
                self.token = result.get("data")
                print(f"✓ Authentication successful! Token: {self.token[:20]}...")
                return True
            else:
                print(f"✗ Authentication failed: {result.get('message')}")
                return False
                
        except requests.exceptions.Timeout:
            print("✗ Request timeout, please check network connection")
            return False
        except requests.exceptions.RequestException as e:
            print(f"✗ Request exception: {str(e)}")
            return False
        except json.JSONDecodeError:
            print("✗ Response parsing failed, invalid data format")
            return False
    
    def get_headers(self) -> dict:
        """
        Get request headers with authentication info
        
        Returns:
            dict: Request headers including Authorization
        """
        if not self.token:
            raise ValueError("Token not initialized, please call authenticate() first")
        
        return {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.token}"
        }

# Usage example
if __name__ == "__main__":
    # Read credentials from environment variables or config file
    client = PangolinAPIClient(
        email="[email protected]",
        password="your_password"
    )
    
    # Perform authentication
    if client.authenticate():
        print("API client initialized, ready to call data interfaces")
    else:
        print("Authentication failed, please check email and password")

This code demonstrates several important best practices: using class encapsulation for improved code reusability, comprehensive exception handling mechanisms, clear logging output, and type annotations for enhanced code readability. The timeout parameter setting is particularly noteworthy as it prevents network issues from causing the program to hang indefinitely. In production environments, it’s recommended to cache the Token to a file or database to avoid frequent authentication API calls.

Step 3: Basic API Calls – Fetching Product Details

After authentication, we can start calling Pangolin’s core functionality. The most common scenario is fetching detailed Amazon product information, including title, price, rating, review count, stock status, and dozens of other fields. Pangolin API supports three data return formats: json (structured data), rawHtml (raw HTML), and markdown (Markdown format). For most application scenarios, json format is the best choice because it’s already cleaned and structured, ready for direct use in business logic.

Let’s write a complete product detail fetching function:

def fetch_product_detail(self, asin: str, zipcode: str = "10041") -> dict:
    """
    Fetch Amazon product details
    
    Args:
        asin: Product ASIN code
        zipcode: ZIP code (for region-specific pricing and stock)
        
    Returns:
        dict: Product detail data
    """
    scrape_url = f"{self.base_url}/api/v1/scrape"
    
    payload = {
        "url": f"https://www.amazon.com/dp/{asin}",
        "parserName": "amzProductDetail",
        "format": "json",
        "bizContext": {
            "zipcode": zipcode
        }
    }
    
    try:
        response = requests.post(
            scrape_url,
            json=payload,
            headers=self.get_headers(),
            timeout=30  # Product detail pages have more data, extend timeout
        )
        
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 0:
            # Extract core data
            data = result.get("data", {})
            json_data = data.get("json", [{}])[0]
            
            if json_data.get("code") == 0:
                product_info = json_data.get("data", {}).get("results", [{}])[0]
                
                # Extract key fields
                simplified_data = {
                    "asin": product_info.get("asin"),
                    "title": product_info.get("title"),
                    "price": product_info.get("price"),
                    "rating": product_info.get("star"),
                    "review_count": product_info.get("rating"),
                    "brand": product_info.get("brand"),
                    "image": product_info.get("image"),
                    "in_stock": product_info.get("has_cart"),
                    "category": product_info.get("category_name")
                }
                
                print(f"✓ Successfully fetched details for product {asin}")
                return simplified_data
            else:
                print(f"✗ Data parsing failed: {json_data.get('message')}")
                return {}
        else:
            error_msg = result.get("message")
            if result.get("code") == 2001:
                print("✗ Insufficient credits, please recharge to continue")
            elif result.get("code") == 1004:
                print("✗ Invalid token, please re-authenticate")
            else:
                print(f"✗ API call failed: {error_msg}")
            return {}
            
    except requests.exceptions.Timeout:
        print(f"✗ Request timeout: Failed to fetch product {asin} data")
        return {}
    except Exception as e:
        print(f"✗ Unknown error: {str(e)}")
        return {}

# Usage example
if __name__ == "__main__":
    client = PangolinAPIClient("[email protected]", "your_password")
    
    if client.authenticate():
        # Fetch single product details
        product = client.fetch_product_detail("B0DYTF8L2W")
        
        if product:
            print("\nProduct Information:")
            print(f"Title: {product['title']}")
            print(f"Price: {product['price']}")
            print(f"Rating: {product['rating']} ({product['review_count']} reviews)")
            print(f"Brand: {product['brand']}")
            print(f"Stock: {'In Stock' if product['in_stock'] else 'Out of Stock'}")

This example demonstrates several key technical points: first, the use of the parserName parameter, which tells the API which parsing template to use for data extraction. Pangolin supports multiple parsers including product details, keyword search, rankings, etc. Second, the application of bizContext—by specifying a ZIP code, you can get region-specific pricing and stock information, very useful for cross-region price comparisons. Finally, error code handling—different error codes represent different issues requiring targeted handling.

Step 4: Batch Data Collection and Concurrency Optimization

In real applications, we often need to collect data for large numbers of products, such as monitoring all competitors in a category or tracking all SKUs in your own store. Using serial processing to call the API one by one would be extremely inefficient—assuming each request takes 10 seconds, collecting 1,000 products would take nearly 3 hours. Through concurrent processing, we can reduce this time to 10-15 minutes, improving efficiency by over 10 times.

Python offers multiple concurrency solutions. For IO-intensive API call scenarios, using the ThreadPoolExecutor from the concurrent.futures module is the best choice. It provides a clean interface, automatically manages the thread pool, and supports exception handling and result collection. Let’s implement a batch collection function:

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List
import time

def batch_fetch_products(self, asin_list: List[str], max_workers: int = 5) -> List[dict]:
    """
    Batch fetch product details (concurrent version)
    
    Args:
        asin_list: List of ASINs
        max_workers: Maximum concurrent threads (recommended 5-10)
        
    Returns:
        List[dict]: List of product details
    """
    results = []
    failed_asins = []
    
    print(f"Starting batch collection of {len(asin_list)} products...")
    start_time = time.time()
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Submit all tasks
        future_to_asin = {
            executor.submit(self.fetch_product_detail, asin): asin 
            for asin in asin_list
        }
        
        # Collect results
        for future in as_completed(future_to_asin):
            asin = future_to_asin[future]
            try:
                product_data = future.result()
                if product_data:
                    results.append(product_data)
                else:
                    failed_asins.append(asin)
            except Exception as e:
                print(f"✗ Exception processing {asin}: {str(e)}")
                failed_asins.append(asin)
    
    elapsed_time = time.time() - start_time
    
    print(f"\nBatch collection complete!")
    print(f"Success: {len(results)} items")
    print(f"Failed: {len(failed_asins)} items")
    print(f"Time elapsed: {elapsed_time:.2f} seconds")
    print(f"Average speed: {len(asin_list)/elapsed_time:.2f} items/second")
    
    if failed_asins:
        print(f"Failed ASINs: {', '.join(failed_asins)}")
    
    return results

# Usage example
if __name__ == "__main__":
    client = PangolinAPIClient("[email protected]", "your_password")
    
    if client.authenticate():
        # Prepare ASIN list to collect
        asin_list = [
            "B0DYTF8L2W",
            "B08N5WRWNW",
            "B07ZPKN6YR",
            "B08L5VFJ2R",
            "B09JQMJHXY"
        ]
        
        # Batch collection
        products = client.batch_fetch_products(asin_list, max_workers=5)
        
        # Save to CSV file
        if products:
            import pandas as pd
            df = pd.DataFrame(products)
            df.to_csv("products.csv", index=False, encoding="utf-8-sig")
            print(f"\nData saved to products.csv")

The key to this code is the max_workers parameter setting. While theoretically more threads mean faster speeds, in practice you’re limited by the API server’s processing capacity and your network bandwidth. Too many concurrent requests might cause timeouts or rate limiting. Based on testing, 5-10 concurrent threads is the optimal balance, fully utilizing concurrency advantages without overwhelming the server. If you have massive amounts of data to collect (tens of thousands of ASINs), it’s recommended to process in batches of around 1,000 each, with a few seconds interval between batches.

Step 5: Comprehensive Error Handling and Retry Mechanisms

In production environments, network fluctuations, temporary server failures, and data format anomalies are inevitable. A robust system must have comprehensive error handling and automatic retry capabilities. Pangolin API returns very detailed error codes, including insufficient credits (2001), invalid token (1004), expired account (2007), scraping failures (10000/10001), etc. We need to adopt different handling strategies for different error types.

Let’s implement an API call decorator with exponential backoff retry mechanism:

import functools
import time
from typing import Callable, Any

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0
):
    """
    Retry decorator with exponential backoff
    
    Args:
        max_retries: Maximum retry attempts
        base_delay: Base delay time (seconds)
        max_delay: Maximum delay time (seconds)
        exponential_base: Exponential base
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            retries = 0
            
            while retries <= max_retries:
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.Timeout:
                    retries += 1
                    if retries > max_retries:
                        print(f"✗ Exceeded maximum retries ({max_retries}), giving up request")
                        raise
                    
                    # Calculate delay time (exponential backoff)
                    delay = min(
                        base_delay * (exponential_base ** (retries - 1)),
                        max_delay
                    )
                    
                    print(f"⟳ Request timeout, retrying in {delay:.1f}s (attempt {retries})...")
                    time.sleep(delay)
                    
                except requests.exceptions.RequestException as e:
                    # Don't retry for non-timeout network errors
                    print(f"✗ Network error: {str(e)}")
                    raise
                    
        return wrapper
    return decorator

class PangolinAPIClientWithRetry(PangolinAPIClient):
    """API client with retry mechanism"""
    
    @retry_with_backoff(max_retries=3, base_delay=2.0)
    def fetch_product_detail_with_retry(self, asin: str, zipcode: str = "10041") -> dict:
        """
        Fetch product details (with retry)
        
        This method automatically retries on timeout errors,
        with retry intervals using exponential backoff: 2s, 4s, 8s
        """
        return self.fetch_product_detail(asin, zipcode)
    
    def handle_api_error(self, error_code: int, error_message: str) -> str:
        """
        Unified error handling function
        
        Args:
            error_code: API returned error code
            error_message: Error message
            
        Returns:
            str: User-friendly error message
        """
        error_handlers = {
            0: "Operation successful",
            1004: "Invalid token, please re-authenticate",
            2001: "Insufficient credits, please recharge in console",
            2007: "Account expired, please contact support for renewal",
            10000: "Scraping failed, please try again later",
            10001: "Scraping failed, target page may not exist",
            404: "URL address error, please check link format"
        }
        
        user_message = error_handlers.get(
            error_code,
            f"Unknown error (code: {error_code})"
        )
        
        # Log detailed error
        self.log_error(error_code, error_message)
        
        return user_message
    
    def log_error(self, error_code: int, error_message: str):
        """Log error to file"""
        import datetime
        
        log_entry = {
            "timestamp": datetime.datetime.now().isoformat(),
            "error_code": error_code,
            "error_message": error_message
        }
        
        # Can be extended to write to log file or send alerts
        print(f"[ERROR LOG] {log_entry}")

The core idea of exponential backoff strategy is: the first retry waits a short time, and if it still fails, progressively increase the wait time. This both enables quick recovery from temporary failures and avoids making things worse when the server is under heavy load. In actual use, you can adjust retry parameters based on business characteristics—for example, reduce retry attempts for scenarios requiring real-time responses, or increase retry attempts and delays for batch background tasks.

Production Project 1: Real-time Best Seller Ranking Monitor

Let’s integrate the knowledge we’ve learned to build a real application scenario: monitoring changes in Amazon Best Seller rankings. This system helps sellers discover trending products timely, analyze competitor dynamics, and provide data support for product selection and operational decisions. We’ll implement the following features: scheduled ranking data collection, ranking change comparison, new entry identification, and change report generation.

import json
import time
from datetime import datetime, timedelta
from typing import List, Dict

class BestSellerMonitor:
    """Best Seller ranking monitor"""
    
    def __init__(self, api_client: PangolinAPIClient):
        self.client = api_client
        self.history_file = "data/bestseller_history.json"
        self.load_history()
    
    def load_history(self):
        """Load historical data"""
        try:
            with open(self.history_file, 'r', encoding='utf-8') as f:
                self.history = json.load(f)
        except FileNotFoundError:
            self.history = {}
    
    def save_history(self):
        """Save historical data"""
        with open(self.history_file, 'w', encoding='utf-8') as f:
            json.dump(self.history, f, ensure_ascii=False, indent=2)
    
    def fetch_bestseller_list(self, category_url: str) -> List[Dict]:
        """
        Fetch Best Seller ranking
        
        Args:
            category_url: Category URL
            
        Returns:
            List[Dict]: List of ranking products
        """
        scrape_url = f"{self.client.base_url}/api/v1/scrape"
        
        payload = {
            "url": category_url,
            "parserName": "amzBestSellers",
            "format": "json",
            "bizContext": {
                "zipcode": "10041"
            }
        }
        
        try:
            response = requests.post(
                scrape_url,
                json=payload,
                headers=self.client.get_headers(),
                timeout=30
            )
            
            response.raise_for_status()
            result = response.json()
            
            if result.get("code") == 0:
                data = result.get("data", {})
                json_data = data.get("json", [{}])[0]
                
                if json_data.get("code") == 0:
                    products = json_data.get("data", {}).get("results", [])
                    print(f"✓ Successfully fetched {len(products)} ranking products")
                    return products
            
            return []
            
        except Exception as e:
            print(f"✗ Failed to fetch rankings: {str(e)}")
            return []
    
    def compare_rankings(self, current_data: List[Dict], previous_data: List[Dict]) -> Dict:
        """
        Compare ranking changes
        
        Args:
            current_data: Current ranking data
            previous_data: Historical ranking data
            
        Returns:
            Dict: Change analysis results
        """
        changes = {
            "new_entries": [],      # New entries
            "rank_up": [],          # Rank increased
            "rank_down": [],        # Rank decreased
            "dropped_out": []       # Dropped from list
        }
        
        # Build ASIN to rank mapping
        current_ranks = {item['asin']: idx + 1 for idx, item in enumerate(current_data)}
        previous_ranks = {item['asin']: idx + 1 for idx, item in enumerate(previous_data)}
        
        # Identify new entries
        for asin in current_ranks:
            if asin not in previous_ranks:
                product = next(p for p in current_data if p['asin'] == asin)
                changes["new_entries"].append({
                    "asin": asin,
                    "title": product.get("title"),
                    "rank": current_ranks[asin]
                })
        
        # Identify ranking changes
        for asin in current_ranks:
            if asin in previous_ranks:
                rank_change = previous_ranks[asin] - current_ranks[asin]
                
                if rank_change > 0:  # Rank improved
                    product = next(p for p in current_data if p['asin'] == asin)
                    changes["rank_up"].append({
                        "asin": asin,
                        "title": product.get("title"),
                        "from_rank": previous_ranks[asin],
                        "to_rank": current_ranks[asin],
                        "change": rank_change
                    })
                elif rank_change < 0:  # Rank dropped
                    product = next(p for p in current_data if p['asin'] == asin)
                    changes["rank_down"].append({
                        "asin": asin,
                        "title": product.get("title"),
                        "from_rank": previous_ranks[asin],
                        "to_rank": current_ranks[asin],
                        "change": abs(rank_change)
                    })
        
        # Identify dropped products
        for asin in previous_ranks:
            if asin not in current_ranks:
                product = next(p for p in previous_data if p['asin'] == asin)
                changes["dropped_out"].append({
                    "asin": asin,
                    "title": product.get("title"),
                    "previous_rank": previous_ranks[asin]
                })
        
        return changes
    
    def generate_report(self, changes: Dict) -> str:
        """Generate change report"""
        report = []
        report.append("=" * 60)
        report.append(f"Best Seller Ranking Change Report")
        report.append(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append("=" * 60)
        
        if changes["new_entries"]:
            report.append(f"\n🆕 New Entries ({len(changes['new_entries'])} items):")
            for item in changes["new_entries"][:5]:
                report.append(f"  #{item['rank']} - {item['title'][:50]}...")
        
        if changes["rank_up"]:
            report.append(f"\n📈 Rank Improved ({len(changes['rank_up'])} items):")
            for item in sorted(changes["rank_up"], key=lambda x: x['change'], reverse=True)[:5]:
                report.append(f"  {item['title'][:50]}...")
                report.append(f"     {item['from_rank']} → {item['to_rank']} (↑{item['change']})")
        
        if changes["rank_down"]:
            report.append(f"\n📉 Rank Dropped ({len(changes['rank_down'])} items):")
            for item in sorted(changes["rank_down"], key=lambda x: x['change'], reverse=True)[:5]:
                report.append(f"  {item['title'][:50]}...")
                report.append(f"     {item['from_rank']} → {item['to_rank']} (↓{item['change']})")
        
        if changes["dropped_out"]:
            report.append(f"\n❌ Dropped from List ({len(changes['dropped_out'])} items):")
            for item in changes["dropped_out"][:5]:
                report.append(f"  {item['title'][:50]}... (Previous rank #{item['previous_rank']})")
        
        return "\n".join(report)

# Usage example
if __name__ == "__main__":
    client = PangolinAPIClient("[email protected]", "your_password")
    
    if client.authenticate():
        monitor = BestSellerMonitor(client)
        
        # Monitor Electronics category Best Seller rankings
        category_url = "https://www.amazon.com/Best-Sellers-Electronics/zgbs/electronics"
        
        # Fetch current rankings
        current_data = monitor.fetch_bestseller_list(category_url)
        
        if current_data:
            print(f"\nSuccessfully fetched {len(current_data)} products")
            print("Monitoring system is ready!")

This monitoring system demonstrates how to integrate API calls, data processing, change detection, and report generation into a complete application. In actual use, you can further expand functionality, such as adding email or WeChat notifications, storing data in databases, generating visualization charts, etc. The key is maintaining code modularity and extensibility so that adding new features later doesn’t require major refactoring.

Production Project 2: Competitor Price Tracking and Alert System

Price is one of the core elements of e-commerce competition. Timely awareness of competitor price changes is crucial for pricing strategy. We’ll build an automated price tracking system that regularly collects competitor prices, detects abnormal fluctuations, and sends alerts when price changes exceed thresholds. This system helps sellers quickly respond to market changes and avoid sales declines due to price disadvantages.

import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional

class PriceTracker:
    """Competitor price tracker"""
    
    def __init__(self, api_client: PangolinAPIClient):
        self.client = api_client
        self.price_history_file = "data/price_history.csv"
        self.load_price_history()
    
    def load_price_history(self):
        """Load price history data"""
        try:
            self.price_df = pd.read_csv(self.price_history_file)
            self.price_df['timestamp'] = pd.to_datetime(self.price_df['timestamp'])
        except FileNotFoundError:
            self.price_df = pd.DataFrame(columns=[
                'timestamp', 'asin', 'title', 'price', 'currency'
            ])
    
    def save_price_history(self):
        """Save price history"""
        self.price_df.to_csv(self.price_history_file, index=False)
    
    def track_price(self, asin: str) -> Optional[Dict]:
        """
        Track single product price
        
        Args:
            asin: Product ASIN
            
        Returns:
            Optional[Dict]: Price information
        """
        product = self.client.fetch_product_detail(asin)
        
        if product and product.get('price'):
            price_str = product['price']
            
            # Parse price (handle $ symbol and commas)
            try:
                price_value = float(price_str.replace('$', '').replace(',', ''))
                
                price_record = {
                    'timestamp': datetime.now(),
                    'asin': asin,
                    'title': product.get('title', ''),
                    'price': price_value,
                    'currency': 'USD'
                }
                
                # Add to history
                self.price_df = pd.concat([
                    self.price_df,
                    pd.DataFrame([price_record])
                ], ignore_index=True)
                
                return price_record
                
            except ValueError:
                print(f"✗ Price parsing failed: {price_str}")
                return None
        
        return None
    
    def detect_price_changes(self, asin: str, threshold_percent: float = 5.0) -> Dict:
        """
        Detect price changes
        
        Args:
            asin: Product ASIN
            threshold_percent: Change threshold (percentage)
            
        Returns:
            Dict: Change analysis results
        """
        # Get product price history
        product_prices = self.price_df[self.price_df['asin'] == asin].sort_values('timestamp')
        
        if len(product_prices) < 2:
            return {"status": "insufficient_data"}
        
        latest_price = product_prices.iloc[-1]['price']
        previous_price = product_prices.iloc[-2]['price']
        
        price_change = latest_price - previous_price
        price_change_percent = (price_change / previous_price) * 100
        
        result = {
            "asin": asin,
            "title": product_prices.iloc[-1]['title'],
            "current_price": latest_price,
            "previous_price": previous_price,
            "change_amount": price_change,
            "change_percent": price_change_percent,
            "status": "normal"
        }
        
        if abs(price_change_percent) >= threshold_percent:
            result["status"] = "alert"
            result["alert_type"] = "price_increase" if price_change > 0 else "price_decrease"
        
        return result

# Usage example
if __name__ == "__main__":
    client = PangolinAPIClient("[email protected]", "your_password")
    
    if client.authenticate():
        tracker = PriceTracker(client)
        
        # Define competitor ASIN list to track
        competitor_asins = [
            "B08N5WRWNW",
            "B07ZPKN6YR",
            "B08L5VFJ2R"
        ]
        
        # Track prices
        print("Starting competitor price tracking...")
        for asin in competitor_asins:
            tracker.track_price(asin)
            time.sleep(1)
        
        tracker.save_price_history()
        print("Price tracking complete!")

The core value of this price tracking system lies in automation and timeliness. By running it daily through scheduled tasks (using cron or Windows Task Scheduler), you can build a complete price history database to support pricing decisions. Further, you can combine machine learning algorithms to predict price trends, or automatically adjust your own pricing strategy based on competitor prices, achieving truly intelligent operations.

Summary and Best Practice Recommendations

Through this article’s 7 complete code examples, we’ve systematically learned all Python Pangolin API integration skills, from basic environment setup and API authentication to advanced concurrent processing and error retry, and finally to production monitoring and tracking systems. This knowledge applies not only to Pangolin API but also represents universal methodology for developing any API integration project.

In practical applications, several key best practices deserve special attention: First is security—never hardcode API keys in your code; use environment variables or encrypted configuration files. Second is stability—comprehensive error handling and retry mechanisms are essential for production environments. Third is performance—proper use of concurrency and caching can dramatically improve system efficiency. Finally is maintainability—modular code structure and clear documentation comments enable long-term healthy project development.

If you want to start practicing now, I recommend following these steps: First, visit the Pangolinfo official website to register an account and obtain API keys. Second, use the code examples provided in this article to build a basic framework. Third, customize functional modules based on your actual needs. Fourth, validate system stability in a small-scale test environment. Fifth, gradually expand to production. Remember, any complex system starts from a simple MVP and iterates—don’t try to implement all features at once, but continuously optimize and improve.

For more technical details and API parameter descriptions, please refer to the Pangolinfo API Official Documentation. If you encounter issues during development, visit the Developer Console to view API call logs and credit usage. Wishing you smooth development and data-driven business growth!

Start Your Python API Development Journey Today → Visit Pangolinfo Scrape API to get free trial credits, or check the Complete API Documentation for more technical details. Empower your data collection with Python code!

Our solution

Protect your web crawler against blocked requests, proxy failure, IP leak, browser crash and CAPTCHAs!

With AMZ Data Tracker, easily access cross-page, endto-end data, solving data fragmentation andcomplexity, empowering quick, informedbusiness decisions.

Weekly Tutorial

Ready to start your data scraping journey?

Sign up for a free account and instantly experience the powerful web data scraping API – no credit card required.

Scan WhatsApp
to Contact

QR Code
Quick Test