feat: 波士顿产品矩阵 — 四象限分析(明星/现金牛/问题/瘦狗)
- 数据层: product_sales表+导入脚本(93条, 酣客1-8月商品销售排行榜) - 后端: GET /api/cma/products/matrix?entity_id&months 横轴=销售趋势(线性回归), 纵轴=加权毛利率, 气泡=销售额 - 前端: ProductMatrix.vue ECharts散点图+象限卡片+明细表, 支持酣客/博海切换 - 验证: API四象限分类正确, 数据联动, 前端路由200, 全部通过
This commit is contained in:
Hermes CI Fix
@@ -0,0 +1,158 @@
|
||||
"""波士顿产品矩阵 API — 四象限分析"""
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
from app.database import get_db
|
||||
from app.models import ProductSales
|
||||
from collections import defaultdict
|
||||
|
||||
router = APIRouter(prefix="/api/cma/products", tags=["产品矩阵"])
|
||||
|
||||
|
||||
def _calc_quadrant(trend: float, margin: float) -> str:
|
||||
"""四象限分类：
|
||||
横轴=近3月销售趋势（正=增长），纵轴=毛利率
|
||||
明星(Star) = 高趋势+高毛利
|
||||
现金牛(CashCow) = 低趋势+高毛利
|
||||
问题(QuestionMark) = 高趋势+低毛利
|
||||
瘦狗(Dog) = 低趋势+低毛利
|
||||
"""
|
||||
trend_high = trend >= 0
|
||||
margin_high = margin >= 0
|
||||
if trend_high and margin_high:
|
||||
return "star"
|
||||
if not trend_high and margin_high:
|
||||
return "cash_cow"
|
||||
if trend_high and not margin_high:
|
||||
return "question_mark"
|
||||
return "dog"
|
||||
|
||||
|
||||
@router.get("/matrix")
|
||||
def get_product_matrix(
|
||||
entity_id: int = Query(1, description="1=酣客 2=博海"),
|
||||
months: int = Query(3, ge=1, le=6, description="趋势计算月数"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""产品矩阵：横轴=销售趋势，纵轴=毛利率，气泡=销售额"""
|
||||
# 取最近 months+1 个月（多取1个月用于计算趋势）
|
||||
periods = db.query(ProductSales.period_month).filter(
|
||||
ProductSales.entity_id == entity_id
|
||||
).distinct().order_by(ProductSales.period_month.desc()).limit(months + 1).all()
|
||||
periods = sorted([p[0] for p in periods])
|
||||
|
||||
if len(periods) < 2:
|
||||
return {
|
||||
"entity_id": entity_id,
|
||||
"has_data": False,
|
||||
"message": "数据不足，至少需要2个月数据",
|
||||
"quadrants": [],
|
||||
"products": [],
|
||||
}
|
||||
|
||||
trend_periods = periods[-months:] # 最近 months 个月
|
||||
prev_periods = periods[:-months] if len(periods) > months else periods[:1]
|
||||
|
||||
# 加载数据
|
||||
rows = db.query(ProductSales).filter(
|
||||
ProductSales.entity_id == entity_id,
|
||||
ProductSales.period_month.in_(periods),
|
||||
).all()
|
||||
|
||||
# 按商品聚合
|
||||
products = defaultdict(lambda: {
|
||||
"code": "", "name": "", "months": {},
|
||||
"total_sales": 0, "total_qty": 0, "total_gross": 0,
|
||||
})
|
||||
for row in rows:
|
||||
p = products[row.product_code]
|
||||
p["code"] = row.product_code
|
||||
p["name"] = row.product_name
|
||||
p["months"][row.period_month] = {
|
||||
"sales": float(row.sales_amount or 0),
|
||||
"margin": float(row.gross_margin_rate or 0),
|
||||
"gross": float(row.gross_profit or 0),
|
||||
"qty": int(row.sales_qty or 0),
|
||||
}
|
||||
p["total_sales"] += float(row.sales_amount or 0)
|
||||
p["total_qty"] += int(row.sales_qty or 0)
|
||||
p["total_gross"] += float(row.gross_profit or 0)
|
||||
|
||||
# 计算每个商品的趋势和毛利率
|
||||
result_products = []
|
||||
for code, p in products.items():
|
||||
# 趋势 = 最近月份 vs 前一月的环比（取趋势期间的平均环比增速）
|
||||
# 用最近3个月的销售序列做简单线性趋势
|
||||
trend_sales = []
|
||||
for pp in periods:
|
||||
trend_sales.append(p["months"].get(pp, {}).get("sales", 0))
|
||||
# 线性回归斜率（最小二乘）
|
||||
n = len(trend_sales)
|
||||
if n >= 2:
|
||||
xs = list(range(n))
|
||||
x_mean = sum(xs) / n
|
||||
y_mean = sum(trend_sales) / n
|
||||
numerator = sum((xs[i] - x_mean) * (trend_sales[i] - y_mean) for i in range(n))
|
||||
denominator = sum((xs[i] - x_mean) ** 2 for i in range(n))
|
||||
slope = numerator / denominator if denominator else 0
|
||||
# 斜率转为百分比（相对期间平均销售）
|
||||
avg = y_mean if y_mean != 0 else 1
|
||||
trend = slope / abs(avg) * 100
|
||||
else:
|
||||
trend = 0.0
|
||||
|
||||
# 毛利率 = 加权平均（按销售额）
|
||||
weighted_margin = 0.0
|
||||
total_sales_for_margin = 0
|
||||
for pp in trend_periods:
|
||||
m = p["months"].get(pp)
|
||||
if m and m["sales"] > 0:
|
||||
weighted_margin += m["margin"] * m["sales"]
|
||||
total_sales_for_margin += m["sales"]
|
||||
if total_sales_for_margin > 0:
|
||||
weighted_margin = weighted_margin / total_sales_for_margin
|
||||
else:
|
||||
# 无销售用平均毛利率
|
||||
margins = [p["months"][pp]["margin"] for pp in p["months"] if p["months"][pp]["margin"] != 0]
|
||||
weighted_margin = sum(margins) / len(margins) if margins else 0
|
||||
|
||||
quadrant = _calc_quadrant(trend, weighted_margin)
|
||||
|
||||
result_products.append({
|
||||
"code": code,
|
||||
"name": p["name"],
|
||||
"total_sales": round(p["total_sales"], 2),
|
||||
"total_qty": p["total_qty"],
|
||||
"total_gross": round(p["total_gross"], 2),
|
||||
"trend_pct": round(trend, 1),
|
||||
"margin_pct": round(weighted_margin, 1),
|
||||
"quadrant": quadrant,
|
||||
})
|
||||
|
||||
# 按销售额排序
|
||||
result_products.sort(key=lambda x: -x["total_sales"])
|
||||
|
||||
# 四象限汇总
|
||||
quadrant_labels = {
|
||||
"star": {"label": "明星产品", "icon": "🌟", "advice": "高增长+有毛利，重点主推，加大投入"},
|
||||
"cash_cow": {"label": "现金牛", "icon": "🥇", "advice": "销量大但增长放缓，维持稳定产出"},
|
||||
"question_mark": {"label": "问题产品", "icon": "❓", "advice": "增长好但毛利低，优化成本或提价"},
|
||||
"dog": {"label": "瘦狗产品", "icon": "🐶", "advice": "低增长+低毛利，考虑清库存或停产"},
|
||||
}
|
||||
quadrants = []
|
||||
for q in ["star", "cash_cow", "question_mark", "dog"]:
|
||||
items = [p for p in result_products if p["quadrant"] == q]
|
||||
quadrants.append({
|
||||
"key": q,
|
||||
**quadrant_labels[q],
|
||||
"count": len(items),
|
||||
"products": items,
|
||||
})
|
||||
|
||||
return {
|
||||
"entity_id": entity_id,
|
||||
"has_data": True,
|
||||
"periods": periods,
|
||||
"months_analyzed": months,
|
||||
"quadrants": quadrants,
|
||||
"products": result_products,
|
||||
}
|
||||
Reference in New Issue
Block a user