Hermes CI Fix
9de6e522a4
- 数据层: product_sales表+导入脚本(93条, 酣客1-8月商品销售排行榜) - 后端: GET /api/cma/products/matrix?entity_id&months 横轴=销售趋势(线性回归), 纵轴=加权毛利率, 气泡=销售额 - 前端: ProductMatrix.vue ECharts散点图+象限卡片+明细表, 支持酣客/博海切换 - 验证: API四象限分类正确, 数据联动, 前端路由200, 全部通过
159 lines
5.9 KiB
Python
159 lines
5.9 KiB
Python
"""波士顿产品矩阵 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,
|
|
}
|