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):
        """Define security metric"""
        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):
        """Record metric measurement"""
        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):
        """Calculate measurement quality score"""
        metric = self.metrics[metric_id]
        
        # Check if value is within expected ranges
        if metric['type'] == 'operational':
            if metric['unit'] == 'hours' and value > 24:
                return 0.5  # Suspicious value
            elif metric['unit'] == 'percentage' and (value < 0 or value > 100):
                return 0.0  # Invalid value
        
        return 1.0  # Valid value
    
    def calculate_metric_statistics(self, metric_id, days=30):
        """Calculate metric statistics"""
        if metric_id not in self.metrics:
            return None
        
        # Get measurements from last N days
        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):
        """Calculate value trend"""
        if len(values) < 2:
            return 'insufficient_data'
        
        # Calculate slope using simple linear regression
        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):
        """Evaluate KPI performance"""
        if kpi_id not in self.kpis:
            return None
        
        kpi = self.kpis[kpi_id]
        metric_id = kpi['metric_id']
        
        # Get current value
        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']
        
        # Determine KPI status
        if current_value >= target_value:
            status = 'excellent'
        elif current_value >= warning_threshold:
            status = 'good'
        elif current_value >= critical_threshold:
            status = 'warning'
        else:
            status = 'critical'
        
        # Calculate deviation from target
        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):
        """Generate metrics dashboard"""
        dashboard = {
            'generated_date': datetime.now(),
            'summary': {},
            'kpis': {},
            'alerts': [],
            'recommendations': []
        }
        
        # General summary
        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)])
        }
        
        # Evaluate KPIs
        for kpi_id in self.kpis.keys():
            performance = self.evaluate_kpi_performance(kpi_id)
            if performance:
                dashboard['kpis'][kpi_id] = performance
                
                # Generate alerts
                if performance['status'] in ['warning', 'critical']:
                    dashboard['alerts'].append({
                        'type': 'kpi_alert',
                        'kpi_id': kpi_id,
                        'status': performance['status'],
                        'message': f"KPI {kpi_id} is in {performance['status']} status"
                    })
        
        # Generate recommendations
        dashboard['recommendations'] = self.generate_recommendations()
        
        return dashboard
    
    def generate_recommendations(self):
        """Generate recommendations based on metrics"""
        recommendations = []
        
        # Analyze KPIs with issues
        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"Improve {kpi_id} - deviation of {performance['deviation_percentage']:.1f}%"
                })
        
        # Analyze metrics without recent data
        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"Collect data for {metric_id} - no recent measurements"
                })
        
        return recommendations

# Usage example
metrics_mgmt = SecurityMetricsManagement()

# Define metrics
metrics_mgmt.define_metric('METRIC-001', {
    'name': 'Incident Detection Time',
    'description': 'Average time to detect security incidents',
    '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': 'Compliance Rate',
    'description': 'Percentage of security policy compliance',
    'category': 'strategic',
    'type': 'strategic',
    'unit': 'percentage',
    'frequency': 'monthly',
    'is_kpi': True,
    'target_value': 95.0,
    'threshold_warning': 90.0,
    'threshold_critical': 80.0
})

# Record measurements
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))

# Generate dashboard
dashboard = metrics_mgmt.generate_metrics_dashboard()
print(f"Dashboard generated: {dashboard['summary']['total_metrics']} metrics")

  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):
        """Analyze metric trend"""
        if len(measurements) < 10:
            return {'status': 'insufficient_data'}
        
        values = [m['value'] for m in measurements]
        timestamps = [m['timestamp'] for m in measurements]
        
        # Linear trend analysis
        x = np.arange(len(values))
        y = np.array(values)
        
        # Linear regression
        slope, intercept = np.polyfit(x, y, 1)
        r_squared = self.calculate_r_squared(x, y, slope, intercept)
        
        # Seasonality analysis
        seasonal_pattern = self.detect_seasonal_pattern(values, timestamps)
        
        # Anomaly detection
        anomalies = self.detect_anomalies(values)
        
        # Short-term prediction
        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):
        """Calculate R² to evaluate fit quality"""
        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):
        """Detect seasonal pattern"""
        if len(values) < 30:
            return {'has_seasonality': False}
        
        # Group by day of week
        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])
        
        # Calculate variation between days
        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)
        
        # Determine if there is significant seasonality
        has_seasonality = weekday_std > overall_mean * 0.1  # 10% variation
        
        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):
        """Detect anomalies in values"""
        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):
        """Predict future values"""
        if len(values) < 10:
            return []
        
        # Use simple moving average for prediction
        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):
        """Calculate confidence interval"""
        if len(values) < 2:
            return {'lower': 0, 'upper': 0}
        
        mean = np.mean(values)
        std = np.std(values)
        n = len(values)
        
        # Use t-distribution for small samples
        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):
        """Calculate trend confidence level"""
        if len(values) < 10:
            return 'low'
        
        # Based on R² and number of points
        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'

# Usage example
trend_analysis = TrendAnalysis()

# Simulate measurement data
measurements = []
for i in range(30):
    value = 1.0 + np.random.normal(0, 0.2) + i * 0.01  # Increasing trend with noise
    measurements.append({
        'value': value,
        'timestamp': datetime.now() - timedelta(days=30-i)
    })

# Analyze trend
trend_result = trend_analysis.analyze_trend('METRIC-001', measurements)
print(f"Trend analysis: {trend_result['trend_direction']}")
print(f"Confidence: {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):
        """Create executive dashboard"""
        dashboard = {
            'dashboard_id': dashboard_config['dashboard_id'],
            'name': dashboard_config['name'],
            'audience': dashboard_config['audience'],
            'kpis': [],
            'charts': [],
            'alerts': [],
            'summary': {},
            'last_updated': datetime.now()
        }
        
        # Add main KPIs
        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):
        """Generate executive summary"""
        summary = {
            'report_date': datetime.now(),
            'overall_security_status': 'good',
            'key_highlights': [],
            'areas_of_concern': [],
            'recommendations': [],
            'financial_impact': {},
            'risk_assessment': {}
        }
        
        # Evaluate overall security status
        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'
        
        # Generate 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 in excellent status")
        
        # Generate areas of concern
        if critical_kpis:
            summary['areas_of_concern'].append(f"{len(critical_kpis)} critical KPIs require immediate attention")
        
        if warning_kpis:
            summary['areas_of_concern'].append(f"{len(warning_kpis)} KPIs in warning")
        
        # Generate recommendations
        if critical_kpis:
            summary['recommendations'].append("Prioritize resources for critical KPIs")
        
        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("Review security strategy - performance below target")
        
        return summary
    
    def create_kpi_scorecard(self, kpis_data):
        """Create KPI scorecard"""
        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
        }
        
        # Calculate overall score
        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):
        """Generate trend report"""
        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

# Usage example
executive_reporting = ExecutiveReporting()

# Create executive dashboard
dashboard_config = {
    'dashboard_id': 'EXEC-001',
    'name': 'Security Executive Dashboard',
    'audience': 'C-Level',
    'kpis': [
        {
            'kpi_id': 'KPI-001',
            'name': 'Detection Time',
            'current_value': 0.8,
            'target_value': 1.0,
            'status': 'good',
            'trend': 'improving'
        },
        {
            'kpi_id': 'KPI-002',
            'name': 'Compliance',
            'current_value': 92.5,
            'target_value': 95.0,
            'status': 'warning',
            'trend': 'stable'
        }
    ]
}

dashboard = executive_reporting.create_executive_dashboard(dashboard_config)
print(f"Executive dashboard created: {dashboard['name']}")

# Generate executive summary
kpis_data = [
    {'status': 'excellent', 'variance': 5},
    {'status': 'warning', 'variance': -15},
    {'status': 'good', 'variance': 2}
]

summary = executive_reporting.generate_executive_summary({}, kpis_data)
print(f"Security status: {summary['overall_security_status']}")