consultoria-as
a386ca5b4a
- F0 blindaje: Redis/Celery, pg_hba sin trust LAN, secretos a env 600, backups, webhook auth, jobs muertos - F1 caja: bugs POS, reembolsos al corte, apartados, CFDI, topes de descuento en cotizacion/apartado - F2/F7 IA de catalogo: autocategorizacion, dedup de SKUs con embeddings, OCR de facturas, descripciones, claves SAT, limpieza de fitment - F3 IA en el flujo: vision Yonke, preguntas MeLi, venta cruzada, anomalias, sinonimos, busqueda semantica (rerank/embeddings) - F4/F8 escala y cartera: /metrics+Prometheus+Grafana, pool, score de credito, churn, metricas multi-tenant - Nexus Yonke (M1-M6): modelo de datos 1-de-1, alta por donador, vitrina publica, MeLi usado, consignacion, garantia por dias, reportes, etiquetas - Marketplace B2B entre yonkes (catalogo + pedidos cross-tenant) - Seguridad: auth en /api/admin/* (35 endpoints), alertas por tenant - Migraciones v4.27/v4.28; timers expire-demos y rollup-metrics
464 lines
15 KiB
Python
464 lines
15 KiB
Python
"""Credit risk scoring for store credit ("fiado") — A9.
|
|
|
|
Pure rule-based / heuristic scoring: no ML dependencies, works with very little
|
|
history and stays conservative when evidence is scarce. Every point of the score
|
|
is traceable to an explicit rule listed in ``reasons``.
|
|
|
|
Public API:
|
|
score_customer(tenant_conn, customer_id) -> dict
|
|
is_enforcement_enabled(tenant_conn) -> bool
|
|
|
|
The score is 0..100 where **higher is safer**. It is built from up to four
|
|
independent signals, each 0..100, then combined with fixed weights over the
|
|
signals that actually have data (missing signals are not invented):
|
|
|
|
tenure 0.15 months since customers.created_at
|
|
purchase 0.25 frequency + recency + value from sales
|
|
payment 0.35 punctuality on credit sales + open overdue aging
|
|
utilization 0.25 credit_balance / credit_limit
|
|
|
|
Credit sales are assumed to be due 30 days after the sale, matching the
|
|
receivables aging convention in ``blueprints/accounting_bp.py``.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
logger = logging.getLogger('pos')
|
|
|
|
# Credit sale terms used across the app for receivables aging.
|
|
CREDIT_TERMS_DAYS = 30
|
|
|
|
# Weight of each signal in the combined score.
|
|
_WEIGHTS = {
|
|
'tenure': 0.15,
|
|
'purchase': 0.25,
|
|
'payment': 0.35,
|
|
'utilization': 0.25,
|
|
}
|
|
|
|
_TIER_ORDER = {'risk': 0, 'medium': 1, 'good': 2}
|
|
|
|
# Enforcement flag key in tenant_config. Off by default.
|
|
ENFORCE_FLAG = 'credit_scoring_enforce'
|
|
|
|
|
|
def _now():
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def _aware(dt):
|
|
"""Normalize a possibly-naive datetime to UTC-aware."""
|
|
if dt is None:
|
|
return None
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
|
|
|
|
def _clamp(value, low=0, high=100):
|
|
return max(low, min(high, value))
|
|
|
|
|
|
def is_enforcement_enabled(tenant_conn):
|
|
"""Return True when the tenant opted into blocking credit sales for 'risk'.
|
|
|
|
Defaults to False on any error or missing flag so the sale flow never
|
|
breaks because of scoring.
|
|
"""
|
|
try:
|
|
cur = tenant_conn.cursor()
|
|
cur.execute("SELECT value FROM tenant_config WHERE key = %s", (ENFORCE_FLAG,))
|
|
row = cur.fetchone()
|
|
cur.close()
|
|
if not row:
|
|
return False
|
|
return str(row[0]).strip().lower() in ('true', '1', 'yes', 'on')
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
logger.warning("credit_scoring: could not read enforce flag: %s", exc)
|
|
return False
|
|
|
|
|
|
def _collect_evidence(cur, customer_id):
|
|
"""Read every raw signal in a few queries. Returns a dict or None if the
|
|
customer does not exist."""
|
|
cur.execute("""
|
|
SELECT id, name, created_at, credit_limit, credit_balance,
|
|
price_tier, is_active
|
|
FROM customers WHERE id = %s
|
|
""", (customer_id,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
|
|
ev = {
|
|
'customer_id': row[0],
|
|
'name': row[1],
|
|
'created_at': _aware(row[2]),
|
|
'credit_limit': float(row[3] or 0),
|
|
'credit_balance': float(row[4] or 0),
|
|
'price_tier': row[5],
|
|
'is_active': row[6],
|
|
}
|
|
|
|
# Completed sales aggregate (frequency / value / recency).
|
|
cur.execute("""
|
|
SELECT COUNT(*), COALESCE(SUM(total), 0), MIN(created_at),
|
|
MAX(created_at), COALESCE(AVG(total), 0),
|
|
COUNT(*) FILTER (WHERE sale_type = 'credit')
|
|
FROM sales
|
|
WHERE customer_id = %s AND status = 'completed'
|
|
""", (customer_id,))
|
|
total_orders, ltv, first_purchase, last_purchase, aov, credit_sales = cur.fetchone()
|
|
ev.update({
|
|
'total_orders': int(total_orders or 0),
|
|
'ltv': float(ltv or 0),
|
|
'first_purchase': _aware(first_purchase),
|
|
'last_purchase': _aware(last_purchase),
|
|
'avg_order_value': float(aov or 0),
|
|
'credit_sales': int(credit_sales or 0),
|
|
})
|
|
|
|
# Open (unpaid) credit sales: balance and due date for aging.
|
|
cur.execute("""
|
|
SELECT s.id, s.total, s.created_at,
|
|
s.total - COALESCE(SUM(sp.amount), 0) AS balance
|
|
FROM sales s
|
|
LEFT JOIN sale_payments sp ON sp.sale_id = s.id
|
|
WHERE s.customer_id = %s AND s.sale_type = 'credit' AND s.status = 'completed'
|
|
GROUP BY s.id, s.total, s.created_at
|
|
HAVING s.total - COALESCE(SUM(sp.amount), 0) > 0.005
|
|
ORDER BY s.created_at
|
|
""", (customer_id,))
|
|
|
|
now = _now()
|
|
open_sales = []
|
|
for sale_id, total, created_at, balance in cur.fetchall():
|
|
created_at = _aware(created_at)
|
|
due = created_at + timedelta(days=CREDIT_TERMS_DAYS) if created_at else None
|
|
overdue = max(0, (now - due).days) if due else 0
|
|
open_sales.append({
|
|
'sale_id': sale_id,
|
|
'total': float(total or 0),
|
|
'created_at': created_at,
|
|
'balance': float(balance or 0),
|
|
'due_date': due,
|
|
'days_overdue': overdue,
|
|
})
|
|
ev['open_sales'] = open_sales
|
|
ev['open_balance'] = round(sum(s['balance'] for s in open_sales), 2)
|
|
ev['max_overdue_days'] = max((s['days_overdue'] for s in open_sales), default=0)
|
|
ev['overdue_balance'] = round(
|
|
sum(s['balance'] for s in open_sales if s['days_overdue'] > 0), 2
|
|
)
|
|
|
|
# Aging buckets (by days overdue).
|
|
buckets = {'current': 0.0, 'd1_30': 0.0, 'd31_60': 0.0, 'd61_90': 0.0, 'd90_plus': 0.0}
|
|
for s in open_sales:
|
|
d = s['days_overdue']
|
|
if d == 0:
|
|
buckets['current'] += s['balance']
|
|
elif d <= 30:
|
|
buckets['d1_30'] += s['balance']
|
|
elif d <= 60:
|
|
buckets['d31_60'] += s['balance']
|
|
elif d <= 90:
|
|
buckets['d61_90'] += s['balance']
|
|
else:
|
|
buckets['d90_plus'] += s['balance']
|
|
ev['aging'] = {k: round(v, 2) for k, v in buckets.items()}
|
|
|
|
# Settled credit sales: was the final payment within terms?
|
|
cur.execute("""
|
|
SELECT s.created_at, MAX(sp.created_at) AS last_payment
|
|
FROM sales s
|
|
JOIN sale_payments sp ON sp.sale_id = s.id
|
|
WHERE s.customer_id = %s AND s.sale_type = 'credit' AND s.status = 'completed'
|
|
GROUP BY s.id, s.total, s.created_at
|
|
HAVING SUM(sp.amount) >= s.total - 0.005
|
|
""", (customer_id,))
|
|
settled = []
|
|
for created_at, last_payment in cur.fetchall():
|
|
created_at = _aware(created_at)
|
|
last_payment = _aware(last_payment)
|
|
if not created_at or not last_payment:
|
|
continue
|
|
due = created_at + timedelta(days=CREDIT_TERMS_DAYS)
|
|
settled.append({'days_late': max(0, (last_payment - due).days)})
|
|
ev['settled_credit_sales'] = settled
|
|
|
|
# Direct customer payments (abonos) — extra evidence of engagement.
|
|
cur.execute("""
|
|
SELECT COUNT(*), COALESCE(SUM(amount), 0), MAX(created_at)
|
|
FROM customer_payments WHERE customer_id = %s
|
|
""", (customer_id,))
|
|
cp_count, cp_total, cp_last = cur.fetchone()
|
|
ev.update({
|
|
'customer_payments_count': int(cp_count or 0),
|
|
'customer_payments_total': float(cp_total or 0),
|
|
'customer_payments_last': _aware(cp_last),
|
|
})
|
|
|
|
return ev
|
|
|
|
|
|
# ─── Signal scoring ─────────────────────────────────────
|
|
|
|
def _tenure_signal(ev, reasons):
|
|
created_at = ev['created_at']
|
|
if not created_at:
|
|
return None
|
|
days = max(0, (_now() - created_at).days)
|
|
months = days / 30.44
|
|
if months < 3:
|
|
score = 45
|
|
elif months < 6:
|
|
score = 60
|
|
elif months < 12:
|
|
score = 75
|
|
elif months < 24:
|
|
score = 85
|
|
else:
|
|
score = 95
|
|
reasons.append(f"Cliente con {int(months)} meses de antigüedad ({days} días).")
|
|
if ev.get('is_active') is False:
|
|
score = min(score, 40)
|
|
reasons.append("Cliente marcado como inactivo.")
|
|
return score
|
|
|
|
|
|
def _purchase_signal(ev, reasons):
|
|
total_orders = ev['total_orders']
|
|
if total_orders <= 0:
|
|
reasons.append("Sin historial de compras registrado.")
|
|
return None
|
|
|
|
if total_orders == 1:
|
|
freq = 50
|
|
elif total_orders <= 3:
|
|
freq = 65
|
|
elif total_orders <= 9:
|
|
freq = 80
|
|
elif total_orders <= 24:
|
|
freq = 90
|
|
else:
|
|
freq = 95
|
|
|
|
last = ev['last_purchase']
|
|
days_since = (_now() - last).days if last else None
|
|
if days_since is None:
|
|
recency = 60
|
|
elif days_since <= 30:
|
|
recency = 100
|
|
elif days_since <= 90:
|
|
recency = 80
|
|
elif days_since <= 180:
|
|
recency = 60
|
|
elif days_since <= 365:
|
|
recency = 40
|
|
else:
|
|
recency = 25
|
|
|
|
score = round(0.6 * freq + 0.4 * recency)
|
|
reasons.append(
|
|
f"{total_orders} compras por ${ev['ltv']:.2f} "
|
|
f"(ticket promedio ${ev['avg_order_value']:.2f}), "
|
|
f"última hace {days_since} días."
|
|
if days_since is not None else
|
|
f"{total_orders} compras por ${ev['ltv']:.2f}."
|
|
)
|
|
return _clamp(score)
|
|
|
|
|
|
def _payment_signal(ev, reasons):
|
|
has_history = (
|
|
ev['credit_sales'] > 0
|
|
or ev['open_balance'] > 0
|
|
or ev['customer_payments_count'] > 0
|
|
)
|
|
if not has_history:
|
|
reasons.append("Sin historial de crédito ni abonos.")
|
|
return None
|
|
|
|
settled = ev['settled_credit_sales']
|
|
if settled:
|
|
on_time = sum(1 for s in settled if s['days_late'] <= 0)
|
|
late_ratio = (len(settled) - on_time) / len(settled)
|
|
base = 90 - 60 * late_ratio
|
|
reasons.append(
|
|
f"{on_time}/{len(settled)} créditos liquidados dentro del plazo de "
|
|
f"{CREDIT_TERMS_DAYS} días."
|
|
)
|
|
else:
|
|
base = 70 if ev['credit_sales'] > 0 or ev['open_balance'] > 0 else 75
|
|
reasons.append("Sin créditos liquidados aún; sin historial de puntualidad.")
|
|
|
|
max_overdue = ev['max_overdue_days']
|
|
if max_overdue > 0:
|
|
penalty = min(55.0, 15.0 + max_overdue * 0.5)
|
|
base -= penalty
|
|
reasons.append(
|
|
f"Mora abierta: ${ev['overdue_balance']:.2f} con hasta {max_overdue} "
|
|
f"días de atraso."
|
|
)
|
|
|
|
if ev['credit_limit'] > 0:
|
|
util = ev['credit_balance'] / ev['credit_limit']
|
|
if util > 1.0:
|
|
base -= 15
|
|
reasons.append(f"Crédito excedido ({util * 100:.0f}% del límite).")
|
|
elif util > 0.9:
|
|
base -= 8
|
|
reasons.append(f"Crédito casi al límite ({util * 100:.0f}%).")
|
|
|
|
return _clamp(round(base))
|
|
|
|
|
|
def _utilization_signal(ev, reasons):
|
|
limit = ev['credit_limit']
|
|
balance = ev['credit_balance']
|
|
if limit <= 0:
|
|
if balance > 0:
|
|
reasons.append(f"Saldo de ${balance:.2f} sin límite de crédito definido.")
|
|
return 45
|
|
return None
|
|
|
|
util = balance / limit
|
|
if util <= 0.001:
|
|
score = 100
|
|
elif util <= 0.3:
|
|
score = 95
|
|
elif util <= 0.6:
|
|
score = 80
|
|
elif util <= 0.9:
|
|
score = 60
|
|
elif util <= 1.0:
|
|
score = 40
|
|
else:
|
|
score = 15
|
|
reasons.append(f"Utilización de crédito {util * 100:.0f}% (${balance:.2f} / ${limit:.2f}).")
|
|
return score
|
|
|
|
|
|
def _tier_from_score(score, data_points, max_overdue_days, behavioral_evidence):
|
|
if max_overdue_days > 60:
|
|
return 'risk'
|
|
if score >= 75 and data_points >= 2 and behavioral_evidence:
|
|
tier = 'good'
|
|
elif score >= 50:
|
|
tier = 'medium'
|
|
else:
|
|
tier = 'risk'
|
|
# Conservative when mora is building or evidence is thin.
|
|
if max_overdue_days > 30 and _TIER_ORDER[tier] > _TIER_ORDER['medium']:
|
|
tier = 'medium'
|
|
if not behavioral_evidence and _TIER_ORDER[tier] > _TIER_ORDER['medium']:
|
|
tier = 'medium'
|
|
if data_points <= 1 and _TIER_ORDER[tier] > _TIER_ORDER['medium']:
|
|
tier = 'medium'
|
|
return tier
|
|
|
|
|
|
def _recommended_limit(tier, ev):
|
|
if ev['total_orders'] <= 0:
|
|
return 0.0
|
|
aov = ev['avg_order_value'] or (ev['ltv'] / ev['total_orders'])
|
|
if tier == 'good':
|
|
factor = 4.0
|
|
elif tier == 'medium':
|
|
factor = 2.0
|
|
else:
|
|
return 0.0 # risk: do not recommend extending new credit
|
|
suggested = aov * factor
|
|
suggested = round(suggested / 100.0) * 100.0
|
|
return float(min(suggested, 200000.0))
|
|
|
|
|
|
def score_customer(tenant_conn, customer_id):
|
|
"""Score a customer's credit risk.
|
|
|
|
Returns a dict::
|
|
|
|
{
|
|
'score': 0..100, # higher = safer
|
|
'tier': 'good'|'medium'|'risk',
|
|
'recommended_limit': float,
|
|
'reasons': [str, ...],
|
|
'data_points': int, # number of signals with real data (0..4)
|
|
'components': {...}, # per-signal sub-scores (None when absent)
|
|
'metrics': {...}, # raw evidence for auditing
|
|
'computed_at': iso8601 str,
|
|
}
|
|
|
|
Raises ValueError if the customer does not exist.
|
|
"""
|
|
cur = tenant_conn.cursor()
|
|
try:
|
|
ev = _collect_evidence(cur, customer_id)
|
|
finally:
|
|
cur.close()
|
|
|
|
if ev is None:
|
|
raise ValueError(f"Customer {customer_id} not found")
|
|
|
|
reasons = []
|
|
components = {
|
|
'tenure': _tenure_signal(ev, reasons),
|
|
'purchase': _purchase_signal(ev, reasons),
|
|
'payment': _payment_signal(ev, reasons),
|
|
'utilization': _utilization_signal(ev, reasons),
|
|
}
|
|
|
|
available = {k: v for k, v in components.items() if v is not None}
|
|
data_points = len(available)
|
|
|
|
if available:
|
|
weight_sum = sum(_WEIGHTS[k] for k in available)
|
|
score = sum(available[k] * _WEIGHTS[k] for k in available) / weight_sum
|
|
else:
|
|
score = 50.0
|
|
score = int(round(_clamp(score)))
|
|
|
|
behavioral = components['purchase'] is not None or components['payment'] is not None
|
|
|
|
# Hard cap: heavy mora can never yield a high score.
|
|
if ev['max_overdue_days'] > 60:
|
|
score = min(score, 39)
|
|
elif ev['max_overdue_days'] > 30:
|
|
score = min(score, 60)
|
|
|
|
tier = _tier_from_score(score, data_points, ev['max_overdue_days'], behavioral)
|
|
recommended = _recommended_limit(tier, ev)
|
|
|
|
if data_points <= 1:
|
|
reasons.append(
|
|
f"Pocos datos ({data_points} señal{'es' if data_points != 1 else ''}); "
|
|
"evaluación conservadora."
|
|
)
|
|
|
|
return {
|
|
'customer_id': customer_id,
|
|
'score': score,
|
|
'tier': tier,
|
|
'recommended_limit': recommended,
|
|
'reasons': reasons,
|
|
'data_points': data_points,
|
|
'components': components,
|
|
'metrics': {
|
|
'months_active': round((_now() - ev['created_at']).days / 30.44, 1) if ev['created_at'] else None,
|
|
'total_orders': ev['total_orders'],
|
|
'ltv': round(ev['ltv'], 2),
|
|
'avg_order_value': round(ev['avg_order_value'], 2),
|
|
'days_since_last_purchase': (_now() - ev['last_purchase']).days if ev['last_purchase'] else None,
|
|
'credit_limit': ev['credit_limit'],
|
|
'credit_balance': ev['credit_balance'],
|
|
'open_credit_balance': ev['open_balance'],
|
|
'max_overdue_days': ev['max_overdue_days'],
|
|
'overdue_balance': ev['overdue_balance'],
|
|
'aging': ev['aging'],
|
|
'credit_sales': ev['credit_sales'],
|
|
'settled_credit_sales': len(ev['settled_credit_sales']),
|
|
},
|
|
'computed_at': _now().isoformat(),
|
|
}
|