276 lines
9.9 KiB
Python
276 lines
9.9 KiB
Python
import re
|
|
from decimal import Decimal
|
|
|
|
from django.db.models import Count, Q, Sum
|
|
|
|
from .models import Fact
|
|
|
|
METRICS = {
|
|
"revenue": "实际收入",
|
|
"cost": "成本",
|
|
"gross": "毛利贡献",
|
|
"roi": "ROI",
|
|
"customers": "消费人数",
|
|
"basket": "客单价",
|
|
"margin": "毛利率",
|
|
"discount_rate": "产品扣率",
|
|
"self_revenue": "自营收入",
|
|
"upgrade_rate": "升单率",
|
|
"list_basket": "原价客单价",
|
|
}
|
|
|
|
|
|
def dimensions(ds):
|
|
return [
|
|
{"key": c["key"], "name": c["name"], "description": c.get("description", "")}
|
|
for c in ds.mapping["columns"]
|
|
if c["role"] == "dimension"
|
|
]
|
|
|
|
|
|
def default_preferences(ds):
|
|
dims = [d["key"] for d in dimensions(ds)]
|
|
preferred = [
|
|
d
|
|
for d in ["department", "subject", "doctor_level", "doctor", "product", "product_subcategory"]
|
|
if d in dims
|
|
]
|
|
return {
|
|
"entry": "overview",
|
|
"comparison": "yoy",
|
|
"hierarchy": preferred or dims[:5],
|
|
"metric": "revenue",
|
|
"threshold": 0.15,
|
|
"min_impact": 1000,
|
|
"min_samples": 5,
|
|
"show_heatmap": True,
|
|
"show_queue": True,
|
|
"show_copilot": True,
|
|
"alerts": True,
|
|
}
|
|
|
|
|
|
def month_index(s):
|
|
if not isinstance(s, str) or not re.fullmatch(r"\d{4}-\d{2}", s):
|
|
raise ValueError("期间格式必须为 YYYY-MM")
|
|
year, month = map(int, s.split("-"))
|
|
if not 1900 <= year <= 2200 or not 1 <= month <= 12:
|
|
raise ValueError("期间无效")
|
|
return year * 12 + month - 1
|
|
|
|
|
|
def month_text(index):
|
|
year, month = divmod(index, 12)
|
|
return f"{year:04d}-{month + 1:02d}"
|
|
|
|
|
|
def context(ds, body):
|
|
if not isinstance(body, dict):
|
|
raise ValueError("查询格式错误")
|
|
end = body.get("end", ds.periods[-1])
|
|
start = body.get("start", end)
|
|
a, b = month_index(start), month_index(end)
|
|
if b < a or b - a > 59:
|
|
raise ValueError("期间范围必须为 1 至 60 个月")
|
|
mode = body.get("comparison", "yoy")
|
|
if mode not in ("yoy", "mom", "custom"):
|
|
raise ValueError("比较口径无效")
|
|
length = b - a + 1
|
|
base_end = (
|
|
month_index(body.get("baseline_end", ""))
|
|
if mode == "custom"
|
|
else b - (12 if mode == "yoy" else length)
|
|
)
|
|
current = [month_text(i) for i in range(a, b + 1)]
|
|
baseline = [month_text(i) for i in range(base_end - length + 1, base_end + 1)]
|
|
if set(current) & set(baseline):
|
|
raise ValueError("本期与基期不能重叠")
|
|
filters = body.get("filters", {})
|
|
allowed = {d["key"] for d in dimensions(ds)}
|
|
if not isinstance(filters, dict) or any(
|
|
k not in allowed or not isinstance(v, str) or len(v) > 256 for k, v in filters.items()
|
|
):
|
|
raise ValueError("筛选维度无效")
|
|
group = body.get("group", next(iter(dimensions(ds)))["key"])
|
|
if not isinstance(group, str) or group not in allowed:
|
|
raise ValueError("分组维度不存在")
|
|
metric = body.get("metric", "revenue")
|
|
if not isinstance(metric, str) or metric not in METRICS:
|
|
raise ValueError("指标不存在")
|
|
missing = sorted(set(current + baseline) - set(ds.periods))
|
|
return {
|
|
"start": start,
|
|
"end": end,
|
|
"comparison": mode,
|
|
"baseline_end": baseline[-1],
|
|
"current": current,
|
|
"baseline": baseline,
|
|
"missing": missing,
|
|
"filters": filters,
|
|
"group": group,
|
|
"metric": metric,
|
|
}
|
|
|
|
|
|
def scoped_facts(ds, member, filters=None):
|
|
qs = Fact.objects.filter(dataset=ds)
|
|
if member.departments:
|
|
qs = qs.filter(dimensions__department__in=member.departments)
|
|
for key, value in (filters or {}).items():
|
|
qs = qs.filter(**{f"dimensions__{key}": value})
|
|
return qs
|
|
|
|
|
|
def aggregations(products=None):
|
|
return {
|
|
"rows": Count("id"),
|
|
"revenue_sum": Sum("revenue"),
|
|
"cost_sum": Sum("cost"),
|
|
"gross_sum": Sum("gross"),
|
|
"cost_n": Count("cost"),
|
|
"gross_n": Count("gross"),
|
|
"customer_n": Count("customer"),
|
|
"customers": Count("customer", distinct=True),
|
|
"list_sum": Sum("list_amount"),
|
|
"list_n": Count("list_amount"),
|
|
"self_sum": Sum("revenue", filter=Q(dimensions__product__in=products or [])),
|
|
"upgraded": Count("customer", distinct=True, filter=Q(dimensions__product__in=products or [])),
|
|
}
|
|
|
|
|
|
def ratio(a, b):
|
|
return a / b if a is not None and b is not None and b > 0 else None
|
|
|
|
|
|
def metrics(agg, formula, products=None):
|
|
n = agg["rows"]
|
|
revenue = agg["revenue_sum"] or Decimal(0)
|
|
cost = (agg["cost_sum"] or Decimal(0)) if n and agg["cost_n"] == n else None
|
|
gross = (agg["gross_sum"] or Decimal(0)) if n and agg["gross_n"] == n else None
|
|
customers = agg["customers"] if n and agg["customer_n"] == n else None
|
|
profit = revenue - cost if cost is not None else None
|
|
if formula == "gross_less_cost_over_cost":
|
|
profit = gross - cost if gross is not None and cost is not None else None
|
|
return {
|
|
"rows": n,
|
|
"revenue": revenue,
|
|
"cost": cost,
|
|
"gross": gross,
|
|
"customers": customers,
|
|
"roi": ratio(profit, cost),
|
|
"basket": ratio(revenue, customers),
|
|
"margin": ratio(gross, revenue),
|
|
"discount_rate": ratio(revenue, agg["list_sum"]) if n and agg["list_n"] == n else None,
|
|
"self_revenue": (agg["self_sum"] or Decimal(0)) if products else None,
|
|
"upgrade_rate": ratio(Decimal(agg["upgraded"]), customers) if products else None,
|
|
"list_basket": ratio(agg["list_sum"], customers) if n and agg["list_n"] == n else None,
|
|
}
|
|
|
|
|
|
def comparison(current, baseline, comparable):
|
|
result = {}
|
|
for key in METRICS:
|
|
c, b = current.get(key), baseline.get(key)
|
|
delta = c - b if comparable and c is not None and b is not None else None
|
|
change = Decimal(delta) / abs(Decimal(b)) if delta is not None and b != 0 else None
|
|
result[key] = {"current": c, "baseline": b if comparable else None, "delta": delta, "change": change}
|
|
return result
|
|
|
|
|
|
def analyze(ds, member, body, preferences):
|
|
ctx = context(ds, body)
|
|
qs = scoped_facts(ds, member, ctx["filters"])
|
|
formula = ds.mapping["roi_formula"]
|
|
products = ds.mapping.get("self_operated_products", [])
|
|
total_c = metrics(
|
|
qs.filter(period__in=ctx["current"]).aggregate(**aggregations(products)), formula, products
|
|
)
|
|
total_b = metrics(
|
|
qs.filter(period__in=ctx["baseline"]).aggregate(**aggregations(products)), formula, products
|
|
)
|
|
comparable = not ctx["missing"]
|
|
if set(ctx["current"]) - set(ds.periods):
|
|
total_c = {**total_c, **{key: None for key in METRICS}}
|
|
summary = comparison(total_c, total_b, comparable)
|
|
group = "dimensions__" + ctx["group"]
|
|
buckets = {}
|
|
for side in ("current", "baseline"):
|
|
for row in qs.filter(period__in=ctx[side]).values(group).annotate(**aggregations(products)):
|
|
buckets.setdefault(row[group], {})[side] = metrics(row, formula, products)
|
|
|
|
def empty_side(total):
|
|
return {
|
|
"rows": 0,
|
|
**{
|
|
k: Decimal(0)
|
|
if k in ("revenue", "cost", "gross", "customers", "self_revenue") and total.get(k) is not None
|
|
else None
|
|
for k in METRICS
|
|
},
|
|
}
|
|
|
|
rows = []
|
|
metric = ctx["metric"]
|
|
parent_delta = summary[metric]["delta"]
|
|
threshold = Decimal(str(preferences["threshold"]))
|
|
impact_floor = Decimal(str(preferences["min_impact"]))
|
|
for label, sides in buckets.items():
|
|
c, b = sides.get("current", empty_side(total_c)), sides.get("baseline", empty_side(total_b))
|
|
if set(ctx["current"]) - set(ds.periods):
|
|
c = {**c, **{key: None for key in METRICS}}
|
|
values = comparison(c, b, comparable)
|
|
v = values[metric]
|
|
change, delta = v["change"], v["delta"]
|
|
impact = (
|
|
abs(delta)
|
|
if metric in ("revenue", "gross", "cost", "self_revenue") and delta is not None
|
|
else abs(values["revenue"]["delta"] or 0)
|
|
)
|
|
adverse = change is not None and (change >= threshold if metric == "cost" else change <= -threshold)
|
|
anomaly = bool(
|
|
adverse and impact >= impact_floor and min(c["rows"], b["rows"]) >= preferences["min_samples"]
|
|
)
|
|
contribution = (
|
|
delta / parent_delta
|
|
if delta is not None and parent_delta and metric in ("revenue", "cost", "gross", "self_revenue")
|
|
else None
|
|
)
|
|
rows.append(
|
|
{
|
|
"label": label or "（未分类）",
|
|
"values": values,
|
|
"rows": c["rows"],
|
|
"baseline_rows": b["rows"],
|
|
"anomaly": anomaly,
|
|
"impact": impact,
|
|
"contribution": contribution,
|
|
"filters": {**ctx["filters"], ctx["group"]: label or "（未分类）"},
|
|
}
|
|
)
|
|
rows.sort(key=lambda r: (not r["anomaly"], -r["impact"], r["label"]))
|
|
observed = {
|
|
r["period"]: metrics(r, formula, products)
|
|
for r in qs.values("period").annotate(**aggregations(products))
|
|
}
|
|
trend = []
|
|
for index in range(month_index(ds.periods[0]), month_index(ds.periods[-1]) + 1):
|
|
period = month_text(index)
|
|
values = observed.get(period)
|
|
if values is None:
|
|
values = empty_side(total_c) if period in ds.periods else {key: None for key in METRICS}
|
|
trend.append({"period": period, **values})
|
|
return {
|
|
"dataset_id": str(ds.pk),
|
|
"version": ds.sha256,
|
|
"context": ctx,
|
|
"summary": summary,
|
|
"rows": rows,
|
|
"trend": trend,
|
|
"row_count": total_c["rows"],
|
|
"anomaly_count": sum(r["anomaly"] for r in rows),
|
|
"comparable": comparable,
|
|
"rules": preferences,
|
|
"notice": "贡献分解为统计关联，不代表业务因果；退款和冲销保留在净额中。",
|
|
"roi_formula": formula,
|
|
}
|