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
358 lines
15 KiB
Python
358 lines
15 KiB
Python
"""Business anomaly detection (A5): discounts, below-cost sales, cancellations
|
|
and off-hours activity.
|
|
|
|
Works on the tenant DB tables ``sale_items`` (discount_pct, unit_price,
|
|
unit_cost), ``sales`` (hour, employee, total, status) and ``audit_log``
|
|
(SALE / CANCEL). Everything is computed with a robust z-score (median + MAD)
|
|
so a handful of extreme tickets cannot drag the baseline.
|
|
|
|
Cold start is explicit: statistical rules only fire once there are at least
|
|
``min_samples`` observations in the window. The hard rules (a sale below its
|
|
recorded cost) fire regardless, because they do not need a baseline — but
|
|
``unit_cost`` of 0/NULL is treated as "unknown" and never flagged.
|
|
|
|
Public API:
|
|
detect_anomalies(conn, days=90, min_samples=5, z_threshold=3.5) -> dict
|
|
"""
|
|
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
import numpy as np
|
|
|
|
DEFAULT_WINDOW_DAYS = 90
|
|
DEFAULT_MIN_SAMPLES = 5
|
|
DEFAULT_Z = 3.5
|
|
|
|
# Local business hours (in BUSINESS_TZ). Sales outside are "off-hours".
|
|
BUSINESS_TZ = os.environ.get('ANOMALY_TZ', 'America/Mexico_City')
|
|
OFF_HOURS_START = int(os.environ.get('ANOMALY_OFF_HOURS_START', '7'))
|
|
OFF_HOURS_END = int(os.environ.get('ANOMALY_OFF_HOURS_END', '21'))
|
|
|
|
# Minimum repeated cancellations before we raise a flag (cold-start guard).
|
|
MIN_CANCELLATIONS = int(os.environ.get('ANOMALY_MIN_CANCELLATIONS', '3'))
|
|
CANCEL_RATE_THRESHOLD = float(os.environ.get('ANOMALY_CANCEL_RATE', '0.25'))
|
|
|
|
# A discount must be at least this high (percent points) to be worth flagging;
|
|
# keeps the detector from calling a 2% discount "anomalous" in a cash-register
|
|
# that normally discounts 0%.
|
|
DISCOUNT_ALERT_MIN = float(os.environ.get('ANOMALY_DISCOUNT_MIN', '15'))
|
|
|
|
# Off-hours needs at least this many tickets before it is reported, so a
|
|
# single late-night sale does not raise a flag.
|
|
OFF_HOURS_MIN = int(os.environ.get('ANOMALY_OFF_HOURS_MIN', '2'))
|
|
|
|
SEVERITY_POINTS = {'critical': 40, 'high': 25, 'medium': 15, 'low': 8}
|
|
|
|
|
|
def _robust_stats(values):
|
|
"""Return (median, scale) robust stats, or (None, None) for empty input.
|
|
|
|
Scale is the MAD when it is non-zero. When most values are identical
|
|
(MAD == 0 — very common with discounts, where the baseline is 0%), it
|
|
falls back to a scaled mean absolute deviation so a clear outlier is
|
|
still measurable instead of yielding a division by zero.
|
|
"""
|
|
arr = np.asarray(values, dtype=float)
|
|
if arr.size == 0:
|
|
return None, None
|
|
med = float(np.median(arr))
|
|
mad = float(np.median(np.abs(arr - med)))
|
|
if mad > 1e-9:
|
|
return med, mad
|
|
meanad = float(np.mean(np.abs(arr - med)))
|
|
if meanad > 1e-9:
|
|
return med, meanad * 1.2533
|
|
return med, 0.0
|
|
|
|
|
|
def _robust_z(value, med, scale):
|
|
"""Robust z-score (0.6745 * (x - med) / scale), None if scale == 0."""
|
|
if med is None or scale is None or scale <= 1e-9:
|
|
return None
|
|
return 0.6745 * (float(value) - med) / scale
|
|
|
|
|
|
def _is_off_hours(local_hour):
|
|
if local_hour is None:
|
|
return False
|
|
return local_hour < OFF_HOURS_START or local_hour > OFF_HOURS_END
|
|
|
|
|
|
def _anomaly(kind, severity, detail, metric=None, sample=None):
|
|
return {
|
|
'type': kind,
|
|
'severity': severity,
|
|
'detail': detail,
|
|
'metric': metric,
|
|
'sample_size': sample,
|
|
}
|
|
|
|
|
|
def _load_items(cur, days):
|
|
"""Sale items in the window, joined with their sale/employee/hour."""
|
|
cur.execute("""
|
|
SELECT si.sale_id, si.inventory_id, si.part_number, si.name,
|
|
si.quantity, si.unit_price, si.unit_cost, si.discount_pct,
|
|
s.employee_id, COALESCE(e.name, 'Empleado ' || s.employee_id),
|
|
EXTRACT(HOUR FROM (s.created_at AT TIME ZONE %s))::int AS local_hour
|
|
FROM sale_items si
|
|
JOIN sales s ON s.id = si.sale_id
|
|
LEFT JOIN employees e ON e.id = s.employee_id
|
|
WHERE s.created_at >= NOW() - make_interval(days => %s)
|
|
""", (BUSINESS_TZ, days))
|
|
return cur.fetchall()
|
|
|
|
|
|
def _load_cancellations(cur, days):
|
|
"""Cancelled-sale counts per employee (sales.status + audit_log CANCEL)."""
|
|
cur.execute("""
|
|
SELECT s.employee_id, COALESCE(e.name, 'Empleado ' || s.employee_id), COUNT(*)
|
|
FROM sales s
|
|
LEFT JOIN employees e ON e.id = s.employee_id
|
|
WHERE s.status = 'cancelled'
|
|
AND s.created_at >= NOW() - make_interval(days => %s)
|
|
GROUP BY 1, 2
|
|
""", (days,))
|
|
by_sales = {r[0]: (r[1], r[2]) for r in cur.fetchall()}
|
|
|
|
try:
|
|
cur.execute("""
|
|
SELECT employee_id, COUNT(*)
|
|
FROM audit_log
|
|
WHERE action ILIKE '%%CANCEL%%'
|
|
AND created_at >= NOW() - make_interval(days => %s)
|
|
GROUP BY 1
|
|
""", (days,))
|
|
for emp_id, cnt in cur.fetchall():
|
|
name, prev = by_sales.get(emp_id, (None, 0))
|
|
by_sales[emp_id] = (name or f'Empleado {emp_id}', max(prev, cnt))
|
|
except Exception:
|
|
cur.connection.rollback()
|
|
return by_sales
|
|
|
|
|
|
def _load_sale_counts(cur, days):
|
|
"""Total sales count per employee in the window (cancel-rate denominator)."""
|
|
cur.execute("""
|
|
SELECT employee_id, COUNT(*)
|
|
FROM sales
|
|
WHERE created_at >= NOW() - make_interval(days => %s)
|
|
GROUP BY 1
|
|
""", (days,))
|
|
return {r[0]: r[1] for r in cur.fetchall()}
|
|
|
|
|
|
def detect_anomalies(conn, days=DEFAULT_WINDOW_DAYS, min_samples=DEFAULT_MIN_SAMPLES,
|
|
z_threshold=DEFAULT_Z, limit=50):
|
|
"""Scan a tenant for business anomalies.
|
|
|
|
Returns a dict with per-employee and per-SKU findings, plus a summary and
|
|
a ``cold_start`` flag (True when there is not enough data for statistical
|
|
rules — only hard rules are reported then).
|
|
"""
|
|
days = max(1, int(days))
|
|
min_samples = max(1, int(min_samples))
|
|
cur = conn.cursor()
|
|
try:
|
|
items = _load_items(cur, days)
|
|
cancels = _load_cancellations(cur, days)
|
|
sale_counts = _load_sale_counts(cur, days)
|
|
finally:
|
|
cur.close()
|
|
|
|
total_items = len(items)
|
|
total_sales = len({r[0] for r in items})
|
|
employees_seen = {(r[8], r[9]) for r in items if r[8] is not None}
|
|
cold_start = total_items < min_samples
|
|
|
|
emp = {} # employee_id -> {name, anomalies: [], sales_count}
|
|
|
|
def add_emp(emp_id, name, anomaly):
|
|
if emp_id is None:
|
|
emp_id = 0
|
|
name = name or 'Sin empleado'
|
|
entry = emp.setdefault(emp_id, {'employee_id': emp_id, 'employee_name': name,
|
|
'anomalies': []})
|
|
entry['anomalies'].append(anomaly)
|
|
|
|
sku = {} # inventory_id -> {part_number, name, anomalies: [], tickets: n}
|
|
|
|
def add_sku(inv_id, part_number, name, anomaly):
|
|
if inv_id is None:
|
|
return
|
|
entry = sku.setdefault(inv_id, {'inventory_id': inv_id, 'part_number': part_number,
|
|
'name': name, 'anomalies': []})
|
|
entry['anomalies'].append(anomaly)
|
|
|
|
# ── 1. Sales below recorded cost (hard rule, no baseline needed) ──────
|
|
below_cost_emp = {}
|
|
for (sale_id, inv_id, part_number, name, qty, price, cost, disc,
|
|
emp_id, emp_name, hour) in items:
|
|
if cost is None or float(cost) <= 0:
|
|
continue # unknown cost — cannot judge
|
|
effective = float(price or 0) * (1 - float(disc or 0) / 100.0)
|
|
if effective < float(cost):
|
|
margin = effective - float(cost)
|
|
below_cost_emp[emp_id] = below_cost_emp.get(emp_id, 0) + 1
|
|
add_emp(emp_id, emp_name, _anomaly(
|
|
'below_cost', 'high',
|
|
f"Vendió {name} a {effective:.2f} por debajo del costo {float(cost):.2f}",
|
|
metric={'sale_id': sale_id, 'inventory_id': inv_id,
|
|
'unit_price': effective, 'unit_cost': float(cost),
|
|
'margin': round(margin, 2)},
|
|
))
|
|
add_sku(inv_id, part_number, name, _anomaly(
|
|
'below_cost', 'high',
|
|
f"Ticket #{sale_id} vendido bajo costo ({margin:.2f})",
|
|
metric={'sale_id': sale_id, 'margin': round(margin, 2)},
|
|
))
|
|
|
|
# ── 2. Discounts: robust z of the tenant-wide distribution ────────────
|
|
discounts = [float(r[7]) for r in items if r[7] is not None]
|
|
disc_med, disc_scale = _robust_stats(discounts)
|
|
if not cold_start:
|
|
for (sale_id, inv_id, part_number, name, qty, price, cost, disc,
|
|
emp_id, emp_name, hour) in items:
|
|
if disc is None or float(disc) < DISCOUNT_ALERT_MIN:
|
|
continue
|
|
z = _robust_z(float(disc), disc_med, disc_scale)
|
|
if z is not None and z > z_threshold:
|
|
add_emp(emp_id, emp_name, _anomaly(
|
|
'discount_atypical', 'medium',
|
|
f"Descuento {float(disc):.1f}% en {name} (z={z:.1f})",
|
|
metric={'sale_id': sale_id, 'discount_pct': float(disc),
|
|
'z': round(z, 2), 'baseline_median': round(disc_med, 2)},
|
|
sample=len(discounts),
|
|
))
|
|
add_sku(inv_id, part_number, name, _anomaly(
|
|
'discount_atypical', 'medium',
|
|
f"Descuento {float(disc):.1f}% (z={z:.1f})",
|
|
metric={'z': round(z, 2), 'discount_pct': float(disc)},
|
|
sample=len(discounts),
|
|
))
|
|
|
|
# Employee-level: each cajero's median discount vs the tenant baseline.
|
|
per_emp_disc = {}
|
|
for r in items:
|
|
if r[7] is not None and r[8] is not None:
|
|
per_emp_disc.setdefault(r[8], []).append(float(r[7]))
|
|
for emp_id, vals in per_emp_disc.items():
|
|
if len(vals) < min_samples:
|
|
continue
|
|
med = float(np.median(vals))
|
|
z = _robust_z(med, disc_med, disc_scale)
|
|
if med >= DISCOUNT_ALERT_MIN and z is not None and z > z_threshold:
|
|
add_emp(emp_id, dict(employees_seen).get(emp_id, f'Empleado {emp_id}'),
|
|
_anomaly(
|
|
'discount_employee', 'high',
|
|
f"Descuento mediano {med:.1f}% sobre la media del negocio ({disc_med:.1f}%)",
|
|
metric={'median_discount': round(med, 2), 'z': round(z, 2)},
|
|
sample=len(vals),
|
|
))
|
|
|
|
# SKU-level: each SKU's median discount vs the tenant baseline.
|
|
per_sku_disc = {}
|
|
for r in items:
|
|
if r[7] is not None and r[1] is not None:
|
|
per_sku_disc.setdefault(r[1], []).append(float(r[7]))
|
|
for inv_id, vals in per_sku_disc.items():
|
|
if len(vals) < min_samples:
|
|
continue
|
|
med = float(np.median(vals))
|
|
z = _robust_z(med, disc_med, disc_scale)
|
|
if med >= DISCOUNT_ALERT_MIN and z is not None and z > z_threshold:
|
|
part_number = next((r[2] for r in items if r[1] == inv_id), None)
|
|
name = next((r[3] for r in items if r[1] == inv_id), None)
|
|
add_sku(inv_id, part_number, name, _anomaly(
|
|
'discount_sku', 'medium',
|
|
f"Descuento mediano {med:.1f}% sobre la media ({disc_med:.1f}%)",
|
|
metric={'median_discount': round(med, 2), 'z': round(z, 2)},
|
|
sample=len(vals),
|
|
))
|
|
|
|
# ── 3. Repeated cancellations (employee) ──────────────────────────────
|
|
for emp_id, (emp_name, cnt) in cancels.items():
|
|
total = sale_counts.get(emp_id, cnt) + cnt
|
|
rate = cnt / total if total else 0
|
|
repeats = cnt >= MIN_CANCELLATIONS
|
|
proportional = total >= min_samples and rate >= CANCEL_RATE_THRESHOLD
|
|
if repeats or proportional:
|
|
add_emp(emp_id, emp_name, _anomaly(
|
|
'cancellations', 'high' if repeats else 'medium',
|
|
f"{cnt} cancelaciones en {days} días ({rate * 100:.0f}% de sus ventas)",
|
|
metric={'cancellations': cnt, 'sale_rate': round(rate, 3)},
|
|
sample=total,
|
|
))
|
|
|
|
# ── 4. Off-hours sales (rule; needs a repeat to avoid one-off noise) ──
|
|
off_hours = {}
|
|
for (sale_id, inv_id, part_number, name, qty, price, cost, disc,
|
|
emp_id, emp_name, hour) in items:
|
|
if _is_off_hours(hour):
|
|
off_hours.setdefault(emp_id, {'name': emp_name, 'sales': set()})
|
|
off_hours[emp_id]['sales'].add(sale_id)
|
|
for emp_id, info in off_hours.items():
|
|
count = len(info['sales'])
|
|
if count < OFF_HOURS_MIN and total_items >= min_samples:
|
|
continue
|
|
add_emp(emp_id, info['name'], _anomaly(
|
|
'off_hours', 'low' if count < 3 else 'medium',
|
|
f"{count} ventas fuera de horario ({OFF_HOURS_START}:00-{OFF_HOURS_END}:59)",
|
|
metric={'off_hours_sales': count},
|
|
sample=total_items,
|
|
))
|
|
|
|
# ── Score + serialise ─────────────────────────────────────────────────
|
|
def _score(anomalies):
|
|
total = 0.0
|
|
for a in anomalies:
|
|
total += SEVERITY_POINTS.get(a['severity'], 5)
|
|
return min(total, 100.0)
|
|
|
|
employees = []
|
|
for entry in emp.values():
|
|
entry['score'] = round(_score(entry['anomalies']), 1)
|
|
entry['sales_count'] = sale_counts.get(entry['employee_id'], 0)
|
|
employees.append(entry)
|
|
employees.sort(key=lambda e: e['score'], reverse=True)
|
|
|
|
skus = []
|
|
for entry in sku.values():
|
|
entry['score'] = round(_score(entry['anomalies']), 1)
|
|
skus.append(entry)
|
|
skus.sort(key=lambda s: s['score'], reverse=True)
|
|
|
|
by_type = {}
|
|
for entry in employees:
|
|
for a in entry['anomalies']:
|
|
by_type[a['type']] = by_type.get(a['type'], 0) + 1
|
|
for entry in skus:
|
|
for a in entry['anomalies']:
|
|
by_type[a['type']] = by_type.get(a['type'], 0) + 1
|
|
|
|
return {
|
|
'generated_at': datetime.now(timezone.utc).isoformat(),
|
|
'window_days': days,
|
|
'cold_start': cold_start,
|
|
'sample': {
|
|
'sale_items': total_items,
|
|
'sales': total_sales,
|
|
'employees': len(employees_seen),
|
|
},
|
|
'thresholds': {
|
|
'min_samples': min_samples,
|
|
'z_threshold': z_threshold,
|
|
'business_hours': [OFF_HOURS_START, OFF_HOURS_END],
|
|
'timezone': BUSINESS_TZ,
|
|
'min_cancellations': MIN_CANCELLATIONS,
|
|
},
|
|
'employees': employees[:limit],
|
|
'skus': skus[:limit],
|
|
'summary': {
|
|
'flagged_employees': len(employees),
|
|
'flagged_skus': len(skus),
|
|
'by_type': by_type,
|
|
},
|
|
}
|