Ejemplo: Predicción de Flujo de Caja con Python

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler
import joblib

class CashFlowPredictor:
    """Sistema de predicción de flujo de caja usando Random Forest"""
    
    def __init__(self):
        self.model = RandomForestRegressor(
            n_estimators=200,
            max_depth=12,
            min_samples_split=5,
            random_state=42
        )
        self.scaler = StandardScaler()
        self.feature_names = None
        
    def preparar_features(self, df):
        """Prepara features para el modelo predictivo"""
        features = pd.DataFrame({
            # Features temporales
            'mes': df['fecha'].dt.month,
            'trimestre': df['fecha'].dt.quarter,
            'dia_semana': df['fecha'].dt.dayofweek,
            'es_fin_mes': (df['fecha'].dt.day >= 25).astype(int),
            
            # Features de ingresos
            'ingresos_lag_1': df['ingresos'].shift(1),
            'ingresos_lag_7': df['ingresos'].shift(7),
            'ingresos_ma_7': df['ingresos'].rolling(7).mean(),
            'ingresos_ma_30': df['ingresos'].rolling(30).mean(),
            
            # Features de gastos
            'gastos_lag_1': df['gastos'].shift(1),
            'gastos_ma_7': df['gastos'].rolling(7).mean(),
            'gastos_volatilidad': df['gastos'].rolling(30).std(),
            
            # Features de cuentas por cobrar
            'cxc_ratio': df['cuentas_por_cobrar'] / df['ventas'],
            'dias_cobranza': df['cuentas_por_cobrar'] / (df['ventas'] / 30),
            
            # Features macroeconómicos
            'inflacion': df['tasa_inflacion'],
            'tipo_cambio': df['tipo_cambio'],
            'tasa_interes': df['tasa_interes_base'],
            
            # Features de industria
            'indice_industria': df['indice_sector'],
            'estacionalidad': np.sin(2 * np.pi * df['fecha'].dt.dayofyear / 365.25)
        })
        
        return features.fillna(method='forward').fillna(0)
    
    def entrenar_modelo(self, datos_historicos):
        """Entrena el modelo con datos históricos"""
        # Preparar features
        X = self.preparar_features(datos_historicos)
        y = datos_historicos['flujo_caja_neto']
        
        # Eliminar NaN
        mask = ~(X.isna().any(axis=1) | y.isna())
        X_clean = X[mask]
        y_clean = y[mask]
        
        # Escalar features
        X_scaled = self.scaler.fit_transform(X_clean)
        
        # Entrenar modelo
        self.model.fit(X_scaled, y_clean)
        self.feature_names = X_clean.columns.tolist()
        
        # Calcular métricas de evaluación
        score = self.model.score(X_scaled, y_clean)
        print(f"R² Score del modelo: {score:.3f}")
        
        return score
    
    def predecir_flujo_caja(self, datos_futuros, dias_adelante=30):
        """Predice flujo de caja para los próximos días"""
        predicciones = []
        datos_temp = datos_futuros.copy()
        
        for dia in range(dias_adelante):
            # Preparar features para el día actual
            X_day = self.preparar_features(datos_temp.tail(365))  # Usar último año
            X_scaled = self.scaler.transform(X_day.tail(1))
            
            # Predecir flujo de caja
            flujo_predicho = self.model.predict(X_scaled)[0]
            predicciones.append(flujo_predicho)
            
            # Actualizar datos para siguiente predicción
            fecha_siguiente = datos_temp['fecha'].iloc[-1] + pd.Timedelta(days=1)
            nueva_fila = pd.DataFrame({
                'fecha': [fecha_siguiente],
                'flujo_caja_neto': [flujo_predicho],
                # ... otros campos se actualizarían con lógica de negocio
            })
            datos_temp = pd.concat([datos_temp, nueva_fila], ignore_index=True)
        
        return predicciones
    
    def generar_alertas(self, predicciones, umbral_critico=-100000):
        """Genera alertas basadas en predicciones"""
        alertas = []
        saldo_acumulado = 0
        
        for i, flujo in enumerate(predicciones):
            saldo_acumulado += flujo
            
            if saldo_acumulado < umbral_critico:
                alertas.append({
                    'dia': i + 1,
                    'tipo': 'CRÍTICO',
                    'mensaje': f'Saldo proyectado: ${saldo_acumulado:,.0f}',
                    'recomendacion': 'Activar línea de crédito o diferir pagos'
                })
            elif flujo < 0 and abs(flujo) > abs(np.mean(predicciones)):
                alertas.append({
                    'dia': i + 1,
                    'tipo': 'ADVERTENCIA',
                    'mensaje': f'Flujo negativo alto: ${flujo:,.0f}',
                    'recomendacion': 'Revisar pagos programados'
                })
        
        return alertas
    
    def optimizar_inversiones(self, predicciones, tasas_disponibles):
        """Sugiere inversiones óptimas basadas en excedentes proyectados"""
        sugerencias = []
        
        for i, flujo in enumerate(predicciones):
            if flujo > 50000:  # Excedente significativo
                # Encontrar mejor tasa según plazo disponible
                dias_disponibles = len(predicciones) - i
                
                mejor_opcion = None
                mejor_rendimiento = 0
                
                for instrumento, detalles in tasas_disponibles.items():
                    if detalles['plazo_minimo'] <= dias_disponibles:
                        rendimiento = flujo * (detalles['tasa_anual'] / 365) * dias_disponibles
                        if rendimiento > mejor_rendimiento:
                            mejor_rendimiento = rendimiento
                            mejor_opcion = instrumento
                
                if mejor_opcion:
                    sugerencias.append({
                        'dia': i + 1,
                        'monto': flujo * 0.8,  # Conservar 20% como colchón
                        'instrumento': mejor_opcion,
                        'rendimiento_esperado': mejor_rendimiento * 0.8,
                        'plazo': dias_disponibles
                    })
        
        return sugerencias
    
    def generar_reporte_ejecutivo(self, predicciones, alertas, sugerencias):
        """Genera reporte ejecutivo automatizado"""
        flujo_promedio = np.mean(predicciones)
        flujo_total = sum(predicciones)
        dias_negativos = sum(1 for f in predicciones if f < 0)
        
        reporte = {
            'resumen_ejecutivo': {
                'flujo_promedio_diario': flujo_promedio,
                'flujo_total_periodo': flujo_total,
                'dias_con_flujo_negativo': dias_negativos,
                'porcentaje_dias_negativos': (dias_negativos / len(predicciones)) * 100
            },
            'alertas_criticas': len([a for a in alertas if a['tipo'] == 'CRÍTICO']),
            'oportunidades_inversion': len(sugerencias),
            'rendimiento_potencial': sum(s['rendimiento_esperado'] for s in sugerencias)
        }
        
        return reporte

# Ejemplo de uso
predictor = CashFlowPredictor()

# Datos de ejemplo (normalmente vendrían de base de datos)
# datos_historicos = pd.read_csv('cash_flow_history.csv')
# score = predictor.entrenar_modelo(datos_historicos)

# Predicción para próximos 30 días
# datos_actuales = pd.read_csv('current_financial_data.csv')
# predicciones = predictor.predecir_flujo_caja(datos_actuales, 30)
# alertas = predictor.generar_alertas(predicciones)

print("Sistema de predicción de flujo de caja inicializado")