""" =============================================================== analytics.py — Motor de Reportes y Análisis Predictivo =============================================================== Módulo independiente que se importa en main.py. FUNCIONALIDADES: 1. Simulación de registros históricos de recolección (en producción vendría de una tabla real en DB) 2. Detección de días con mayor volumen de residuos 3. Identificación de zonas críticas por colonia 4. Análisis predictivo simple con regresión lineal usando numpy (sin sklearn, más ligero) 5. Recomendaciones automáticas de logística INSTALAR: pip install numpy USO EN main.py: from analytics import generar_reporte_completo, predecir_proxima_semana =============================================================== """ import numpy as np import random from datetime import datetime, timedelta from typing import List, Dict, Any import hashlib # --------------------------------------------------------------- # SEMILLA DETERMINISTA # Los datos simulados son siempre los mismos para el mismo día, # lo que da consistencia en demos sin necesidad de una DB real. # --------------------------------------------------------------- def _semilla_dia(fecha: datetime) -> int: s = fecha.strftime("%Y-%m-%d") return int(hashlib.md5(s.encode()).hexdigest(), 16) % (2**32) # --------------------------------------------------------------- # DATOS DE REFERENCIA # --------------------------------------------------------------- COLONIAS = [ "Zona Centro", "Las Arboledas", "Trojes", "San Juanico", "Los Olivos", "Rancho Seco", "Las Insurgentes", ] RUTAS = { "Zona Centro": "RUTA-01", "Las Arboledas": "RUTA-01", "Trojes": "RUTA-13", "San Juanico": "RUTA-03", "Los Olivos": "RUTA-04", "Rancho Seco": "RUTA-05", "Las Insurgentes": "RUTA-12", } DIAS_SEMANA = ["Lunes", "Martes", "Miércoles", "Jueves", "Viernes", "Sábado", "Domingo"] # Factores base por colonia (densidad poblacional simulada) FACTOR_COLONIA = { "Zona Centro": 1.45, "Las Arboledas": 1.10, "Trojes": 0.85, "San Juanico": 0.90, "Los Olivos": 1.00, "Rancho Seco": 0.75, "Las Insurgentes": 1.20, } # Factores por día de la semana (lunes tras el finde = más basura) FACTOR_DIA = { 0: 1.35, # Lunes 1: 1.05, # Martes 2: 1.00, # Miércoles 3: 0.95, # Jueves 4: 1.15, # Viernes 5: 1.25, # Sábado 6: 0.80, # Domingo } # --------------------------------------------------------------- # GENERADOR DE HISTÓRICO SIMULADO # Genera 90 días de datos de recolección por colonia # --------------------------------------------------------------- def _generar_historico(dias: int = 90) -> List[Dict]: registros = [] hoy = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) for d in range(dias, 0, -1): fecha = hoy - timedelta(days=d) rng = random.Random(_semilla_dia(fecha)) dia_semana = fecha.weekday() for colonia in COLONIAS: # Volumen base en kg: entre 800 y 2500 kg por día base = rng.uniform(800, 2500) factor = FACTOR_COLONIA[colonia] * FACTOR_DIA[dia_semana] # Ruido aleatorio ±15% ruido = rng.uniform(0.85, 1.15) volumen = round(base * factor * ruido, 1) # Incidencias: reportes de basura tirada fuera de horario incidencias = rng.randint(0, int(factor * 5)) # Tiempo de recolección en minutos tiempo_min = round(30 + (volumen / 100) * rng.uniform(0.8, 1.2), 0) registros.append({ "fecha": fecha, "fecha_str": fecha.strftime("%Y-%m-%d"), "dia_semana": dia_semana, "dia_nombre": DIAS_SEMANA[dia_semana], "colonia": colonia, "ruta_id": RUTAS[colonia], "volumen_kg": volumen, "incidencias": incidencias, "tiempo_recoleccion_min": tiempo_min, }) return registros # --------------------------------------------------------------- # ANÁLISIS: Días con más residuos # --------------------------------------------------------------- def analisis_por_dia_semana(historico: List[Dict]) -> List[Dict]: """Agrupa el volumen promedio y total por día de la semana.""" acumulado = {i: {"total_kg": 0.0, "count": 0, "incidencias": 0} for i in range(7)} for r in historico: d = r["dia_semana"] acumulado[d]["total_kg"] += r["volumen_kg"] acumulado[d]["count"] += 1 acumulado[d]["incidencias"] += r["incidencias"] resultado = [] for dia_idx in range(7): v = acumulado[dia_idx] count = v["count"] if v["count"] > 0 else 1 resultado.append({ "dia": DIAS_SEMANA[dia_idx], "dia_idx": dia_idx, "promedio_kg": round(v["total_kg"] / count, 1), "total_kg": round(v["total_kg"], 1), "promedio_incidencias": round(v["incidencias"] / count, 1), "semanas_registradas": count // len(COLONIAS), }) # Ordenar por promedio descendente para ranking resultado.sort(key=lambda x: x["promedio_kg"], reverse=True) return resultado # --------------------------------------------------------------- # ANÁLISIS: Zonas críticas por colonia # --------------------------------------------------------------- def analisis_zonas_criticas(historico: List[Dict]) -> List[Dict]: """Identifica colonias con mayor volumen e incidencias.""" acumulado: Dict[str, Dict] = {} for r in historico: c = r["colonia"] if c not in acumulado: acumulado[c] = { "total_kg": 0.0, "total_incidencias": 0, "total_tiempo": 0.0, "count": 0, } acumulado[c]["total_kg"] += r["volumen_kg"] acumulado[c]["total_incidencias"] += r["incidencias"] acumulado[c]["total_tiempo"] += r["tiempo_recoleccion_min"] acumulado[c]["count"] += 1 resultado = [] max_vol = max(v["total_kg"] / v["count"] for v in acumulado.values()) for colonia, v in acumulado.items(): count = v["count"] if v["count"] > 0 else 1 promedio_kg = v["total_kg"] / count promedio_incidencias = v["total_incidencias"] / count indice_criticidad = round((promedio_kg / max_vol) * 100, 1) # Nivel de criticidad if indice_criticidad >= 80: nivel = "CRÍTICO" elif indice_criticidad >= 60: nivel = "ALTO" elif indice_criticidad >= 40: nivel = "MEDIO" else: nivel = "BAJO" resultado.append({ "colonia": colonia, "ruta_id": RUTAS[colonia], "promedio_kg_dia": round(promedio_kg, 1), "total_kg_90dias": round(v["total_kg"], 1), "promedio_incidencias_dia": round(promedio_incidencias, 2), "promedio_tiempo_min": round(v["total_tiempo"] / count, 1), "indice_criticidad": indice_criticidad, "nivel_criticidad": nivel, }) resultado.sort(key=lambda x: x["indice_criticidad"], reverse=True) return resultado # --------------------------------------------------------------- # ANÁLISIS PREDICTIVO: Regresión lineal simple con numpy # Predice el volumen de los próximos 7 días por colonia # --------------------------------------------------------------- def predecir_proxima_semana(historico: List[Dict]) -> List[Dict]: """ Regresión lineal por colonia sobre los últimos 30 días. Retorna predicción para los próximos 7 días. MÉTODO: - X = número de día (0..N) - Y = volumen_kg - Ajustamos y = mx + b con numpy.polyfit(deg=1) - Proyectamos para los días N+1 .. N+7 - Aplicamos el factor de día de la semana para ajustar estacionalidad """ hoy = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) predicciones = [] for colonia in COLONIAS: # Filtrar últimos 30 días de esta colonia datos_colonia = [ r for r in historico if r["colonia"] == colonia ][-30:] # últimos 30 registros if len(datos_colonia) < 7: continue x = np.array(range(len(datos_colonia)), dtype=float) y = np.array([r["volumen_kg"] for r in datos_colonia], dtype=float) # Regresión lineal coefs = np.polyfit(x, y, 1) # [pendiente, intercepto] pendiente = coefs[0] intercepto = coefs[1] # R² para medir calidad del ajuste y_pred = np.polyval(coefs, x) ss_res = np.sum((y - y_pred) ** 2) ss_tot = np.sum((y - np.mean(y)) ** 2) r2 = round(1 - (ss_res / ss_tot) if ss_tot > 0 else 0, 3) # Proyección próximos 7 días n = len(datos_colonia) dias_pred = [] for i in range(7): fecha_pred = hoy + timedelta(days=i + 1) dia_semana = fecha_pred.weekday() vol_base = np.polyval(coefs, n + i) # Ajustar por factor estacional del día vol_ajustado = max(0, vol_base * FACTOR_DIA[dia_semana]) dias_pred.append({ "fecha": fecha_pred.strftime("%Y-%m-%d"), "dia": DIAS_SEMANA[dia_semana], "volumen_predicho_kg": round(float(vol_ajustado), 1), "confianza": "Alta" if r2 > 0.7 else ("Media" if r2 > 0.4 else "Baja"), }) predicciones.append({ "colonia": colonia, "ruta_id": RUTAS[colonia], "tendencia": "CRECIENTE" if pendiente > 10 else ("DECRECIENTE" if pendiente < -10 else "ESTABLE"), "pendiente_diaria_kg": round(float(pendiente), 2), "r2": r2, "prediccion_7dias": dias_pred, "volumen_predicho_total_kg": round(sum(d["volumen_predicho_kg"] for d in dias_pred), 1), }) predicciones.sort(key=lambda x: x["volumen_predicho_total_kg"], reverse=True) return predicciones # --------------------------------------------------------------- # RECOMENDACIONES AUTOMÁTICAS DE LOGÍSTICA # --------------------------------------------------------------- def generar_recomendaciones( zonas: List[Dict], dias: List[Dict], predicciones: List[Dict], ) -> List[Dict]: """Genera recomendaciones concretas basadas en el análisis.""" recomendaciones = [] # 1. Zonas críticas → refuerzo de camiones criticas = [z for z in zonas if z["nivel_criticidad"] == "CRÍTICO"] for z in criticas: recomendaciones.append({ "tipo": "REFUERZO_RUTA", "prioridad": "ALTA", "emoji": "🚛", "titulo": f"Refuerzo en {z['colonia']}", "descripcion": ( f"{z['colonia']} genera en promedio {z['promedio_kg_dia']:.0f} kg/día " f"con {z['promedio_incidencias_dia']:.1f} incidencias. " f"Se recomienda asignar un segundo camión en días pico." ), "ruta_afectada": z["ruta_id"], }) # 2. Días pico → ajuste de horarios dia_pico = dias[0] if dias else None if dia_pico: recomendaciones.append({ "tipo": "HORARIO_PICO", "prioridad": "MEDIA", "emoji": "⏰", "titulo": f"{dia_pico['dia']} es el día con más residuos", "descripcion": ( f"El {dia_pico['dia']} se recolectan en promedio " f"{dia_pico['promedio_kg']:.0f} kg por colonia. " f"Considera iniciar rutas 30 minutos antes ese día." ), "ruta_afectada": None, }) # 3. Colonias con tendencia creciente → alerta temprana crecientes = [p for p in predicciones if p["tendencia"] == "CRECIENTE"] for p in crecientes[:2]: # máximo 2 alertas recomendaciones.append({ "tipo": "TENDENCIA_CRECIENTE", "prioridad": "MEDIA", "emoji": "📈", "titulo": f"Aumento en {p['colonia']}", "descripcion": ( f"{p['colonia']} muestra una tendencia creciente de " f"+{p['pendiente_diaria_kg']:.1f} kg/día. " f"Volumen predicho esta semana: {p['volumen_predicho_total_kg']:.0f} kg." ), "ruta_afectada": p["ruta_id"], }) # 4. Colonias con muchas incidencias → campaña de concientización alta_incidencia = sorted(zonas, key=lambda x: x["promedio_incidencias_dia"], reverse=True) if alta_incidencia: z = alta_incidencia[0] recomendaciones.append({ "tipo": "CONCIENTIZACION", "prioridad": "BAJA", "emoji": "📢", "titulo": f"Campaña en {z['colonia']}", "descripcion": ( f"{z['colonia']} registra {z['promedio_incidencias_dia']:.1f} incidencias/día " f"de basura fuera de horario. Se recomienda campaña de concientización ciudadana." ), "ruta_afectada": z["ruta_id"], }) return recomendaciones # --------------------------------------------------------------- # FUNCIÓN PRINCIPAL: Genera el reporte completo # --------------------------------------------------------------- def generar_reporte_completo(dias_historico: int = 90) -> Dict[str, Any]: """ Punto de entrada principal. Genera todos los análisis y los empaqueta. Cachea el resultado por 10 minutos para no recalcular en cada request. """ historico = _generar_historico(dias_historico) dias_semana = analisis_por_dia_semana(historico) zonas = analisis_zonas_criticas(historico) predicciones = predecir_proxima_semana(historico) recomendaciones = generar_recomendaciones(zonas, dias_semana, predicciones) # Resumen ejecutivo total_kg_periodo = sum(r["volumen_kg"] for r in historico) promedio_diario = total_kg_periodo / dias_historico if dias_historico > 0 else 0 return { "generado_en": datetime.now().isoformat(), "periodo_dias": dias_historico, "resumen": { "total_kg_recolectados": round(total_kg_periodo, 1), "promedio_kg_dia": round(promedio_diario, 1), "colonias_analizadas": len(COLONIAS), "zonas_criticas": len([z for z in zonas if z["nivel_criticidad"] == "CRÍTICO"]), "dia_pico": dias_semana[0]["dia"] if dias_semana else "N/A", "colonia_mayor_volumen": zonas[0]["colonia"] if zonas else "N/A", }, "dias_semana": dias_semana, "zonas_criticas": zonas, "prediccion_proxima_semana": predicciones, "recomendaciones": recomendaciones, }