"""Remediation service — generates fix suggestions for detected leaks.
Each detector defines its own remediation templates. The engine picks the
best match based on severity, loss, and historical context.
Public API:
get_remediation_suggestions(leak) -> list of suggestion dicts
apply_suggestion(leak_id, suggestion) -> upsert OptimizationMove
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
import logging
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Remediation templates per detector type
# ---------------------------------------------------------------------------
# Each entry maps a detector_id to a list of suggestion templates.
# Templates are dicts with:
# - title: short action label
# - description: what to do and why
# - priority: high|medium|low
# - effort: quick|moderate|complex
# - impact: expected improvement range (e.g. '10-25% recovery')
# - move_type: OptimizationMove type if applicable
# - conditions: optional filter on leak fields (severity, estimated_loss)
REMEDIATION_TEMPLATES: Dict[str, List[Dict[str, Any]]] = {
# -- QuickBooks detectors --
"qb_overdue_30d": [
{
"title": "Send Past-Due Invoice Reminders",
"description": (
"Automate follow-up emails for invoices past due. "
"Use QuickBooks built-in reminders or a third-party "
"dunning service for higher recovery rates."
),
"priority": "high",
"effort": "quick",
"impact": "5-15% faster collection",
"move_type": "channel_change",
},
{
"title": "Implement Automatic Payment Plans",
"description": (
"Offer customers installment plans for large overdue "
"balances to recover partial revenue immediately."
),
"priority": "medium",
"effort": "moderate",
"impact": "20-40% of overdue balance recovered",
},
],
"qb_overdue_60d": [
{
"title": "Escalate to Collections Process",
"description": (
"Invoices over 60 days need aggressive follow-up. "
"Consider engaging a collections agency for amounts "
"over $5,000."
),
"priority": "critical",
"effort": "moderate",
"impact": "30-50% recovery on aging debt",
},
],
"qb_draft": [
{
"title": "Convert Draft Invoices to Sent",
"description": (
"Draft invoices represent completed work that hasn't "
"been billed. Review and send all drafts older than "
"7 days."
),
"priority": "high",
"effort": "quick",
"impact": "Immediate revenue recognition",
"move_type": "budget_shift",
},
],
# -- CRM detectors --
"crm_stale_deal_14d": [
{
"title": "Re-engage Stale Deals",
"description": (
"Deals with no activity for 14+ days are at risk. "
"Send personalized follow-up emails referencing the "
"last conversation point."
),
"priority": "medium",
"effort": "quick",
"impact": "15-30% re-engagement rate",
},
{
"title": "Set Automated Follow-Up Reminders",
"description": (
"Configure CRM to trigger follow-up tasks when deals "
"go inactive for 7 days. Prevents future staleness."
),
"priority": "medium",
"effort": "moderate",
"impact": "25% reduction in stale deals",
},
],
"crm_stale_deal_30d": [
{
"title": "Executive Outreach for Stale Deals",
"description": (
"Deals stuck for 30+ days likely need decision-maker "
"intervention. Have a senior rep or exec reach out "
"directly to the prospect's champion."
),
"priority": "high",
"effort": "moderate",
"impact": "20-35% conversion of stale deals",
},
],
"crm_slipped_close": [
{
"title": "Reschedule Missed Close Dates",
"description": (
"Deals past their target close date need immediate "
"attention. Re-set close dates with commitment "
"milestones."
),
"priority": "high",
"effort": "quick",
"impact": "10-20% pipeline recovery",
},
],
"crm_orphaned": [
{
"title": "Reassign Orphaned Deals",
"description": (
"Orphaned deals have no owner. Reassign to active "
"reps or move to a shared pool for team pickup."
),
"priority": "high",
"effort": "quick",
"impact": "40-60% of orphaned deals recoverable",
},
],
# -- Ads detectors --
"ads_zero_conv_14d": [
{
"title": "Pause or Restructure Underperforming Ads",
"description": (
"Ads with zero conversions in 14+ days are burning "
"budget. Pause and restructure with new creative or "
"targeting."
),
"priority": "high",
"effort": "quick",
"impact": "Immediate budget savings",
"move_type": "budget_shift",
},
{
"title": "Redirect Budget to Winning Campaigns",
"description": (
"Move spend from zero-conversion campaigns to your "
"top-performing campaigns to maximize ROI."
),
"priority": "high",
"effort": "quick",
"impact": "15-30% ROAS improvement",
"move_type": "budget_shift",
},
],
"ads_zero_conv_30d": [
{
"title": "Kill Long-Term Zero-Conversion Campaigns",
"description": (
"Campaigns with zero conversions for 30+ days are "
"pure waste. Shut them down and reallocate 100% "
"of the budget."
),
"priority": "critical",
"effort": "quick",
"impact": "100% of wasted spend recovered",
"move_type": "budget_shift",
},
],
# -- Leads detectors --
"leads_unworked_24h": [
{
"title": "Assign Unworked Leads Immediately",
"description": (
"Leads older than 24h without contact are cooling "
"off. Assign to available reps now — response time "
"is the #1 predictor of conversion."
),
"priority": "high",
"effort": "quick",
"impact": "5x higher conversion if contacted within 1h",
},
],
# -- Speed to lead --
"speed_to_lead_slow": [
{
"title": "Reduce First Response Time",
"description": (
"Your team takes too long to respond to inbound "
"leads. Implement automated SMS/email responses "
"while waiting for a human."
),
"priority": "high",
"effort": "moderate",
"impact": "2-3x conversion lift with <15min response",
},
],
# -- Phase 2 behavioral --
"price_drop_pattern": [
{
"title": "Stop Discounting — Add Value Instead",
"description": (
"Repeated price drops train customers to wait for "
"discounts. Replace with value-add bundles instead."
),
"priority": "high",
"effort": "moderate",
"impact": "10-20% margin improvement",
},
],
"payment_decline_loop": [
{
"title": "Update Payment Methods on Decline",
"description": (
"Repeated payment declines suggest expired cards. "
"Send automatic card-update reminders instead of "
"retrying the same payment."
),
"priority": "high",
"effort": "quick",
"impact": "60% of declines are fixable with new card info",
},
],
"seasonal_dip": [
{
"title": "Pre-Season Campaign Planning",
"description": (
"Historical seasonal dips detected. Launch "
"counter-seasonal offers or push booking ahead "
"of the slow period."
),
"priority": "medium",
"effort": "moderate",
"impact": "10-20% reduction in seasonal variance",
},
],
"high_refund_rate": [
{
"title": "Analyze and Fix Refund Root Causes",
"description": (
"Refund rate above 15% signals product or "
"expectation mismatch. Review refund reasons "
"and fix the top cause."
),
"priority": "critical",
"effort": "complex",
"impact": "20-40% refund reduction possible",
},
],
# -- Phase 3 cross-connector --
"channel_quality_low": [
{
"title": "Shift Budget from Low-Quality Channels",
"description": (
"This channel generates leads that don't convert. "
"Reduce spend by 50% and redirect to higher-quality "
"sources."
),
"priority": "high",
"effort": "quick",
"impact": "15-25% ROAS improvement",
"move_type": "budget_shift",
},
],
"sms_response_time_slow": [
{
"title": "Improve SMS Response SLA",
"description": (
"SMS response time exceeding 30 minutes. Set up "
"auto-replies and alert on-call staff when "
"response time exceeds 10 minutes."
),
"priority": "high",
"effort": "moderate",
"impact": "2x engagement with <10min response",
},
],
"forecast_gap_large": [
{
"title": "Address Revenue Forecast Gap",
"description": (
"Significant gap between forecasted and actual "
"revenue. Review pipeline accuracy, close date "
"realism, and win rates."
),
"priority": "critical",
"effort": "moderate",
"impact": "More accurate forecasting = better decisions",
},
],
"customer_churn_risk_high": [
{
"title": "Proactive Retention Outreach",
"description": (
"At-risk customers identified. Launch targeted "
"retention campaign: check-in calls, special "
"offers, or success plan reviews."
),
"priority": "critical",
"effort": "moderate",
"impact": "15-30% churn reduction on targeted accounts",
},
],
"budget_misallocation": [
{
"title": "Reallocate Budget Based on ROI",
"description": (
"Budget allocation doesn't match channel ROI. "
"Shift 20-40% from low-ROI channels to "
"high-performing ones."
),
"priority": "high",
"effort": "quick",
"impact": "20-35% overall ROAS improvement",
"move_type": "budget_shift",
},
],
}
# Default fallback for unknown detectors
DEFAULT_SUGGESTIONS: List[Dict[str, Any]] = [
{
"title": "Review and Address This Leak",
"description": (
"This leak pattern doesn't have a pre-defined fix. "
"Review the details and determine the best action."
),
"priority": "medium",
"effort": "moderate",
"impact": "Variable",
},
{
"title": "Set Up Monitoring to Prevent Recurrence",
"description": (
"Once addressed, configure alerts to catch this "
"pattern earlier next time."
),
"priority": "medium",
"effort": "quick",
"impact": "Prevents future occurrences",
},
]
def get_remediation_suggestions(leak: Dict[str, Any]) -> List[Dict[str, Any]]:
"""Get fix suggestions for a detected leak.
Args:
leak: Revenue leak dict with detector_id, severity, estimated_loss, etc.
Returns:
List of suggestion dicts sorted by priority.
"""
detector_id = leak.get("detector_id", "")
severity = leak.get("severity", "medium")
templates = REMEDIATION_TEMPLATES.get(
detector_id, DEFAULT_SUGGESTIONS
)
# Filter by severity — critical leaks get all suggestions,
# others get suggestions matching their priority level
priority_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}
leak_priority = priority_order.get(severity, 2)
suggestions = []
for t in templates:
suggestion = {
"title": t["title"],
"description": t["description"],
"priority": t.get("priority", "medium"),
"effort": t.get("effort", "moderate"),
"impact": t.get("impact", "Variable"),
"move_type": t.get("move_type"),
"source": "auto_remediation",
}
suggestions.append(suggestion)
# Sort by priority
suggestions.sort(
key=lambda s: priority_order.get(s["priority"], 2)
)
return suggestions
def apply_suggestion(
company_id: str,
leak_id: str,
suggestion: Dict[str, Any],
user_id: Optional[str] = None,
) -> Optional[str]:
"""Apply a remediation suggestion by creating an OptimizationMove.
Returns the new OptimizationMove id, or None if creation failed.
"""
from app.models import db, OptimizationMove
move = OptimizationMove(
company_id=company_id,
move_type=suggestion.get("move_type") or "budget_shift",
description=(
f"[Remediation] {suggestion['title']}\n\n"
f"{suggestion['description']}\n\n"
f"Priority: {suggestion.get('priority', 'medium')}\n"
f"Effort: {suggestion.get('effort', 'moderate')}\n"
f"Expected impact: {suggestion.get('impact', 'Variable')}"
),
status="recommended",
)
# Store leak reference in metadata
if not hasattr(move, "metadata_json"):
# Add if not already there
pass
db.session.add(move)
db.session.commit()
logger.info(
"Created optimization move %s for leak %s",
move.id,
leak_id,
)
return move.id