LoopyLuci CRM and Sales Pipeline Management
Building and managing customer relationship management (CRM) systems and sales pipelines — from deal tracking and pipeline analytics through automation and team collaboration.
When to Use
- Implementing a CRM for your sales team
- Building custom pipeline tracking (deals, stages, probability)
- Automating sales tasks (follow-ups, task assignment, notifications)
- Analyzing pipeline health (velocity, conversion, bottlenecks)
- Integrating CRM with other tools (email, calendar, marketing)
Pipeline Architecture
Lead → MQL → SQL → Opportunity → Proposal → Negotiation → Closed Won
↑ ↑ ↑ ↑ ↑ ↑ ↓
0% 10% 25% 50% 75% 90% 100%
Pipeline Management
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from enum import Enum
import json
class DealStage(str, Enum):
LEAD = 'lead'
MQL = 'marketing_qualified_lead'
SQL = 'sales_qualified_lead'
OPPORTUNITY = 'opportunity'
PROPOSAL = 'proposal'
NEGOTIATION = 'negotiation'
CLOSED_WON = 'closed_won'
CLOSED_LOST = 'closed_lost'
STAGE_PROBABILITY = {
DealStage.LEAD: 0.05,
DealStage.MQL: 0.10,
DealStage.SQL: 0.25,
DealStage.OPPORTUNITY: 0.50,
DealStage.PROPOSAL: 0.75,
DealStage.NEGOTIATION: 0.90,
DealStage.CLOSED_WON: 1.0,
DealStage.CLOSED_LOST: 0.0,
}
class SalesPipeline:
"""End-to-end sales pipeline management."""
def __init__(self, name: str = "Main Pipeline"):
self.name = name
self.deals = {}
self.stages = [s.value for s in DealStage if s.value != 'closed_lost']
def create_deal(self, name: str, value: float, contact: Dict,
stage: str = 'lead', notes: str = "") -> str:
"""Create a new deal in the pipeline."""
import uuid
deal_id = str(uuid.uuid4())[:8]
deal = {
'id': deal_id,
'name': name,
'value': value,
'weighted_value': value * STAGE_PROBABILITY.get(stage, 0),
'contact': contact,
'company': contact.get('company', ''),
'stage': stage,
'probability': STAGE_PROBABILITY.get(stage, 0),
'created_at': datetime.now().isoformat(),
'expected_close': None,
'notes': notes,
'tags': [],
'activities': [],
'tasks': [],
'owner': None,
}
self.deals[deal_id] = deal
self._log_activity(deal_id, 'created', f'Deal created at stage: {stage}')
return deal_id
def update_stage(self, deal_id: str, new_stage: str, reason: str = ""):
"""Move deal to a new stage and update weighted value."""
if deal_id not in self.deals:
return False
old_stage = self.deals[deal_id]['stage']
self.deals[deal_id]['stage'] = new_stage
self.deals[deal_id]['probability'] = STAGE_PROBABILITY.get(new_stage, 0)
self.deals[deal_id]['weighted_value'] = self.deals[deal_id]['value'] * self.deals[deal_id]['probability']
self._log_activity(deal_id, 'stage_change',
f'{old_stage} → {new_stage}: {reason}')
return True
def add_activity(self, deal_id: str, activity_type: str, description: str):
"""Log an activity (call, email, meeting) on a deal."""
if deal_id in self.deals:
self._log_activity(deal_id, activity_type, description)
def _log_activity(self, deal_id: str, activity_type: str, description: str):
self.deals[deal_id]['activities'].append({
'timestamp': datetime.now().isoformat(),
'type': activity_type,
'description': description,
})
def get_pipeline_metrics(self) -> Dict:
"""Calculate key pipeline metrics."""
total_value = 0
weighted_value = 0
stage_counts = {s: 0 for s in self.stages}
for deal in self.deals.values():
if deal['stage'] != 'closed_lost' and deal['stage'] != 'closed_won':
total_value += deal['value']
weighted_value += deal['weighted_value']
if deal['stage'] in stage_counts:
stage_counts[deal['stage']] += 1
# Velocity (avg days from lead to proposal)
velocities = []
for deal in self.deals.values():
if deal['stage'] in ('proposal', 'negotiation', 'closed_won'):
created = datetime.fromisoformat(deal['created_at'])
# Find when it hit proposal stage
for activity in deal['activities']:
if activity['type'] == 'stage_change' and 'lead → proposal' in activity['description']:
proposal_time = datetime.fromisoformat(activity['timestamp'])
velocities.append((proposal_time - created).days)
avg_velocity = sum(velocities) / len(velocities) if velocities else 0
return {
'total_pipeline_value': total_value,
'weighted_pipeline_value': weighted_value,
'deal_count': len(self.deals),
'active_deals': sum(1 for d in self.deals.values()
if d['stage'] not in ('closed_won', 'closed_lost')),
'stage_distribution': stage_counts,
'avg_velocity_days': round(avg_velocity, 1),
'conversion_rate': self._calculate_conversion(),
}
def _calculate_conversion(self) -> float:
"""Calculate lead-to-close conversion rate."""
won = sum(1 for d in self.deals.values() if d['stage'] == 'closed_won')
lost = sum(1 for d in self.deals.values() if d['stage'] == 'closed_lost')
total_closed = won + lost
return won / total_closed if total_closed > 0 else 0
def get_stuck_deals(self, days_in_stage=14) -> List[Dict]:
"""Find deals that haven't moved stages recently."""
stuck = []
now = datetime.now()
for deal in self.deals.values():
if deal['stage'] in ('closed_won', 'closed_lost'):
continue
# Find last stage change
last_change = None
for activity in reversed(deal['activities']):
if activity['type'] == 'stage_change':
last_change = datetime.fromisoformat(activity['timestamp'])
break
if last_change and (now - last_change).days >= days_in_stage:
stuck.append({
'deal_id': deal['id'],
'name': deal['name'],
'stage': deal['stage'],
'days_stuck': (now - last_change).days,
'value': deal['value'],
})
return sorted(stuck, key=lambda x: x['days_stuck'], reverse=True)
def create_task(self, deal_id: str, task_name: str,
due_date: str, assigned_to: str):
"""Create a task associated with a deal."""
if deal_id in self.deals:
task = {
'id': str(uuid.uuid4())[:8],
'name': task_name,
'due_date': due_date,
'assigned_to': assigned_to,
'completed': False,
'created_at': datetime.now().isoformat(),
}
self.deals[deal_id]['tasks'].append(task)
return task['id']
return None
Forecasting
class SalesForecast:
"""Forecast future revenue based on pipeline."""
@staticmethod
def forecast(pipeline: SalesPipeline, months_ahead: int = 3) -> Dict:
"""Generate revenue forecast from pipeline."""
monthly_forecast = {m: 0 for m in range(1, months_ahead + 1)}
for deal in pipeline.deals.values():
if deal['stage'] in ('closed_won', 'closed_lost'):
continue
expected_close = deal.get('expected_close')
if not expected_close:
continue
close_date = datetime.fromisoformat(expected_close)
month = close_date.month - datetime.now().month + 1
if 1 <= month <= months_ahead:
monthly_forecast[month] += deal['weighted_value']
total = sum(monthly_forecast.values())
return {
'monthly_forecast': monthly_forecast,
'total_forecast': total,
'confidence': 'high' if total > 0 and pipeline._calculate_conversion() > 0.3 else 'medium',
}
Common Pitfalls
- Stale pipeline — deals that should be closed-lost stay in pipeline; set inactivity alerts
- Over-optimistic probability — every deal is "50% likely"; use data-driven stage probabilities
- No lead source tracking — can't optimize spend; tag every deal with source
- Manual data entry burden — reps avoid CRM if it's tedious; integrate with email and calendar
- Stage definition disputes — sales and marketing disagree on MQL vs SQL; document clear criteria
- Pipeline ≠ forecast — pipeline shows all deals; forecast should use weighted values and historical conversion
Verification Checklist
- Pipeline stages defined with clear entry/exit criteria
- Deal source tracked for every entry
- Probability per stage reflects historical data
- Pipeline velocity tracked (avg days per stage)
- Stuck deal alerts configured (e.g., 14 days without movement)
- Forecast based on weighted pipeline, not raw pipeline
- Integration with email/calendar for auto-logging
- Conversion rate by source, owner, and deal size available
See Also
- real-estate-crm-leads — real estate specific CRM
- email-marketing-campaigns — email sequences for sales
- business-metrics-kpis — tracking sales metrics
- customer-segmentation-analysis — segmenting the pipeline