商品销量数据是电商分析中的核心指标，item_get_sales 接口专门用于获取指定商品的销量详情，包括历史销量趋势、时段销量分布、规格销量占比等精细化数据。该接口为销售策略优化、库存管理、竞品分析提供数据支持，广泛应用于电商运营和市场分析场景。
一、接口核心特性分析
1. 接口功能与定位
2. 认证机制
3. 核心参数与响应结构
请求参数
响应核心字段
二、Python 脚本实现
import requests
import time
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, Optional, List
from requests.exceptions import RequestException
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class ItemSalesAPI:
def __init__(self, appkey: str, appsecret: str, platform: str = "general"):
"""
初始化商品销量API客户端
:param appkey: 开放平台appkey
:param appsecret: 开放平台appsecret
:param platform: 平台标识，如"taobao"、"jd"、"vip"等
"""
self.appkey = appkey
self.appsecret = appsecret
self.platform = platform
self.base_urls = {
"taobao": "https://eco.taobao.com",
"jd": "https://api.jd.com",
"vip": "https://api.vip.com",
"mogujie": "https://api.mogujie.com",
"general": "https://api.ecommerce.com" # 通用平台地址
}
self.base_url = self.base_urls.get(platform, self.base_urls["general"])
self.access_token = None
self.token_expires_at = 0 # token过期时间戳
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
})
def _get_access_token(self) -> Optional[str]:
"""获取访问令牌（适配多平台）"""
# 检查token是否有效
if self.access_token and self.token_expires_at > time.time() + 60:
return self.access_token
logging.info(f"获取{self.platform}平台新的access_token")
# 不同平台的token获取地址和参数可能不同
if self.platform == "taobao":
url = f"{self.base_url}/oauth/token"
params = {
"grant_type": "client_credentials",
"client_id": self.appkey,
"client_secret": self.appsecret
}
elif self.platform == "jd":
url = f"{self.base_url}/oauth2/token"
params = {
"grant_type": "client_credentials",
"appkey": self.appkey,
"appsecret": self.appsecret
}
else: # 通用平台
url = f"{self.base_url}/oauth/token"
params = {
"grant_type": "client_credentials",
"client_id": self.appkey,
"client_secret": self.appsecret
}
try:
# 淘宝等平台用post，京东部分接口用get
method = "post" if self.platform in ["taobao", "mogujie"] else "get"
if method == "post":
response = self.session.post(url, data=params, timeout=10)
else:
response = self.session.get(url, params=params, timeout=10)
response.raise_for_status()
result = response.json()
if "access_token" in result:
self.access_token = result["access_token"]
# 设置过期时间，默认2小时
self.token_expires_at = time.time() + result.get("expires_in", 7200)
return self.access_token
else:
logging.error(f"获取access_token失败: {result.get('error_description', '未知错误')}")
return None
except RequestException as e:
logging.error(f"获取access_token请求异常: {str(e)}")
return None
def get_item_sales(self,
item_id: str,
time_range: str = "30d",
granularity: str = "day",
include_sku: bool = False) -> Optional[Dict]:
"""
获取商品销量详情
:param item_id: 商品ID
:param time_range: 时间范围，如"7d"、"30d"、"90d"
:param granularity: 时间粒度，"day"、"week"、"month"
:param include_sku: 是否包含SKU销量数据
:return: 销量详情数据
"""
# 验证参数
valid_ranges = ["7d", "30d", "90d", "180d", "365d"]
if time_range not in valid_ranges:
logging.error(f"无效的时间范围: {time_range}，支持: {valid_ranges}")
return None
valid_granularities = ["day", "week", "month"]
if granularity not in valid_granularities:
logging.error(f"无效的时间粒度: {granularity}，支持: {valid_granularities}")
return None
# 获取有效的access_token
if not self._get_access_token():
return None
# 不同平台的接口路径可能不同
if self.platform == "taobao":
url = f"{self.base_url}/router/rest"
params = {
"method": "taobao.item.sales.get",
"format": "json",
"v": "2.0",
"item_id": item_id,
"time_range": time_range,
"granularity": granularity,
"include_sku": "true" if include_sku else "false",
"access_token": self.access_token
}
else: # 通用接口参数
url = f"{self.base_url}/item/get_sales"
params = {
"item_id": item_id,
"time_range": time_range,
"granularity": granularity,
"include_sku": include_sku,
"access_token": self.access_token
}
try:
response = self.session.get(url, params=params, timeout=15)
response.raise_for_status()
result = response.json()
# 处理不同平台的响应格式
if self.platform == "taobao" and "error_response" in result:
logging.error(f"获取销量失败: {result['error_response'].get('msg')} (错误码: {result['error_response'].get('code')})")
return None
elif self.platform != "taobao" and result.get("code") != 0:
logging.error(f"获取销量失败: {result.get('message')} (错误码: {result.get('code')})")
return None
# 提取核心数据（不同平台数据路径可能不同）
raw_data = result.get("item_sales_get_response", {}).get("result", {}) \
if self.platform == "taobao" else result.get("data", {})
if not raw_data:
logging.warning("未获取到销量数据")
return None
# 格式化销量数据
return self._format_sales_data(raw_data, time_range, granularity)
except RequestException as e:
logging.error(f"获取销量请求异常: {str(e)}")
return None
except json.JSONDecodeError:
logging.error(f"销量响应解析失败: {response.text[:200]}...")
return None
def _format_sales_data(self, sales_data: Dict, time_range: str, granularity: str) -> Dict:
"""格式化销量数据"""
# 基础销量信息
base_info = {
"item_id": sales_data.get("item_id"),
"total_sales": int(sales_data.get("total_sales", 0)), # 累计销量
"total_revenue": float(sales_data.get("total_revenue", 0)), # 累计销售额
"avg_price": round(
float(sales_data.get("total_revenue", 0)) / max(int(sales_data.get("total_sales", 0)), 1),
2
), # 平均客单价
"time_range": time_range,
"update_time": sales_data.get("update_time", datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
}
# 时间趋势数据
trend_data = {
"granularity": granularity,
"dates": sales_data.get("dates", []), # 日期标签
"sales_trend": [int(num) for num in sales_data.get("sales_trend", [])], # 销量趋势
"revenue_trend": [float(num) for num in sales_data.get("revenue_trend", [])], # 销售额趋势
"growth_rate": round(float(sales_data.get("growth_rate", 0)), 4) # 环比增长率
}
# 计算周期内的关键指标
if trend_data["sales_trend"]:
trend_data["max_sales"] = max(trend_data["sales_trend"]) # 周期内最大销量
trend_data["min_sales"] = min(trend_data["sales_trend"]) # 周期内最小销量
trend_data["avg_daily_sales"] = round(
sum(trend_data["sales_trend"]) / len(trend_data["sales_trend"]),
1
) # 日均销量
# SKU销量分布（如包含）
sku_sales = []
if sales_data.get("sku_sales"):
total_sku_sales = sum(int(sku.get("sales", 0)) for sku in sales_data["sku_sales"])
for sku in sales_data["sku_sales"]:
sales = int(sku.get("sales", 0))
sku_sales.append({
"sku_id": sku.get("sku_id"),
"sku_name": sku.get("sku_name"),
"sales": sales,
"revenue": float(sku.get("revenue", 0)),
"proportion": round(sales / max(total_sku_sales, 1), 4) if total_sku_sales > 0 else 0, # 销量占比
"price": float(sku.get("price", 0))
})
# 按销量排序
sku_sales.sort(key=lambda x: x["sales"], reverse=True)
return {
"base_info": base_info,
"trend_data": trend_data,
"sku_sales": sku_sales,
"raw_data": sales_data # 保留原始数据
}
def get_sales_comparison(self, item_id: str, periods: List[str] = ["7d", "30d"]) -> Optional[Dict]:
"""
获取多个周期的销量对比数据
:param item_id: 商品ID
:param periods: 周期列表
:return: 对比数据
"""
comparison = {}
for period in periods:
logging.info(f"获取{period}销量数据用于对比")
sales_data = self.get_item_sales(item_id, time_range=period)
if sales_data:
comparison[period] = {
"total_sales": sales_data["base_info"]["total_sales"],
"total_revenue": sales_data["base_info"]["total_revenue"],
"avg_daily_sales": sales_data["trend_data"].get("avg_daily_sales", 0)
}
# 控制请求频率
time.sleep(1)
return comparison if comparison else None
# 示例调用
if __name__ == "__main__":
# 替换为实际的appkey和appsecret
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 替换为目标商品ID
ITEM_ID = "12345678"
# 平台标识，如"taobao"、"jd"、"vip"等
PLATFORM = "taobao"
# 初始化API客户端
api = ItemSalesAPI(APPKEY, APPSECRET, platform=PLATFORM)
# 获取单个周期的销量数据
sales_details = api.get_item_sales(
item_id=ITEM_ID,
time_range="30d",
granularity="day",
include_sku=True
)
if sales_details:
print(f"=== 商品销量详情 ({item_id}) ===")
print(f"累计销量: {sales_details['base_info']['total_sales']}件")
print(f"累计销售额: {sales_details['base_info']['total_revenue']}元")
print(f"平均客单价: {sales_details['base_info']['avg_price']}元")
print(f"时间范围: {sales_details['base_info']['time_range']}")
print(f"环比增长率: {sales_details['trend_data']['growth_rate']*100}%")
print(f"日均销量: {sales_details['trend_data'].get('avg_daily_sales', 0)}件")
# 打印最近5天的销量
if sales_details['trend_data']['dates'] and sales_details['trend_data']['sales_trend']:
print("\n最近5天销量趋势:")
recent_dates = sales_details['trend_data']['dates'][-5:]
recent_sales = sales_details['trend_data']['sales_trend'][-5:]
for date, sale in zip(recent_dates, recent_sales):
print(f" {date}: {sale}件")
# 打印热销规格
if sales_details['sku_sales']:
print("\n热销规格TOP3:")
for i, sku in enumerate(sales_details['sku_sales'][:3], 1):
print(f" {i}. {sku['sku_name']}: {sku['sales']}件 ({sku['proportion']*100:.1f}%)")
# 获取多周期销量对比
# sales_compare = api.get_sales_comparison(ITEM_ID, periods=["7d", "30d", "90d"])
# if sales_compare:
# print("\n=== 多周期销量对比 ===")
# for period, data in sales_compare.items():
# print(f"{period}销量: {data['total_sales']}件, 日均: {data['avg_daily_sales']}件")