"""Detector: Project Schedule Variance (Slippage). Compares estimated completion dates to actual/forecasted completion for active and recently completed projects. Flags projects that are >10% behind schedule — critical for design-build companies (Tundraland, Renuity) where each week of slippage eats margin through crew costs, equipment rentals, and overhead. Data sources: - Project: start_date, end_date (planned), completed_date (actual) - JobberJob (P2): estimated_end_date, actual_end_date (if Jobber connected) Works with zero, one, or multiple connectors — falls back gracefully to local Project data when no CRM/PM connector is available. """ from __future__ import annotations from datetime import timedelta from typing import Any, Dict, List, Optional from .base import BaseDetector, LeakCandidate from ._common import connector_available, any_connector_available, utcnow, as_aware, days_since class ProjectScheduleVarianceDetector(BaseDetector): id = "project_schedule_variance" name = "Project Schedule Variance" connector = None # cross-connector / composite enabled_by_default = True description = ( "Compares planned completion dates to actual/forecasted completion. " "Flags projects >10% behind schedule. Tracks slippage by project type " "and project manager — margin protection for design-build ops." ) default_params = { "slippage_threshold_pct": 10, "min_slippage_days": 3, # Ignore slippage < 3 days (noise floor) "lookback_days": 365, "active_only": False, # If True, only check active projects } def check(self, company_id: str) -> List[LeakCandidate]: from app.models import Project, JobberJob now = utcnow() cutoff = now - timedelta(days=self.default_params.get("lookback_days", 365)) threshold_pct = self.default_params.get("slippage_threshold_pct", 10) min_slippage_days = self.default_params.get("min_slippage_days", 3) active_only = self.default_params.get("active_only", False) candidates: List[LeakCandidate] = [] # -- 1. Check local Project data ------------------------------------ project_query = Project.query.filter( Project.company_id == company_id, Project.start_date.isnot(None), Project.end_date.isnot(None), ) if active_only: project_query = project_query.filter( Project.status.in_(["active", "planning"]) ) else: project_query = project_query.filter( Project.status.in_(["active", "planning", "completed", "on_hold"]) ) projects = project_query.all() # Filter to lookback window recent_projects = [ p for p in projects if as_aware(p.start_date) is not None and as_aware(p.start_date) >= cutoff ] slipped_projects: Dict[str, Dict[str, Any]] = {} for proj in recent_projects: start = as_aware(proj.start_date) planned_end = as_aware(proj.end_date) if start is None or planned_end is None: continue # Calculate planned duration planned_duration = (planned_end - start).total_seconds() / 86400.0 if planned_duration <= 0: continue # Determine actual or forecasted completion completed = as_aware(proj.completed_date) if completed is not None: # Project is done — use actual completion actual_end = completed else: # Project still active — estimate current progress # Use elapsed time since start as proxy for progress elapsed = (now - start).total_seconds() / 86400.0 # If we're past the planned end date, the project is slipping if now > planned_end: actual_end = now # Calculate how far behind days_behind = (now - planned_end).days slipped_projects[f"project_{proj.id}"] = { "project_id": proj.id, "project_name": proj.name, "status": proj.status, "start_date": start.isoformat(), "planned_end": planned_end.isoformat(), "planned_duration_days": round(planned_duration, 1), "days_behind": days_behind, "slippage_pct": round((days_behind / planned_duration) * 100, 1), "forecast_actual_end": None, "estimated_daily_cost": self._estimate_daily_cost(proj, planned_duration), "estimated_margin_impact": round(days_behind * self._estimate_daily_cost(proj, planned_duration), 2), "assigned_to": proj.assigned_to, "budget": proj.budget, } continue # Project not past planned end yet — check if it's on track # Use a simple heuristic: if elapsed > planned_duration * 0.8 # and project still active, flag as at-risk progress_ratio = elapsed / planned_duration if planned_duration > 0 else 0 if progress_ratio > 0.8 and proj.status in ("active", "planning"): # Near deadline but not done — potential risk days_remaining = (planned_end - now).days if days_remaining < min_slippage_days: slipped_projects[f"project_{proj.id}_at_risk"] = { "project_id": proj.id, "project_name": proj.name, "status": proj.status, "start_date": start.isoformat(), "planned_end": planned_end.isoformat(), "planned_duration_days": round(planned_duration, 1), "days_behind": 0, "days_remaining": days_remaining, "slippage_pct": 0, "at_risk": True, "progress_ratio": round(progress_ratio, 2), "estimated_daily_cost": self._estimate_daily_cost(proj, planned_duration), "estimated_margin_impact": 0, "assigned_to": proj.assigned_to, "budget": proj.budget, } continue # Completed project — compare actual vs. planned actual_duration = (actual_end - start).total_seconds() / 86400.0 days_over = actual_duration - planned_duration if days_over <= 0: continue slippage_pct = (days_over / planned_duration) * 100 # Only flag if above threshold and above noise floor if slippage_pct < threshold_pct or days_over < min_slippage_days: continue daily_cost = self._estimate_daily_cost(proj, planned_duration) slipped_projects[f"project_{proj.id}"] = { "project_id": proj.id, "project_name": proj.name, "status": proj.status, "start_date": start.isoformat(), "planned_end": planned_end.isoformat(), "actual_end": actual_end.isoformat(), "planned_duration_days": round(planned_duration, 1), "actual_duration_days": round(actual_duration, 1), "days_behind": round(days_over, 1), "slippage_pct": round(slippage_pct, 1), "estimated_daily_cost": daily_cost, "estimated_margin_impact": round(days_over * daily_cost, 2), "assigned_to": proj.assigned_to, "budget": proj.budget, "actual_cost": proj.actual_cost, } # -- 2. Check Jobber jobs if connected ------------------------------ if connector_available(company_id, "jobber"): jobber_jobs = JobberJob.query.filter( JobberJob.company_id == company_id, JobberJob.estimated_end_date.isnot(None), ).all() for job in jobber_jobs: planned_end = as_aware(job.estimated_end_date) if planned_end is None: continue actual_end = as_aware(job.actual_end_date) if actual_end is None: continue start_date = as_aware(job.start_date) or as_aware(job.created_at) if start_date is None: continue planned_duration = (planned_end - start_date).total_seconds() / 86400.0 if planned_duration <= 0: continue actual_duration = (actual_end - start_date).total_seconds() / 86400.0 days_over = actual_duration - planned_duration if days_over <= 0: continue slippage_pct = (days_over / planned_duration) * 100 if slippage_pct < threshold_pct or days_over < min_slippage_days: continue job_budget = job.total_amount or 0.0 daily_cost = round(job_budget / planned_duration, 2) if planned_duration > 0 else 0 key = f"jobber_{job.external_id}" if key not in slipped_projects: slipped_projects[key] = { "project_id": job.external_id, "project_name": job.job_name or f"Jobber #{job.external_id}", "status": job.status, "source": "jobber", "start_date": start_date.isoformat(), "planned_end": planned_end.isoformat(), "actual_end": actual_end.isoformat(), "planned_duration_days": round(planned_duration, 1), "actual_duration_days": round(actual_duration, 1), "days_behind": round(days_over, 1), "slippage_pct": round(slippage_pct, 1), "estimated_daily_cost": daily_cost, "estimated_margin_impact": round(days_over * daily_cost, 2), "budget": job_budget, } # -- 3. Emit candidates ---------------------------------------------- if not slipped_projects: return [] # Aggregate totals total_slippage_days = round( sum(p.get("days_behind", 0) for p in slipped_projects.values()), 1 ) total_margin_impact = round( sum(p.get("estimated_margin_impact", 0) for p in slipped_projects.values()), 2 ) at_risk_count = sum( 1 for p in slipped_projects.values() if p.get("at_risk", False) ) slipped_count = len(slipped_projects) - at_risk_count # Severity based on total margin impact if total_margin_impact > 50000: severity = "critical" elif total_margin_impact > 15000: severity = "high" elif total_margin_impact > 3000: severity = "medium" else: severity = "low" # Aggregate leak candidate (weekly snapshot) project_names = ", ".join( p.get("project_name", "Unknown") for p in list(slipped_projects.values())[:5] ) extra = "" if len(slipped_projects) > 5: extra = f" and {len(slipped_projects) - 5} more" desc_parts = [] if slipped_count > 0: desc_parts.append( f"{slipped_count} project(s) behind schedule " f"(>{threshold_pct}% slippage): " f"{total_slippage_days} total days late, " f"~${total_margin_impact:,.2f} estimated margin impact" ) if at_risk_count > 0: desc_parts.append( f"{at_risk_count} project(s) at risk of missing deadline" ) candidates.append(LeakCandidate( detector_id=self.id, source="Project Schedule Variance", description=" | ".join(desc_parts) + f". Affected: {project_names}{extra}.", estimated_loss=total_margin_impact, severity=severity, metadata_json={ "dedupe_key": f"{self.id}:weekly_snapshot", "slipped_count": slipped_count, "at_risk_count": at_risk_count, "total_projects_checked": len(recent_projects), "total_slippage_days": total_slippage_days, "total_margin_impact": total_margin_impact, "threshold_pct": threshold_pct, "projects": slipped_projects, }, source_connector=None, rule_params=self.default_params, )) # Individual project candidates (for drill-down) for key, proj_info in slipped_projects.items(): proj_name = proj_info.get("project_name", "Unknown") slippage_pct = proj_info.get("slippage_pct", 0) days_behind = proj_info.get("days_behind", 0) margin_impact = proj_info.get("estimated_margin_impact", 0) if proj_info.get("at_risk"): proj_severity = "medium" desc = ( f"Project '{proj_name}' — approaching deadline " f"({proj_info.get('days_remaining', 0)} days left) " f"with {proj_info.get('progress_ratio', 0) * 100:.0f}% elapsed. " f"At risk of slippage." ) else: proj_severity = ( "high" if margin_impact > 10000 else "medium" if margin_impact > 2000 else "low" ) planned_end_str = proj_info.get("planned_end", "").split("T")[0] if proj_info.get("planned_end") else "" desc = ( f"Project '{proj_name}' — planned completion {planned_end_str}, " f"now {days_behind:.0f} days behind ({slippage_pct:.1f}% slippage). " f"Estimated margin impact: ${margin_impact:,.2f}." ) candidates.append(LeakCandidate( detector_id=self.id, source=f"Schedule Variance: {proj_name}", description=desc, estimated_loss=margin_impact if not proj_info.get("at_risk") else None, severity=proj_severity, metadata_json={ "dedupe_key": f"{self.id}:{key}", "project_key": key, **proj_info, }, source_connector=proj_info.get("source"), rule_params=self.default_params, )) return candidates @staticmethod def _estimate_daily_cost(project: Any, planned_duration_days: float) -> float: """Estimate daily cost of a project based on budget or actual_cost.""" if planned_duration_days <= 0: return 0.0 # Prefer actual_cost if available (more accurate for ongoing projects) cost_basis = getattr(project, "actual_cost", None) if cost_basis and cost_basis > 0: return round(cost_basis / planned_duration_days, 2) if project.budget and project.budget > 0: return round(project.budget / planned_duration_days, 2) # Fallback: estimate $500/day overhead for typical remodeling project return 500.0