# Crm Sales Pipeline > Use when building CRM and sales pipeline management systems. - Skill: `loopyluci/crm-sales-pipeline` (Agent Skill) - Install (CLI): `npx skillmds@latest add loopyluci/crm-sales-pipeline` - Raw SKILL.md: https://api.skillmd.com/api/skills/loopyluci/crm-sales-pipeline/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - License: MIT - Author: LoopyLuci (https://skillmd.com/u/loopyluci) - Updated: 2026-09-08 - Page: https://skillmd.com/skills/loopyluci/crm-sales-pipeline --- # 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 ```python 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 ```python 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 1. **Stale pipeline** — deals that should be closed-lost stay in pipeline; set inactivity alerts 2. **Over-optimistic probability** — every deal is "50% likely"; use data-driven stage probabilities 3. **No lead source tracking** — can't optimize spend; tag every deal with source 4. **Manual data entry burden** — reps avoid CRM if it's tedious; integrate with email and calendar 5. **Stage definition disputes** — sales and marketing disagree on MQL vs SQL; document clear criteria 6. **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