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