"""Detector: Margin Per Job Tracking.
Compares estimated costs to actual costs per project. Flags jobs where
margin falls below a configurable threshold (default: 15%) — critical
for remodeling ops where each project's P&L determines whether the
company actually made money on the work.
Data sources:
- Project: budget (estimated revenue), actual_cost, revenue
- Estimate: estimate_amount, estimated_cost, estimated_margin (P2)
- QuickbooksExpense: expenses synced from QB Online (cross-reference)
Works with zero, one, or multiple connectors — falls back gracefully
to local Project data when no accounting 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 ProjectMarginLeakDetector(BaseDetector):
id = "project_margin_leak"
name = "Margin Per Job Tracking"
connector = None # cross-connector / composite
enabled_by_default = True
description = (
"Compares estimated vs. actual cost per project. Flags jobs with "
"margin below threshold (default: 15%). Tracks margin by project type "
"and aggregates negative-margin projects — margin protection for "
"design-build and remodeling ops."
)
default_params = {
"margin_threshold_pct": 15,
"min_project_value": 5000,
"lookback_days": 365,
"include_active": False, # If True, also check active projects (estimated)
}
def check(self, company_id: str) -> List[LeakCandidate]:
from app.models import Project, QuickbooksExpense, Estimate
now = utcnow()
cutoff = now - timedelta(days=self.default_params.get("lookback_days", 365))
threshold_pct = self.default_params.get("margin_threshold_pct", 15)
min_value = self.default_params.get("min_project_value", 5000)
include_active = self.default_params.get("include_active", False)
candidates: List[LeakCandidate] = []
# -- 1. Check completed projects with actual cost data --------------
project_query = Project.query.filter(
Project.company_id == company_id,
Project.actual_cost.isnot(None),
Project.actual_cost > 0,
)
if include_active:
project_query = project_query.filter(
Project.status.in_(["active", "planning", "completed", "on_hold"])
)
else:
project_query = project_query.filter(
Project.status == "completed"
)
projects = project_query.all()
# Filter to lookback window
recent_projects = []
for p in projects:
completed = as_aware(p.completed_date)
if completed is not None and completed >= cutoff:
recent_projects.append(p)
elif p.status != "completed":
# Active projects always included if include_active
if include_active:
recent_projects.append(p)
else:
# Fallback: use updated_at
updated = as_aware(p.updated_at)
if updated is not None and updated >= cutoff:
recent_projects.append(p)
low_margin_projects: Dict[str, Dict[str, Any]] = {}
for proj in recent_projects:
actual_cost = proj.actual_cost or 0.0
# Use revenue if available, else budget as the revenue proxy
revenue = proj.revenue or proj.budget or 0.0
# Skip if no revenue data
if revenue <= 0:
continue
# Skip if below min project value
if revenue < min_value:
continue
# Calculate margin
gross_profit = revenue - actual_cost
margin_pct = (gross_profit / revenue) * 100
# Determine if this is a low-margin project
is_low_margin = margin_pct < threshold_pct
# Also flag projects where cost exceeded estimate (budget)
budget = proj.budget or revenue
cost_variance = 0.0
cost_variance_pct = 0.0
if budget > 0:
cost_variance = actual_cost - (budget * (100 - threshold_pct) / 100)
cost_variance_pct = ((actual_cost - budget) / budget) * 100
if not is_low_margin:
continue
# Categorize severity
if margin_pct < 0:
proj_severity = "critical" # Losing money
elif margin_pct < threshold_pct * 0.5:
proj_severity = "high"
else:
proj_severity = "medium"
low_margin_projects[f"project_{proj.id}"] = {
"project_id": proj.id,
"project_name": proj.name,
"status": proj.status,
"project_type": "general",
"budget": round(budget, 2),
"revenue": round(revenue, 2),
"actual_cost": round(actual_cost, 2),
"gross_profit": round(gross_profit, 2),
"margin_pct": round(margin_pct, 1),
"target_margin_pct": threshold_pct,
"margin_shortfall_pct": round(threshold_pct - margin_pct, 1),
"cost_variance_vs_budget": round(actual_cost - budget, 2) if budget else None,
"assigned_to": proj.assigned_to,
"start_date": proj.start_date.isoformat() if proj.start_date else None,
"completed_date": proj.completed_date.isoformat() if proj.completed_date else None,
}
# -- 2. Cross-reference with Estimate data if available -------------
# Updated for new Estimate schema (total_value, estimated_cost, estimated_margin)
estimates = Estimate.query.filter(
Estimate.company_id == company_id,
Estimate.estimated_cost.isnot(None),
Estimate.estimated_cost > 0,
Estimate.stage.in_(["accepted", "scheduled", "closed"]),
).all()
for est in estimates:
estimate_value = est.total_value or 0.0
estimated_cost = est.estimated_cost or 0.0
estimated_margin = est.estimated_margin or 0.0
if estimate_value <= 0:
continue
# If estimate has explicit margin, use it; otherwise calculate
if estimated_margin is None and estimate_value > 0:
estimated_margin = ((estimate_value - estimated_cost) / estimate_value) * 100
# Try to match estimate to project
matched_proj = None
for proj in recent_projects:
value_ratio = abs((proj.revenue or proj.budget or 0) - estimate_value) / estimate_value if estimate_value > 0 else 1
if value_ratio < 0.2:
matched_proj = proj
break
if matched_proj:
key = f"project_{matched_proj.id}"
if key in low_margin_projects:
continue
if estimated_margin < threshold_pct and estimate_value >= min_value:
key = f"estimate_{est.id}"
low_margin_projects[key] = {
"project_id": est.id,
"project_name": est.prospect_name or f"Estimate {est.estimate_number}",
"status": est.stage,
"project_type": est.project_type or "general",
"budget": round(estimate_value, 2),
"revenue": round(estimate_value, 2),
"actual_cost": round(estimated_cost, 2),
"gross_profit": round(estimate_value - estimated_cost, 2),
"margin_pct": round(estimated_margin, 1),
"target_margin_pct": threshold_pct,
"margin_shortfall_pct": round(threshold_pct - estimated_margin, 1),
"source": "estimate",
}
# -- 3. Cross-reference with QuickBooks expenses --------------------
if connector_available(company_id, "quickbooks"):
qb_expenses = QuickbooksExpense.query.filter(
QuickbooksExpense.company_id == company_id,
QuickbooksExpense.tx_date.isnot(None),
QuickbooksExpense.tx_date >= cutoff,
QuickbooksExpense.amount.isnot(None),
).all()
# Aggregate expenses by category/date range for trend analysis
total_qb_expenses = round(
sum(e.amount or 0 for e in qb_expenses), 2
)
# If we have low margin projects, add QB expense context
if low_margin_projects:
# Group expenses by category for reporting
expenses_by_category: Dict[str, float] = {}
for exp in qb_expenses:
cat = exp.category or "uncategorized"
expenses_by_category[cat] = round(
(expenses_by_category.get(cat, 0) or 0) + (exp.amount or 0), 2
)
# Add this context to the aggregate candidate (below)
low_margin_projects["_qb_context"] = {
"total_expenses": total_qb_expenses,
"by_category": expenses_by_category,
"lookback_days": self.default_params.get("lookback_days", 365),
}
# -- 4. Emit candidates ----------------------------------------------
if not low_margin_projects or (len(low_margin_projects) == 1 and "_qb_context" in low_margin_projects):
# Filter out _qb_context from count
real_projects = {k: v for k, v in low_margin_projects.items() if not k.startswith("_")}
if not real_projects:
return []
# Separate real projects from context data
real_projects = {k: v for k, v in low_margin_projects.items() if not k.startswith("_")}
context_data = {k: v for k, v in low_margin_projects.items() if k.startswith("_")}
if not real_projects:
return []
# Aggregate totals
total_revenue = round(
sum(p.get("revenue", 0) for p in real_projects.values()), 2
)
total_cost = round(
sum(p.get("actual_cost", 0) for p in real_projects.values()), 2
)
total_profit = round(total_revenue - total_cost, 2)
aggregate_margin = round((total_profit / total_revenue) * 100, 1) if total_revenue > 0 else 0
# Count by severity
critical_count = sum(
1 for p in real_projects.values()
if p.get("margin_pct", 0) < 0
)
high_count = sum(
1 for p in real_projects.values()
if 0 <= p.get("margin_pct", 0) < threshold_pct * 0.5
)
medium_count = len(real_projects) - critical_count - high_count
# Rollup by project type
margin_by_type: Dict[str, Dict[str, Any]] = {}
for proj_info in real_projects.values():
ptype = proj_info.get("project_type", "general")
if ptype not in margin_by_type:
margin_by_type[ptype] = {
"count": 0,
"total_revenue": 0,
"total_cost": 0,
"avg_margin": 0,
}
margin_by_type[ptype]["count"] += 1
margin_by_type[ptype]["total_revenue"] += proj_info.get("revenue", 0) or 0
margin_by_type[ptype]["total_cost"] += proj_info.get("actual_cost", 0) or 0
# Calculate average margin per type
for ptype, data in margin_by_type.items():
if data["total_revenue"] > 0:
data["avg_margin"] = round(
((data["total_revenue"] - data["total_cost"]) / data["total_revenue"]) * 100, 1
)
data["total_revenue"] = round(data["total_revenue"], 2)
data["total_cost"] = round(data["total_cost"], 2)
# Overall severity
if critical_count > 0:
severity = "critical"
elif high_count > 2:
severity = "high"
elif len(real_projects) > 0:
severity = "medium"
else:
severity = "low"
# Aggregate leak candidate (weekly snapshot)
project_names = ", ".join(
p.get("project_name", "Unknown")
for p in list(real_projects.values())[:5]
)
extra = ""
if len(real_projects) > 5:
extra = f" and {len(real_projects) - 5} more"
losing_money = []
for p in real_projects.values():
if p.get("margin_pct", 0) < 0:
losing_money.append(p.get("project_name", "Unknown"))
desc_parts = []
desc_parts.append(
f"{len(real_projects)} project(s) with margin below "
f"{threshold_pct}% target. Aggregate margin: {aggregate_margin:.1f}% "
f"(${total_profit:,.2f} profit on ${total_revenue:,.2f} revenue)"
)
if losing_money:
desc_parts.append(
f"{len(losing_money)} project(s) losing money: {', '.join(losing_money)}"
)
candidates.append(LeakCandidate(
detector_id=self.id,
source="Margin Per Job Tracking",
description=" | ".join(desc_parts) + f". Affected: {project_names}{extra}.",
estimated_loss=abs(total_profit) if total_profit < 0 else None,
severity=severity,
metadata_json={
"dedupe_key": f"{self.id}:weekly_snapshot",
"low_margin_count": len(real_projects),
"critical_count": critical_count,
"high_count": high_count,
"medium_count": medium_count,
"total_revenue": total_revenue,
"total_cost": total_cost,
"total_profit": total_profit,
"aggregate_margin_pct": aggregate_margin,
"margin_threshold_pct": threshold_pct,
"margin_by_type": margin_by_type,
"projects": {k: v for k, v in real_projects.items()},
},
source_connector=None,
rule_params=self.default_params,
))
# Add QB context if available
if context_data:
for key, ctx in context_data.items():
candidates[-1].metadata_json[key] = ctx
# Individual project candidates (for drill-down)
for key, proj_info in real_projects.items():
proj_name = proj_info.get("project_name", "Unknown")
margin_pct = proj_info.get("margin_pct", 0)
revenue = proj_info.get("revenue", 0)
cost = proj_info.get("actual_cost", 0)
profit = proj_info.get("gross_profit", 0)
shortfall = proj_info.get("margin_shortfall_pct", 0)
if margin_pct < 0:
proj_severity = "critical"
desc = (
f"Project '{proj_name}' — LOSING MONEY. "
f"Revenue ${revenue:,.2f}, cost ${cost:,.2f}. "
f"Negative margin: {margin_pct:.1f}%. "
f"Lost ${abs(profit):,.2f} on this job."
)
estimated_loss = abs(profit)
else:
proj_severity = (
"high" if margin_pct < threshold_pct * 0.5 else "medium"
)
desc = (
f"Project '{proj_name}' — margin {margin_pct:.1f}% "
f"(target: {threshold_pct}%). "
f"Revenue ${revenue:,.2f}, cost ${cost:,.2f}, "
f"profit ${profit:,.2f}. {shortfall:.1f}% below target."
)
estimated_loss = None
candidates.append(LeakCandidate(
detector_id=self.id,
source=f"Margin Leak: {proj_name}",
description=desc,
estimated_loss=estimated_loss,
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