How to Build an AI Deal Risk Alert System

Stalled deals are silent revenue killers. An ai deal risk alert system watches your pipeline 24/7 and pings you the moment a deal shows risk signals like a stalled pipeline. I built this after watching a client lose a $50K deal that sat untouched for three weeks.

What You Need Before Starting

• Python 3.8+
• CRM API access (HubSpot, Salesforce, etc.)
• Slack webhook URL or SMTP credentials
• A scheduled task runner (cron)

Step 1: Define Risk Signals

A deal is at risk when activity stops. Define your thresholds clearly.

RISK_RULES = {
    "no_activity": {"days": 7, "severity": "high"},
    "stage_stalled": {"days": 10, "severity": "medium"},
    "no_meeting": {"days": 14, "severity": "high"},
    "value_decreased": {"threshold": 0.1, "severity": "medium"},
}

Step 2: Scan the Pipeline

Pull all open deals and check each one against your risk rules.

from datetime import datetime

def scan_for_risks(deals, risk_rules):
    flagged = []
    for deal in deals:
        risks = []
        days_since_activity = (
            datetime.now() - deal["last_activity_date"]
        ).days

        if days_since_activity >= risk_rules["no_activity"]["days"]:
            risks.append({"type": "no_activity", "days": days_since_activity})

        days_in_stage = (datetime.now() - deal["stage_changed_at"]).days
        if days_in_stage >= risk_rules["stage_stalled"]["days"]:
            risks.append({"type": "stage_stalled", "days": days_in_stage})

        if risks:
            flagged.append({"deal": deal, "risks": risks})
    return flagged

Step 3: Score Risk Severity

Not all risks are equal. A $100K deal stalled for 10 days is more urgent than a $5K deal with no email for a week.

def score_risk(flagged_deal):
    deal = flagged_deal["deal"]
    base_score = 0
    for risk in flagged_deal["risks"]:
        if risk["type"] == "no_activity":
            base_score += 30
        elif risk["type"] == "stage_stalled":
            base_score += 20
    value_multiplier = min(deal["amount"] / 10000, 3.0)
    return base_score * value_multiplier

Step 4: Send Targeted Alerts

Route alerts to the right person. The deal owner gets the alert, their manager gets a summary.

import requests

def send_slack_alert(webhook_url, deal, risks, score):
    risk_text = "\n".join(
        [f"  - {r['type']}: {r.get('days', 'N/A')} days" for r in risks]
    )
    payload = {
        "text": f"*Deal Risk Alert* (Score: {score:.0f})\n"
                f"*{deal['name']}* - ${deal['amount']:,.0f}\n"
                f"Owner: {deal['owner']}\nRisks:\n{risk_text}"
    }
    requests.post(webhook_url, json=payload)

Step 5: Run on a Schedule

Check every 4 hours during business hours. Deals can go cold fast.

# Check at 8am, 12pm, and 4pm weekdays
0 8,12,16 * * 1-5 cd /root/deal-risk && python scan_risks.py

What to Build Next

Connect the risk alerts back to your CRM. When a deal gets flagged, automatically create a task for the owner to re-engage the prospect.

Related Reading

• The Risk Framework for AI Implementation - risk framework ai implementation
• AI for Sales Pipeline Management - ai sales pipeline management
• How to Build Automated Alerts That Actually Help - automated alerts that help