Coverage for python/pyairflowtester/dependency_intelligence/intelligence.py: 80%
209 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 20:43 +0530
1"""Intelligence engines for advanced analysis (Phase 3: Weeks 9-12)."""
3import logging
4from dataclasses import dataclass
5from datetime import datetime, timedelta
6from typing import Dict, List, Optional
8from .analytics import TestCoverageAnalyzer
9from .graph import DependencyGraphEngine
10from .models import (
11 DependencyGraph,
12 NodeSeverity,
13)
15logger = logging.getLogger(__name__)
18@dataclass
19class FailurePrediction:
20 """Failure prediction for a node."""
22 node_id: str
23 failure_probability: float # 0.0-1.0
24 confidence: float
25 contributing_factors: List[str]
26 predicted_at: datetime
27 time_to_failure: Optional[timedelta]
30@dataclass
31class AnomalyDetection:
32 """Detected anomalies in dependency patterns."""
34 anomaly_type: str # unusual_connectivity, missing_tests, abandoned_node
35 node_id: str
36 severity: NodeSeverity
37 details: str
38 detected_at: datetime
41@dataclass
42class Recommendation:
43 """Intelligence-based recommendation."""
45 recommendation_type: str # refactor, add_sla, increase_tests, optimize
46 node_id: str
47 priority: str # critical, high, medium, low
48 action: str
49 expected_benefit: str
50 effort: str
53@dataclass
54class HealthScore:
55 """Overall health score for dependency graph."""
57 overall_score: float # 0-100
58 coverage_score: float
59 connectivity_score: float
60 ownership_score: float
61 test_score: float
62 issues_count: int
63 critical_issues: int
66class FailurePredictionEngine:
67 """Predict likelihood of node failures."""
69 def __init__(
70 self,
71 graph: DependencyGraph,
72 test_analyzer: Optional[TestCoverageAnalyzer] = None,
73 ):
74 self.graph = graph
75 self.engine = DependencyGraphEngine(graph)
76 self.test_analyzer = test_analyzer or TestCoverageAnalyzer(graph)
77 self.failure_history: Dict[str, List[datetime]] = {}
79 def record_failure(self, node_id: str) -> None:
80 """Record a failure event."""
81 if node_id not in self.failure_history:
82 self.failure_history[node_id] = []
83 self.failure_history[node_id].append(datetime.utcnow())
85 def predict_node_failure(self, node_id: str) -> FailurePrediction:
86 """
87 Predict failure probability for a node.
89 Factors:
90 - Historical failure rate
91 - Dependencies on failing nodes
92 - Test coverage
93 - Complexity
94 - Age of last change
95 """
96 node = self.graph.nodes.get(node_id)
97 if not node: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true
98 return FailurePrediction(
99 node_id=node_id,
100 failure_probability=0.0,
101 confidence=0.0,
102 contributing_factors=[],
103 predicted_at=datetime.utcnow(),
104 time_to_failure=None,
105 )
107 probability = 0.0
108 confidence = 0.5
109 factors = []
111 # Factor 1: Historical failure rate
112 if node_id in self.failure_history:
113 failures = self.failure_history[node_id]
114 if len(failures) > 0: 114 ↛ 123line 114 didn't jump to line 123 because the condition on line 114 was always true
115 # Calculate failure frequency
116 days_of_data = 30 # Assume 30 days of history
117 failure_rate = len(failures) / days_of_data
118 probability += min(0.3, failure_rate)
119 factors.append(f"Historical failure rate: {failure_rate:.1%}")
120 confidence = min(1.0, confidence + 0.2)
122 # Factor 2: Upstream failures impact
123 upstream = self.engine.get_upstream_nodes(node_id)
124 failing_upstream = [
125 u for u in upstream if u in self.failure_history and len(self.failure_history[u]) > 0
126 ]
127 if failing_upstream:
128 probability += 0.2 * (len(failing_upstream) / max(1, len(upstream)))
129 factors.append(f"{len(failing_upstream)} upstream nodes have failed")
130 confidence = min(1.0, confidence + 0.1)
132 # Factor 3: Test coverage
133 test_count = self.test_analyzer.analyze_coverage(node_id).total_tests
134 if test_count == 0:
135 probability += 0.15
136 factors.append("No test coverage")
138 # Factor 4: Complexity (high downstream = higher risk)
139 downstream = self.engine.get_downstream_nodes(node_id)
140 if len(downstream) > 100: 140 ↛ 141line 140 didn't jump to line 141 because the condition on line 140 was never true
141 probability += 0.1
142 factors.append(f"High complexity: {len(downstream)} downstream nodes")
144 # Factor 5: Node severity (critical nodes maintained better)
145 if node.severity == NodeSeverity.LOW: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 probability += 0.05
147 factors.append("Low severity (may lack maintenance)")
149 probability = min(1.0, probability)
151 # Estimate time to failure based on frequency
152 time_to_failure = None
153 if probability > 0.3:
154 # Convert probability to estimated hours until failure
155 hours_until = int(max(1, (1.0 - probability) * 168)) # 168 hours/week
156 time_to_failure = timedelta(hours=hours_until)
158 return FailurePrediction(
159 node_id=node_id,
160 failure_probability=probability,
161 confidence=confidence,
162 contributing_factors=factors,
163 predicted_at=datetime.utcnow(),
164 time_to_failure=time_to_failure,
165 )
167 def predict_all_nodes(self) -> Dict[str, FailurePrediction]:
168 """Predict failures for all nodes."""
169 return {node_id: self.predict_node_failure(node_id) for node_id in self.graph.nodes}
171 def get_high_risk_nodes(self, threshold: float = 0.5) -> List[FailurePrediction]:
172 """Get nodes with failure probability above threshold."""
173 predictions = self.predict_all_nodes()
174 return sorted(
175 [p for p in predictions.values() if p.failure_probability >= threshold],
176 key=lambda p: p.failure_probability,
177 reverse=True,
178 )
181class AnomalyDetector:
182 """Detect anomalies in dependency patterns."""
184 def __init__(self, graph: DependencyGraph):
185 self.graph = graph
186 self.engine = DependencyGraphEngine(graph)
188 def detect_all_anomalies(self) -> List[AnomalyDetection]:
189 """Detect all anomalies in the graph."""
190 anomalies = []
192 # Detect isolated nodes
193 orphans = self.engine.detect_orphans()
194 for isolated in orphans["isolated"]:
195 node = self.graph.nodes.get(isolated)
196 if node: 196 ↛ 194line 196 didn't jump to line 194 because the condition on line 196 was always true
197 anomalies.append(
198 AnomalyDetection(
199 anomaly_type="isolated_node",
200 node_id=isolated,
201 severity=(
202 NodeSeverity.HIGH
203 if node.severity == NodeSeverity.CRITICAL
204 else NodeSeverity.MEDIUM
205 ),
206 details=f"Node {node.name} has no dependencies or dependents",
207 detected_at=datetime.utcnow(),
208 )
209 )
211 # Detect unusual connectivity patterns
212 centrality = self.engine.get_node_centrality()
213 avg_centrality = sum(centrality.values()) / len(centrality) if centrality else 0
214 high_centrality = {n: c for n, c in centrality.items() if c > avg_centrality * 3}
216 for node_id, score in high_centrality.items(): 216 ↛ 217line 216 didn't jump to line 217 because the loop on line 216 never started
217 node = self.graph.nodes.get(node_id)
218 if node:
219 anomalies.append(
220 AnomalyDetection(
221 anomaly_type="high_centrality",
222 node_id=node_id,
223 severity=NodeSeverity.MEDIUM,
224 details=f"Node {node.name} has unusually high connectivity ({score:.1%})",
225 detected_at=datetime.utcnow(),
226 )
227 )
229 # Detect cycles (inherent anomalies)
230 cycles = self.engine.detect_cycles()
231 if cycles: 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true
232 for cycle in cycles[:5]: # Report first 5 cycles
233 if cycle:
234 anomalies.append(
235 AnomalyDetection(
236 anomaly_type="circular_dependency",
237 node_id=cycle[0],
238 severity=NodeSeverity.HIGH,
239 details=f"Circular dependency: {' -> '.join(cycle[:3])}...",
240 detected_at=datetime.utcnow(),
241 )
242 )
244 # Detect unowned critical nodes
245 for node in self.graph.nodes.values():
246 if node.severity == NodeSeverity.CRITICAL and not node.owner:
247 anomalies.append(
248 AnomalyDetection(
249 anomaly_type="unowned_critical",
250 node_id=node.id,
251 severity=NodeSeverity.HIGH,
252 details=f"Critical node {node.name} has no owner assigned",
253 detected_at=datetime.utcnow(),
254 )
255 )
257 return anomalies
260class RecommendationEngine:
261 """Generate intelligence-based recommendations."""
263 def __init__(self, graph: DependencyGraph):
264 self.graph = graph
265 self.engine = DependencyGraphEngine(graph)
267 def generate_recommendations(self) -> List[Recommendation]:
268 """Generate all recommendations."""
269 recommendations = []
271 # Recommendation 1: Refactor high-centrality nodes
272 centrality = self.engine.get_node_centrality()
273 avg_centrality = sum(centrality.values()) / len(centrality) if centrality else 0
274 high_centrality_nodes = sorted(
275 [(n, c) for n, c in centrality.items() if c > avg_centrality * 2],
276 key=lambda x: x[1],
277 reverse=True,
278 )
280 for node_id, score in high_centrality_nodes[:5]: 280 ↛ 281line 280 didn't jump to line 281 because the loop on line 280 never started
281 node = self.graph.nodes.get(node_id)
282 if node:
283 recommendations.append(
284 Recommendation(
285 recommendation_type="refactor_centrality",
286 node_id=node_id,
287 priority="high" if score > avg_centrality * 3 else "medium",
288 action=f"Refactor {node.name} to reduce connectivity",
289 expected_benefit="Reduced coupling, easier maintenance",
290 effort="2-3 days",
291 )
292 )
294 # Recommendation 2: Add SLAs to critical nodes
295 critical_without_sla = [
296 n.id for n in self.graph.nodes.values() if n.severity == NodeSeverity.CRITICAL
297 ]
299 for node_id in critical_without_sla[:10]:
300 node = self.graph.nodes.get(node_id)
301 if node: 301 ↛ 299line 301 didn't jump to line 299 because the condition on line 301 was always true
302 downstream_count = len(self.engine.get_downstream_nodes(node_id))
303 recommendations.append(
304 Recommendation(
305 recommendation_type="add_sla",
306 node_id=node_id,
307 priority="critical",
308 action=f"Define SLA for {node.name}",
309 expected_benefit=f"Monitor {downstream_count} downstream nodes",
310 effort="1-2 hours",
311 )
312 )
314 # Recommendation 3: Improve test coverage
315 # (Would integrate with TestCoverageAnalyzer)
316 poorly_tested = [
317 n.id
318 for n in self.graph.nodes.values()
319 if n.severity in (NodeSeverity.CRITICAL, NodeSeverity.HIGH)
320 ][:5]
322 for node_id in poorly_tested:
323 node = self.graph.nodes.get(node_id)
324 if node: 324 ↛ 322line 324 didn't jump to line 322 because the condition on line 324 was always true
325 recommendations.append(
326 Recommendation(
327 recommendation_type="improve_tests",
328 node_id=node_id,
329 priority="high",
330 action=f"Add tests for {node.name}",
331 expected_benefit="Reduce failure probability by 30%",
332 effort="2-3 days",
333 )
334 )
336 # Recommendation 4: Establish ownership
337 unowned = [n for n in self.graph.nodes.values() if not n.owner]
338 for node in unowned[:5]: 338 ↛ 339line 338 didn't jump to line 339 because the loop on line 338 never started
339 recommendations.append(
340 Recommendation(
341 recommendation_type="establish_ownership",
342 node_id=node.id,
343 priority="high" if node.severity == NodeSeverity.CRITICAL else "medium",
344 action=f"Assign owner to {node.name}",
345 expected_benefit="Clear responsibility, faster response",
346 effort="1 hour",
347 )
348 )
350 return sorted(
351 recommendations,
352 key=lambda r: {"critical": 0, "high": 1, "medium": 2, "low": 3}.get(r.priority, 4),
353 )
355 def get_top_recommendations(self, limit: int = 10) -> List[Recommendation]:
356 """Get top N recommendations."""
357 return self.generate_recommendations()[:limit]
360class HealthScoreCalculator:
361 """Calculate overall health score for dependency graph."""
363 def __init__(
364 self,
365 graph: DependencyGraph,
366 test_analyzer: Optional[TestCoverageAnalyzer] = None,
367 ):
368 self.graph = graph
369 self.engine = DependencyGraphEngine(graph)
370 self.test_analyzer = test_analyzer or TestCoverageAnalyzer(graph)
372 def calculate_health_score(self) -> HealthScore:
373 """
374 Calculate comprehensive health score.
376 Components:
377 - Coverage: percentage of nodes with metadata
378 - Connectivity: balance between isolated and over-connected
379 - Ownership: percentage of nodes with assigned owners
380 - Tests: percentage of nodes with test coverage
381 """
382 # Coverage score (0-20)
383 coverage_score = self._calculate_coverage_score()
385 # Connectivity score (0-20)
386 connectivity_score = self._calculate_connectivity_score()
388 # Ownership score (0-20)
389 ownership_score = self._calculate_ownership_score()
391 # Test score (0-20)
392 test_score = self._calculate_test_score()
394 # SLA score (0-20)
395 sla_score = self._calculate_sla_score()
397 overall = coverage_score + connectivity_score + ownership_score + test_score + sla_score
399 # Count issues
400 issues = self._count_issues()
402 return HealthScore(
403 overall_score=overall,
404 coverage_score=coverage_score,
405 connectivity_score=connectivity_score,
406 ownership_score=ownership_score,
407 test_score=test_score,
408 issues_count=issues["total"],
409 critical_issues=issues["critical"],
410 )
412 def _calculate_coverage_score(self) -> float:
413 """Score completeness of metadata."""
414 nodes_with_description = sum(1 for n in self.graph.nodes.values() if n.description)
415 coverage = nodes_with_description / len(self.graph.nodes) if self.graph.nodes else 0
416 return min(20.0, coverage * 20)
418 def _calculate_connectivity_score(self) -> float:
419 """Score graph connectivity health."""
420 components = self.engine.detect_disconnected_components()
422 if len(components) <= 1: 422 ↛ 424line 422 didn't jump to line 424 because the condition on line 422 was always true
423 return 20.0 # Fully connected
424 elif len(components) <= 3:
425 return 15.0
426 elif len(components) <= 10:
427 return 10.0
428 else:
429 return 5.0
431 def _calculate_ownership_score(self) -> float:
432 """Score ownership coverage."""
433 owned = sum(1 for n in self.graph.nodes.values() if n.owner)
434 ownership = owned / len(self.graph.nodes) if self.graph.nodes else 0
435 return min(20.0, ownership * 20)
437 def _calculate_test_score(self) -> float:
438 """Score test coverage using real data from TestCoverageAnalyzer."""
439 if not self.graph.nodes: 439 ↛ 440line 439 didn't jump to line 440 because the condition on line 439 was never true
440 return 0.0
441 nodes_with_tests = sum(
442 1
443 for node_id in self.graph.nodes
444 if self.test_analyzer.analyze_coverage(node_id).total_tests > 0
445 )
446 coverage = nodes_with_tests / len(self.graph.nodes)
447 return min(20.0, coverage * 20)
449 def _calculate_sla_score(self) -> float:
450 """Score SLA coverage for critical nodes."""
451 critical = sum(1 for n in self.graph.nodes.values() if n.severity == NodeSeverity.CRITICAL)
452 return min(20.0, (critical / max(1, critical)) * 10)
454 def _count_issues(self) -> Dict[str, int]:
455 """Count critical and total issues."""
456 cycles = len(self.engine.detect_cycles())
457 orphans = len(self.engine.detect_orphans()["isolated"])
458 unowned_critical = sum(
459 1
460 for n in self.graph.nodes.values()
461 if n.severity == NodeSeverity.CRITICAL and not n.owner
462 )
464 return {
465 "total": cycles + orphans + unowned_critical,
466 "critical": cycles + unowned_critical,
467 }
469 def get_health_summary(self) -> Dict[str, str]:
470 """Get human-readable health summary."""
471 score = self.calculate_health_score()
473 if score.overall_score >= 80: 473 ↛ 474line 473 didn't jump to line 474 because the condition on line 473 was never true
474 status = "Excellent"
475 color = "green"
476 elif score.overall_score >= 60: 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true
477 status = "Good"
478 color = "yellow"
479 elif score.overall_score >= 40: 479 ↛ 483line 479 didn't jump to line 483 because the condition on line 479 was always true
480 status = "Fair"
481 color = "orange"
482 else:
483 status = "Poor"
484 color = "red"
486 return {
487 "status": status,
488 "score": f"{score.overall_score:.1f}/100",
489 "color": color,
490 "critical_issues": str(score.critical_issues),
491 "total_issues": str(score.issues_count),
492 }