TokenToSea提示词广场
English 控制台

销售研究

商业咨询·系统提示 / 角色

在聊天中使用

技能式方法论，指导调研销售潜在客户，包括公司研究、联系人画像和购买信号识别。

---
name: sales-research
description: 本技能提供研究销售线索的方法论和最佳实践。
---

# 销售研究

## 概述

本技能提供研究销售线索的方法论和最佳实践。涵盖公司研究、联系人画像和信号检测，以挖掘可操作的情报。

## 使用方式

当 company-researcher 和 contact-researcher 子代理执行以下操作时，会引用本技能：
- 研究新线索
- 查找公司信息
- 为个人联系人建立画像
- 检测购买信号

## 研究方法论

### 公司研究清单

1. **基础资料**
   - 公司名称、行业、规模（员工数、收入）
   - 总部及主要办公地点
   - 成立日期、发展阶段

2. **近期动态**
   - 融资公告（过去 12 个月）
   - 并购活动
   - 高管变动
   - 产品发布

3. **技术栈**
   - 已知技术（BuiltWith、StackShare）
   - 招聘岗位中提及的工具
   - 集成合作伙伴

4. **信号**
   - 招聘岗位（扩张 = 机会）
   - Glassdoor 评论（痛点）
   - 新闻提及（背景信息）
   - 社交媒体动态

### 联系人研究清单

1. **职业背景**
   - 当前职位及任职时长
   - 以往公司及职位
   - 教育背景

2. **影响力指标**
   - 汇报关系
   - 决策权
   - 预算权

3. **互动切入点**
   - 最近的 LinkedIn 帖子
   - 发表的文章
   - 演讲活动
   - 共同联系人

## 资源

- `resources/signal-indicators.md` - 购买信号分类法
- `resources/research-checklist.md` - 完整研究清单

## 脚本

- `scripts/company-enricher.py` - 从多个来源聚合公司数据
- `scripts/linkedin-parser.py` - 结构化 LinkedIn 个人资料数据
FILE:company-enricher.py
#!/usr/bin/env python3
"""
company-enricher.py - 从多个来源聚合公司数据

输入：
  - company_name: 字符串
  - domain: 字符串（可选）

输出：
  - profile:
      name: 字符串
      industry: 字符串
      size: 字符串
      funding: 字符串
      tech_stack: [字符串]
      recent_news: [新闻条目]

依赖项：
  - requests, beautifulsoup4
"""

# 要求：requests, beautifulsoup4

import json
from typing import Any
from dataclasses import dataclass, asdict
from datetime import datetime


@dataclass
class NewsItem:
    title: str
    date: str
    source: str
    url: str
    summary: str


@dataclass
class CompanyProfile:
    name: str
    domain: str
    industry: str
    size: str
    location: str
    founded: str
    funding: str
    tech_stack: list[str]
    recent_news: list[dict]
    competitors: list[str]
    description: str


def search_company_info(company_name: str, domain: str = None) -> dict:
    """
    搜索公司基本信息。
    在生产环境中，将调用 Clearbit、Crunchbase 等 API。
    """
    # TODO: 实现实际的 API 调用
    # 占位符返回结构
    return {
        "name": company_name,
        "domain": domain or f"{company_name.lower().replace(' ', '')}.com",
        "industry": "Technology",  # 将来自 API
        "size": "Unknown",
        "location": "Unknown",
        "founded": "Unknown",
        "description": f"Information about {company_name}"
    }


def search_funding_info(company_name: str) -> dict:
    """
    搜索融资信息。
    在生产环境中，将调用 Crunchbase、PitchBook 等。
    """
    # TODO: 实现实际的 API 调用
    return {
        "total_funding": "Unknown",
        "last_round": "Unknown",
        "last_round_date": "Unknown",
        "investors": []
    }


def search_tech_stack(domain: str) -> list[str]:
    """
    检测技术栈。
    在生产环境中，将调用 BuiltWith、Wappalyzer 等。
    """
    # TODO: 实现实际的 API 调用
    return []


def search_recent_news(company_name: str, days: int = 90) -> list[dict]:
    """
    搜索公司近期新闻。
    在生产环境中，将调用新闻 API。
    """
    # TODO: 实现实际的 API 调用
    return []


def main(
    company_name: str,
    domain: str = None
) -> dict[str, Any]:
    """
    从多个来源聚合公司数据。

    参数：
        company_name: 要研究的公司名称
        domain: 公司域名（可选，将被推断）

    返回：
        包含行业、规模、融资、技术栈、新闻的公司画像字典
    """
    # 获取公司基本信息
    basic_info = search_company_info(company_name, domain)

    # 获取融资信息
    funding_info = search_funding_info(company_name)

    # 检测技术栈
    company_domain = basic_info.get("domain", domain)
    tech_stack = search_tech_stack(company_domain) if company_domain else []

    # 获取近期新闻
    news = search_recent_news(company_name)

    # 汇总公司画像
    profile = CompanyProfile(
        name=basic_info["name"],
        domain=basic_info["domain"],
        industry=basic_info["industry"],
        size=basic_info["size"],
        location=basic_info["location"],
        founded=basic_info["founded"],
        funding=funding_info.get("total_funding", "Unknown"),
        tech_stack=tech_stack,
        recent_news=news,
        competitors=[],  # 将从行业分析中丰富
        description=basic_info["description"]
    )

    return {
        "profile": asdict(profile),
        "funding_details": funding_info,
        "enriched_at": datetime.now().isoformat(),
        "sources_checked": ["company_info", "funding", "tech_stack", "news"]
    }


if __name__ == "__main__":
    import sys

    # 示例用法
    result = main(
        company_name="DataFlow Systems",
        domain="dataflow.io"
    )
    print(json.dumps(result, indent=2))
FILE:linkedin-parser.py
#!/usr/bin/env python3
"""
linkedin-parser.py - 结构化 LinkedIn 个人资料数据

输入：
  - profile_url: 字符串
  - 或 name + company: 字符串

输出：
  - contact:
      name: 字符串
      title: 字符串
      tenure: 字符串
      previous_roles: [角色对象]
      mutual_connections: [字符串]
      recent_activity: [帖子摘要]

依赖项：
  - requests
"""

# 要求：requests

import json
from typing import Any
from dataclasses import dataclass, asdict
from datetime import datetime


@dataclass
class PreviousRole:
    title: str
    company: str
    duration: str
    description: str


@dataclass
class RecentPost:
    date: str
    content_preview: str
    engagement: int
    topic: str


@dataclass
class ContactProfile:
    name: str
    title: str
    company: str
    location: str
    tenure: str
    previous_roles: list[dict]
    education: list[str]
    mutual_connections: list[str]
    recent_activity: list[dict]
    profile_url: str
    headline: str


def search_linkedin_profile(name: str = None, company: str = None, profile_url: str = None) -> dict:
    """
    搜索 LinkedIn 个人资料信息。
    在生产环境中，将使用 LinkedIn API 或 Sales Navigator。
    """
    # TODO: 实现实际的 LinkedIn API 集成
    # 注意：LinkedIn 的 API 有严格的服务条款

    return {
        "found": False,
        "name": name or "Unknown",
        "title": "Unknown",
        "company": company or "Unknown",
        "location": "Unknown",
        "headline": "",
        "tenure": "Unknown",
        "profile_url": profile_url or ""
    }


def get_career_history(profile_data: dict) -> list[dict]:
    """
    从个人资料中提取职业历史。
    """
    # TODO: 实现职业经历提取
    return []


def get_mutual_connections(profile_data: dict, user_network: list = None) -> list[str]:
    """
    查找共同联系人。
    """
    # TODO: 实现共同联系人检测
    return []


def get_recent_activity(profile_data: dict, days: int = 30) -> list[dict]:
    """
    获取最近的帖子和活动。
    """
    # TODO: 实现活动提取
    return []


def main(
    name: str = None,
    company: str = None,
    profile_url: str = None
) -> dict[str, Any]:
    """
    为销售准备结构化 LinkedIn 个人资料数据。

    参数：
        name: 人员姓名
        company: 他们就职的公司
        profile_url: 直接的 LinkedIn 个人资料 URL

    返回：
        包含结构化联系人资料的字典
    """
    if not profile_url and not (name and company):
        return {"error": "请提供 profile_url 或 name + company"}

    # 搜索个人资料
    profile_data = search_linkedin_profile(
        name=name,
        company=company,
        profile_url=profile_url
    )

    if not profile_data.get("found"):
        return {
            "found": False,
            "name": name or "Unknown",
            "company": company or "Unknown",
            "message": "未找到个人资料或访问受限",
            "suggestions": [
                "尝试直接在 LinkedIn 上搜索",
                "检查是否有其他拼写方式",
                "确认此人是否仍在该公司工作"
            ]
        }

    # 获取职业历史
    previous_roles = get_career_history(profile_data)

    # 查找共同联系人
    mutual_connections = get_mutual_connections(profile_data)

    # 获取最近活动
    recent_activity = get_recent_activity(profile_data)

    # 汇总联系人资料
    contact = ContactProfile(
        name=profile_data["name"],
        title=profile_data["title"],
        company=profile_data["company"],
        location=profile_data["location"],
        tenure=profile_data["tenure"],
        previous_roles=previous_roles,
        education=[],  # 将从个人资料中提取
        mutual_connections=mutual_connections,
        recent_activity=recent_activity,
        profile_url=profile_data["profile_url"],
        headline=profile_data["headline"]
    )

    return {
        "found": True,
    "contact": asdict(contact),
    "research_date": datetime.now().isoformat(),
    "data_completeness": calculate_completeness(contact)
}


def calculate_completeness(contact: ContactProfile) -> dict:
    """计算个人资料数据的完整程度。"""
    fields = {
        "basic_info": bool(contact.name and contact.title and contact.company),
        "career_history": len(contact.previous_roles) > 0,
        "mutual_connections": len(contact.mutual_connections) > 0,
        "recent_activity": len(contact.recent_activity) > 0,
        "education": len(contact.education) > 0
    }

    complete_count = sum(fields.values())
    return {
        "fields": fields,
        "score": f"{complete_count}/{len(fields)}",
        "percentage": int((complete_count / len(fields)) * 100)
    }


if __name__ == "__main__":
    import sys

    # 示例用法
    result = main(
        name="Sarah Chen",
        company="DataFlow Systems"
    )
    print(json.dumps(result, indent=2))
FILE:priority-scorer.py
#!/usr/bin/env python3
"""
priority-scorer.py - 计算并排序潜在客户优先级

输入：
  - prospects: [包含信号的潜在客户对象]
  - weights: {deal_size, timing, warmth, signals}

输出：
  - ranked: [包含分数和推理的潜在客户]

依赖：
  - (无 - 纯 Python)
"""

import json
from typing import Any
from dataclasses import dataclass


# 默认评分权重
DEFAULT_WEIGHTS = {
    "deal_size": 0.25,
    "timing": 0.30,
    "warmth": 0.20,
    "signals": 0.25
}

# 信号评分映射
SIGNAL_SCORES = {
    # 高意向信号
    "recent_funding": 10,
    "leadership_change": 8,
    "job_postings_relevant": 9,
    "expansion_news": 7,
    "competitor_mention": 6,

    # 中等意向信号
    "general_hiring": 4,
    "industry_event": 3,
    "content_engagement": 3,

    # 关系信号
    "mutual_connection": 5,
    "previous_contact": 6,
    "referred_lead": 8,

    # 负面信号
    "recent_layoffs": -3,
    "budget_freeze_mentioned": -5,
    "competitor_selected": -7,
}


@dataclass
class ScoredProspect:
    company: str
    contact: str
    call_time: str
    raw_score: float
    normalized_score: int
    priority_rank: int
    score_breakdown: dict
    reasoning: str
    is_followup: bool


def score_deal_size(prospect: dict) -> tuple[float, str]:
    """基于预估交易规模进行评分。"""
    size_indicators = prospect.get("size_indicators", {})

    employee_count = size_indicators.get("employees", 0)
    revenue_estimate = size_indicators.get("revenue", 0)

    # 基于公司规模的简单评分
    if employee_count > 1000 or revenue_estimate > 100_000_000:
        return 10.0, "Enterprise-scale opportunity"
    elif employee_count > 200 or revenue_estimate > 20_000_000:
        return 7.0, "Mid-market opportunity"
    elif employee_count > 50:
        return 5.0, "SMB opportunity"
    else:
        return 3.0, "Small business"


def score_timing(prospect: dict) -> tuple[float, str]:
    """基于时间信号进行评分。"""
    timing_signals = prospect.get("timing_signals", [])

    score = 5.0  # 基础分
    reasons = []

    for signal in timing_signals:
        if signal == "budget_cycle_q4":
            score += 3
            reasons.append("Q4 budget planning")
        elif signal == "contract_expiring":
            score += 4
            reasons.append("Contract expiring soon")
        elif signal == "active_evaluation":
            score += 5
            reasons.append("Actively evaluating")
        elif signal == "just_funded":
            score += 3
            reasons.append("Recently funded")

    return min(score, 10.0), "; ".join(reasons) if reasons else "Standard timing"


def score_warmth(prospect: dict) -> tuple[float, str]:
    """基于关系亲密度进行评分。"""
    relationship = prospect.get("relationship", {})

    if relationship.get("is_followup"):
        last_outcome = relationship.get("last_outcome", "neutral")
        if last_outcome == "positive":
            return 9.0, "Warm follow-up (positive last contact)"
        elif last_outcome == "neutral":
            return 7.0, "Follow-up (neutral last contact)"
        else:
            return 5.0, "Follow-up (needs re-engagement)"

    if relationship.get("referred"):
        return 8.0, "Referred lead"

    if relationship.get("mutual_connections", 0) > 0:
        return 6.0, f"{relationship['mutual_connections']} mutual connections"

    if relationship.get("inbound"):
        return 7.0, "Inbound interest"

    return 4.0, "Cold outreach"


def score_signals(prospect: dict) -> tuple[float, str]:
    """基于检测到的购买信号进行评分。"""
    signals = prospect.get("signals", [])

    total_score = 0
    signal_reasons = []

    for signal in signals:
        signal_score = SIGNAL_SCORES.get(signal, 0)
        total_score += signal_score
        if signal_score > 0:
            signal_reasons.append(signal.replace("_", " "))
    # 归一化到 0-10 分数
    normalized = min(max(total_score / 2, 0), 10)

    reason = f"信号：{', '.join(signal_reasons)}" if signal_reasons else "无强信号"
    return normalized, reason


def calculate_priority_score(
    prospect: dict,
    weights: dict = None
) -> ScoredProspect:
    """计算潜在客户的总体优先级分数。"""
    weights = weights or DEFAULT_WEIGHTS

    # 计算各组件分数
    deal_score, deal_reason = score_deal_size(prospect)
    timing_score, timing_reason = score_timing(prospect)
    warmth_score, warmth_reason = score_warmth(prospect)
    signal_score, signal_reason = score_signals(prospect)

    # 加权总分
    raw_score = (
        deal_score * weights["deal_size"] +
        timing_score * weights["timing"] +
        warmth_score * weights["warmth"] +
        signal_score * weights["signals"]
    )

    # 汇总理由
    reasons = []
    if timing_score >= 8:
        reasons.append(timing_reason)
    if signal_score >= 7:
        reasons.append(signal_reason)
    if warmth_score >= 7:
        reasons.append(warmth_reason)
    if deal_score >= 8:
        reasons.append(deal_reason)

    return ScoredProspect(
        company=prospect.get("company", "Unknown"),
        contact=prospect.get("contact", "Unknown"),
        call_time=prospect.get("call_time", "Unknown"),
        raw_score=round(raw_score, 2),
        normalized_score=int(raw_score * 10),
        priority_rank=0,  # 排序后设置
        score_breakdown={
            "deal_size": {"score": deal_score, "reason": deal_reason},
            "timing": {"score": timing_score, "reason": timing_reason},
            "warmth": {"score": warmth_score, "reason": warmth_reason},
            "signals": {"score": signal_score, "reason": signal_reason}
        },
        reasoning="; ".join(reasons) if reasons else "标准优先级",
        is_followup=prospect.get("relationship", {}).get("is_followup", False)
    )


def main(
    prospects: list[dict],
    weights: dict = None
) -> dict[str, Any]:
    """
    计算并排序潜在客户的优先级。

    参数:
        prospects: 包含信号的潜在客户对象列表
        weights: 可选的评分组件自定义权重

    返回:
        包含排序后潜在客户和评分详情的字典
    """
    weights = weights or DEFAULT_WEIGHTS

    # 对所有潜在客户评分
    scored = [calculate_priority_score(p, weights) for p in prospects]

    # 按原始分数降序排序
    scored.sort(key=lambda x: x.raw_score, reverse=True)

    # 分配排名
    for i, prospect in enumerate(scored, 1):
        prospect.priority_rank = i

    # 转换为字典以便 JSON 序列化
    ranked = []
    for s in scored:
        ranked.append({
            "company": s.company,
            "contact": s.contact,
            "call_time": s.call_time,
            "priority_rank": s.priority_rank,
            "score": s.normalized_score,
            "reasoning": s.reasoning,
            "is_followup": s.is_followup,
            "breakdown": s.score_breakdown
        })

    return {
        "ranked": ranked,
        "weights_used": weights,
        "total_prospects": len(prospects)
    }


if __name__ == "__main__":
    import sys

    # 示例用法
    example_prospects = [
        {
            "company": "DataFlow Systems",
            "contact": "Sarah Chen",
            "call_time": "2pm",
            "size_indicators": {"employees": 200, "revenue": 25_000_000},
            "timing_signals": ["just_funded", "active_evaluation"],
            "signals": ["recent_funding", "job_postings_relevant"],
            "relationship": {"is_followup": False, "mutual_connections": 2}
        },
        {
            "company": "Acme Manufacturing",
            "contact": "Tom Bradley",
            "call_time": "10am",
            "size_indicators": {"employees": 500},
            "timing_signals": ["contract_expiring"],
            "signals": [],
            "relationship": {"is_followup": True, "last_outcome": "neutral"}
        },
        {
            "company": "FirstRate Financial",
            "contact": "Linda Thompson",
            "call_time": "4pm",
            "size_indicators": {"employees": 300},
            "timing_signals": [],
            "signals": [],
            "relationship": {"is_followup": False}
        }
    ]

    result = main(prospects=example_prospects)
    print(json.dumps(result, indent=2))
## 高意向信号

### 职位发布
- **发布 3 个以上相关职位** = 正在推进主动举措，已有预算分配
- **在你的领域招聘高级职位** = 战略重点
- **紧急语言（“ASAP”、“立即”）** = 痛点紧迫
- **提及特定工具** = 对竞品或品类有认知

### 财务事件
- **B 轮及以上融资** = 拥有增长资金，购买力强
- **准备上市（IPO）** = 需要运营成熟度
- **宣布收购** = 即将面临整合挑战
- **营收里程碑新闻稿** = 有可用预算

### 领导层变动
- **你的领域内新任 CXO** = 正处于 90 天优先事项设定期
- **新任 CRO/CMO** = 很可能正在评估技术栈
- **创始人转任 CEO** = 正在推动运营专业化

## 中等意向信号

### 扩张信号
- **开设新办公室** = 存在基础设施需求
- **国际化扩张** = 本地化与合规需求
- **新产品发布** = 面临扩展挑战
- **赢得重要客户** = 交付压力增大

### 技术信号
- **发布 RFP** = 正处于活跃采购流程
- **提及供应商评审** = 正在比价选型
- **技术栈变更** = 存在集成机会
- **对遗留系统抱怨** = 有现代化需求

### 内容信号
- **发布与你主题相关的博客文章** = 正在自我教育
- **参加网络研讨会** = 兴趣已确认
- **下载白皮书** = 已意识到问题
- **在会议上发言** = 追求思想领导力与曝光度

## 低意向信号（培育中）

### 一般性活动
- **参加行业活动** = 市场参与者
- **常规招聘** = 公司在增长
- **正面新闻报道** = 公司健康
- **社交媒体活跃** = 领导层积极参与

## 信号评分

| 信号类型 | 分数 | 行动 |
|-------------|-------|--------|
| 职位发布（相关） | +3 | 优先触达 |
| 近期融资 | +3 | 在对话中引用 |
| 领导层变动 | +2 | 时效性机会 |
| 扩张新闻 | +2 | 从增长角度切入 |
| 负面评价 | +2 | 从痛点角度切入 |
| 内容参与 | +1 | 纳入培育流程 |
| 无信号 | 0 | 聚焦探索发现 |

相关推荐

更多