app.py

#!/usr/bin/env python3
import json
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
from scipy import stats
from scipy.signal import find_peaks
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import DBSCAN
import gradio as gr

# Store global para mantener los datos analizados
analysis_store = {}

class TrendsAnalyzer:
    def __init__(self):
        self.scaler = StandardScaler()
        
    def detect_trends(self, data: pd.Series, window: int = 7) -> Dict[str, Any]:
        """Detecta tendencias en la serie temporal"""
        # Suavizado con media móvil
        smoothed = data.rolling(window=window, center=True).mean()
        
        # Calcular pendientes
        slopes = []
        for i in range(len(smoothed) - window):
            x = np.arange(window)
            y = smoothed.iloc[i:i+window].values
            if not np.isnan(y).all():
                slope, _, _, _, _ = stats.linregress(x, y)
                slopes.append(slope)
            else:
                slopes.append(0)
        
        # Clasificar tendencias
        trend_threshold = np.std(slopes) * 0.5
        trends = []
        for slope in slopes:
            if slope > trend_threshold:
                trends.append("creciente")
            elif slope < -trend_threshold:
                trends.append("decreciente")
            else:
                trends.append("estable")
        
        return {
            "slopes": slopes,
            "trends": trends,
            "smoothed_data": smoothed.tolist(),
            "trend_strength": np.std(slopes)
        }
    
    def detect_anomalies(self, data: pd.Series, method: str = "zscore") -> Dict[str, Any]:
        """Detecta anomalías en los datos"""
        if method == "zscore":
            z_scores = np.abs(stats.zscore(data.dropna()))
            anomalies = z_scores > 2.5
            anomaly_indices = data.index[anomalies].tolist()
            
        elif method == "iqr":
            Q1 = data.quantile(0.25)
            Q3 = data.quantile(0.75)
            IQR = Q3 - Q1
            lower_bound = Q1 - 1.5 * IQR
            upper_bound = Q3 + 1.5 * IQR
            anomalies = (data < lower_bound) | (data > upper_bound)
            anomaly_indices = data.index[anomalies].tolist()
            
        elif method == "isolation":
            # Usar DBSCAN como aproximación
            values = data.values.reshape(-1, 1)
            values_scaled = self.scaler.fit_transform(values)
            clustering = DBSCAN(eps=0.5, min_samples=5).fit(values_scaled)
            anomalies = clustering.labels_ == -1
            anomaly_indices = data.index[anomalies].tolist()
        
        return {
            "anomaly_indices": anomaly_indices,
            "anomaly_values": data.iloc[anomalies].tolist() if len(anomaly_indices) > 0 else [],
            "total_anomalies": len(anomaly_indices)
        }
    
    def detect_seasonality(self, data: pd.Series, periods: List[int] = [7, 30, 365]) -> Dict[str, Any]:
        """Detecta patrones estacionales"""
        seasonality_results = {}
        
        for period in periods:
            if len(data) >= period * 2:
                # Autocorelación para detectar periodicidad
                autocorr = []
                for lag in range(1, min(period + 1, len(data) // 2)):
                    corr = data.autocorr(lag=lag)
                    autocorr.append(corr if not np.isnan(corr) else 0)
                
                max_corr = max(autocorr) if autocorr else 0
                seasonality_results[f"period_{period}"] = {
                    "strength": max_corr,
                    "detected": max_corr > 0.3
                }
        
        return seasonality_results
    
    def find_peaks_valleys(self, data: pd.Series) -> Dict[str, Any]:
        """Encuentra picos y valles en la serie"""
        # Encontrar picos
        peaks, peak_properties = find_peaks(data.values, prominence=np.std(data) * 0.5)
        
        # Encontrar valles (picos invertidos)
        valleys, valley_properties = find_peaks(-data.values, prominence=np.std(data) * 0.5)
        
        return {
            "peaks": {
                "indices": data.index[peaks].tolist(),
                "values": data.iloc[peaks].tolist(),
                "count": len(peaks)
            },
            "valleys": {
                "indices": data.index[valleys].tolist(),
                "values": data.iloc[valleys].tolist(),
                "count": len(valleys)
            }
        }
    
    def generate_forecast(self, data: pd.Series, periods: int = 30) -> Dict[str, Any]:
        """Genera pronóstico simple usando tendencia lineal"""
        # Ajustar modelo lineal simple
        x = np.arange(len(data))
        y = data.values
        
        # Remover NaN
        mask = ~np.isnan(y)
        if mask.sum() < 2:
            return {"error": "Insuficientes datos para pronóstico"}
        
        slope, intercept, r_value, _, _ = stats.linregress(x[mask], y[mask])
        
        # Generar pronóstico
        future_x = np.arange(len(data), len(data) + periods)
        forecast = slope * future_x + intercept
        
        # Calcular intervalos de confianza (simplificados)
        residuals = y[mask] - (slope * x[mask] + intercept)
        mse = np.mean(residuals ** 2)
        std_error = np.sqrt(mse)
        
        return {
            "forecast_values": forecast.tolist(),
            "confidence_upper": (forecast + 1.96 * std_error).tolist(),
            "confidence_lower": (forecast - 1.96 * std_error).tolist(),
            "r_squared": r_value ** 2,
            "trend_slope": slope
        }

# Instancia global del analizador
analyzer = TrendsAnalyzer()

def create_visualizations(df: pd.DataFrame, column: str, analysis_results: Dict[str, Any]) -> go.Figure:
    """Crea las visualizaciones de la serie temporal"""
    fig = make_subplots(
        rows=3, cols=2,
        subplot_titles=(
            'Serie Temporal Original', 'Tendencias Detectadas',
            'Anomalías', 'Picos y Valles',
            'Pronóstico', 'Distribución de Valores'
        ),
        specs=[[{"secondary_y": False}, {"secondary_y": False}],
               [{"secondary_y": False}, {"secondary_y": False}],
               [{"secondary_y": False}, {"secondary_y": False}]]
    )
    
    # Serie original
    fig.add_trace(
        go.Scatter(x=df.index, y=df[column], name="Original", line=dict(color="blue")),
        row=1, col=1
    )
    
    # Tendencias
    if 'trends' in analysis_results:
        smoothed = analysis_results['trends']['smoothed_data']
        fig.add_trace(
            go.Scatter(x=df.index, y=smoothed, name="Suavizada", line=dict(color="orange")),
            row=1, col=2
        )
    
    # Anomalías
    if 'anomalies' in analysis_results and analysis_results['anomalies']['anomaly_indices']:
        anomaly_indices = analysis_results['anomalies']['anomaly_indices']
        anomaly_values = analysis_results['anomalies']['anomaly_values']
        
        fig.add_trace(
            go.Scatter(x=df.index, y=df[column], name="Serie", line=dict(color="blue")),
            row=2, col=1
        )
        fig.add_trace(
            go.Scatter(
                x=anomaly_indices, y=anomaly_values, 
                mode="markers", name="Anomalías",
                marker=dict(color="red", size=8)
            ),
            row=2, col=1
        )
    
    # Picos y valles
    if 'peaks_valleys' in analysis_results:
        fig.add_trace(
            go.Scatter(x=df.index, y=df[column], name="Serie", line=dict(color="blue")),
            row=2, col=2
        )
        
        peaks = analysis_results['peaks_valleys']['peaks']
        if peaks['indices']:
            fig.add_trace(
                go.Scatter(
                    x=peaks['indices'], y=peaks['values'],
                    mode="markers", name="Picos",
                    marker=dict(color="green", size=8, symbol="triangle-up")
                ),
                row=2, col=2
            )
        
        valleys = analysis_results['peaks_valleys']['valleys']
        if valleys['indices']:
            fig.add_trace(
                go.Scatter(
                    x=valleys['indices'], y=valleys['values'],
                    mode="markers", name="Valles",
                    marker=dict(color="red", size=8, symbol="triangle-down")
                ),
                row=2, col=2
            )
    
    # Pronóstico
    if 'forecast' in analysis_results and 'error' not in analysis_results['forecast']:
        forecast_data = analysis_results['forecast']
        last_date = df.index[-1]
        
        # Generar fechas futuras
        if isinstance(last_date, pd.Timestamp):
            future_dates = pd.date_range(
                start=last_date + pd.Timedelta(days=1),
                periods=len(forecast_data['forecast_values']),
                freq='D'
            )
        else:
            future_dates = range(len(df), len(df) + len(forecast_data['forecast_values']))
        
        # Serie histórica
        fig.add_trace(
            go.Scatter(x=df.index, y=df[column], name="Histórico", line=dict(color="blue")),
            row=3, col=1
        )
        
        # Pronóstico
        fig.add_trace(
            go.Scatter(
                x=future_dates, y=forecast_data['forecast_values'],
                name="Pronóstico", line=dict(color="red", dash="dash")
            ),
            row=3, col=1
        )
        
        # Intervalos de confianza
        fig.add_trace(
            go.Scatter(
                x=future_dates, y=forecast_data['confidence_upper'],
                fill=None, mode='lines', line=dict(color='rgba(0,0,0,0)'),
                showlegend=False
            ),
            row=3, col=1
        )
        fig.add_trace(
            go.Scatter(
                x=future_dates, y=forecast_data['confidence_lower'],
                fill='tonexty', mode='lines', line=dict(color='rgba(0,0,0,0)'),
                name='Intervalo de Confianza', fillcolor='rgba(255,0,0,0.2)'
            ),
            row=3, col=1
        )
    
    # Distribución
    fig.add_trace(
        go.Histogram(x=df[column], name="Distribución", nbinsx=30),
        row=3, col=2
    )
    
    fig.update_layout(height=1200, showlegend=True, title_text="Análisis de Tendencias Temporales")
    return fig

def analyze_time_series(file, column_name, trend_window, anomaly_method, forecast_periods):
    """ Analiza una serie temporal para detectar tendencias, anomalías, estacionalidad y picos/valles."""
    try:
        # Leer archivo
        if file.name.endswith('.csv'):
            df = pd.read_csv(file.name)
        elif file.name.endswith(('.xlsx', '.xls')):
            df = pd.read_excel(file.name)
        else:
            return "Error: Formato de archivo no soportado", None, None
        
        # Validar columna
        if column_name not in df.columns:
            return f"Error: Columna '{column_name}' no encontrada", None, None
        
        # Intentar convertir índice a datetime si es posible
        if 'fecha' in df.columns or 'date' in df.columns:
            date_col = 'fecha' if 'fecha' in df.columns else 'date'
            df[date_col] = pd.to_datetime(df[date_col])
            df.set_index(date_col, inplace=True)
        
        # Preparar serie
        series = df[column_name].astype(float)
        
        # Realizar análisis
        results = {}
        
        # Análisis de tendencias
        results['trends'] = analyzer.detect_trends(series, window=trend_window)
        
        # Detección de anomalías
        results['anomalies'] = analyzer.detect_anomalies(series, method=anomaly_method)
        
        # Detección de estacionalidad
        results['seasonality'] = analyzer.detect_seasonality(series)
        
        # Picos y valles
        results['peaks_valleys'] = analyzer.find_peaks_valleys(series)
        
        # Pronóstico
        results['forecast'] = analyzer.generate_forecast(series, periods=forecast_periods)
        
        # Guardar resultados
        analysis_id = f"analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        analysis_store[analysis_id] = {
            'data': df,
            'column': column_name,
            'results': results
        }
        
        # Crear visualización
        fig = create_visualizations(df, column_name, results)
        
        # Generar reporte
        report = generate_report(results, series)
        
        return report, fig, analysis_id
        
    except Exception as e:
        return f"Error en el análisis: {str(e)}", None, None

def generate_report(results: Dict[str, Any], series: pd.Series) -> str:
    """Genera un reporte textual del análisis"""
    report = "# Reporte de Análisis de Tendencias Temporales\n\n"
    
    # Estadísticas básicas
    report += "## Estadísticas Básicas\n"
    report += f"- Número de observaciones: {len(series)}\n"
    report += f"- Media: {series.mean():.2f}\n"
    report += f"- Desviación estándar: {series.std():.2f}\n"
    report += f"- Mínimo: {series.min():.2f}\n"
    report += f"- Máximo: {series.max():.2f}\n\n"
    
    # Tendencias
    if 'trends' in results:
        trend_strength = results['trends']['trend_strength']
        report += "## Análisis de Tendencias\n"
        report += f"- Fuerza de la tendencia: {trend_strength:.4f}\n"
        if trend_strength > 0.1:
            report += "- La serie presenta tendencias significativas\n\n"
        else:
            report += "- La serie presenta tendencias débiles o estables\n\n"
    
    # Anomalías
    if 'anomalies' in results:
        anomaly_count = results['anomalies']['total_anomalies']
        report += "## Detección de Anomalías\n"
        report += f"- Anomalías detectadas: {anomaly_count}\n"
        report += f"- Porcentaje de anomalías: {(anomaly_count/len(series)*100):.1f}%\n\n"
    
    # Estacionalidad
    if 'seasonality' in results:
        report += "## Análisis de Estacionalidad\n"
        for period, data in results['seasonality'].items():
            period_name = period.replace('period_', '')
            if data['detected']:
                report += f"- Patrón estacional de {period_name} períodos detectado (fuerza: {data['strength']:.3f})\n"
        report += "\n"
    
    # Picos y valles
    if 'peaks_valleys' in results:
        peaks_count = results['peaks_valleys']['peaks']['count']
        valleys_count = results['peaks_valleys']['valleys']['count']
        report += "## Picos y Valles\n"
        report += f"- Picos detectados: {peaks_count}\n"
        report += f"- Valles detectados: {valleys_count}\n\n"
    
    # Pronóstico
    if 'forecast' in results and 'error' not in results['forecast']:
        r_squared = results['forecast']['r_squared']
        slope = results['forecast']['trend_slope']
        report += "## Pronóstico\n"
        report += f"- R² del modelo: {r_squared:.3f}\n"
        report += f"- Pendiente de tendencia: {slope:.4f}\n"
        if slope > 0:
            report += "- Tendencia proyectada: Creciente\n"
        elif slope < 0:
            report += "- Tendencia proyectada: Decreciente\n"
        else:
            report += "- Tendencia proyectada: Estable\n"
    
    return report

# Funciones MCP para Gradio 5
def mcp_analyze_trends(data: List[float], dates: Optional[List[str]] = None, 
                      window: int = 7, anomaly_method: str = "zscore") -> str:
    """ Analiza una serie temporal para detectar tendencias, anomalías, estacionalidad y picos/valles.

    Parámetros:
    - data (List[float]): Serie de valores numéricos a analizar.
    - dates (List[str], opcional): Fechas correspondientes a los datos. Si se omite, se usará índice por posición.
    - window (int): Ventana de suavizado para detección de tendencias.
    - anomaly_method (str): Método de detección de anomalías. Opciones: "zscore", "iqr", "isolation".

    Retorna:
    - str: JSON con resumen del análisis, resultados completos y un ID único."""
    
    def make_json_serializable(obj):
        """Convierte objetos no serializables a JSON"""
        if isinstance(obj, np.integer):
            return int(obj)
        elif isinstance(obj, np.floating):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, pd.Timestamp):
            return obj.isoformat()
        elif isinstance(obj, pd.Index):
            return obj.tolist()
        elif isinstance(obj, dict):
            return {key: make_json_serializable(value) for key, value in obj.items()}
        elif isinstance(obj, list):
            return [make_json_serializable(item) for item in obj]
        elif isinstance(obj, (bool, np.bool_)):
            return bool(obj)
        elif pd.isna(obj):
            return None
        else:
            return obj
    
    try:
        # Crear serie pandas
        if dates:
            index = pd.to_datetime(dates)
            series = pd.Series(data, index=index)
        else:
            series = pd.Series(data)
        
        # Realizar análisis
        results = {}
        results['trends'] = analyzer.detect_trends(series, window=window)
        results['anomalies'] = analyzer.detect_anomalies(series, method=anomaly_method)
        results['seasonality'] = analyzer.detect_seasonality(series)
        results['peaks_valleys'] = analyzer.find_peaks_valleys(series)
        
        # Convertir a JSON serializable
        results_serializable = make_json_serializable(results)
        
        # Guardar análisis
        analysis_id = f"mcp_analysis_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        analysis_store[analysis_id] = {
            'data': series.to_frame('value'),
            'column': 'value',
            'results': results
        }
        
        return json.dumps({
            "analysis_id": analysis_id,
            "summary": {
                "total_points": len(data),
                "anomalies_detected": int(results['anomalies']['total_anomalies']),
                "peaks_count": int(results['peaks_valleys']['peaks']['count']),
                "valleys_count": int(results['peaks_valleys']['valleys']['count']),
                "trend_strength": float(results['trends']['trend_strength'])
            },
            "results": results_serializable
        }, indent=2)
        
    except Exception as e:
        return f"Error en análisis MCP: {str(e)}"

def mcp_get_analysis_report(analysis_id: str) -> str:
    """Recupera el reporte textual completo de un análisis previo a partir de su ID.

    Parámetros:
    - analysis_id (str): Identificador único del análisis almacenado, obtenido de mcp_analyze_trends. Formato: "mcp_analysis_YYYYMMDD_HHMMSS"

    Retorna:
    - str: Reporte completo en formato Markdown con:
        * Estadísticas básicas (media, desviación, min/max)
        * Análisis de tendencias (fuerza y dirección)
        * Detección de anomalías (cantidad y porcentaje)
        * Análisis de estacionalidad (patrones detectados)
        * Picos y valles (conteo de extremos)
        * Información de pronóstico si está disponible
    
    En caso de error o ID no encontrado, devuelve mensaje de error."""
    try:
        if analysis_id not in analysis_store:
            return "Error: Análisis no encontrado"
        
        stored_analysis = analysis_store[analysis_id]
        series = stored_analysis['data'][stored_analysis['column']]
        results = stored_analysis['results']
        
        report = generate_report(results, series)
        return report
        
    except Exception as e:
        return f"Error obteniendo reporte: {str(e)}"

def mcp_forecast_series(data: List[float], periods: int = 30) -> str:
    """ Genera un pronóstico lineal para una serie temporal basado en regresión lineal simple.

    Parámetros:
    - data (List[float]): Serie de datos históricos numéricos. Mínimo 2 valores requeridos. Ejemplo: [100, 105, 110, 108, 115, 120]
    - periods (int): Número de períodos futuros a predecir (5-100). Por defecto: 30

    Retorna:
    - str: JSON con pronóstico que incluye:
        * forecast_values: Array de valores pronosticados
        * confidence_upper: Límite superior del intervalo de confianza (95%)
        * confidence_lower: Límite inferior del intervalo de confianza (95%)
        * r_squared: Coeficiente de determinación del modelo (0-1, donde 1 es perfecto)
        * trend_slope: Pendiente de la tendencia (positivo=creciente, negativo=decreciente, ~0=estable)
    
    Usa regresión lineal simple con intervalos de confianza del 95%. En caso de datos insuficientes, retorna error."""
    try:
        series = pd.Series(data)
        forecast_results = analyzer.generate_forecast(series, periods=periods)
        return json.dumps(forecast_results, indent=2)
        
    except Exception as e:
        return f"Error en pronóstico: {str(e)}"

# Interfaz Gradio con funciones MCP integradas
def create_gradio_interface():
    with gr.Blocks(title="Análisis de Tendencias Temporales MCP", theme=gr.themes.Soft()) as interface:
        gr.Markdown("# 📈 Análisis de Tendencias Temporales con MCP")
        gr.Markdown("Servidor de análisis con capacidades MCP integradas para series temporales.")
        
        with gr.Tab("🔍 Análisis de Archivos"):
            with gr.Row():
                with gr.Column(scale=1):
                    file_input = gr.File(
                        label="Subir archivo (CSV/Excel)",
                        file_types=[".csv", ".xlsx", ".xls"]
                    )
                    column_input = gr.Textbox(
                        label="Nombre de la columna a analizar",
                        placeholder="ej: ventas, temperatura, precio"
                    )
                    
                    with gr.Row():
                        trend_window = gr.Slider(
                            minimum=3, maximum=30, value=7,
                            label="Ventana para detectar tendencias"
                        )
                        forecast_periods = gr.Slider(
                            minimum=5, maximum=100, value=30,
                            label="Períodos a pronosticar"
                        )
                    
                    anomaly_method = gr.Dropdown(
                        choices=["zscore", "iqr", "isolation"],
                        value="zscore",
                        label="Método de detección de anomalías"
                    )
                    
                    analyze_btn = gr.Button("🔍 Analizar Serie Temporal", variant="primary")
                
                with gr.Column(scale=2):
                    report_output = gr.Markdown(label="Reporte de Análisis")
                    analysis_id_output = gr.Textbox(label="ID de Análisis", visible=True)
            
            with gr.Row():
                plot_output = gr.Plot(label="Visualizaciones")
            
            analyze_btn.click(
                analyze_time_series,
                inputs=[file_input, column_input, trend_window, anomaly_method, forecast_periods],
                outputs=[report_output, plot_output, analysis_id_output]
            )
        
        with gr.Tab("🤖 Funciones MCP"):
            gr.Markdown("## Funciones MCP para Análisis Programático")
            
            with gr.Row():
                with gr.Column():
                    gr.Markdown("### Analizar Serie Temporal")
                    mcp_data_input = gr.Textbox(
                        label="Datos (JSON array)",
                        placeholder='[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]',
                        lines=3
                    )
                    mcp_dates_input = gr.Textbox(
                        label="Fechas (opcional, JSON array)",
                        placeholder='["2024-01-01", "2024-01-02", ...]',
                        lines=2
                    )
                    mcp_window = gr.Number(label="Ventana", value=7)
                    mcp_method = gr.Dropdown(
                        choices=["zscore", "iqr", "isolation"],
                        value="zscore",
                        label="Método de anomalías"
                    )
                    mcp_analyze_btn = gr.Button("📊 Analizar con MCP")
                
                with gr.Column():
                    gr.Markdown("### Obtener Reporte")
                    mcp_analysis_id = gr.Textbox(
                        label="ID de Análisis",
                        placeholder="analysis_20241201_123456"
                    )
                    mcp_report_btn = gr.Button("📄 Obtener Reporte")
                    
                    gr.Markdown("### Generar Pronóstico")
                    mcp_forecast_data = gr.Textbox(
                        label="Datos para Pronóstico",
                        placeholder='[10, 12, 14, 16, 18, 20]',
                        lines=2
                    )
                    mcp_periods = gr.Number(label="Períodos", value=10)
                    mcp_forecast_btn = gr.Button("🔮 Pronosticar")
            
            mcp_output = gr.Textbox(
                label="Resultado MCP",
                lines=15,
                show_copy_button=True
            )
            
            def handle_mcp_analyze(data_str, dates_str, window, method):
                """Función auxiliar para interfaz Gradio. Convierte strings JSON a listas y llama mcp_analyze_trends.
    
                    Parámetros:
                    - data_str (str): JSON string con array de números
                    - dates_str (str): JSON string con array de fechas (opcional)
                    - window (int): Ventana de tendencias
                    - method (str): Método de detección de anomalías
    
                    Retorna resultado de mcp_analyze_trends o mensaje de error."""
                try:
                    data = json.loads(data_str)
                    dates = json.loads(dates_str) if dates_str.strip() else None
                    return mcp_analyze_trends(data, dates, int(window), method)
                except Exception as e:
                    return f"Error: {str(e)}"
            
            def handle_mcp_forecast(data_str, periods):
                """Función auxiliar para interfaz Gradio. Convierte string JSON a lista y llama mcp_forecast_series.
    
                    Parámetros:
                    - data_str (str): JSON string con array de números históricos
                    - periods (int): Períodos a pronosticar
                    
                    Retorna resultado de mcp_forecast_series o mensaje de error."""
                try:
                    data = json.loads(data_str)
                    return mcp_forecast_series(data, int(periods))
                except Exception as e:
                    return f"Error: {str(e)}"
            
            mcp_analyze_btn.click(
                handle_mcp_analyze,
                inputs=[mcp_data_input, mcp_dates_input, mcp_window, mcp_method],
                outputs=mcp_output
            )
            
            mcp_report_btn.click(
                mcp_get_analysis_report,
                inputs=mcp_analysis_id,
                outputs=mcp_output
            )
            
            mcp_forecast_btn.click(
                handle_mcp_forecast,
                inputs=[mcp_forecast_data, mcp_periods],
                outputs=mcp_output
            )
        
        with gr.Tab("📚 Documentación"):
            gr.Markdown("""
            ## Available MCP Functions

            ### 1. `analyze_trends`
            Analyzes a time series to detect trends, anomalies, and patterns.

            **Parameters:**
            - `data`: Array of numbers (required)
            - `dates`: Array of dates (optional)
            - `window`: Trend smoothing window (default: 7)
            - `anomaly_method`: Detection method ("zscore", "iqr", "isolation")

            **Returns:** JSON with analysis ID and full results

            ### 2. `get_analysis_report`
            Retrieves the textual report of a previous analysis.

            **Parameters:**
            - `analysis_id`: Analysis ID (required)

            **Returns:** Report in Markdown format

            ### 3. `forecast_series`
            Generates forecasts for a time series.

            **Parameters:**
            - `data`: Array of historical numbers (required)
            - `periods`: Number of periods to forecast (default: 30)

            **Returns:** JSON with forecasted values and confidence intervals

            ## Analysis Features

            - 🔍 **Trend detection** with moving average smoothing  
            - ⚠️ **Anomaly identification** using multiple methods  
            - 📅 **Seasonality analysis** for recurring patterns  
            - 📊 **Peak and valley detection** using scipy  
            - 🔮 **Linear forecasting** with confidence intervals  
            - 📈 **Interactive visualizations** with Plotly

            ## Use from External Applications

            This server can be used as a full MCP server, providing  
            time series analysis through structured function calls.
            """)
    
    return interface

def main():
    """Función principal"""
    print("🚀 Iniciando Servidor de Análisis de Tendencias Temporales")
    print("📊 Con capacidades MCP integradas en Gradio 5")
    
    # Crear y lanzar interfaz con MCP habilitado
    interface = create_gradio_interface()
    
    # Lanzar con MCP=True para habilitar capacidades MCP en Gradio 5
    interface.launch(
        share=True,
        mcp_server=True  # Habilita capacidades MCP en Gradio 5
    )

if __name__ == "__main__":
    main()