为AI Agent提供高质量训练数据：Pangolinfo的电商AI应用场景

import requests
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
import time

class AmazonDataCollector:
    """亚马逊数据批量采集器"""
    
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.pangolinfo.com/scrape"
    
    def fetch_product(self, asin, domain="amazon.com"):
        """获取单个产品数据"""
        params = {
            "api_key": self.api_key,
            "amazon_domain": domain,
            "asin": asin,
            "type": "product",
            "output": "json"
        }
        
        try:
            response = requests.get(self.base_url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"Error fetching {asin}: {str(e)}")
            return None
    
    def batch_fetch(self, asin_list, max_workers=5):
        """批量获取产品数据"""
        results = []
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {executor.submit(self.fetch_product, asin): asin 
                      for asin in asin_list}
            
            for future in futures:
                result = future.result()
                if result:
                    results.append(result)
        
        return results
    
    def save_to_dataset(self, data, filename="training_data.csv"):
        """保存为训练数据集"""
        df = pd.DataFrame(data)
        
        # 选择关键字段
        columns = [
            'asin', 'title', 'price', 'currency', 
            'bsr_rank', 'category', 'rating', 'review_count',
            'availability', 'brand', 'images'
        ]
        
        df = df[columns]
        df.to_csv(filename, index=False, encoding='utf-8')
        print(f"Saved {len(df)} products to {filename}")

# 使用示例
collector = AmazonDataCollector(api_key="your_api_key")

# 批量采集
asin_list = ["B08XYZ123", "B07ABC456", "B09DEF789"]  # 替换为实际ASIN列表
products = collector.batch_fetch(asin_list, max_workers=10)

# 保存数据集
collector.save_to_dataset(products)
        

import pandas as pd
import re

def clean_training_data(df):
    """清洗训练数据"""
    
    # 1. 去除重复数据
    df = df.drop_duplicates(subset=['asin'], keep='last')
    
    # 2. 处理缺失值
    df['price'] = df['price'].fillna(0)
    df['rating'] = df['rating'].fillna(0)
    df['review_count'] = df['review_count'].fillna(0)
    
    # 3. 数据类型转换
    df['price'] = pd.to_numeric(df['price'], errors='coerce')
    df['bsr_rank'] = pd.to_numeric(df['bsr_rank'], errors='coerce')
    
    # 4. 文本清洗
    df['title'] = df['title'].apply(lambda x: re.sub(r'[^\w\s]', '', str(x)))
    
    # 5. 异常值处理
    df = df[df['price'] > 0]  # 移除价格为0的数据
    df = df[df['price'] < 10000]  # 移除异常高价
    
    # 6. 标准化
    df['price_normalized'] = (df['price'] - df['price'].mean()) / df['price'].std()
    
    return df

# 使用
df = pd.read_csv("training_data.csv")
df_clean = clean_training_data(df)
df_clean.to_csv("training_data_clean.csv", index=False)
        

def annotate_for_recommendation(df):
    """为推荐系统添加标注"""
    
    # 1. 价格区间标注
    df['price_tier'] = pd.cut(df['price'], 
                              bins=[0, 20, 50, 100, float('inf')],
                              labels=['budget', 'mid', 'premium', 'luxury'])
    
    # 2. 热度标注
    df['popularity'] = pd.cut(df['review_count'],
                              bins=[0, 100, 1000, 10000, float('inf')],
                              labels=['niche', 'growing', 'popular', 'bestseller'])
    
    # 3. 质量标注
    df['quality_score'] = df['rating'] * (df['review_count'] ** 0.5) / 100
    
    # 4. 竞争力标注
    df['competitiveness'] = df.groupby('category')['bsr_rank'].rank(pct=True)
    
    return df

# 使用
df_annotated = annotate_for_recommendation(df_clean)
        

from sklearn.model_selection import train_test_split

# 划分数据集（70% 训练，15% 验证，15% 测试）
train_val, test = train_test_split(df_annotated, test_size=0.15, random_state=42)
train, val = train_test_split(train_val, test_size=0.176, random_state=42)  # 0.176 ≈ 15/85

# 保存
train.to_csv("train.csv", index=False)
val.to_csv("val.csv", index=False)
test.to_csv("test.csv", index=False)

print(f"训练集: {len(train)} 样本")
print(f"验证集: {len(val)} 样本")
print(f"测试集: {len(test)} 样本")
        

def assess_data_quality(df):
    """评估数据质量基线"""
    
    quality_report = {
        '总样本数': len(df),
        '重复率': df.duplicated().sum() / len(df) * 100,
        '缺失率': {
            col: df[col].isnull().sum() / len(df) * 100 
            for col in df.columns
        },
        '异常值率': {
            'price': ((df['price'] < 0) | (df['price'] > 10000)).sum() / len(df) * 100,
            'rating': ((df['rating'] < 0) | (df['rating'] > 5)).sum() / len(df) * 100
        }
    }
    
    return quality_report

# 使用
baseline = assess_data_quality(df)
print(f"数据质量基线: {baseline}")
        

# ❌ 错误做法：直接修改原始数据
df['price'] = df['price'].fillna(0)

# ✅ 正确做法：创建新列
df['price_clean'] = df['price'].fillna(df['price'].median())
df['price_original'] = df['price']  # 保留原始值
        

def auto_annotate_with_confidence(df):
    """自动标注并计算置信度"""
    
    # 自动标注
    df['category_auto'] = df['title'].apply(classify_category)
    df['sentiment_auto'] = df['reviews'].apply(analyze_sentiment)
    
    # 计算置信度
    df['annotation_confidence'] = df.apply(calculate_confidence, axis=1)
    
    # 标记需要人工审核的样本
    df['needs_review'] = df['annotation_confidence'] < 0.8
    
    return df

# 导出需要人工审核的样本
df_review = df[df['needs_review']]
df_review.to_csv("for_manual_review.csv", index=False)
        

import matplotlib.pyplot as plt
import seaborn as sns

def create_quality_dashboard(df):
    """创建数据质量监控Dashboard"""
    
    fig, axes = plt.subplots(2, 2, figsize=(15, 10))
    
    # 1. 缺失值热图
    sns.heatmap(df.isnull(), cbar=False, ax=axes[0, 0])
    axes[0, 0].set_title('缺失值分布')
    
    # 2. 价格分布
    df['price'].hist(bins=50, ax=axes[0, 1])
    axes[0, 1].set_title('价格分布')
    
    # 3. 评分分布
    df['rating'].value_counts().sort_index().plot(kind='bar', ax=axes[1, 0])
    axes[1, 0].set_title('评分分布')
    
    # 4. 数据质量趋势
    quality_scores = df.groupby('date')['quality_score'].mean()
    quality_scores.plot(ax=axes[1, 1])
    axes[1, 1].set_title('数据质量趋势')
    
    plt.tight_layout()
    plt.savefig('quality_dashboard.png')

create_quality_dashboard(df)