sales-bot-stacks/sales-bot/analytics/predictions.py

"""
Sales prediction algorithms for Sales Bot
Uses moving average and linear regression for basic predictions
"""

import logging
from typing import List, Dict, Optional
from datetime import datetime, timedelta

logger = logging.getLogger(__name__)

# Try to import scipy for linear regression
try:
    from scipy import stats
    import numpy as np
    SCIPY_AVAILABLE = True
except ImportError:
    SCIPY_AVAILABLE = False
    logger.warning("scipy not available, using basic prediction methods")


def prediccion_basica(ventas_diarias: List[float], dias_prediccion: int = 7) -> Dict:
    """
    Genera predicciones combinando promedio móvil con tendencia lineal.

    Args:
        ventas_diarias: Lista de montos de ventas por día
        dias_prediccion: Número de días a predecir

    Returns:
        dict con predicciones, tendencia y confianza
    """
    if not ventas_diarias or len(ventas_diarias) < 3:
        return {
            'predicciones': [],
            'next_day': 0,
            'next_week': 0,
            'tendencia': 'stable',
            'confidence': 0,
            'error': 'Datos insuficientes para predicción'
        }

    try:
        # Calcular promedio móvil (últimos 7 días o menos si no hay suficientes)
        ventana = min(7, len(ventas_diarias))
        promedio_movil = sum(ventas_diarias[-ventana:]) / ventana

        # Calcular tendencia
        if SCIPY_AVAILABLE:
            # Usar regresión lineal con scipy
            x = np.arange(len(ventas_diarias))
            y = np.array(ventas_diarias)
            slope, intercept, r_value, p_value, std_err = stats.linregress(x, y)
            confianza = abs(r_value)
        else:
            # Método básico sin scipy
            slope, intercept, confianza = _calcular_regresion_simple(ventas_diarias)

        # Generar predicciones
        predicciones = []
        for i in range(1, dias_prediccion + 1):
            # Predicción lineal
            pred_lineal = slope * (len(ventas_diarias) + i) + intercept

            # Combinar promedio móvil con tendencia lineal
            pred_combinada = (promedio_movil + pred_lineal) / 2

            # No permitir predicciones negativas
            predicciones.append(max(0, pred_combinada))

        # Determinar tendencia
        if slope > 0.5:
            tendencia = 'increasing'
        elif slope < -0.5:
            tendencia = 'decreasing'
        else:
            tendencia = 'stable'

        return {
            'predicciones': predicciones,
            'next_day': round(predicciones[0], 2) if predicciones else 0,
            'next_week': round(sum(predicciones[:7]), 2) if len(predicciones) >= 7 else round(sum(predicciones), 2),
            'tendencia': tendencia,
            'trend': tendencia,  # Alias for frontend
            'confidence': round(confianza, 2),
            'promedio_movil': round(promedio_movil, 2),
            'slope': round(slope, 4)
        }

    except Exception as e:
        logger.error(f"Error en predicción: {str(e)}")
        return {
            'predicciones': [],
            'next_day': 0,
            'next_week': 0,
            'tendencia': 'stable',
            'trend': 'stable',
            'confidence': 0,
            'error': str(e)
        }


def _calcular_regresion_simple(datos: List[float]) -> tuple:
    """
    Calcula regresión lineal simple sin scipy.

    Returns:
        (slope, intercept, r_value)
    """
    n = len(datos)
    if n < 2:
        return 0, datos[0] if datos else 0, 0

    x = list(range(n))
    y = datos

    # Calcular medias
    x_mean = sum(x) / n
    y_mean = sum(y) / n

    # Calcular slope
    numerador = sum((x[i] - x_mean) * (y[i] - y_mean) for i in range(n))
    denominador = sum((x[i] - x_mean) ** 2 for i in range(n))

    if denominador == 0:
        return 0, y_mean, 0

    slope = numerador / denominador
    intercept = y_mean - slope * x_mean

    # Calcular R-squared (coeficiente de determinación)
    y_pred = [slope * xi + intercept for xi in x]
    ss_res = sum((y[i] - y_pred[i]) ** 2 for i in range(n))
    ss_tot = sum((y[i] - y_mean) ** 2 for i in range(n))

    r_squared = 1 - (ss_res / ss_tot) if ss_tot > 0 else 0
    r_value = r_squared ** 0.5 if r_squared >= 0 else 0

    return slope, intercept, r_value


def calcular_tendencia(ventas_diarias: List[float]) -> Dict:
    """
    Analiza la tendencia de los datos de ventas.

    Returns:
        dict con información de tendencia
    """
    if not ventas_diarias or len(ventas_diarias) < 2:
        return {
            'direccion': 'stable',
            'cambio_porcentual': 0,
            'volatilidad': 0
        }

    # Dividir en dos mitades
    mitad = len(ventas_diarias) // 2
    primera_mitad = ventas_diarias[:mitad] if mitad > 0 else ventas_diarias[:1]
    segunda_mitad = ventas_diarias[mitad:] if mitad > 0 else ventas_diarias[1:]

    promedio_primera = sum(primera_mitad) / len(primera_mitad) if primera_mitad else 0
    promedio_segunda = sum(segunda_mitad) / len(segunda_mitad) if segunda_mitad else 0

    # Calcular cambio porcentual
    if promedio_primera > 0:
        cambio_pct = ((promedio_segunda - promedio_primera) / promedio_primera) * 100
    else:
        cambio_pct = 0

    # Determinar dirección
    if cambio_pct > 5:
        direccion = 'increasing'
    elif cambio_pct < -5:
        direccion = 'decreasing'
    else:
        direccion = 'stable'

    # Calcular volatilidad (desviación estándar relativa)
    media = sum(ventas_diarias) / len(ventas_diarias)
    if media > 0:
        varianza = sum((x - media) ** 2 for x in ventas_diarias) / len(ventas_diarias)
        desviacion = varianza ** 0.5
        volatilidad = (desviacion / media) * 100
    else:
        volatilidad = 0

    return {
        'direccion': direccion,
        'cambio_porcentual': round(cambio_pct, 2),
        'volatilidad': round(volatilidad, 2),
        'promedio_anterior': round(promedio_primera, 2),
        'promedio_actual': round(promedio_segunda, 2)
    }


def generar_prediccion_extendida(
    ventas_diarias: List[float],
    dias_historico: int = 30,
    dias_futuro: int = 14
) -> Dict:
    """
    Genera predicción extendida con más detalles.

    Args:
        ventas_diarias: Datos históricos
        dias_historico: Días de historia a considerar
        dias_futuro: Días a predecir

    Returns:
        dict con predicción completa
    """
    # Limitar datos al histórico solicitado
    datos = ventas_diarias[-dias_historico:] if len(ventas_diarias) > dias_historico else ventas_diarias

    # Obtener predicción básica
    prediccion = prediccion_basica(datos, dias_futuro)

    # Agregar análisis de tendencia
    tendencia_info = calcular_tendencia(datos)

    # Calcular estadísticas adicionales
    if datos:
        maximo = max(datos)
        minimo = min(datos)
        promedio = sum(datos) / len(datos)
        mejor_dia_idx = datos.index(maximo)
    else:
        maximo = minimo = promedio = 0
        mejor_dia_idx = 0

    return {
        **prediccion,
        'tendencia_detalle': tendencia_info,
        'estadisticas': {
            'maximo': round(maximo, 2),
            'minimo': round(minimo, 2),
            'promedio': round(promedio, 2),
            'mejor_dia_indice': mejor_dia_idx,
            'dias_analizados': len(datos)
        }
    }