"""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