"""Detector: Change Order Revenue Leakage.
Compares original estimate amounts to final invoice amounts per project.
Flags projects where actual scope exceeded the estimate by more than a
configurable threshold (default: 15%) but the invoice didn't reflect the
increase — indicating uncaptured change orders.
Data sources:
- CrmDeal: original estimate amounts (from ServiceTitan/HubSpot)
- Project: estimated budget, actual cost, final revenue
- QuickbooksInvoice: final invoiced amounts per project
Typical leak: A $500k kitchen remodel ends up with $650k in work but
only $500k invoiced because the scope changes weren't captured on paper.
"""
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 ChangeOrderLeakDetector(BaseDetector):
id = "change_order_leak"
name = "Change Order Revenue Leakage"
connector = None # cross-connector
enabled_by_default = True
description = (
"Compares original estimates to final invoices per project. Flags "
"projects where scope increased but the invoice didn't — uncaptured "
"change orders costing 10-15% of project revenue."
)
default_params = {
"scope_variance_threshold_pct": 15,
"min_project_value": 5000,
"lookback_days": 365,
}
def check(self, company_id: str) -> List[LeakCandidate]:
has_crm = connector_available(company_id, "servicetitan") or \
connector_available(company_id, "hubspot") or \
connector_available(company_id, "jobber")
has_qb = connector_available(company_id, "quickbooks")
if not has_crm and not has_qb:
# Fall back to comparing Project.budget vs Project.revenue
pass
from app.models import CrmDeal, Project, QuickbooksInvoice
now = utcnow()
cutoff = now - timedelta(
days=self.default_params.get("lookback_days", 365)
)
threshold_pct = self.default_params.get("scope_variance_threshold_pct", 15)
min_value = self.default_params.get("min_project_value", 5000)
candidates: List[LeakCandidate] = []
seen_projects: Dict[str, Dict[str, Any]] = {}
# -- 1. Check completed projects for budget vs. revenue variance --
completed_projects = Project.query.filter(
Project.company_id == company_id,
Project.status == "completed",
Project.completed_date.isnot(None),
Project.budget.isnot(None),
).all()
# Filter to lookback window (naive datetime safety)
recent_completed = [
p for p in completed_projects
if p.completed_date and as_aware(p.completed_date) >= cutoff
]
for proj in recent_completed:
if (proj.budget or 0) < min_value:
continue
estimate = proj.budget or 0.0
final_revenue = proj.revenue or 0.0
# Skip if no revenue data
if final_revenue == 0:
continue
variance_pct = ((final_revenue - estimate) / estimate) * 100 if estimate > 0 else 0
# Flag if revenue is significantly LOWER than budget
# (estimate was higher than what was actually invoiced)
# OR if actual_cost >> revenue (margin squeeze from scope creep)
if variance_pct < -threshold_pct:
uncaptured = round(estimate - final_revenue, 2)
key = f"project_{proj.id}"
seen_projects[key] = {
"project_id": proj.id,
"project_name": proj.name,
"estimate": estimate,
"final_revenue": final_revenue,
"uncaptured": uncaptured,
"variance_pct": variance_pct,
}
elif proj.actual_cost and proj.actual_cost > final_revenue:
# Actual costs exceeded invoiced amount — margin leak
margin_leak = round(proj.actual_cost - final_revenue, 2)
if margin_leak > 0:
key = f"project_{proj.id}_margin"
seen_projects[key] = {
"project_id": proj.id,
"project_name": proj.name,
"estimate": estimate,
"final_revenue": final_revenue,
"actual_cost": proj.actual_cost,
"margin_leak": margin_leak,
}
# -- 2. Cross-reference CRM deals with QB invoices -----------------
if has_qb:
# Get paid invoices from the lookback window
paid_invoices = QuickbooksInvoice.query.filter(
QuickbooksInvoice.company_id == company_id,
QuickbooksInvoice.status == "Paid",
QuickbooksInvoice.tx_date.isnot(None),
QuickbooksInvoice.tx_date >= cutoff,
).all()
# Build lookup: qb_doc_id → total_amount
invoice_lookup: Dict[str, float] = {}
for inv in paid_invoices:
ext_id = inv.qb_doc_id or ""
if ext_id:
# Sum multiple invoices per qb_doc_id
invoice_lookup[ext_id] = invoice_lookup.get(ext_id, 0) + (inv.total_amount or 0)
# Check CRM deals that are won
CLOSED_WON = {"Closed Won", "closedwon", "closed_won", "won"}
won_deals = CrmDeal.query.filter(
CrmDeal.company_id == company_id,
CrmDeal.stage.in_(list(CLOSED_WON)),
CrmDeal.amount.isnot(None),
CrmDeal.amount >= min_value,
CrmDeal.updated_at >= cutoff,
).all()
for deal in won_deals:
# Try to match deal → invoice by external_id
invoiced_total = invoice_lookup.get(deal.external_id, 0)
if invoiced_total == 0:
continue
estimate = deal.amount or 0.0
variance_pct = ((invoiced_total - estimate) / estimate) * 100 if estimate > 0 else 0
if variance_pct < -threshold_pct:
uncaptured = round(estimate - invoiced_total, 2)
key = f"deal_{deal.external_id}"
seen_projects[key] = {
"source": "crm_deal",
"external_id": deal.external_id,
"project_name": deal.name,
"estimate": estimate,
"final_revenue": invoiced_total,
"uncaptured": uncaptured,
"variance_pct": variance_pct,
}
# -- 3. Emit candidates --------------------------------------------
if not seen_projects:
return []
# Aggregate total uncaptured
total_uncaptured = round(
sum(
p.get("uncaptured", p.get("margin_leak", 0))
for p in seen_projects.values()
),
2,
)
# Determine severity based on total uncaptured amount
if total_uncaptured > 100000:
severity = "critical"
elif total_uncaptured > 25000:
severity = "high"
elif total_uncaptured > 5000:
severity = "medium"
else:
severity = "low"
# Aggregate leak candidate
project_names = ", ".join(
p.get("project_name", "Unknown") for p in list(seen_projects.values())[:5]
)
extra = ""
if len(seen_projects) > 5:
extra = f" and {len(seen_projects) - 5} more"
candidates.append(LeakCandidate(
detector_id=self.id,
source="Change Order Revenue Leakage",
description=(
f"{len(seen_projects)} project(s) with uncaptured change orders: "
f"${total_uncaptured:,.2f} total. "
f"Affected: {project_names}{extra}. "
f"Original estimates exceeded final invoices by >{threshold_pct}%."
),
estimated_loss=total_uncaptured,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:monthly_aggregate",
"project_count": len(seen_projects),
"total_uncaptured": total_uncaptured,
"threshold_pct": threshold_pct,
"projects": seen_projects,
},
source_connector=None,
rule_params=self.default_params,
))
# Individual project candidates (for drill-down)
for key, proj_info in seen_projects.items():
uncaptured = proj_info.get("uncaptured", proj_info.get("margin_leak", 0))
proj_name = proj_info.get("project_name", "Unknown")
if "margin_leak" in proj_info:
proj_severity = "medium"
desc = (
f"Project '{proj_name}' — actual cost ${proj_info['actual_cost']:,.2f} "
f"exceeded invoiced ${proj_info['final_revenue']:,.2f}. "
f"Margin leak: ${uncaptured:,.2f}. Scope creep not captured on invoice."
)
else:
proj_severity = (
"high" if uncaptured > 20000 else
"medium" if uncaptured > 5000 else "low"
)
variance = proj_info.get("variance_pct", 0)
desc = (
f"Project '{proj_name}' — estimated ${proj_info['estimate']:,.2f}, "
f"invoked ${proj_info['final_revenue']:,.2f}. "
f"~${uncaptured:,.2f} ({abs(variance):.0f}%) in uncaptured change orders."
)
candidates.append(LeakCandidate(
detector_id=self.id,
source=f"Change Order Leak: {proj_name}",
description=desc,
estimated_loss=uncaptured,
severity=proj_severity,
metadata_json={
"dedupe_key": f"{self.id}:{key}",
"project_key": key,
**proj_info,
},
source_connector=None,
rule_params=self.default_params,
))
return candidates