256 lines
9.7 KiB
Python
256 lines
9.7 KiB
Python
"""AI 应收风险预警（纯统计引擎，不依赖 LLM）。
|
||
|
||
对标结论：竞品 AI 全在开单/问答，没人做应收风险——这是差异化卖点。
|
||
本模块只做统计评分（可离线、可测试、无 KEY 也能跑），LLM 仅作为话术增强（见 `advice.py`）。
|
||
|
||
三因子加权 → 0-100 风险分：
|
||
1. 回款周期漂移（权重 40）：客户近期实际回款天数 vs 历史均值，变慢得高分
|
||
2. 欠款趋势（权重 35）：未结余额环比上升幅度
|
||
3. 开单频次骤降（权重 25）：欠款在账但近期不再开单（可能已流失/绕单）
|
||
|
||
评分分档：high ≥70 / medium 40-69 / low <40
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal
|
||
from typing import Optional
|
||
|
||
from django.db.models import Sum
|
||
|
||
# 因子权重（和为 1.0）
|
||
WEIGHT_CYCLE_DRIFT = Decimal("0.40")
|
||
WEIGHT_OUTSTANDING_TREND = Decimal("0.35")
|
||
WEIGHT_ORDER_DROP = Decimal("0.25")
|
||
|
||
# 分档阈值
|
||
THRESHOLD_HIGH = Decimal("70")
|
||
THRESHOLD_MEDIUM = Decimal("40")
|
||
|
||
RECENT_WINDOW_DAYS = 90
|
||
BASELINE_WINDOW_DAYS = 365
|
||
|
||
|
||
def _d(value) -> Decimal:
|
||
return Decimal(str(value or 0))
|
||
|
||
|
||
def _payback_days_by_receivable(tenant, customer, *, since: Optional[date] = None) -> list:
|
||
"""返回该客户每张已核销应收的"从开单到收齐"天数列表。
|
||
|
||
用 Allocation（收款核销）里最晚一笔核销日期 - 应收单开单日期近似回款周期；
|
||
未结清的应收不参与（它们没有回款周期，但会进"欠款趋势"因子）。
|
||
"""
|
||
from apps.finance.models import Allocation
|
||
|
||
qs = Allocation.objects.filter(
|
||
tenant=tenant,
|
||
kind="receipt",
|
||
receivable__customer=customer,
|
||
receivable__isnull=False,
|
||
is_deleted=False,
|
||
).select_related("receipt", "receivable")
|
||
|
||
if since is not None:
|
||
qs = qs.filter(receivable__bill_date__gte=since)
|
||
|
||
latest = {}
|
||
for a in qs:
|
||
rid = a.receivable_id
|
||
rd = a.receipt.bill_date if a.receipt and a.receipt.bill_date else None
|
||
if rd is None:
|
||
continue
|
||
if rid not in latest or rd > latest[rid][1]:
|
||
latest[rid] = (a.receivable.bill_date, rd)
|
||
|
||
days = []
|
||
for bill_date, receipt_date in latest.values():
|
||
delta = (receipt_date - bill_date).days
|
||
if delta >= 0:
|
||
days.append(delta)
|
||
return days
|
||
|
||
|
||
def cycle_drift_score(tenant, customer, *, today: Optional[date] = None) -> dict:
|
||
"""回款周期漂移因子：近期平均回款天数 vs 历史基线。"""
|
||
today = today or date.today()
|
||
recent_start = today - timedelta(days=RECENT_WINDOW_DAYS)
|
||
baseline_start = today - timedelta(days=BASELINE_WINDOW_DAYS)
|
||
|
||
recent = _payback_days_by_receivable(tenant, customer, since=recent_start)
|
||
baseline_all = _payback_days_by_receivable(tenant, customer, since=baseline_start)
|
||
# 基线排除近期，避免自己比自己
|
||
baseline = [d for d in baseline_all if d not in recent] or baseline_all
|
||
|
||
if not recent or not baseline:
|
||
return {"score": Decimal("0"), "recent_days": None, "baseline_days": None,
|
||
"reason": "样本不足，回款周期因子不计分"}
|
||
|
||
recent_avg = Decimal(sum(recent)) / Decimal(len(recent))
|
||
base_avg = Decimal(sum(baseline)) / Decimal(len(baseline))
|
||
if base_avg <= 0:
|
||
return {"score": Decimal("0"), "recent_days": recent_avg,
|
||
"baseline_days": base_avg, "reason": "基线为 0，不计分"}
|
||
|
||
ratio = recent_avg / base_avg
|
||
# ratio ≤1 → 0 分；1.0→1.5 线性到 100 分；≥1.5 → 100
|
||
if ratio <= 1:
|
||
score = Decimal("0")
|
||
else:
|
||
score = min(Decimal("100"), (ratio - 1) / Decimal("0.5") * Decimal("100"))
|
||
return {
|
||
"score": score.quantize(Decimal("0.01")),
|
||
"recent_days": float(recent_avg),
|
||
"baseline_days": float(base_avg),
|
||
"reason": (f"近 {RECENT_WINDOW_DAYS} 天平均回款 {recent_avg:.0f} 天，"
|
||
f"历史均值 {base_avg:.0f} 天，"
|
||
f"{'变慢' if ratio > 1 else '未变慢'} {abs(ratio - 1) * 100:.0f}%"),
|
||
}
|
||
|
||
|
||
def outstanding_trend_score(tenant, customer, *, today: Optional[date] = None) -> dict:
|
||
"""欠款趋势因子：本月未结 vs 上月同口径，上升得高分。"""
|
||
from apps.finance.models import Receivable
|
||
|
||
today = today or date.today()
|
||
this_month_start = today.replace(day=1)
|
||
last_month_end = this_month_start - timedelta(days=1)
|
||
last_month_start = last_month_end.replace(day=1)
|
||
|
||
def _outstanding(as_of: date) -> Decimal:
|
||
agg = Receivable.objects.filter(
|
||
tenant=tenant, customer=customer,
|
||
status__in=["open", "partial"], is_deleted=False,
|
||
bill_date__lte=as_of,
|
||
).aggregate(amount=Sum("total_amount"), paid=Sum("paid_amount"))
|
||
return _d(agg["amount"]) - _d(agg["paid"])
|
||
|
||
now_val = _outstanding(today)
|
||
prev_val = _outstanding(last_month_end)
|
||
|
||
if prev_val <= 0:
|
||
# 上期无欠款、本期有 → 新增欠款，按金额档给分（不直接满分）
|
||
if now_val > 0:
|
||
return {"score": Decimal("50"), "current": float(now_val),
|
||
"previous": 0.0, "reason": f"上月无欠款，本月新增未结 ¥{now_val:.2f}"}
|
||
return {"score": Decimal("0"), "current": 0.0, "previous": 0.0,
|
||
"reason": "两期均无欠款"}
|
||
|
||
growth = (now_val - prev_val) / prev_val
|
||
if growth <= 0:
|
||
score = Decimal("0")
|
||
else:
|
||
# 增长 0→50% 线性到 100 分
|
||
score = min(Decimal("100"), growth / Decimal("0.5") * Decimal("100"))
|
||
return {
|
||
"score": score.quantize(Decimal("0.01")),
|
||
"current": float(now_val),
|
||
"previous": float(prev_val),
|
||
"reason": (f"未结余额 ¥{prev_val:.2f} → ¥{now_val:.2f}，"
|
||
f"{'上升' if growth > 0 else '下降'} {abs(growth) * 100:.0f}%"),
|
||
}
|
||
|
||
|
||
def order_drop_score(tenant, customer, *, today: Optional[date] = None) -> dict:
|
||
"""开单频次骤降因子：欠款在账但连续 N 天无新单。"""
|
||
from apps.sales.models import SalesBill
|
||
|
||
today = today or date.today()
|
||
last_bill = SalesBill.objects.filter(
|
||
tenant=tenant, customer=customer, is_deleted=False,
|
||
).exclude(state="cancelled").order_by("-bill_date").first()
|
||
|
||
from apps.partner.services import credit_usage
|
||
usage = credit_usage(tenant=tenant, customer=customer)
|
||
outstanding = _d(usage["outstanding"])
|
||
|
||
if outstanding <= 0:
|
||
return {"score": Decimal("0"), "days_since_last_bill": None,
|
||
"reason": "无欠款，频次因子不适用"}
|
||
|
||
if last_bill is None:
|
||
return {"score": Decimal("80"), "days_since_last_bill": None,
|
||
"reason": "有欠款但从未开单（历史导入或异常）"}
|
||
|
||
gap = (today - last_bill.bill_date).days
|
||
if gap <= 30:
|
||
score = Decimal("0")
|
||
elif gap >= 90:
|
||
score = Decimal("100")
|
||
else:
|
||
score = (Decimal(gap - 30) / Decimal("60") * Decimal("100"))
|
||
return {
|
||
"score": score.quantize(Decimal("0.01")),
|
||
"days_since_last_bill": gap,
|
||
"last_bill_no": last_bill.bill_no,
|
||
"reason": (f"已 {gap} 天未开新单，但仍有 ¥{outstanding:.2f} 欠款在账"
|
||
if gap > 30 else f"最近 {gap} 天内有开单，频次正常"),
|
||
}
|
||
|
||
|
||
def score_customer(tenant, customer, *, today: Optional[date] = None) -> dict:
|
||
"""单客户风险评分：三因子加权 → 0-100 + 分档 + 理由。"""
|
||
cycle = cycle_drift_score(tenant, customer, today=today)
|
||
trend = outstanding_trend_score(tenant, customer, today=today)
|
||
drop = order_drop_score(tenant, customer, today=today)
|
||
|
||
total = (
|
||
_d(cycle["score"]) * WEIGHT_CYCLE_DRIFT
|
||
+ _d(trend["score"]) * WEIGHT_OUTSTANDING_TREND
|
||
+ _d(drop["score"]) * WEIGHT_ORDER_DROP
|
||
).quantize(Decimal("0.01"))
|
||
|
||
if total >= THRESHOLD_HIGH:
|
||
level = "high"
|
||
elif total >= THRESHOLD_MEDIUM:
|
||
level = "medium"
|
||
else:
|
||
level = "low"
|
||
|
||
from apps.partner.services import credit_usage
|
||
usage = credit_usage(tenant=tenant, customer=customer)
|
||
|
||
return {
|
||
"customer_id": customer.id,
|
||
"customer_code": customer.code,
|
||
"customer_name": customer.name,
|
||
"score": float(total),
|
||
"level": level,
|
||
"outstanding": float(_d(usage["outstanding"])),
|
||
"limit": float(_d(usage["limit"])),
|
||
"factors": {
|
||
"cycle_drift": {**cycle, "score": float(cycle["score"]),
|
||
"weight": float(WEIGHT_CYCLE_DRIFT)},
|
||
"outstanding_trend": {**trend, "score": float(trend["score"]),
|
||
"weight": float(WEIGHT_OUTSTANDING_TREND)},
|
||
"order_drop": {**drop, "score": float(drop["score"]),
|
||
"weight": float(WEIGHT_ORDER_DROP)},
|
||
},
|
||
}
|
||
|
||
|
||
def risk_ranking(tenant, *, top_n: int = 5, min_outstanding: Decimal = Decimal("0"),
|
||
today: Optional[date] = None) -> list:
|
||
"""全租户客户风险排行（默认只看有欠款的客户），按分数降序取前 N。"""
|
||
from apps.finance.models import Receivable
|
||
from apps.partner.models import Customer
|
||
from django.db.models import Q
|
||
|
||
customer_ids = list(
|
||
Receivable.objects.filter(
|
||
tenant=tenant, status__in=["open", "partial"], is_deleted=False,
|
||
).values_list("customer_id", flat=True).distinct()
|
||
)
|
||
if not customer_ids:
|
||
return []
|
||
|
||
qs = Customer.objects.filter(tenant=tenant, id__in=customer_ids, is_deleted=False)
|
||
if min_outstanding > 0:
|
||
qs = qs.filter(Q(credit_limit__gt=0) | Q(id__in=customer_ids))
|
||
|
||
results = [score_customer(tenant, c, today=today) for c in qs]
|
||
results = [r for r in results if _d(r["outstanding"]) > min_outstanding]
|
||
results.sort(key=lambda r: r["score"], reverse=True)
|
||
return results[:top_n]
|