"""Detector: Backlog Burn-Down Tracking.
Calculates a company's project backlog, weekly burn rate, and weeks of
coverage. Flags when coverage drops below a configurable threshold —
critical for storm restoration (Omnia) and remodeling ops where lead
pipeline must feed project scheduling.
Data sources:
- CrmDeal: active pipeline deals (backlog)
- Project: active/scheduled projects with budgets
- QuickbooksInvoice: completed project revenue for burn rate
Works with zero, one, or multiple connectors — falls back gracefully.
"""
from __future__ import annotations
from datetime import timedelta
from typing import List, Optional
from .base import BaseDetector, LeakCandidate
from ._common import connector_available, any_connector_available, utcnow, as_aware
class BacklogBurndownDetector(BaseDetector):
id = "backlog_burndown"
name = "Backlog Burn-Down Tracking"
connector = None # cross-connector / composite
enabled_by_default = True
description = (
"Calculates active backlog value, weekly burn rate, and weeks of "
"coverage. Alerts when coverage drops below threshold (default: 4 weeks)."
)
default_params = {
"coverage_threshold_weeks": 4,
"burn_rate_lookback_weeks": 12,
"backlog_stages": None, # None = use CRM closed_won + active projects
}
def check(self, company_id: str) -> List[LeakCandidate]:
has_any = any_connector_available(
company_id, ("servicetitan", "hubspot", "quickbooks", "jobber")
)
if not has_any:
# Still run if we have local Project data
pass
from app.models import CrmDeal, Project, QuickbooksInvoice
now = utcnow()
lookback = self.default_params.get("burn_rate_lookback_weeks", 12)
threshold = self.default_params.get("coverage_threshold_weeks", 4)
# -- 1. Calculate backlog value -----------------------------------
# Backlog = deals in active stages (not won/lost) + active/planning projects
# CRM deals in pipeline (excluding won/lost)
CLOSED_STAGES = {
"Closed Won", "Closed Lost",
"closedwon", "closedlost",
"closed_won", "closed_lost",
"won", "lost",
}
pipeline_deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.notin_(list(CLOSED_STAGES)),
CrmDeal.amount.isnot(None),
).all()
backlog_from_deals = round(
sum((d.amount or 0) for d in pipeline_deals), 2
)
# Active/planning projects
active_projects = Project.query.filter(
Project.company_id == company_id,
Project.status.in_(["active", "planning"]),
Project.budget.isnot(None),
).all()
backlog_from_projects = round(
sum((p.budget or 0) for p in active_projects), 2
)
total_backlog = round(backlog_from_deals + backlog_from_projects, 2)
active_project_count = len(active_projects)
pipeline_deal_count = len(pipeline_deals)
# -- 2. Calculate weekly burn rate --------------------------------
# Burn rate = completed project revenue over the lookback period
completed_projects = Project.query.filter(
Project.company_id == company_id,
Project.status == "completed",
Project.completed_date.isnot(None),
Project.revenue.isnot(None),
).all()
# Filter to lookback window
cutoff = now - timedelta(weeks=lookback)
recent_completed = [
p for p in completed_projects
if p.completed_date and as_aware(p.completed_date) >= cutoff
]
recent_revenue = round(
sum((p.revenue or 0) for p in recent_completed), 2
)
# Fallback: use QuickBooks invoices if we have few completed projects
if len(recent_completed) < 3 and connector_available(company_id, "quickbooks"):
invoices = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.status == "Paid",
QuickbooksInvoice.due_date.isnot(None),
QuickbooksInvoice.due_date >= cutoff,
).all()
qb_revenue = round(sum((i.total_amount or 0) for i in invoices), 2)
if qb_revenue > recent_revenue:
recent_revenue = qb_revenue
weekly_burn_rate = round(recent_revenue / max(lookback, 1), 2)
# -- 3. Calculate weeks of coverage --------------------------------
if weekly_burn_rate > 0:
weeks_coverage = round(total_backlog / weekly_burn_rate, 1)
else:
weeks_coverage = None # infinite if no burn rate
# -- 4. Determine severity and emit --------------------------------
severity = "low"
description = ""
if weeks_coverage is None:
if total_backlog > 0:
severity = "low"
description = (
f"Backlog: ${total_backlog:,.2f} across {active_project_count} "
f"active projects and {pipeline_deal_count} pipeline deals. "
f"No burn rate baseline yet (need more completed project data)."
)
else:
# No backlog at all — critical for production companies
description = (
"No active backlog. Zero scheduled projects or pipeline deals. "
"Immediate lead generation needed."
)
severity = "critical"
elif weeks_coverage < threshold:
if weeks_coverage < threshold * 0.5:
severity = "critical"
elif weeks_coverage < threshold * 0.75:
severity = "high"
else:
severity = "medium"
description = (
f"Backlog coverage: {weeks_coverage} weeks (below {threshold}-week threshold). "
f"Active backlog: ${total_backlog:,.2f} | "
f"Weekly burn rate: ${weekly_burn_rate:,.2f} | "
f"Projects: {active_project_count} active, "
f"{pipeline_deal_count} pipeline deals."
)
else:
description = (
f"Backlog coverage: {weeks_coverage} weeks (above {threshold}-week threshold). "
f"Active backlog: ${total_backlog:,.2f} | "
f"Weekly burn rate: ${weekly_burn_rate:,.2f}"
)
candidates: List[LeakCandidate] = []
candidates.append(LeakCandidate(
detector_id=self.id,
source="Backlog Burn-Down",
description=description,
estimated_loss=None if severity == "low" else None,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:weekly_snapshot",
"total_backlog": total_backlog,
"backlog_from_deals": backlog_from_deals,
"backlog_from_projects": backlog_from_projects,
"weekly_burn_rate": weekly_burn_rate,
"weeks_coverage": weeks_coverage,
"coverage_threshold_weeks": threshold,
"burn_rate_lookback_weeks": lookback,
"active_project_count": active_project_count,
"pipeline_deal_count": pipeline_deal_count,
"recent_completed_count": len(recent_completed),
"recent_revenue": recent_revenue,
},
source_connector=None,
rule_params=self.default_params,
))
# If coverage is critically low, also estimate the revenue at risk
if weeks_coverage is not None and weeks_coverage < threshold:
weeks_shortfall = threshold - weeks_coverage
revenue_at_risk = round(weekly_burn_rate * weeks_shortfall, 2)
candidates.append(LeakCandidate(
detector_id=self.id,
source="Backlog Burn-Down: Revenue At Risk",
description=(
f"Backlog is {weeks_shortfall:.1f} weeks below target. "
f"Estimated revenue at risk: ${revenue_at_risk:,.2f} "
f"over the next {threshold} weeks."
),
estimated_loss=revenue_at_risk,
severity="high" if weeks_coverage > 0 else "critical",
metadata_json={
"dedupe_key": f"{self.id}:revenue_at_risk",
"weeks_shortfall": weeks_shortfall,
"revenue_at_risk": revenue_at_risk,
"weekly_burn_rate": weekly_burn_rate,
},
source_connector=None,
rule_params=self.default_params,
))
return candidates