# 1688 选品 + 淘宝比价 + 京东竞品监控，一条数据流跑起来（附 python 源码）
> 前面写了很多篇怎么把数据拿下来：签名、限流、稳定性、字段对齐。但拿下来之后呢？
>
> 数据本身不产生价值。**用三平台的数据回答三个问题**才产生价值：这个货最低多少钱能拿到（1688）、同款现在卖多少钱（淘宝）、对手在做什么动作（京东）。
>
> 这篇把三个问题串成一条流，从采集到输出一张「能不能做」的清单。中间最难的环节不是采集——是**同款识别**。三个平台对同一个商品的描述方式完全不同，认错了，后面全错。
## 一、先说清这条流要输出什么
技术方案要服务于一个具体产物。我们做这条流，最后只输出一张表，每天早上推到企业微信：
| 商品 | 1688 拿货价 | 起订量 | 淘宝价格带（P25/中位/P75） | 京东在售竞品 | 预估净利率 | 退货率红线 | 建议 |
| ------------ | -------- | ------ | ----------------- | ------ | ----- | ----- | -- |
| 不锈钢保温杯 500ml | 28.0 | 2 件代发 | 49 / 65 / 89 | 7 家 | 28.8% | 125% | 可做 |
| 硅胶折叠水壶 | 41.5 | 50 件起批 | 39 / 45 / 52 | 12 家 | -2.1% | — | 放弃 |
| 玻璃冷水壶 1.2L | 19.8 | 5 件代发 | 45 / 58 / 79 | 3 家 | 31.5% | 88% | 重点 |
这张表的四列数据来源不同、更新频率不同，但**必须在同一行上对齐**——这是同款识别要解决的问题。
三段的职责：
- **1688（货源）**：成本价、起订量、是否支持一件代发。日级更新就够，价格不常动。
- **淘宝（比价）**：不是找最低价，是找**价格带**。最低价往往是不能做的货（亏本引流、规格不同、二手翻新）。
- **京东（竞品）**：不是看有多少竞品，是看竞品**做了什么动作**——降价、上新、断货。
## 二、最难的环节：同款识别
同一个保温杯，三个平台可能这样写：
```
淘宝： 【爆款】316不锈钢保温杯男女士大容量水杯便携 500ml 送礼定制
京东： 富光保温杯 316不锈钢 500ml 白色 FGL-500
1688： 厂家直销 316不锈钢真空保温杯 500ml 定制LOGO 2件起批
```
纯标题匹配会把这当成三个完全不同的商品——关键词几乎没有一个完全重合。而实际上它们就是同一个货。
### 三级匹配：类目 → 标题 → 主图
我们的做法是三道筛子，逐级加严：
| 级别 | 手段 | 作用 | 成本 |
| -- | -------------------- | ------------ | ------ |
| 一级 | 叶子类目 + 价格量级 | 粗筛，砍掉 90% 噪音 | 极低 |
| 二级 | 标题归一 + 字符 2-gram 相似度 | 找候选 | 低 |
| 三级 | 主图 dHash 汉明距离 | 定案 | 需要下载图片 |
实测数据：**纯标题匹配的误判率约 40%，加上主图后降到 8%**。所以主图那一步不能省——图片是唯一跨平台稳定的特征，同一张图在三个平台都会出现（哪怕是供应商给的白底图）。
### 标题归一：去营销词，但必须保留规格词
这一步的规则是**反直觉的**：营销词要全部删掉（它们纯属噪音），但规格词必须留下并归一化——`500ml`、`0.5L`、`500毫升` 是同一个东西，而 `500ml` 和 `750ml` 是两个不同的货，价格能差 30%。
```python
# matching.py —— 标题归一：去噪音，保留并统一规格
import re
NOISE = ["包邮", "爆款", "新款", "正品", "厂家直销", "一件代发", "买一送一",
"限时", "特价", "热销", "网红", "同款", "批发", "定制", "送礼", "便携"]
UNIT = {"ml": ("ml", 1), "毫升": ("ml", 1), "l": ("ml", 1000), "升": ("ml", 1000),
"g": ("g", 1), "克": ("g", 1), "kg": ("g", 1000), "千克": ("g", 1000)}
SPEC_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(ml|毫升|l|升|g|克|kg|千克)")
def normalize_title(title: str) -> str:
t = title.lower()
for w in NOISE:
t = t.replace(w, "")
t = re.sub(r"[【】\[\]（）()《》\s\-_/|,，、!！?？+*~]+", " ", t)
def repl(m):
num = float(m.group(1))
unit, factor = UNIT[m.group(2)]
return f" {int(num * factor)}{unit} " # 0.5l / 500毫升 -> 500ml
return " ".join(SPEC_RE.sub(repl, t).split())
```
跑一下上面三个标题，归一后是这样：
```
不锈钢保温杯男女士大容量水杯 500ml
富光保温杯不锈钢 500ml白色 fgl-500
不锈钢真空保温杯 500ml logo
```
现在「保温杯」和「500ml」都浮出来了，2-gram 相似度就能算得比较准。
### 相似度：中文用字符 2-gram，不要用空格分词
中文商品标题没有空格分词，用 `split()` 分出来的往往是一整个长串。字符 2-gram 简单粗暴但效果好，而且零依赖：
```python
def bigrams(s: str) -> set[str]:
s = re.sub(r"\s+", "", s)
return {s[i:i + 2] for i in range(len(s) - 1)} if len(s) > 1 else {s}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def jaccard(a: str, b: str) -> float:
A, B = bigrams(a), bigrams(b)
return len(A & B) / len(A | B) if A | B else 0.0
```
### 主图 dHash：64 位指纹，汉明距离 ≤10 判相似
dHash 的思路是把图缩到 9×8 灰度，逐像素和右边邻居比较，得到一个 64 位整数。它只关心**明暗结构的梯度**，所以对缩放、轻微裁剪、压缩都不敏感——正好适合跨平台的主图。
```python
# image_hash.py
from PIL import Image
def dhash(path: str, size: int = 8) -> int:
img = Image.open(path).convert("L").resize((size + 1, size), Image.LANCZOS)
px = list(img.getdata())
bits = 0
for row in range(size):
for col in range(size):
i = row * (size + 1) + col
bits = (bits << 1) | (1 if px[i] > px[i + 1] else 0)
return bits
def hamming(a: int, b: int) -> int:
return bin(a ^ b).count("1")
```
`hamming ≤ 10` 判为相似（64 位里最多差 10 位）。这个阈值我们是拿 500 对人工标注样本调出来的，你可以按自己的品类重调——**服装类要放宽到 14**，因为模特姿势和背景差异更大。
### 组合打分：权重和阈值都要保守
```python
from decimal import Decimal
W = {"title": 0.55, "image": 0.35, "price": 0.10}
def price_plausible(cost: Decimal, retail: Decimal) -> float:
"""成本价与售价的倍数关系：正常 1.5~5 倍；<1 或 >8 基本是不同货或不同规格"""
if cost <= 0 or retail <= 0:
return 0.0
ratio = float(retail / cost)
if ratio < 1.0 or ratio > 8.0:
return 0.0
return 1.0
def match_score(title_sim: float, image_sim: float, price_score: float) -> float:
"""image_sim: 1 - hamming/64；无图时传 None 走保守降权"""
if image_sim is None:
# 没图不能直接当 0 分，按标题相似度打折，且总分封顶 0.7（进人工复核）
return min(title_sim * 0.9, 0.7)
return (W["title"] * title_sim + W["image"] * image_sim
+ W["price"] * price_score)
def is_same(score: float, threshold: float = 0.72) -> bool:
return score >= threshold
```
**阈值偏保守是有意为之。** 漏匹配的代价是少一个候选，人工还能补；误匹配的代价是按错的价格算利润，最后备货备错——后者贵得多。所以我们把 0.6~0.72 这段叫「灰区」，全部丢进人工复核队列，一天几十条，运营顺手就过了。
## 三、利润模型：把隐性成本算进去，否则算出来的都是幻觉
选品最容易犯的错是「售价减拿货价，毛利 40%，能做」。真实情况是：**毛利 36% 的大件，退货率 25% 的时候净利率只剩 10.7%**，再扣人工仓储就是负的。
四个必须进模型的隐性成本：
1. **平台佣金**（约 5%，各平台各品类不同）
2. **推广费**（新店起量期 10~20%，成熟期 5%）
3. **退货成本**——不只是退钱，是「发货运费 + 回程运费 + 包装折损 + 商品可能无法二次销售」
4. **资金占用**——批量拿货便宜 20%，但货压在仓里，两个月卖不掉就是负现金流
```python
# profit.py
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class CostModel:
price: Decimal # 售价
supply: Decimal # 1688 拿货价
ship_out: Decimal # 发货运费
ship_back: Decimal # 退货回程运费
pack_loss: Decimal = Decimal("3") # 退货包装/折损
commission_rate: Decimal = Decimal("0.05")
ad_rate: Decimal = Decimal("0.10")
return_rate: Decimal = Decimal("0.15")
def breakdown(self) -> dict:
gross = self.price - self.supply - self.ship_out
commission = self.price * self.commission_rate
ad = self.price * self.ad_rate
per_return = self.ship_out + self.ship_back + self.pack_loss
return_loss = self.return_rate * per_return
net = gross - commission - ad - return_loss
return {
"毛利": gross,
"佣金": commission,
"推广": ad,
"退货损失": return_loss,
"净利": net,
"净利率": net / self.price if self.price else Decimal("0"),
}
def breakeven_return_rate(self) -> Decimal:
"""
退货率红线：净利归零时的退货率。
选品时看这个数比看净利率有用——它告诉你「这个货能承受多差的售后」
"""
gross = self.price - self.supply - self.ship_out
after_fixed = gross - self.price * (self.commission_rate + self.ad_rate)
per_return = self.ship_out + self.ship_back + self.pack_loss
if per_return <= 0:
return Decimal("1")
return max(Decimal("0"), after_fixed / per_return)
```
拿三个真实的品算一遍，都是我们做过或评估过的（费率统一按佣金 5%、推广 10%）：
```
【保温杯】售价 65，拿货 28，发 6，回 8，折损 3，退货率 15%
毛利 31.00 (47.7%)
佣金+推广 9.75
退货损失 2.55 (15% × 17)
净利 18.70 (28.8%) 退货率红线 125% → 健康
【服装】售价 129，拿货 45，发 5，回 9，折损 8，退货率 40%
毛利 79.00 (61.2%)
佣金+推广 19.35
退货损失 8.80 (40% × 22)
净利 50.85 (39.4%) 退货率红线 271% → 扛得住
【大件家居】售价 599，拿货 320，发 60，回 120，折损 80，退货率 25%
毛利 219.00 (36.6%)
佣金+推广 89.85
退货损失 65.00 (25% × 260)
净利 64.15 (10.7%) 退货率红线 49.7% → 危险
```
第三个才是真正的杀手。**它单件退货成本 260 元，比毛利 219 元还高**——退一单等于白卖一单还倒贴。所以退货率红线只有 50%，而大件的行业退货率经常就在 30% 上下，稍有波动就到线。
注意这三个品的**毛利率排序和净利率排序是反的**：服装毛利 61% 但净利 39%，大件毛利 37% 净利只剩 11%。只看「毛利率」选品，会系统性地偏向高毛利品类，而高毛利品类往往正是退货成本最高的。
所以 `breakeven_return_rate()` 的真正价值不是「算红线」，而是**逼你把退货率填对**。我们内部的做法：新品按类目历史退货率的 1.2 倍填，换供应商时重算，大件类目额外单列「退货物流成本」字段——因为它的运费可能比货值还高。
## 四、竞品监控：监控「变化」，不是监控「数据」
这是另一个常见的浪费：写个脚本每天把竞品价格、销量、评价数存一遍，看板上堆满数字，然后没人看。
**人要的不是数据，是「发生了什么」。** 所以我们只输出五类事件：
| 事件 | 触发条件 | 业务含义 |
| ------ | ---------------------------- | ---------------- |
| 降价 | 幅度 ≥3% **且** 超过该商品 30 天波动 3σ | 对手在抢量，可能要跟 |
| 涨价 | 同上（反向） | 对手在提利润，是机会窗口 |
| SKU 减少 | SKU 集合出现消失项 | 对手断货或砍款，可能是我们的机会 |
| 断货 | `in_stock` 由 true 变 false | 立刻可抢排名 |
| 评价增速异常 | 日增评价数的 z-score > 3 | 对手在冲量（刷单或大促） |
关键在**阈值要用该商品自己的历史波动做基线**，不能拍脑袋定 3%：
```python
# monitor.py
from decimal import Decimal
def detect_events(prev: dict, cur: dict, sigma: Decimal) -> list[tuple[str, dict]]:
"""prev/cur: {"price", "in_stock", "sku_ids", "review_count"}"""
events = []
# 价格：幅度门槛 + 基线门槛，双条件才报，避免 99 -> 98.9 这种噪音
if prev["price"] and cur["price"]:
delta = (cur["price"] - prev["price"]) / prev["price"]
if abs(delta) >= Decimal("0.03") and abs(delta) >= 3 * sigma:
events.append(("PRICE_DROP" if delta < 0 else "PRICE_UP",
{"delta": round(float(delta), 4),
"from": prev["price"], "to": cur["price"]}))
if prev["in_stock"] and not cur["in_stock"]:
events.append(("OUT_OF_STOCK", {"price": cur["price"]}))
gone = set(prev["sku_ids"]) - set(cur["sku_ids"])
if gone:
events.append(("SKU_REMOVED", {"sku_ids": sorted(gone)}))
# 评价增速：用「日增评价数」的 z-score，不是评价总数
prev_v, cur_v = prev.get("review_velocity"), cur.get("review_velocity")
review_sigma = prev.get("review_sigma") # 该商品历史日增评价数的标准差
if prev_v is not None and cur_v is not None and review_sigma:
z = (cur_v - prev_v) / review_sigma
if z > 3:
events.append(("REVIEW_SURGE", {"z": round(float(z), 2),
"velocity": cur_v}))
return events
```
这个设计救过我们一次：最初阈值定的是「价格变动 >1%」，结果一个竞品每天微调 1 块钱（99→98→97.5），一天推 300 条告警，运营把通知静音了，三周后对手直接降到 79 没人知道。加了「3σ 基线 + 3% 幅度」双条件之后，一天最多 5 条，每条都值得看。
**监控清单是人定的，不是全类目扫。** 一个类目全扫是几万次调用，其中 99% 你根本不关心。我们的做法是运营每周挑 20~60 个 SKU 进监控池，只盯这些。
## 五、把三段串起来：两级池 + 分级频率
调用量是这条流最大的成本，靠「两级池」收敛：
```
全类目候选 20,000 ──粗筛（日级，1 次/天）──▶ 候选池 800
│
┌─────────┴─────────┐
淘宝比价（4 次/天） 京东竞品（48 次/天，仅精选 60）
│ │
└────▶ 决策表 + 事件推送
```
- **候选池 800 个**：1688 日扫一次（800）+ 淘宝 6 小时比一次（3,200）→ 约 4,000 次/天
- **精选池 60 个**：运营确认过的，京东 30 分钟一次 → 2,880 次/天
- 合计约 6,900 次/天；如果全量按 30 分钟扫 2 万个商品，是 96 万次/天
**调用量降到约 1/140，监控频率反而从 12 小时提升到 30 分钟。** 这就是第 13 篇那条铁律的又一次应用：按变更频率分级调度，只对值得高频的对象高频。
调度骨架可以直接复用第 13 篇的 `DueScheduler`，只需换一张频率表：
```python
LEVELS = {
"supply": {"interval_s": 86400, "pool": "candidate"}, # 1688 货源
"retail": {"interval_s": 21600, "pool": "candidate"}, # 淘宝比价
"rival": {"interval_s": 1800, "pool": "selected"}, # 京东竞品
}
```
## 六、三个结论
1. **同款识别是地基。** 认错了，比价是错的、利润是错的、备货也是错的。宁可用保守阈值 + 人工复核，也不要让误匹配混进决策表。
2. **利润要算到净利，退货率是隐藏杀手。** 填 5% 退货率的模型得出的结论，不如不填；`breakeven_return_rate()` 的用处是逼你把假设填对。
3. **监控要监控事件，不要监控数字。** 每天的 300 条噪音会让运营关掉通知，那比不监控更糟。
最后一句合规提醒：这套流用的都是平台授权接口返回的公开商品数据，**不采集买家信息、不抓取评价内容原文**。竞品监控是看公开的挂牌价格和库存状态，这和抓取用户数据是两件事，别混在一起做。
---
*如果不想自己维护这套同款识别和分级调度，我们的三平台接口服务已经把同款聚合、价格带统计、竞品事件都做成了标准输出，接进来自带上面这张决策表。评论区或私信聊。*
