第3课：AI自动化选品
一、选品逻辑详解
1.1 选品的核心指标
选品不是随便找个高佣商品就行，需要综合考虑多个维度：
好商品 = 高佣金 + 高销量 + 好评多 + 有优惠券 + 竞争小
核心筛选条件：
| 指标 | 最低要求 | 理想值 | 说明 |
|---|---|---|---|
| 佣金率 | ≥20% | ≥30% | 佣金率太低不赚钱 |
| 月销量 | ≥1000 | ≥5000 | 卖得动说明需求大 |
| 好评率 | ≥4.8 | ≥4.9 | 差评多的商品推广了会掉粉 |
| 优惠券 | ≥10元 | ≥30元 | 有券才有吸引力 |
| 价格区间 | 30-300元 | 50-150元 | 太便宜佣金少，太贵转化低 |
| 店铺评分 | ≥4.7 | ≥4.8 | 店铺评分低影响售后 |
1.2 选品公式
def calculate_product_score(product):
"""
商品综合评分公式
满分100分，60分以上值得推广
"""
score = 0
# 佣金率评分（40分）
commission_rate = float(product.get('commission_rate', 0))
if commission_rate >= 30:
score += 40
elif commission_rate >= 25:
score += 35
elif commission_rate >= 20:
score += 30
elif commission_rate >= 15:
score += 20
else:
score += 10
# 月销量评分（25分）
sales = int(product.get('sales', 0))
if sales >= 10000:
score += 25
elif sales >= 5000:
score += 20
elif sales >= 1000:
score += 15
elif sales >= 500:
score += 10
else:
score += 5
# 优惠券金额评分（20分）
coupon = float(product.get('coupon_amount', 0))
price = float(product.get('price', 0))
coupon_ratio = coupon / price if price > 0 else 0
if coupon_ratio >= 0.3:
score += 20
elif coupon_ratio >= 0.2:
score += 15
elif coupon_ratio >= 0.1:
score += 10
else:
score += 5
# 价格区间评分（15分）
if 50 <= price <= 150:
score += 15
elif 30 <= price <= 300:
score += 10
else:
score += 5
return score
1.3 不同品类的选品策略
美妆护肤：
├── 佣金率：30%-50%（高佣品类）
├── 价格区间：50-200元
├── 关注点：品牌、成分、口碑
└── 旺季：618、双11、年货节
服饰鞋包：
├── 佣金率：20%-40%
├── 价格区间：80-300元
├── 关注点：款式、尺码、退货率
└── 旺季：换季前后
食品生鲜：
├── 佣金率：15%-30%
├── 价格区间：30-100元
├── 关注点：保质期、口碑、复购率
└── 旺季：节日、年货
数码家电：
├── 佣金率：5%-15%（低佣但客单价高）
├── 价格区间：100-2000元
├── 关注点：品牌、参数、售后
└── 旺季：618、双11
家居日用：
├── 佣金率：20%-40%
├── 价格区间：30-150元
├── 关注点：实用性、颜值、口碑
└── 旺季：全年平稳
二、用Hermes Agent自动抓取高佣商品
2.1 选品脚本框架
#!/usr/bin/env python3
"""
AI自动化选品系统
功能：
1. 自动抓取高佣商品
2. 智能评分筛选
3. 自动分类
4. 爆品预测
5. 数据库存储
"""
import json
import time
import sqlite3
import requests
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
from dataclasses import dataclass, asdict
from collections import defaultdict
@dataclass
class Product:
"""商品数据类"""
item_id: str
title: str
price: float
original_price: float
coupon_amount: float
commission_rate: float
sales: int
shop_title: str
user_type: int # 0=集市, 1=天猫
category: str
pict_url: str
score: float = 0
trend_score: float = 0
created_at: str = ''
def to_dict(self):
return asdict(self)
class ProductScorer:
"""商品评分器"""
# 权重配置
WEIGHTS = {
'commission': 0.35,
'sales': 0.25,
'coupon': 0.20,
'price': 0.10,
'user_type': 0.10,
}
@classmethod
def score(cls, product: Product) -> float:
"""
计算商品综合评分
"""
scores = {}
# 佣金率评分
rate = product.commission_rate
if rate >= 50:
scores['commission'] = 100
elif rate >= 40:
scores['commission'] = 90
elif rate >= 30:
scores['commission'] = 80
elif rate >= 25:
scores['commission'] = 70
elif rate >= 20:
scores['commission'] = 60
elif rate >= 15:
scores['commission'] = 40
else:
scores['commission'] = 20
# 销量评分
sales = product.sales
if sales >= 50000:
scores['sales'] = 100
elif sales >= 20000:
scores['sales'] = 90
elif sales >= 10000:
scores['sales'] = 80
elif sales >= 5000:
scores['sales'] = 70
elif sales >= 1000:
scores['sales'] = 60
elif sales >= 500:
scores['sales'] = 40
else:
scores['sales'] = 20
# 优惠券评分
if product.price > 0:
coupon_ratio = product.coupon_amount / product.price
if coupon_ratio >= 0.5:
scores['coupon'] = 100
elif coupon_ratio >= 0.3:
scores['coupon'] = 80
elif coupon_ratio >= 0.2:
scores['coupon'] = 60
elif coupon_ratio >= 0.1:
scores['coupon'] = 40
else:
scores['coupon'] = 20
else:
scores['coupon'] = 0
# 价格区间评分
price = product.price
if 50 <= price <= 150:
scores['price'] = 100
elif 30 <= price <= 300:
scores['price'] = 80
elif 10 <= price <= 500:
scores['price'] = 60
else:
scores['price'] = 30
# 卖家类型评分（天猫加分）
scores['user_type'] = 90 if product.user_type == 1 else 60
# 加权求和
total = sum(scores[k] * cls.WEIGHTS[k] for k in cls.WEIGHTS)
return round(total, 2)
class CategoryClassifier:
"""商品分类器"""
# 类目关键词映射
CATEGORY_KEYWORDS = {
'美妆': ['口红', '眼影', '粉底', '腮红', '睫毛膏', '眉笔', '遮瑕', '定妆', '高光', '修容'],
'护肤': ['面膜', '精华', '面霜', '乳液', '爽肤水', '防晒', '眼霜', '洁面', '卸妆', '水乳'],
'服饰': ['连衣裙', 'T恤', '衬衫', '外套', '裤子', '裙子', '卫衣', '毛衣', '羽绒服', '牛仔'],
'鞋包': ['运动鞋', '高跟鞋', '单肩包', '双肩包', '手提包', '斜挎包', '凉鞋', '靴子', '拖鞋'],
'数码': ['手机', '耳机', '充电器', '数据线', '平板', '笔记本', '键盘', '鼠标', '音箱'],
'食品': ['零食', '坚果', '饼干', '巧克力', '糖果', '果干', '肉脯', '饮料', '茶叶'],
'家居': ['床单', '枕头', '毛巾', '收纳', '置物架', '垃圾桶', '拖鞋', '地毯', '窗帘'],
'母婴': ['奶粉', '纸尿裤', '婴儿', '童装', '玩具', '辅食', '奶瓶', '推车'],
'家电': ['电饭煲', '吹风机', '吸尘器', '空气净化器', '洗衣机', '冰箱', '空调'],
'运动': ['瑜伽', '跑步', '健身', '球类', '游泳', '户外', '登山', '自行车'],
}
@classmethod
def classify(cls, title: str) -> str:
"""
根据标题自动分类
"""
title_lower = title.lower()
for category, keywords in cls.CATEGORY_KEYWORDS.items():
for keyword in keywords:
if keyword in title_lower:
return category
return '其他'
class TrendAnalyzer:
"""趋势分析器"""
def __init__(self, db_path: str):
self.db_path = db_path
def get_historical_sales(self, item_id: str, days: int = 7) -> List[int]:
"""
获取商品历史销量
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT sales, created_at
FROM product_history
WHERE item_id = ?
AND created_at >= datetime('now', ?)
ORDER BY created_at ASC
''', (item_id, f'-{days} days'))
results = cursor.fetchall()
conn.close()
return [r[0] for r in results]
def calculate_trend(self, item_id: str) -> float:
"""
计算销量趋势得分
正数表示上升趋势，负数表示下降趋势
"""
sales_history = self.get_historical_sales(item_id)
if len(sales_history) < 2:
return 0
# 计算增长率
first = sales_history[0] if sales_history[0] > 0 else 1
last = sales_history[-1]
growth_rate = (last - first) / first
# 计算波动性（标准差）
import statistics
if len(sales_history) > 1:
std_dev = statistics.stdev(sales_history)
avg = statistics.mean(sales_history)
volatility = std_dev / avg if avg > 0 else 0
else:
volatility = 0
# 趋势得分 = 增长率 × 100 - 波动性 × 50
trend_score = growth_rate * 100 - volatility * 50
return round(trend_score, 2)
def predict_hot_products(self, products: List[Product]) -> List[Product]:
"""
预测爆品
结合当前评分和趋势得分
"""
for product in products:
trend = self.calculate_trend(product.item_id)
product.trend_score = trend
# 综合得分 = 基础评分 × 0.7 + 趋势得分 × 0.3
# 趋势得分归一化到0-100
normalized_trend = max(0, min(100, (trend + 50)))
product.score = product.score * 0.7 + normalized_trend * 0.3
return sorted(products, key=lambda p: p.score, reverse=True)
class ProductDatabase:
"""商品数据库"""
def __init__(self, db_path: str = "products.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""初始化数据库"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# 商品主表
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
item_id TEXT PRIMARY KEY,
title TEXT,
price REAL,
original_price REAL,
coupon_amount REAL,
commission_rate REAL,
sales INTEGER,
shop_title TEXT,
user_type INTEGER,
category TEXT,
pict_url TEXT,
score REAL,
trend_score REAL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 商品历史表（用于趋势分析）
cursor.execute('''
CREATE TABLE IF NOT EXISTS product_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT,
sales INTEGER,
price REAL,
commission_rate REAL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 推广记录表
cursor.execute('''
CREATE TABLE IF NOT EXISTS promote_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT,
channel TEXT,
tpwd TEXT,
clicks INTEGER DEFAULT 0,
orders INTEGER DEFAULT 0,
commission REAL DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# 创建索引
cursor.execute('CREATE INDEX IF NOT EXISTS idx_products_score ON products(score DESC)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_products_category ON products(category)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_products_commission ON products(commission_rate DESC)')
cursor.execute('CREATE INDEX IF NOT EXISTS idx_history_item ON product_history(item_id)')
conn.commit()
conn.close()
def save_product(self, product: Product):
"""保存商品"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO products
(item_id, title, price, original_price, coupon_amount,
commission_rate, sales, shop_title, user_type, category,
pict_url, score, trend_score, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
''', (
product.item_id, product.title, product.price,
product.original_price, product.coupon_amount,
product.commission_rate, product.sales,
product.shop_title, product.user_type,
product.category, product.pict_url,
product.score, product.trend_score
))
# 保存历史记录
cursor.execute('''
INSERT INTO product_history (item_id, sales, price, commission_rate)
VALUES (?, ?, ?, ?)
''', (product.item_id, product.sales, product.price, product.commission_rate))
conn.commit()
conn.close()
def save_products(self, products: List[Product]):
"""批量保存商品"""
for product in products:
self.save_product(product)
def get_top_products(
self,
limit: int = 50,
category: str = None,
min_score: float = 60
) -> List[Dict]:
"""获取Top商品"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
query = '''
SELECT * FROM products
WHERE score >= ?
'''
params = [min_score]
if category:
query += ' AND category = ?'
params.append(category)
query += ' ORDER BY score DESC LIMIT ?'
params.append(limit)
cursor.execute(query, params)
columns = [description[0] for description in cursor.description]
results = [dict(zip(columns, row)) for row in cursor.fetchall()]
conn.close()
return results
def get_category_stats(self) -> Dict[str, Dict]:
"""获取各类目统计"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT
category,
COUNT(*) as count,
AVG(commission_rate) as avg_commission,
AVG(sales) as avg_sales,
AVG(score) as avg_score
FROM products
GROUP BY category
ORDER BY avg_score DESC
''')
stats = {}
for row in cursor.fetchall():
stats[row[0]] = {
'count': row[1],
'avg_commission': round(row[2], 2),
'avg_sales': round(row[2], 0),
'avg_score': round(row[4], 2),
}
conn.close()
return stats
class AutoProductSelector:
"""自动选品系统"""
def __init__(self, app_key: str, app_secret: str, adzone_id: str):
"""
初始化选品系统
"""
self.api = TaoBaoKeAPI(app_key, app_secret, adzone_id)
self.db = ProductDatabase()
self.scorer = ProductScorer()
self.classifier = CategoryClassifier()
self.trend_analyzer = TrendAnalyzer("products.db")
def select_products(
self,
keywords: List[str],
min_rate: int = 2000,
min_sales: int = 1000,
min_score: float = 60,
limit_per_keyword: int = 30
) -> List[Product]:
"""
自动选品
Args:
keywords: 搜索关键词列表
min_rate: 最低佣金率（万分比）
min_sales: 最低月销量
min_score: 最低评分
limit_per_keyword: 每个关键词返回数量
Returns:
筛选后的商品列表
"""
all_products = []
for keyword in keywords:
print(f"\n🔍 搜索: {keyword}")
# 搜索商品
results = self.api.search_high_commission_products(
keyword=keyword,
min_rate=min_rate,
limit=limit_per_keyword
)
for item in results:
# 创建商品对象
product = Product(
item_id=str(item.get('item_id', '')),
title=item.get('title', ''),
price=float(item.get('price', 0)),
original_price=float(item.get('original_price', 0)),
coupon_amount=float(item.get('coupon_amount', 0) or 0),
commission_rate=float(item.get('commission_rate', 0)),
sales=int(item.get('sales', 0)),
shop_title=item.get('shop_title', ''),
user_type=int(item.get('user_type', 0)),
category='',
pict_url=item.get('pict_url', ''),
created_at=datetime.now().strftime('%Y-%m-%d %H:%M:%S')
)
# 筛选条件
if product.sales < min_sales:
continue
# 自动分类
product.category = self.classifier.classify(product.title)
# 计算评分
product.score = self.scorer.score(product)
if product.score >= min_score:
all_products.append(product)
print(f" 找到 {len([p for p in all_products if p.category != '其他'])} 个符合条件的商品")
# 控制请求频率
time.sleep(1)
# 趋势分析和爆品预测
print("\n📈 进行趋势分析...")
all_products = self.trend_analyzer.predict_hot_products(all_products)
# 保存到数据库
print("💾 保存到数据库...")
self.db.save_products(all_products)
return all_products
def get_daily_top50(self) -> List[Dict]:
"""
获取今日Top50高佣商品
"""
return self.db.get_top_products(limit=50, min_score=60)
def get_category_top(self, category: str, limit: int = 10) -> List[Dict]:
"""
获取指定类目Top商品
"""
return self.db.get_top_products(limit=limit, category=category, min_score=60)
def generate_report(self) -> str:
"""
生成选品报告
"""
top_products = self.get_daily_top50()
category_stats = self.db.get_category_stats()
report = f"""# 今日选品报告
生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
## 一、Top50高佣商品
| 排名 | 商品 | 价格 | 佣金率 | 月销 | 评分 | 类目 |
|------|------|------|--------|------|------|------|
"""
for i, product in enumerate(top_products[:50], 1):
report += f"| {i} | {product['title'][:20]}... | ¥{product['price']:.0f} | {product['commission_rate']:.0f}% | {product['sales']} | {product['score']:.1f} | {product['category']} |\n"
report += f"""
## 二、类目统计
| 类目 | 商品数 | 平均佣金 | 平均销量 | 平均评分 |
|------|--------|----------|----------|----------|
"""
for category, stats in category_stats.items():
report += f"| {category} | {stats['count']} | {stats['avg_commission']:.1f}% | {stats['avg_sales']:.0f} | {stats['avg_score']:.1f} |\n"
report += f"""
## 三、推荐推广商品
基于评分和趋势，以下商品最值得推广：
"""
for i, product in enumerate(top_products[:10], 1):
final_price = product['price'] - product['coupon_amount']
est_commission = final_price * product['commission_rate'] / 10000
report += f"""### {i}. {product['title'][:40]}
- **价格**: ¥{product['price']:.0f} → ¥{final_price:.0f}（券后）
- **佣金**: {product['commission_rate']:.0f}%（约¥{est_commission:.2f}/单）
- **月销**: {product['sales']}
- **评分**: {product['score']:.1f}
- **类目**: {product['category']}
"""
return report
# ========== 主程序 ==========
if __name__ == '__main__':
# 配置
APP_KEY = "your_app_key"
APP_SECRET="your..."
ADZONE_ID = "your_adzone_id"
# 创建选品系统
selector = AutoProductSelector(APP_KEY, APP_SECRET, ADZONE_ID)
# 定义搜索关键词
keywords = [
# 美妆护肤
"面膜", "口红", "防晒霜", "精华液", "洗面奶",
# 服饰
"连衣裙", "T恤女", "牛仔裤",
# 食品
"零食大礼包", "坚果",
# 家居
"收纳盒", "床上四件套",
# 数码
"蓝牙耳机", "充电宝",
]
# 执行选品
print("🚀 开始自动选品...\n")
products = selector.select_products(
keywords=keywords,
min_rate=2000, # 佣金率>=20%
min_sales=1000, # 月销>=1000
min_score=60, # 评分>=60
limit_per_keyword=20
)
print(f"\n✅ 选品完成！共筛选出 {len(products)} 个商品")
# 生成报告
report = selector.generate_report()
# 保存报告
report_file = f"选品报告_{datetime.now().strftime('%Y%m%d')}.md"
with open(report_file, 'w', encoding='utf-8') as f:
f.write(report)
print(f"📊 报告已保存到: {report_file}")
# 打印Top10
print("\n🏆 今日Top10高佣商品：\n")
top10 = selector.get_daily_top50()[:10]
for i, p in enumerate(top10, 1):
print(f"{i}. {p['title'][:30]}...")
print(f" 价格: ¥{p['price']:.0f} | 佣金: {p['commission_rate']:.0f}% | 评分: {p['score']:.1f}")
三、自动分类系统
3.1 基于关键词的分类
class KeywordClassifier:
"""基于关键词的分类器"""
# 分类规则（优先级从高到低）
RULES = [
# 美妆
{
'category': '美妆',
'keywords': ['口红', '眼影', '粉底', '腮红', '睫毛膏', '眉笔', '遮瑕', '高光', '修容', '散粉'],
'exclude': ['套装', '礼盒'] # 排除词
},
# 护肤
{
'category': '护肤',
'keywords': ['面膜', '精华', '面霜', '乳液', '爽肤水', '防晒', '眼霜', '洁面', '卸妆', '水乳', '护肤'],
'exclude': []
},
# 服饰
{
'category': '服饰',
'keywords': ['连衣裙', 'T恤', '衬衫', '外套', '裤子', '裙子', '卫衣', '毛衣', '羽绒服', '牛仔', '上衣', '短袖'],
'exclude': []
},
# 鞋靴
{
'category': '鞋靴',
'keywords': ['运动鞋', '高跟鞋', '凉鞋', '靴子', '拖鞋', '板鞋', '帆布鞋', '皮鞋'],
'exclude': []
},
# 箱包
{
'category': '箱包',
'keywords': ['单肩包', '双肩包', '手提包', '斜挎包', '钱包', '行李箱', '拉杆箱'],
'exclude': []
},
# 数码
{
'category': '数码',
'keywords': ['手机', '耳机', '充电器', '数据线', '平板', '键盘', '鼠标', '音箱', '充电宝', '蓝牙'],
'exclude': []
},
# 食品
{
'category': '食品',
'keywords': ['零食', '坚果', '饼干', '巧克力', '糖果', '果干', '肉脯', '饮料', '茶叶', '速食'],
'exclude': []
},
# 家居
{
'category': '家居',
'keywords': ['床单', '枕头', '毛巾', '收纳', '置物架', '地毯', '窗帘', '四件套', '被子'],
'exclude': []
},
# 母婴
{
'category': '母婴',
'keywords': ['奶粉', '纸尿裤', '婴儿', '童装', '玩具', '辅食', '奶瓶', '推车', '宝宝'],
'exclude': []
},
# 家电
{
'category': '家电',
'keywords': ['电饭煲', '吹风机', '吸尘器', '空气净化器', '洗衣机', '冰箱', '空调', '微波炉'],
'exclude': []
},
]
@classmethod
def classify(cls, title: str) -> str:
"""根据标题分类"""
title_lower = title.lower()
for rule in cls.RULES:
# 检查排除词
if any(ex in title_lower for ex in rule['exclude']):
continue
# 检查关键词
if any(kw in title_lower for kw in rule['keywords']):
return rule['category']
return '其他'
3.2 基于机器学习的分类（进阶）
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
import pickle
class MLClassifier:
"""基于机器学习的分类器"""
def __init__(self, model_path: str = "classifier_model.pkl"):
self.model_path = model_path
self.pipeline = None
self._load_or_create()
def _load_or_create(self):
"""加载或创建模型"""
try:
with open(self.model_path, 'rb') as f:
self.pipeline = pickle.load(f)
print("✓ 已加载分类模型")
except FileNotFoundError:
self.pipeline = Pipeline([
('tfidf', TfidfVectorizer(max_features=10000, ngram_range=(1, 2))),
('clf', MultinomialNB())
])
print("✓ 已创建新分类模型")
def train(self, titles: List[str], categories: List[str]):
"""
训练模型
Args:
titles: 商品标题列表
categories: 对应类目列表
"""
self.pipeline.fit(titles, categories)
# 保存模型
with open(self.model_path, 'wb') as f:
pickle.dump(self.pipeline, f)
print(f"✓ 模型已保存，训练样本数: {len(titles)}")
def predict(self, title: str) -> str:
"""预测分类"""
return self.pipeline.predict([title])[0]
def predict_batch(self, titles: List[str]) -> List[str]:
"""批量预测"""
return self.pipeline.predict(titles).tolist()
def predict_proba(self, title: str) -> Dict[str, float]:
"""预测各类别的概率"""
probas = self.pipeline.predict_proba([title])[0]
classes = self.pipeline.classes_
return dict(zip(classes, probas))
# 使用示例
def train_classifier():
"""训练分类器"""
# 准备训练数据（实际使用时应该有更多数据）
training_data = [
("MAC子弹头口红持久不脱色", "美妆"),
("兰蔻小黑瓶精华液", "护肤"),
("优衣库联名T恤女", "服饰"),
("Nike Air Max运动鞋", "鞋靴"),
("小米蓝牙耳机", "数码"),
("三只松鼠零食大礼包", "食品"),
("全棉时代毛巾", "家居"),
# ... 更多训练数据
]
titles = [d[0] for d in training_data]
categories = [d[1] for d in training_data]
classifier = MLClassifier()
classifier.train(titles, categories)
# 测试
test_titles = [
"YSL小金条口红",
"资生堂防晒霜",
"阿迪达斯运动裤",
]
for title in test_titles:
category = classifier.predict(title)
print(f"{title} → {category}")
四、爆品预测
4.1 趋势分析算法
import statistics
from typing import List, Tuple
class TrendPredictor:
"""趋势预测器"""
def __init__(self, db_path: str):
self.db_path = db_path
def get_sales_history(self, item_id: str, days: int = 30) -> List[Tuple[str, int]]:
"""
获取销量历史
"""
import sqlite3
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
SELECT DATE(created_at) as date, MAX(sales) as sales
FROM product_history
WHERE item_id = ?
AND created_at >= datetime('now', ?)
GROUP BY DATE(created_at)
ORDER BY date ASC
''', (item_id, f'-{days} days'))
results = cursor.fetchall()
conn.close()
return results
def calculate_growth_rate(self, sales_history: List[Tuple[str, int]]) -> float:
"""
计算增长率
"""
if len(sales_history) < 2:
return 0
sales_values = [s[1] for s in sales_history]
# 计算日均增长
daily_growths = []
for i in range(1, len(sales_values)):
if sales_values[i-1] > 0:
growth = (sales_values[i] - sales_values[i-1]) / sales_values[i-1]
daily_growths.append(growth)
if not daily_growths:
return 0
return statistics.mean(daily_growths)
def calculate_volatility(self, sales_history: List[Tuple[str, int]]) -> float:
"""
计算波动性（变异系数）
"""
sales_values = [s[1] for s in sales_history]
if len(sales_values) < 2:
return 0
avg = statistics.mean(sales_values)
std = statistics.stdev(sales_values)
return std / avg if avg > 0 else 0
def calculate_momentum(self, sales_history: List[Tuple[str, int]], window: int = 7) -> float:
"""
计算动量指标
使用最近window天的平均增长 vs 之前的平均增长
"""
sales_values = [s[1] for s in sales_history]
if len(sales_values) < window * 2:
return 0
# 最近window天的增长
recent = sales_values[-window:]
recent_growth = (recent[-1] - recent[0]) / recent[0] if recent[0] > 0 else 0
# 之前window天的增长
previous = sales_values[-window*2:-window]
previous_growth = (previous[-1] - previous[0]) / previous[0] if previous[0] > 0 else 0
# 动量 = 最近增长 - 之前增长
return recent_growth - previous_growth
def predict_hot_score(self, item_id: str) -> dict:
"""
计算爆品得分
"""
history = self.get_sales_history(item_id, days=30)
if len(history) < 3:
return {
'score': 0,
'growth_rate': 0,
'volatility': 0,
'momentum': 0,
'prediction': '数据不足'
}
growth_rate = self.calculate_growth_rate(history)
volatility = self.calculate_volatility(history)
momentum = self.calculate_momentum(history)
# 综合得分
# 增长率权重50%，动量权重30%，稳定性权重20%
score = (
growth_rate * 50 +
momentum * 30 +
(1 - volatility) * 20
) * 100
# 归一化到0-100
score = max(0, min(100, score + 50))
# 预测
if score >= 80:
prediction = '🔥 爆品潜力大'
elif score >= 60:
prediction = '📈 上升趋势'
elif score >= 40:
prediction = '➡️ 平稳'
else:
prediction = '📉 下降趋势'
return {
'score': round(score, 2),
'growth_rate': round(growth_rate * 100, 2),
'volatility': round(volatility, 2),
'momentum': round(momentum * 100, 2),
'prediction': prediction,
'data_points': len(history)
}
4.2 爆品特征识别
class HotProductFeatures:
"""爆品特征识别"""
# 爆品特征权重
FEATURES = {
'high_commission': 0.20, # 高佣金
'high_sales': 0.20, # 高销量
'rising_trend': 0.20, # 上升趋势
'good_price': 0.15, # 合适价格
'has_coupon': 0.10, # 有优惠券
'tmall_shop': 0.10, # 天猫店铺
'good_title': 0.05, # 标题质量
}
@classmethod
def analyze(cls, product: dict) -> dict:
"""
分析商品是否具有爆品特征
"""
scores = {}
# 高佣金
commission = product.get('commission_rate', 0)
scores['high_commission'] = min(commission / 50, 1.0)
# 高销量
sales = product.get('sales', 0)
scores['high_sales'] = min(sales / 10000, 1.0)
# 上升趋势
trend = product.get('trend_score', 0)
scores['rising_trend'] = max(0, min((trend + 50) / 100, 1.0))
# 合适价格
price = product.get('price', 0)
if 50 <= price <= 150:
scores['good_price'] = 1.0
elif 30 <= price <= 300:
scores['good_price'] = 0.7
else:
scores['good_price'] = 0.3
# 有优惠券
coupon = product.get('coupon_amount', 0)
scores['has_coupon'] = min(coupon / 50, 1.0) if coupon > 0 else 0
# 天猫店铺
scores['tmall_shop'] = 1.0 if product.get('user_type') == 1 else 0.5
# 标题质量（长度适中，有关键词）
title = product.get('title', '')
if 15 <= len(title) <= 50:
scores['good_title'] = 1.0
else:
scores['good_title'] = 0.5
# 加权得分
total_score = sum(scores[k] * cls.FEATURES[k] for k in cls.FEATURES)
return {
'score': round(total_score * 100, 2),
'features': scores,
'is_hot': total_score >= 0.7
}
五、选品数据库
5.1 数据库设计
-- 商品主表
CREATE TABLE products (
item_id TEXT PRIMARY KEY, -- 商品ID
title TEXT NOT NULL, -- 商品标题
price REAL, -- 当前价格
original_price REAL, -- 原价
coupon_amount REAL DEFAULT 0, -- 优惠券金额
commission_rate REAL, -- 佣金比例
sales INTEGER DEFAULT 0, -- 月销量
shop_title TEXT, -- 店铺名称
user_type INTEGER DEFAULT 0, -- 0=集市, 1=天猫
category TEXT, -- 分类
pict_url TEXT, -- 图片URL
score REAL DEFAULT 0, -- 综合评分
trend_score REAL DEFAULT 0, -- 趋势得分
tpwd TEXT, -- 淘口令
promo_link TEXT, -- 推广链接
status INTEGER DEFAULT 1, -- 1=正常, 0=下架
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 价格历史表
CREATE TABLE price_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT,
price REAL,
coupon_amount REAL,
commission_rate REAL,
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (item_id) REFERENCES products(item_id)
);
-- 推广记录表
CREATE TABLE promote_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT,
channel TEXT, -- 推广渠道
tpwd TEXT, -- 使用的淘口令
clicks INTEGER DEFAULT 0, -- 点击数
orders INTEGER DEFAULT 0, -- 订单数
commission REAL DEFAULT 0, -- 佣金收入
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (item_id) REFERENCES products(item_id)
);
-- 选品规则表
CREATE TABLE selection_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT, -- 规则名称
category TEXT, -- 适用类目
min_commission_rate REAL, -- 最低佣金率
min_sales INTEGER, -- 最低销量
min_score REAL, -- 最低评分
keywords TEXT, -- 关键词（JSON数组）
is_active BOOLEAN DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 创建索引
CREATE INDEX idx_products_score ON products(score DESC);
CREATE INDEX idx_products_category ON products(category);
CREATE INDEX idx_products_commission ON products(commission_rate DESC);
CREATE INDEX idx_products_sales ON products(sales DESC);
CREATE INDEX idx_history_item ON price_history(item_id);
CREATE INDEX idx_promote_item ON promote_records(item_id);
CREATE INDEX idx_promote_channel ON promote_records(channel);
5.2 数据库操作封装
import sqlite3
import json
from typing import List, Dict, Optional
from contextlib import contextmanager
class ProductDB:
"""商品数据库操作类"""
def __init__(self, db_path: str = "products.db"):
self.db_path = db_path
self._init_db()
@contextmanager
def _get_conn(self):
"""获取数据库连接（上下文管理器）"""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def _init_db(self):
"""初始化数据库"""
with self._get_conn() as conn:
cursor = conn.cursor()
# 创建表（使用上面的SQL）
cursor.executesuite('''
-- ... 上面的建表SQL ...
''')
def upsert_product(self, product: dict):
"""插入或更新商品"""
with self._get_conn() as conn:
cursor = conn.cursor()
cursor.execute('''
INSERT INTO products
(item_id, title, price, original_price, coupon_amount,
commission_rate, sales, shop_title, user_type, category,
pict_url, score, trend_score, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(item_id) DO UPDATE SET
price = excluded.price,
coupon_amount = excluded.coupon_amount,
commission_rate = excluded.commission_rate,
sales = excluded.sales,
score = excluded.score,
trend_score = excluded.trend_score,
updated_at = CURRENT_TIMESTAMP
''', (
product['item_id'],
product['title'],
product['price'],
product.get('original_price', product['price']),
product.get('coupon_amount', 0),
product['commission_rate'],
product['sales'],
product.get('shop_title', ''),
product.get('user_type', 0),
product.get('category', ''),
product.get('pict_url', ''),
product.get('score', 0),
product.get('trend_score', 0),
))
def batch_upsert(self, products: List[dict]):
"""批量插入或更新"""
with self._get_conn() as conn:
cursor = conn.cursor()
for product in products:
self.upsert_product(product)
def get_top(
self,
limit: int = 50,
category: Optional[str] = None,
min_score: float = 0,
min_commission: float = 0
) -> List[dict]:
"""获取Top商品"""
with self._get_conn() as conn:
cursor = conn.cursor()
query = "SELECT * FROM products WHERE 1=1"
params = []
if category:
query += " AND category = ?"
params.append(category)
if min_score > 0:
query += " AND score >= ?"
params.append(min_score)
if min_commission > 0:
query += " AND commission_rate >= ?"
params.append(min_commission)
query += " ORDER BY score DESC LIMIT ?"
params.append(limit)
cursor.execute(query, params)
return [dict(row) for row in cursor.fetchall()]
def search(self, keyword: str, limit: int = 20) -> List[dict]:
"""搜索商品"""
with self._get_conn() as conn:
cursor = conn.cursor()
cursor.execute('''
SELECT * FROM products
WHERE title LIKE ?
ORDER BY score DESC
LIMIT ?
''', (f'%{keyword}%', limit))
return [dict(row) for row in cursor.fetchall()]
def get_stats(self) -> dict:
"""获取统计信息"""
with self._get_conn() as conn:
cursor = conn.cursor()
# 总商品数
cursor.execute("SELECT COUNT(*) FROM products")
total = cursor.fetchone()[0]
# 各类目数量
cursor.execute('''
SELECT category, COUNT(*) as count, AVG(score) as avg_score
FROM products
GROUP BY category
ORDER BY count DESC
''')
categories = [dict(row) for row in cursor.fetchall()]
# 平均佣金
cursor.execute("SELECT AVG(commission_rate) FROM products")
avg_commission = cursor.fetchone()[0] or 0
return {
'total': total,
'categories': categories,
'avg_commission': round(avg_commission, 2)
}
def export_csv(self, filename: str, limit: int = 1000):
"""导出为CSV"""
import csv
products = self.get_top(limit=limit)
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
if products:
writer = csv.DictWriter(f, fieldnames=products[0].keys())
writer.writeheader()
writer.writerows(products)
print(f"✓ 已导出 {len(products)} 条记录到 {filename}")
六、实操：跑一次自动选品
6.1 完整选品脚本
#!/usr/bin/env python3
"""
完整自动选品脚本
输出今日Top50高佣商品
"""
import json
from datetime import datetime
from auto_product_selector import AutoProductSelector
def main():
# 配置
APP_KEY = "your_app_key"
APP_SECRET="your..."
ADZONE_ID = "your_adzone_id"
# 创建选品系统
selector = AutoProductSelector(APP_KEY, APP_SECRET, ADZONE_ID)
# 定义搜索关键词（覆盖主要品类）
keywords = {
'美妆': ['口红', '眼影', '粉底液', '腮红'],
'护肤': ['面膜', '精华液', '防晒霜', '面霜', '洗面奶'],
'服饰': ['连衣裙', 'T恤', '牛仔裤', '卫衣'],
'数码': ['蓝牙耳机', '充电宝', '数据线', '手机壳'],
'食品': ['零食', '坚果', '饼干', '巧克力'],
'家居': ['收纳盒', '四件套', '毛巾', '枕头'],
'母婴': ['纸尿裤', '婴儿衣服', '玩具'],
'家电': ['电饭煲', '吹风机', '加湿器'],
}
# 执行选品
print("=" * 60)
print("🚀 AI自动化选品系统")
print(f"📅 日期：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 60)
all_products = []
for category, kws in keywords.items():
print(f"\n📦 正在选品：{category}")
for kw in kws:
print(f" 🔍 搜索：{kw}")
products = selector.select_products(
keywords=[kw],
min_rate=2000,
min_sales=1000,
min_score=60,
limit_per_keyword=10
)
all_products.extend(products)
# 去重
seen = set()
unique_products = []
for p in all_products:
if p.item_id not in seen:
seen.add(p.item_id)
unique_products.append(p)
# 按评分排序
unique_products.sort(key=lambda p: p.score, reverse=True)
# 取Top50
top50 = unique_products[:50]
# 生成报告
report = generate_report(top50)
# 保存报告
filename = f"选品报告_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
with open(filename, 'w', encoding='utf-8') as f:
f.write(report)
print(f"\n✅ 选品完成！")
print(f"📊 共筛选 {len(unique_products)} 个商品，Top50已保存到 {filename}")
# 打印Top10
print("\n🏆 今日Top10高佣商品：")
print("-" * 60)
for i, p in enumerate(top50[:10], 1):
final_price = p.price - p.coupon_amount
est_commission = final_price * p.commission_rate / 10000
print(f"{i:2d}. {p.title[:35]}")
print(f" 💰 ¥{p.price:.0f} → ¥{final_price:.0f} | 佣金 {p.commission_rate:.0f}% (¥{est_commission:.2f}) | 月销 {p.sales}")
print()
def generate_report(products):
"""生成Markdown报告"""
report = f"""# 📊 今日选品报告
> 生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
> 筛选条件：佣金率≥20% | 月销≥1000 | 综合评分≥60
---
## 🏆 Top50高佣商品
| 排名 | 商品 | 价格 | 券后价 | 佣金率 | 预估佣金 | 月销 | 评分 | 类目 |
|------|------|------|--------|--------|----------|------|------|------|
"""
for i, p in enumerate(products[:50], 1):
final_price = p.price - p.coupon_amount
est_commission = final_price * p.commission_rate / 10000
report += f"| {i} | {p.title[:25]}... | ¥{p.price:.0f} | ¥{final_price:.0f} | {p.commission_rate:.0f}% | ¥{est_commission:.2f} | {p.sales} | {p.score:.1f} | {p.category} |\n"
# 按类目统计
category_stats = {}
for p in products:
if p.category not in category_stats:
category_stats[p.category] = {
'count': 0,
'total_commission': 0,
'avg_score': 0
}
category_stats[p.category]['count'] += 1
category_stats[p.category]['total_commission'] += p.commission_rate
category_stats[p.category]['avg_score'] += p.score
for cat in category_stats:
count = category_stats[cat]['count']
category_stats[cat]['avg_commission'] = category_stats[cat]['total_commission'] / count
category_stats[cat]['avg_score'] = category_stats[cat]['avg_score'] / count
report += f"""
---
## 📈 类目统计
| 类目 | 商品数 | 平均佣金 | 平均评分 |
|------|--------|----------|----------|
"""
for cat, stats in sorted(category_stats.items(), key=lambda x: x[1]['avg_score'], reverse=True):
report += f"| {cat} | {stats['count']} | {stats['avg_commission']:.1f}% | {stats['avg_score']:.1f} |\n"
report += f"""
---
## 🎯 推荐推广商品（Top10详细）
"""
for i, p in enumerate(products[:10], 1):
final_price = p.price - p.coupon_amount
est_commission = final_price * p.commission_rate / 10000
report += f"""### {i}. {p.title[:50]}
| 指标 | 值 |
|------|-----|
| 商品ID | {p.item_id} |
| 原价 | ¥{p.price:.2f} |
| 优惠券 | ¥{p.coupon_amount:.2f} |
| 到手价 | ¥{final_price:.2f} |
| 佣金比例 | {p.commission_rate:.0f}% |
| 预估佣金/单 | ¥{est_commission:.2f} |
| 月销量 | {p.sales} |
| 综合评分 | {p.score:.1f} |
| 趋势得分 | {p.trend_score:.1f} |
| 店铺 | {p.shop_title} |
| 类目 | {p.category} |
---
"""
report += f"""
## 💡 选品建议
1. **优先推广评分80+的商品**：这些商品佣金高、销量好、转化率高
2. **关注趋势得分高的商品**：正在上升的商品，推广效果更好
3. **搭配优惠券使用**：有券的商品转化率更高
4. **多品类组合**：不要只推一个品类，分散风险
5. **定期更新选品**：每周至少更新一次选品列表
---
> 📌 本报告由AI自动化选品系统生成
> 🔄 建议每天运行一次，保持选品新鲜度
"""
return report
if __name__ == '__main__':
main()
6.2 运行命令
# 安装依赖
pip install requests
# 配置API信息
# 编辑脚本中的 APP_KEY, APP_SECRET, ADZONE_ID
# 运行选品
python3 run_selection.py
# 查看报告
cat 选品报告_*.md
七、Hermes Agent集成
7.1 创建Hermes Skill
# ~/.hermes/skills/product-selector/skill.yaml
name: product-selector
description: AI自动化选品工具
version: 1.0.0
triggers:
- pattern: "选品|选商品|找商品|高佣商品"
action: run_selection
commands:
run_selection:
description: 运行自动选品
script: |
cd /root/tools/courses/code
python3 run_selection.py
get_top10:
description: 获取今日Top10
script: |
cd /root/tools/courses/code
python3 -c "
from auto_product_selector import AutoProductSelector
selector = AutoProductSelector(APP_KEY, APP_SECRET, ADZONE_ID)
top10 = selector.get_daily_top50()[:10]
for i, p in enumerate(top10, 1):
print(f'{i}. {p[\"title\"][:40]}')
"
search:
description: 搜索商品
parameters:
keyword:
type: string
required: true
description: 搜索关键词
script: |
cd /root/tools/courses/code
python3 -c "
from auto_product_selector import AutoProductSelector
selector = AutoProductSelector(APP_KEY, APP_SECRET, ADZONE_ID)
products = selector.select_products(keywords=['$keyword'], limit_per_keyword=10)
for p in products[:10]:
print(f'{p.title} | ¥{p.price} | 佣金{p.commission_rate}%')
"
7.2 定时任务配置
# ~/.hermes/cron/product-selector.yaml
jobs:
daily_selection:
name: 每日自动选品
schedule: "0 8 * * *" # 每天早上8点
command: |
cd /root/tools/courses/code
python3 run_selection.py
notify: true
weekly_report:
name: 每周选品报告
schedule: "0 9 * * 1" # 每周一早上9点
command: |
cd /root/tools/courses/code
python3 weekly_report.py
notify: true
八、课后作业
作业1：理解选品逻辑
□ 理解6个核心指标的含义
□ 能手动计算商品评分
□ 理解不同品类的选品策略差异
作业2：实现选品系统
□ 搭建数据库（SQLite）
□ 实现商品评分器
□ 实现自动分类器
□ 实现趋势分析器
作业3：运行自动选品
□ 配置API信息
□ 运行选品脚本
□ 生成今日Top50报告
□ 分析报告结果
九、本课小结
✅ 掌握了选品的6个核心指标
✅ 实现了商品评分算法
✅ 学会了基于关键词的自动分类
✅ 实现了趋势分析和爆品预测
✅ 搭建了选品数据库
✅ 完成了自动选品脚本
✅ 学会了Hermes Agent集成
下一课预告：我们将学习如何通过多个渠道推广选好的商品，实现流量变现。
十、代码文件清单
本课涉及的代码文件：
product_scorer.py- 商品评分器category_classifier.py- 分类器trend_analyzer.py- 趋势分析器product_database.py- 数据库操作auto_product_selector.py- 自动选品系统run_selection.py- 选品主程序
建议保存到 /root/tools/courses/code/ 目录。
💡 实战提示：选品是淘客的核心竞争力。好的选品能让你事半功倍，差的选品再怎么推广也没用。坚持每天选品，积累数据，你会发现选品越来越准。
评论