"""
Multi-Location Rollup Detector
Aggregates metrics across company locations to identify:
- Revenue concentration risk (single location >50% of total)
- Location revenue variance (top vs bottom performers)
- Underperforming locations (below company average by threshold)
- Location-level project health rollup
Works with both formal Location model records and denormalized
location_id/location_name fields on RevenueRecord.
"""
from datetime import datetime, timezone, timedelta
from typing import List, Dict, Any, Optional
from sqlalchemy import func
from .base import BaseDetector, LeakCandidate
from ...models import db, Company, Location, RevenueRecord, Project, Estimate
class MultiLocationRollupDetector(BaseDetector):
"""Analyze multi-location performance, concentration risk, and cross-location variance."""
id = "multi_location_rollup"
name = "Multi-Location Rollup"
connector = None
enabled_by_default = True
description = (
"Aggregates metrics per location, flags concentration risk, "
"underperformers, and cross-location variance."
)
default_params = {
"concentration_threshold": 0.50,
"underperformer_threshold": 0.80,
"min_locations": 2,
"lookback_days": 90,
}
# ---- check() — BaseDetector contract ----
def check(self, company_id: str) -> List[LeakCandidate]:
"""BaseDetector interface — returns LeakCandidate list."""
company = Company.query.get(company_id)
if not company:
return []
locations = self._get_locations(company_id)
if len(locations) < self.default_params.get("min_locations", 2):
return []
location_metrics = self._compute_location_metrics(company_id, locations)
if not location_metrics:
return []
candidates: List[LeakCandidate] = []
# Concentration findings
conc_findings = self._check_concentration(location_metrics)
for f in conc_findings:
candidates.append(LeakCandidate(
detector_id=self.id,
source=f["source"],
description=f["description"],
estimated_loss=f.get("estimated_impact"),
severity=f["severity"],
metadata_json={
"dedupe_key": f"{self.id}:concentration:{f['metadata_json']['location_id']}",
**f["metadata_json"],
},
source_connector=None,
rule_params=self.default_params,
))
# Underperformer findings
under_findings = self._check_underperformers(location_metrics, company)
for f in under_findings:
candidates.append(LeakCandidate(
detector_id=self.id,
source=f["source"],
description=f["description"],
estimated_loss=f.get("estimated_impact"),
severity=f["severity"],
metadata_json={
"dedupe_key": f"{self.id}:underperformer:{f['metadata_json']['location_id']}",
**f["metadata_json"],
},
source_connector=None,
rule_params=self.default_params,
))
# Variance findings
var_findings = self._check_variance(location_metrics)
for f in var_findings:
candidates.append(LeakCandidate(
detector_id=self.id,
source=f["source"],
description=f["description"],
estimated_loss=f.get("estimated_impact"),
severity=f["severity"],
metadata_json={
"dedupe_key": f"{self.id}:variance",
**f["metadata_json"],
},
source_connector=None,
rule_params=self.default_params,
))
return candidates
# ---- Legacy run() — returns raw findings dicts (for API rollup) ----
def run(self, company_id: str) -> List[Dict[str, Any]]:
"""Legacy/convenience method — returns raw dict findings for API use."""
findings: List[Dict[str, Any]] = []
company = Company.query.get(company_id)
if not company:
return findings
# Gather locations from both formal model and denormalized revenue records
locations = self._get_locations(company_id)
if len(locations) < self.default_params.get("min_locations", 2):
# Not enough locations to do meaningful comparison
return findings
# Compute per-location metrics
location_metrics = self._compute_location_metrics(company_id, locations)
if not location_metrics:
return findings
# Run checks
findings.extend(self._check_concentration(location_metrics))
findings.extend(self._check_underperformers(location_metrics, company))
findings.extend(self._check_variance(location_metrics))
findings.extend(self._check_project_health(company_id, locations))
return findings
# ---- Location Resolution ----
def _get_locations(self, company_id: str) -> List[Location]:
"""Get locations from both formal model and denormalized RevenueRecord data."""
# Start with formal Location records
formal = Location.query.filter_by(
company_id=company_id,
is_active=True,
).all()
location_map: Dict[str, Location] = {}
for loc in formal:
key = loc.id if loc.id else loc.code or loc.name
location_map[key] = loc
# Also extract unique locations from RevenueRecord denormalized fields
revenue_locations = db.session.query(
func.distinct(RevenueRecord.location_id)
).filter(
RevenueRecord.company_id == company_id,
RevenueRecord.location_id.isnot(None),
RevenueRecord.location_id != '',
).all()
for (loc_id,) in revenue_locations:
if loc_id not in location_map:
# Create a synthetic Location from denormalized data
rr = RevenueRecord.query.filter_by(
company_id=company_id,
location_id=loc_id,
).first()
if rr:
synthetic = Location(
id=loc_id,
company_id=company_id,
name=rr.location_name or f"Location {loc_id}",
code=loc_id,
is_active=True,
)
location_map[loc_id] = synthetic
# If still no formal locations, create one "HQ" bucket for unassigned records
if not location_map:
return []
return list(location_map.values())
# ---- Metric Computation ----
def _compute_location_metrics(self, company_id: str, locations: List[Location]) -> List[Dict[str, Any]]:
"""Compute revenue, project count, and estimate metrics per location."""
now = datetime.now(timezone.utc)
start = now - timedelta(days=self.default_params.get("lookback_days", 90))
metrics: List[Dict[str, Any]] = []
for loc in locations:
loc_id = loc.id
# Revenue from RevenueRecord
revenue_result = (
db.session.query(
func.coalesce(func.sum(RevenueRecord.amount), 0.0),
func.count(RevenueRecord.id),
)
.filter(
RevenueRecord.company_id == company_id,
RevenueRecord.location_id == loc_id,
RevenueRecord.transaction_date >= start,
RevenueRecord.status == 'completed',
)
.first()
)
total_revenue = float(revenue_result[0]) if revenue_result else 0.0
transaction_count = revenue_result[1] if revenue_result else 0
# Projects linked to this location via custom_fields or denormalized
project_count = Project.query.filter_by(
company_id=company_id,
status='active',
).count()
# For now, projects without location_id are company-wide
# TODO: Add location_id to Project model when schema migration is available
# Estimates for this location
estimate_count = Estimate.query.filter_by(
company_id=company_id,
).count()
metrics.append({
"location_id": loc_id,
"location_name": loc.name,
"location_code": getattr(loc, 'code', ''),
"total_revenue": total_revenue,
"transaction_count": transaction_count,
"project_count": project_count,
"estimate_count": estimate_count,
"is_formal": bool(hasattr(loc, 'address') and loc.address),
})
return metrics
# ---- Concentration Check ----
def _check_concentration(self, metrics: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Flag if any single location accounts for >threshold of total revenue."""
findings: List[Dict[str, Any]] = []
total_revenue = sum(m["total_revenue"] for m in metrics)
if total_revenue == 0:
return findings
concentration_threshold = self.default_params.get("concentration_threshold", 0.50)
for m in metrics:
share = m["total_revenue"] / total_revenue
if share > concentration_threshold:
findings.append({
"detector_id": self.id,
"source": "multi_location_concentration",
"description": (
f'Location "{m["location_name"]}" accounts for '
f'{share * 100:.1f}% of total revenue '
f'(${m["total_revenue"]:,.2f} of ${total_revenue:,.2f}). '
f'Threshold: {concentration_threshold * 100:.0f}%.'
),
"estimated_impact": total_revenue * share * 0.10,
"severity": self._concentration_severity(share),
"metadata_json": {
"location_id": m["location_id"],
"location_name": m["location_name"],
"revenue_share": round(share, 4),
"total_revenue": round(total_revenue, 2),
"location_revenue": round(m["total_revenue"], 2),
"threshold": concentration_threshold,
},
})
return findings
@staticmethod
def _concentration_severity(share: float) -> str:
if share > 0.75:
return "critical"
if share > 0.65:
return "high"
if share > 0.55:
return "medium"
return "low"
# ---- Underperformer Check ----
def _check_underperformers(
self, metrics: List[Dict[str, Any]], company: Company
) -> List[Dict[str, Any]]:
"""Flag locations performing below company average by threshold."""
findings: List[Dict[str, Any]] = []
revenues = [m["total_revenue"] for m in metrics if m["total_revenue"] > 0]
if not revenues:
return findings
avg_revenue = sum(revenues) / len(revenues)
if avg_revenue == 0:
return findings
underperformer_threshold = self.default_params.get("underperformer_threshold", 0.80)
for m in metrics:
if m["total_revenue"] < avg_revenue * underperformer_threshold:
deficit = avg_revenue - m["total_revenue"]
findings.append({
"detector_id": self.id,
"source": "multi_location_underperformer",
"description": (
f'Location "{m["location_name"]}" is underperforming: '
f'${m["total_revenue"]:,.2f} vs avg ${avg_revenue:,.2f} '
f'(deficit: ${deficit:,.2f}).'
),
"estimated_impact": deficit,
"severity": self._underperformer_severity(m["total_revenue"] / avg_revenue),
"metadata_json": {
"location_id": m["location_id"],
"location_name": m["location_name"],
"location_revenue": round(m["total_revenue"], 2),
"average_revenue": round(avg_revenue, 2),
"deficit": round(deficit, 2),
"performance_ratio": round(m["total_revenue"] / avg_revenue, 4) if avg_revenue > 0 else 0,
},
})
return findings
@staticmethod
def _underperformer_severity(ratio: float) -> str:
if ratio < 0.30:
return "critical"
if ratio < 0.50:
return "high"
if ratio < 0.70:
return "medium"
return "low"
# ---- Variance Check ----
def _check_variance(self, metrics: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Flag high variance between top and bottom performers."""
findings: List[Dict[str, Any]] = []
revenues = sorted([m["total_revenue"] for m in metrics], reverse=True)
if len(revenues) < 2:
return findings
top = revenues[0]
bottom = revenues[-1]
if bottom == 0:
# Check if top location has revenue and bottom has none
if top > 0:
findings.append({
"detector_id": self.id,
"source": "multi_location_variance",
"description": (
f"Extreme revenue variance: top location at "
f"${top:,.2f} vs bottom at $0.00."
),
"estimated_impact": top * 0.15,
"severity": "critical",
"metadata_json": {
"top_revenue": round(top, 2),
"bottom_revenue": 0.0,
"variance_ratio": float("inf"),
"num_locations": len(revenues),
},
})
return findings
variance_ratio = top / bottom if bottom > 0 else float("inf")
if variance_ratio > 5.0:
severity = "critical"
elif variance_ratio > 3.0:
severity = "high"
elif variance_ratio > 2.0:
severity = "medium"
else:
severity = "low"
if variance_ratio > 2.0:
findings.append({
"detector_id": self.id,
"source": "multi_location_variance",
"description": (
f"Revenue variance between locations: "
f"top ${top:,.2f} vs bottom ${bottom:,.2f} "
f"(ratio: {variance_ratio:.1f}x)."
),
"estimated_impact": (top - bottom) * 0.10,
"severity": severity,
"metadata_json": {
"top_revenue": round(top, 2),
"bottom_revenue": round(bottom, 2),
"variance_ratio": round(variance_ratio, 2),
"num_locations": len(revenues),
},
})
return findings
# ---- Project Health Rollup ----
def _check_project_health(
self, company_id: str, locations: List[Location]
) -> List[Dict[str, Any]]:
"""Check project health metrics across locations.
Since Project model doesn't have location_id yet, this produces
a company-wide summary finding when multiple locations exist.
"""
findings: List[Dict[str, Any]] = []
# Count active projects and their health
active_projects = Project.query.filter_by(
company_id=company_id,
status='active',
).all()
if not active_projects:
return findings
total_budget = sum(p.budget or 0 for p in active_projects)
total_actual = sum(p.actual_cost or 0 for p in active_projects)
if total_budget > 0:
budget_utilization = total_actual / total_budget
if budget_utilization > 1.1:
findings.append({
"detector_id": self.id,
"source": "multi_location_project_health",
"description": (
f"Company-wide active projects are {budget_utilization * 100:.0f}% "
f"of budget (${total_actual:,.2f} of ${total_budget:,.2f}). "
f"{len(locations)} locations affected."
),
"estimated_impact": total_actual - total_budget,
"severity": "high" if budget_utilization > 1.2 else "medium",
"metadata_json": {
"active_projects": len(active_projects),
"total_budget": round(total_budget, 2),
"total_actual_cost": round(total_actual, 2),
"budget_utilization": round(budget_utilization, 4),
"num_locations": len(locations),
},
})
return findings
# ---- Rollup Summary (for API) ----
def get_rollup_summary(self, company_id: str) -> Dict[str, Any]:
"""Return a full rollup summary dict suitable for API responses."""
company = Company.query.get(company_id)
if not company:
return {"error": "Company not found"}
locations = self._get_locations(company_id)
metrics = self._compute_location_metrics(company_id, locations)
total_revenue = sum(m["total_revenue"] for m in metrics)
total_transactions = sum(m["transaction_count"] for m in metrics)
if metrics:
revenues = [m["total_revenue"] for m in metrics]
avg_revenue = sum(revenues) / len(revenues) if revenues else 0
max_revenue = max(revenues) if revenues else 0
min_revenue = min(revenues) if revenues else 0
variance_ratio = max_revenue / min_revenue if min_revenue > 0 else float("inf")
else:
avg_revenue = 0
max_revenue = 0
min_revenue = 0
variance_ratio = 0
findings = self.run(company_id)
return {
"company_id": company_id,
"company_name": company.name,
"num_locations": len(locations),
"lookback_days": self.default_params.get("lookback_days", 90),
"total_revenue": round(total_revenue, 2),
"total_transactions": total_transactions,
"average_location_revenue": round(avg_revenue, 2),
"max_location_revenue": round(max_revenue, 2),
"min_location_revenue": round(min_revenue, 2),
"variance_ratio": round(variance_ratio, 2) if variance_ratio != float("inf") else None,
"location_metrics": sorted(metrics, key=lambda m: m["total_revenue"], reverse=True),
"findings_count": len(findings),
"findings": findings,
}