智能决策数据源：利用 1688 商品详情 API 构建实时比价与供应链分析系统

pip install requests pandas matplotlib python-dotenv

import requests
import time
import hashlib
import json
import pandas as pd
import matplotlib.pyplot as plt
from dotenv import load_dotenv
import os

# 加载环境变量（避免硬编码密钥）
load_dotenv()

# 1688 API配置
APP_KEY = os.getenv("1688_APP_KEY")
APP_SECRET = os.getenv("1688_APP_SECRET")
API_URL = "https://gw.open.1688.com/openapi/param2/1/aliexpress.open/api.sdk.top/1688.item.detail.get/{}"

class Ali1688API:
    """1688商品详情API调用类"""
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
    
    def _generate_sign(self, params):
        """生成API请求签名（1688 API要求的签名规则）"""
        # 按参数名排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接字符串
        sign_str = self.app_secret
        for k, v in sorted_params:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        # MD5加密并转大写
        sign = hashlib.md5(sign_str.encode()).hexdigest().upper()
        return sign
    
    def get_item_detail(self, item_id):
        """
        获取商品详情
        :param item_id: 1688商品ID
        :return: 结构化的商品详情字典
        """
        # 构造请求参数
        params = {
            "app_key": self.app_key,
            "method": "1688.item.detail.get",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "format": "json",
            "v": "2.0",
            "item_id": item_id
        }
        # 生成签名
        params["sign"] = self._generate_sign(params)
        
        try:
            # 发送请求
            response = requests.get(API_URL.format(self.app_key), params=params, timeout=10)
            response.raise_for_status()  # 抛出HTTP异常
            result = response.json()
            
            # 解析核心数据
            if result.get("success"):
                data = result["data"]
                return {
                    "item_id": item_id,
                    "title": data.get("title"),
                    "price": float(data.get("price", 0)),  # 批发价
                    "retail_price": float(data.get("retail_price", 0)),  # 零售价
                    "supplier": {
                        "id": data.get("seller", {}).get("userId"),
                        "name": data.get("seller", {}).get("companyName"),
                        "location": data.get("seller", {}).get("address", {}).get("province") + "-" + 
                                    data.get("seller", {}).get("address", {}).get("city", "")
                    },
                    "sales": int(data.get("sales", 0)),  # 销量
                    "stock": int(data.get("stock", 0)),  # 库存
                    "delivery_time": data.get("deliveryTime", "")  # 发货时间
                }
            else:
                print(f"获取商品{item_id}失败：{result.get('errorMessage')}")
                return None
        except Exception as e:
            print(f"API请求异常：{str(e)}")
            return None

class SupplyChainAnalyzer:
    """供应链分析与比价工具类"""
    def __init__(self, api_client):
        self.api_client = api_client
        self.data = pd.DataFrame()  # 存储商品数据的DataFrame
    
    def batch_get_items(self, item_ids):
        """
        批量获取商品数据
        :param item_ids: 商品ID列表
        :return: 填充后的DataFrame
        """
        items_data = []
        for item_id in item_ids:
            print(f"正在获取商品{item_id}数据...")
            item_detail = self.api_client.get_item_detail(item_id)
            if item_detail:
                items_data.append(item_detail)
                time.sleep(1)  # 控制API调用频率，避免超限
        
        self.data = pd.DataFrame(items_data)
        return self.data
    
    def price_comparison(self):
        """
        实时比价分析
        :return: 按价格升序排列的比价结果
        """
        if self.data.empty:
            print("暂无商品数据，无法比价")
            return None
        
        # 按批发价排序
        comparison_result = self.data[["item_id", "title", "price", "supplier.name", "supplier.location"]].sort_values(by="price")
        print("\n=== 商品实时比价结果 ===")
        print(comparison_result)
        return comparison_result
    
    def supplier_analysis(self):
        """
        供应商维度分析（销量、库存、价格）
        :return: 供应商评分DataFrame
        """
        if self.data.empty:
            print("暂无商品数据，无法分析供应商")
            return None
        
        # 构建供应商评分（综合销量、库存、价格）
        supplier_df = self.data.groupby("supplier.name").agg({
            "sales": "sum",  # 总销量
            "stock": "mean",  # 平均库存
            "price": "mean"   # 平均价格
        }).reset_index()
        
        # 标准化评分（0-10分，价格越低、销量越高、库存越充足评分越高）
        supplier_df["sales_score"] = (supplier_df["sales"] / supplier_df["sales"].max()) * 4
        supplier_df["stock_score"] = (supplier_df["stock"] / supplier_df["stock"].max()) * 3
        supplier_df["price_score"] = (1 - supplier_df["price"] / supplier_df["price"].max()) * 3
        supplier_df["total_score"] = supplier_df["sales_score"] + supplier_df["stock_score"] + supplier_df["price_score"]
        
        # 按总分排序
        supplier_df = supplier_df.sort_values(by="total_score", ascending=False)
        print("\n=== 供应商评分结果（总分越高越优质） ===")
        print(supplier_df[["supplier.name", "sales", "stock", "price", "total_score"]])
        return supplier_df
    
    def plot_price_trend(self, item_ids=None):
        """
        绘制价格趋势图（模拟历史数据，实际可结合定时采集的历史数据）
        :param item_ids: 指定商品ID列表，默认全部
        """
        if self.data.empty:
            print("暂无商品数据，无法绘制趋势图")
            return
        
        # 模拟7天价格数据（实际场景需定时采集存储）
        if not item_ids:
            item_ids = self.data["item_id"].tolist()
        
        plt.rcParams["font.sans-serif"] = ["SimHei"]  # 支持中文
        plt.figure(figsize=(12, 6))
        
        for item_id in item_ids:
            item_data = self.data[self.data["item_id"] == item_id]
            if not item_data.empty:
                # 模拟7天价格波动（实际替换为真实历史数据）
                dates = pd.date_range(start="2025-01-01", periods=7)
                base_price = item_data["price"].iloc[0]
                prices = [base_price * (1 + (i % 3 - 1) * 0.05) for i in range(7)]
                
                plt.plot(dates, prices, marker="o", label=f"商品{item_id}")
        
        plt.title("1688商品价格趋势")
        plt.xlabel("日期")
        plt.ylabel("批发价格（元）")
        plt.legend()
        plt.grid(True, alpha=0.3)
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig("price_trend.png")
        print("价格趋势图已保存为price_trend.png")

# ==================== 系统使用示例 ====================
if __name__ == "__main__":
    # 初始化API客户端
    api_client = Ali1688API(APP_KEY, APP_SECRET)
    
    # 初始化分析器
    analyzer = SupplyChainAnalyzer(api_client)
    
    # 批量获取商品数据（替换为实际需要分析的商品ID）
    target_item_ids = ["123456789", "987654321", "112233445"]
    analyzer.batch_get_items(target_item_ids)
    
    # 1. 实时比价
    analyzer.price_comparison()
    
    # 2. 供应商分析
    analyzer.supplier_analysis()
    
    # 3. 价格趋势可视化
    analyzer.plot_price_trend()

1688_APP_KEY=你的App Key
1688_APP_SECRET=你的App Secret