Métricas y KPIs

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

class SecurityMetricsManagement:
    def __init__(self):
        self.metrics = {}
        self.kpis = {}
        self.measurements = {}
        self.baselines = {}
        self.targets = {}
        self.trends = {}
    
    def define_metric(self, metric_id, metric_config):
        """Definir métrica de seguridad"""
        self.metrics[metric_id] = {
            'metric_id': metric_id,
            'name': metric_config['name'],
            'description': metric_config['description'],
            'category': metric_config['category'],
            'type': metric_config['type'],  # 'operational', 'strategic', 'tactical'
            'unit': metric_config.get('unit', 'count'),
            'frequency': metric_config.get('frequency', 'daily'),
            'data_source': metric_config.get('data_source', 'manual'),
            'calculation_method': metric_config.get('calculation_method', 'sum'),
            'is_kpi': metric_config.get('is_kpi', False),
            'created_date': datetime.now()
        }
        
        if metric_config.get('is_kpi', False):
            self.kpis[metric_id] = {
                'metric_id': metric_id,
                'target_value': metric_config.get('target_value', 0),
                'threshold_warning': metric_config.get('threshold_warning', 0),
                'threshold_critical': metric_config.get('threshold_critical', 0),
                'owner': metric_config.get('owner', 'Security Team'),
                'review_frequency': metric_config.get('review_frequency', 'monthly')
            }
    
    def record_measurement(self, metric_id, value, timestamp=None, metadata=None):
        """Registrar medición de métrica"""
        if metric_id not in self.metrics:
            return False
        
        if timestamp is None:
            timestamp = datetime.now()
        
        measurement_id = f"MEAS-{len(self.measurements) + 1}"
        
        self.measurements[measurement_id] = {
            'measurement_id': measurement_id,
            'metric_id': metric_id,
            'value': value,
            'timestamp': timestamp,
            'metadata': metadata or {},
            'quality_score': self.calculate_quality_score(value, metric_id)
        }
        
        return True
    
    def calculate_quality_score(self, value, metric_id):
        """Calcular score de calidad de la medición"""
        metric = self.metrics[metric_id]
        
        # Verificar si el valor está dentro de rangos esperados
        if metric['type'] == 'operational':
            if metric['unit'] == 'hours' and value > 24:
                return 0.5  # Valor sospechoso
            elif metric['unit'] == 'percentage' and (value < 0 or value > 100):
                return 0.0  # Valor inválido
        
        return 1.0  # Valor válido
    
    def calculate_metric_statistics(self, metric_id, days=30):
        """Calcular estadísticas de métrica"""
        if metric_id not in self.metrics:
            return None
        
        # Obtener mediciones de los últimos N días
        cutoff_date = datetime.now() - timedelta(days=days)
        recent_measurements = [
            m for m in self.measurements.values()
            if m['metric_id'] == metric_id and m['timestamp'] >= cutoff_date
        ]
        
        if not recent_measurements:
            return None
        
        values = [m['value'] for m in recent_measurements]
        
        statistics = {
            'metric_id': metric_id,
            'period_days': days,
            'total_measurements': len(values),
            'mean': np.mean(values),
            'median': np.median(values),
            'std_dev': np.std(values),
            'min': np.min(values),
            'max': np.max(values),
            'latest_value': values[-1] if values else None,
            'trend': self.calculate_trend(values)
        }
        
        return statistics
    
    def calculate_trend(self, values):
        """Calcular tendencia de valores"""
        if len(values) < 2:
            return 'insufficient_data'
        
        # Calcular pendiente usando regresión lineal simple
        x = np.arange(len(values))
        y = np.array(values)
        
        if len(x) > 1:
            slope = np.polyfit(x, y, 1)[0]
            
            if slope > 0.1:
                return 'increasing'
            elif slope < -0.1:
                return 'decreasing'
            else:
                return 'stable'
        
        return 'stable'
    
    def evaluate_kpi_performance(self, kpi_id):
        """Evaluar rendimiento de KPI"""
        if kpi_id not in self.kpis:
            return None
        
        kpi = self.kpis[kpi_id]
        metric_id = kpi['metric_id']
        
        # Obtener valor actual
        recent_measurements = [
            m for m in self.measurements.values()
            if m['metric_id'] == metric_id
        ]
        
        if not recent_measurements:
            return None
        
        current_value = recent_measurements[-1]['value']
        target_value = kpi['target_value']
        warning_threshold = kpi['threshold_warning']
        critical_threshold = kpi['threshold_critical']
        
        # Determinar estado del KPI
        if current_value >= target_value:
            status = 'excellent'
        elif current_value >= warning_threshold:
            status = 'good'
        elif current_value >= critical_threshold:
            status = 'warning'
        else:
            status = 'critical'
        
        # Calcular desviación del objetivo
        deviation = ((current_value - target_value) / target_value * 100) if target_value > 0 else 0
        
        performance = {
            'kpi_id': kpi_id,
            'current_value': current_value,
            'target_value': target_value,
            'deviation_percentage': deviation,
            'status': status,
            'last_updated': recent_measurements[-1]['timestamp']
        }
        
        return performance
    
    def generate_metrics_dashboard(self):
        """Generar dashboard de métricas"""
        dashboard = {
            'generated_date': datetime.now(),
            'summary': {},
            'kpis': {},
            'alerts': [],
            'recommendations': []
        }
        
        # Resumen general
        total_metrics = len(self.metrics)
        total_kpis = len(self.kpis)
        total_measurements = len(self.measurements)
        
        dashboard['summary'] = {
            'total_metrics': total_metrics,
            'total_kpis': total_kpis,
            'total_measurements': total_measurements,
            'last_24h_measurements': len([m for m in self.measurements.values() 
                                        if m['timestamp'] >= datetime.now() - timedelta(hours=24)])
        }
        
        # Evaluar KPIs
        for kpi_id in self.kpis.keys():
            performance = self.evaluate_kpi_performance(kpi_id)
            if performance:
                dashboard['kpis'][kpi_id] = performance
                
                # Generar alertas
                if performance['status'] in ['warning', 'critical']:
                    dashboard['alerts'].append({
                        'type': 'kpi_alert',
                        'kpi_id': kpi_id,
                        'status': performance['status'],
                        'message': f"KPI {kpi_id} está en estado {performance['status']}"
                    })
        
        # Generar recomendaciones
        dashboard['recommendations'] = self.generate_recommendations()
        
        return dashboard
    
    def generate_recommendations(self):
        """Generar recomendaciones basadas en métricas"""
        recommendations = []
        
        # Analizar KPIs con problemas
        for kpi_id, kpi in self.kpis.items():
            performance = self.evaluate_kpi_performance(kpi_id)
            if performance and performance['status'] in ['warning', 'critical']:
                recommendations.append({
                    'type': 'kpi_improvement',
                    'kpi_id': kpi_id,
                    'recommendation': f"Mejorar {kpi_id} - desviación del {performance['deviation_percentage']:.1f}%"
                })
        
        # Analizar métricas sin datos recientes
        for metric_id, metric in self.metrics.items():
            recent_measurements = [
                m for m in self.measurements.values()
                if m['metric_id'] == metric_id and 
                m['timestamp'] >= datetime.now() - timedelta(days=7)
            ]
            
            if not recent_measurements:
                recommendations.append({
                    'type': 'data_collection',
                    'metric_id': metric_id,
                    'recommendation': f"Recopilar datos para {metric_id} - sin mediciones recientes"
                })
        
        return recommendations

# Ejemplo de uso
metrics_mgmt = SecurityMetricsManagement()

# Definir métricas
metrics_mgmt.define_metric('METRIC-001', {
    'name': 'Tiempo de Detección de Incidentes',
    'description': 'Tiempo promedio para detectar incidentes de seguridad',
    'category': 'operational',
    'type': 'operational',
    'unit': 'hours',
    'frequency': 'daily',
    'is_kpi': True,
    'target_value': 1.0,
    'threshold_warning': 2.0,
    'threshold_critical': 4.0
})

metrics_mgmt.define_metric('METRIC-002', {
    'name': 'Tasa de Cumplimiento',
    'description': 'Porcentaje de cumplimiento de políticas de seguridad',
    'category': 'strategic',
    'type': 'strategic',
    'unit': 'percentage',
    'frequency': 'monthly',
    'is_kpi': True,
    'target_value': 95.0,
    'threshold_warning': 90.0,
    'threshold_critical': 80.0
})

# Registrar mediciones
metrics_mgmt.record_measurement('METRIC-001', 0.5, datetime.now() - timedelta(hours=1))
metrics_mgmt.record_measurement('METRIC-001', 1.2, datetime.now() - timedelta(hours=2))
metrics_mgmt.record_measurement('METRIC-002', 92.5, datetime.now() - timedelta(days=1))

# Generar dashboard
dashboard = metrics_mgmt.generate_metrics_dashboard()
print(f"Dashboard generado: {dashboard['summary']['total_metrics']} métricas")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173

class TrendAnalysis:
    def __init__(self):
        self.trend_data = {}
        self.seasonal_patterns = {}
        self.anomaly_detection = {}
    
    def analyze_trend(self, metric_id, measurements, period_days=90):
        """Analizar tendencia de métrica"""
        if len(measurements) < 10:
            return {'status': 'insufficient_data'}
        
        values = [m['value'] for m in measurements]
        timestamps = [m['timestamp'] for m in measurements]
        
        # Análisis de tendencia lineal
        x = np.arange(len(values))
        y = np.array(values)
        
        # Regresión lineal
        slope, intercept = np.polyfit(x, y, 1)
        r_squared = self.calculate_r_squared(x, y, slope, intercept)
        
        # Análisis de estacionalidad
        seasonal_pattern = self.detect_seasonal_pattern(values, timestamps)
        
        # Detección de anomalías
        anomalies = self.detect_anomalies(values)
        
        # Predicción a corto plazo
        forecast = self.forecast_values(values, days_ahead=7)
        
        trend_analysis = {
            'metric_id': metric_id,
            'period_days': period_days,
            'trend_direction': 'increasing' if slope > 0.01 else 'decreasing' if slope < -0.01 else 'stable',
            'trend_strength': abs(slope),
            'r_squared': r_squared,
            'seasonal_pattern': seasonal_pattern,
            'anomalies': anomalies,
            'forecast': forecast,
            'confidence': self.calculate_confidence(values, slope, r_squared)
        }
        
        return trend_analysis
    
    def calculate_r_squared(self, x, y, slope, intercept):
        """Calcular R² para evaluar calidad del ajuste"""
        y_pred = slope * x + intercept
        ss_res = np.sum((y - y_pred) ** 2)
        ss_tot = np.sum((y - np.mean(y)) ** 2)
        
        if ss_tot == 0:
            return 0
        
        return 1 - (ss_res / ss_tot)
    
    def detect_seasonal_pattern(self, values, timestamps):
        """Detectar patrón estacional"""
        if len(values) < 30:
            return {'has_seasonality': False}
        
        # Agrupar por día de la semana
        weekday_values = {}
        for i, timestamp in enumerate(timestamps):
            weekday = timestamp.weekday()
            if weekday not in weekday_values:
                weekday_values[weekday] = []
            weekday_values[weekday].append(values[i])
        
        # Calcular variación entre días
        weekday_means = {day: np.mean(vals) for day, vals in weekday_values.items()}
        weekday_std = np.std(list(weekday_means.values()))
        overall_mean = np.mean(values)
        
        # Determinar si hay estacionalidad significativa
        has_seasonality = weekday_std > overall_mean * 0.1  # 10% de variación
        
        return {
            'has_seasonality': has_seasonality,
            'weekday_pattern': weekday_means,
            'variation_coefficient': weekday_std / overall_mean if overall_mean > 0 else 0
        }
    
    def detect_anomalies(self, values, threshold=2.0):
        """Detectar anomalías en los valores"""
        if len(values) < 10:
            return []
        
        mean = np.mean(values)
        std = np.std(values)
        
        anomalies = []
        for i, value in enumerate(values):
            z_score = abs((value - mean) / std) if std > 0 else 0
            
            if z_score > threshold:
                anomalies.append({
                    'index': i,
                    'value': value,
                    'z_score': z_score,
                    'severity': 'high' if z_score > 3 else 'medium'
                })
        
        return anomalies
    
    def forecast_values(self, values, days_ahead=7):
        """Predecir valores futuros"""
        if len(values) < 10:
            return []
        
        # Usar media móvil simple para predicción
        window_size = min(7, len(values) // 2)
        recent_values = values[-window_size:]
        
        forecast = []
        for i in range(days_ahead):
            predicted_value = np.mean(recent_values)
            forecast.append({
                'day': i + 1,
                'predicted_value': predicted_value,
                'confidence_interval': self.calculate_confidence_interval(recent_values)
            })
        
        return forecast
    
    def calculate_confidence_interval(self, values, confidence=0.95):
        """Calcular intervalo de confianza"""
        if len(values) < 2:
            return {'lower': 0, 'upper': 0}
        
        mean = np.mean(values)
        std = np.std(values)
        n = len(values)
        
        # Usar distribución t para muestras pequeñas
        from scipy import stats
        t_value = stats.t.ppf((1 + confidence) / 2, n - 1)
        margin_error = t_value * (std / np.sqrt(n))
        
        return {
            'lower': mean - margin_error,
            'upper': mean + margin_error
        }
    
    def calculate_confidence(self, values, slope, r_squared):
        """Calcular nivel de confianza de la tendencia"""
        if len(values) < 10:
            return 'low'
        
        # Basado en R² y número de puntos
        if r_squared > 0.8 and len(values) > 30:
            return 'high'
        elif r_squared > 0.6 and len(values) > 20:
            return 'medium'
        else:
            return 'low'

# Ejemplo de uso
trend_analysis = TrendAnalysis()

# Simular datos de mediciones
measurements = []
for i in range(30):
    value = 1.0 + np.random.normal(0, 0.2) + i * 0.01  # Tendencia creciente con ruido
    measurements.append({
        'value': value,
        'timestamp': datetime.now() - timedelta(days=30-i)
    })

# Analizar tendencia
trend_result = trend_analysis.analyze_trend('METRIC-001', measurements)
print(f"Análisis de tendencia: {trend_result['trend_direction']}")
print(f"Confianza: {trend_result['confidence']}")

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

class ExecutiveReporting:
    def __init__(self):
        self.report_templates = {}
        self.kpi_dashboards = {}
        self.alert_systems = {}
    
    def create_executive_dashboard(self, dashboard_config):
        """Crear dashboard ejecutivo"""
        dashboard = {
            'dashboard_id': dashboard_config['dashboard_id'],
            'name': dashboard_config['name'],
            'audience': dashboard_config['audience'],
            'kpis': [],
            'charts': [],
            'alerts': [],
            'summary': {},
            'last_updated': datetime.now()
        }
        
        # Añadir KPIs principales
        for kpi_config in dashboard_config.get('kpis', []):
            kpi = {
                'kpi_id': kpi_config['kpi_id'],
                'name': kpi_config['name'],
                'current_value': kpi_config.get('current_value', 0),
                'target_value': kpi_config.get('target_value', 0),
                'status': kpi_config.get('status', 'unknown'),
                'trend': kpi_config.get('trend', 'stable'),
                'variance': kpi_config.get('variance', 0)
            }
            dashboard['kpis'].append(kpi)
        
        return dashboard
    
    def generate_executive_summary(self, metrics_data, kpis_data):
        """Generar resumen ejecutivo"""
        summary = {
            'report_date': datetime.now(),
            'overall_security_status': 'good',
            'key_highlights': [],
            'areas_of_concern': [],
            'recommendations': [],
            'financial_impact': {},
            'risk_assessment': {}
        }
        
        # Evaluar estado general de seguridad
        critical_kpis = [kpi for kpi in kpis_data if kpi.get('status') == 'critical']
        warning_kpis = [kpi for kpi in kpis_data if kpi.get('status') == 'warning']
        
        if len(critical_kpis) > 0:
            summary['overall_security_status'] = 'critical'
        elif len(warning_kpis) > 2:
            summary['overall_security_status'] = 'warning'
        
        # Generar highlights
        excellent_kpis = [kpi for kpi in kpis_data if kpi.get('status') == 'excellent']
        if excellent_kpis:
            summary['key_highlights'].append(f"{len(excellent_kpis)} KPIs en estado excelente")
        
        # Generar áreas de preocupación
        if critical_kpis:
            summary['areas_of_concern'].append(f"{len(critical_kpis)} KPIs críticos requieren atención inmediata")
        
        if warning_kpis:
            summary['areas_of_concern'].append(f"{len(warning_kpis)} KPIs en advertencia")
        
        # Generar recomendaciones
        if critical_kpis:
            summary['recommendations'].append("Priorizar recursos para KPIs críticos")
        
        if len(kpis_data) > 0:
            avg_performance = sum(kpi.get('variance', 0) for kpi in kpis_data) / len(kpis_data)
            if avg_performance < -10:
                summary['recommendations'].append("Revisar estrategia de seguridad - rendimiento por debajo del objetivo")
        
        return summary
    
    def create_kpi_scorecard(self, kpis_data):
        """Crear scorecard de KPIs"""
        scorecard = {
            'total_kpis': len(kpis_data),
            'excellent_count': len([kpi for kpi in kpis_data if kpi.get('status') == 'excellent']),
            'good_count': len([kpi for kpi in kpis_data if kpi.get('status') == 'good']),
            'warning_count': len([kpi for kpi in kpis_data if kpi.get('status') == 'warning']),
            'critical_count': len([kpi for kpi in kpis_data if kpi.get('status') == 'critical']),
            'overall_score': 0
        }
        
        # Calcular score general
        if scorecard['total_kpis'] > 0:
            score = (scorecard['excellent_count'] * 4 + 
                    scorecard['good_count'] * 3 + 
                    scorecard['warning_count'] * 2 + 
                    scorecard['critical_count'] * 1) / scorecard['total_kpis']
            scorecard['overall_score'] = score
        
        return scorecard
    
    def generate_trend_report(self, trend_data):
        """Generar reporte de tendencias"""
        trend_report = {
            'report_date': datetime.now(),
            'trends_analyzed': len(trend_data),
            'improving_trends': [],
            'declining_trends': [],
            'stable_trends': [],
            'anomalies_detected': 0
        }
        
        for trend in trend_data:
            if trend['trend_direction'] == 'increasing':
                trend_report['improving_trends'].append(trend['metric_id'])
            elif trend['trend_direction'] == 'decreasing':
                trend_report['declining_trends'].append(trend['metric_id'])
            else:
                trend_report['stable_trends'].append(trend['metric_id'])
            
            trend_report['anomalies_detected'] += len(trend.get('anomalies', []))
        
        return trend_report

# Ejemplo de uso
executive_reporting = ExecutiveReporting()

# Crear dashboard ejecutivo
dashboard_config = {
    'dashboard_id': 'EXEC-001',
    'name': 'Security Executive Dashboard',
    'audience': 'C-Level',
    'kpis': [
        {
            'kpi_id': 'KPI-001',
            'name': 'Tiempo de Detección',
            'current_value': 0.8,
            'target_value': 1.0,
            'status': 'good',
            'trend': 'improving'
        },
        {
            'kpi_id': 'KPI-002',
            'name': 'Cumplimiento',
            'current_value': 92.5,
            'target_value': 95.0,
            'status': 'warning',
            'trend': 'stable'
        }
    ]
}

dashboard = executive_reporting.create_executive_dashboard(dashboard_config)
print(f"Dashboard ejecutivo creado: {dashboard['name']}")

# Generar resumen ejecutivo
kpis_data = [
    {'status': 'excellent', 'variance': 5},
    {'status': 'warning', 'variance': -15},
    {'status': 'good', 'variance': 2}
]

summary = executive_reporting.generate_executive_summary({}, kpis_data)
print(f"Estado de seguridad: {summary['overall_security_status']}")