apify-buying-signal-detection

购买意向信号检测

将 ICP 描述转化为定期运行的流水线，从三种信号中找出表现出购买意向的公司——招聘信息、融资事件和 LinkedIn 内容——并将其追加到统一去重的 leads.csv，可直接导入 CRM。

此技能的功能（以及刻意不做的事）

此技能会设置并运行定时工作流。它不会撰写冷邮件、按匹配度为潜在客户评分，也不会将行推送到 CRM。它的输出是整洁且附有证据链接的 leads.csv——也就是你现有外联流程的输入。针对这些潜在客户撰写冷邮件是刻意分开的工作（这正是 apify-link-prospecting-outreach 等技能的用途）。

开始之前，值得了解以下三项设计原则：

1. Apify 端和 Claude 端各自按独立计划运行。 Apify 按自己的 cron 计划运行 Actors；Claude 按自己的 cron 计划运行聚合。运行 Actors 时，Claude Code 无需启动。这种解耦让流水线真正实现定期运行，而不只是“你得记得触发它”。
2. 每周幂等性在聚合时强制执行。 如果潜在客户 CSV 中已有本 ISO 周的条目，聚合脚本就会提前退出，不会发起任何 HTTP 调用。Claude 端的计划可以每周运行多次（作为安全保障），且不会产生额外费用。
3. 去重时以首次发现为准。 如果某条招聘信号在周一发现了某个潜在客户，即使同一家公司在周三出现在融资数据流中，它仍会保留为招聘信号潜在客户。最初发现该公司的信号更有用。

前提条件

• Apify 账户（注册）
• 通过以下任一方式进行身份验证：
  • apify login（使用 Apify CLI 时采用 OAuth）
  • APIFY_TOKEN 环境变量
  • 从 Apify Console → Settings → Integrations 获取令牌
• Python 3.10+（用于 scripts/aggregate.py 和 scripts/setup_apify_tasks.py；仅需标准库——有 requests 时会使用，否则会回退到 urllib）
• 可选：用于按计划触发 Claude 端聚合的方式——使用 Claude Code 时，推荐采用本地 scheduled-tasks MCP；如果更倾向于无头运行，也可以使用 cron / Task Scheduler / GitHub Actions

工作流

复制此清单并标记已完成的项目：

Task Progress:
- [ ] Step 1: Collect ICP inputs (block on these)
- [ ] Step 2: Write icp.json + blacklist.csv (if any)
- [ ] Step 3: Provision Apify Actor Tasks (setup_apify_tasks.py)
- [ ] Step 4: Verify Actor picks in the Apify Console
- [ ] Step 5: Register the Claude-side aggregation schedule
- [ ] Step 6: First manual run of aggregate.py — sanity check the output

第 1 步：收集 ICP 输入（这些信息是必需的）

在写入任何文件之前，询问用户以下所有信息。设置脚本需要每个字段才能正确路由；任务配置完成后再修改定时任务，就只能在 Apify Console 中编辑，或重新运行设置——这两种做法都不如一次问清楚。

1. 活动名称 — 简短的 slug（小写字母，以连字符分隔）。用作每个 Apify Task 名称的前缀（例如 emea-saas-hiring-aes-bebity-linkedin-jobs-scraper）。如果用户已经运行多个活动，提前避免名称冲突。
2. 要跟踪的信号 — 从 ["jobs", "funding", "linkedin_content"] 中选择子集。活动很少只需要一种信号；此工作流的优势在于结合每家公司对应的多种信号。除非有明确的成本顾虑，否则建议选择全部三种。
3. 地理区域（ISO 国家代码） — 大写的两字母代码。决定区域 Actor 路由（德国、奥地利和比利时使用 Stepstone，澳大利亚和新西兰使用 Seek，法国使用 France Travail，法国融资信息使用 Maddyness）。全球活动应列出用户实际开展业务的所有国家——传入 ["US", "GB", "DE", "FR", "AU"] 会分发到五个区域职位 Actor，周成本将是 5 倍。参见 references/gotchas.md。
4. 行业关键词 — 类别描述。作为 industry 传递给融资跟踪工具；未提供明确的内容搜索词时，作为 keywords 传递给 LinkedIn；未提供角色职位名称时，作为职位抓取工具的备用关键词。
5. Persona（启用职位信号时） — ICP 招聘的职位名称。具体职位名称优于类别："Account Executive"、"SDR"、"BDR" 是有效匹配；"sales" 则会产生噪声。可选的资历级别（entry、mid、senior、manager、director、vp、cxo）和公司规模范围（"11-50" 等）会在汇总器中事后应用。
6. 融资配置（启用融资信号时） — 阶段（seed、series_a、series_b 等）和 max_days_since_announcement（默认值为 90）。资金到账时间越新、预算越充足，筛选窗口越紧凑，效果越好。
7. LinkedIn 内容配置（启用 linkedin_content 信号时） — 搜索短语。这是影响质量的最大因素；宽泛的词语（"sales"）会浪费预算。具体的痛点短语优于类别名称——参见 references/actors.md。还可以设置用于后置筛选的 min_reactions（默认值为 5）和 posted_within_days（默认值为 14）。
8. 潜在客户存储位置 — CSV 文件路径。默认使用活动目录下的 ./leads.csv。此文件用于保存各次运行间的流程记忆；请将其纳入版本控制（或至少做好备份），以便磁盘重置后仍能使用去重保护。
9. 黑名单 CSV 路径 — 可选。CSV 文件需包含 domain,company,reason 列。与精确域名或规范化公司名称匹配的行会在追加前被剔除。如果用户没有黑名单，请询问是否要先添加明显的排除项（现有客户、自有域名、主要竞争对手）。
10. 定时计划 — apify_side_cron（Apify 运行 Actor 的时间）和 claude_side_cron（Claude 汇总数据的时间）。默认值：0 6 * * 1（Apify 于 UTC 周一 06:00 运行）和 0 8 * * 1（Claude 于 UTC 周一 08:00 运行）。两者之间预留两小时缓冲，以应对 Actor 运行较慢的情况。

完整架构请参阅 references/icp-config-schema.md。示例见 examples/icp.example.json。

第 2 步：编写 icp.json 和 blacklist.csv

写入活动目录内容：

<campaign-dir>/
  icp.json           ← the config from Step 1
  blacklist.csv      ← optional; columns: domain,company,reason
  leads.csv          ← created empty; the aggregator will populate it

leads.csv 只需从表头行开始（架构见 references/csv-schema.md），这样汇总器首次运行时就不必处理文件缺失的情况：

detected_at,company,domain,signal_type,signal_detail,signal_source_actor,signal_date,evidence_url,geo,notes

第 3 步：配置 Apify Actor 任务

运行设置脚本：

APIFY_TOKEN=$APIFY_TOKEN \
python ${CLAUDE_PLUGIN_ROOT}/scripts/setup_apify_tasks.py \
  --config ./icp.json

此脚本会：

• 读取 icp.json，并根据 references/actors.md 中的路由表选择 Actor——始终选择全局 Actor，并额外选择与地区列表匹配的地区 Actor。
• 为每个选中的 Actor，创建或更新一个名为 <campaign>-<actor-slug> 的 Apify Actor 任务，输入负载根据 ICP 生成。重新运行脚本会原地更新现有任务，不会创建重复任务。
• 写入旁路文件 <campaign-dir>/.<campaign-name>.tasks.json，记录任务 ID。aggregate.py 会读取此文件，以确定要从哪些任务中提取数据集条目。
• 如果 ICP 中设置了 schedule.apify_side_cron（默认为启用），则创建或更新一个 Apify Schedule，按该 cron 表达式触发所有任务。

实用标志：

• --dry-run — 输出选取列表和输入负载，不进行任何 API 调用。创建新 campaign 时，务必先运行一次。
• --no-schedule — 配置任务，但跳过 Schedule 创建（适用于校准期间需要手动触发运行的情况）。

第 4 步：在 Apify Console 中验证 Actor 选择

打开 console.apify.com/actors/tasks。按 campaign 前缀筛选。检查以下三项：

1. 选择了正确的 Actor — 地区 Actor（Stepstone / Seek / France Travail / Maddyness）只应在指定地区触发。如果看到 Seek，但 ICP 中没有 ANZ 国家，则说明存在问题。
2. 输入负载正确 — 点击各个任务，查看其输入 JSON。关键词、职位和阶段应从 ICP 填充；不应出现 null。
3. Apify Schedule 已启用 — 在 Schedules 下找到 <campaign>-schedule，确认其已开启且列出了所有任务。

如果发现任何问题，请编辑 icp.json 并重新运行 setup_apify_tasks.py——此脚本具有幂等性。

第 5 步：注册 Claude 侧聚合计划任务

注册一个计划任务，在 campaign 的 claude_side_cron 时间运行聚合器。根据你的环境选择合适的运行方式：

选项 A — Claude Code 中的 scheduled-tasks MCP。 该 MCP 提供 mcp__scheduled-tasks__create_scheduled_task。具体调用如下：

{
  "name": "<campaign-name>-aggregate",
  "cron_expression": "<value of schedule.claude_side_cron from icp.json>",
  "timezone": "UTC",
  "prompt": "Run the buying-signal aggregator. Execute exactly: python ${CLAUDE_PLUGIN_ROOT}/scripts/aggregate.py --config /abs/path/to/icp.json. Requires APIFY_TOKEN env var. On non-zero exit, surface the stderr in the notification body — do not attempt to reinterpret the error."
}

调用前，将 <campaign-name> 和 /abs/path/to/icp.json 替换为实际值。使用 mcp__scheduled-tasks__list_scheduled_tasks 验证任务已创建，并确认 cron 表达式与 icp.json 一致。

选项 B — 无头 cron / Task Scheduler / CI。 添加一个普通的操作系统级计划任务条目：

# Linux crontab entry
0 8 * * 1 APIFY_TOKEN=$APIFY_TOKEN /path/to/python /path/to/aggregate.py --config /path/to/icp.json >> /path/to/aggregate.log 2>&1

或者使用 GitHub Actions 工作流（.github/workflows/aggregate.yml）：

on:
  schedule:
    - cron: '0 8 * * 1'
jobs:
  aggregate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - env: { APIFY_TOKEN: '${{ secrets.APIFY_TOKEN }}' }
        run: python scripts/aggregate.py --config icp.json

两种运行方式都可以安全地以高于每周一次的频率调用——每周幂等性检查会跳过本周已有条目的运行。仅在校准期间使用 --force。

第 6 步：首次手动运行 aggregate.py

在等待计划任务触发之前，手动运行一次以确认配置正确。这也会为 leads.csv 写入初始数据，供每周检查在下周进行比较：

APIFY_TOKEN=$APIFY_TOKEN \
python ${CLAUDE_PLUGIN_ROOT}/scripts/aggregate.py --config ./icp.json

预期输出格式：

{
  "summary": {
    "appended": 47,
    "fetched_by_signal": {"jobs": 320, "funding": 88, "linkedin_content": 210},
    "dropped": {"blacklist": 2, "dup_domain": 156, "dup_url": 401, "post_filter": 12, "unmappable": 0, "linkedin_no_domain": 8},
    "linkedin_profile_lookups": 142,
    "linkedin_profile_resolved": 134,
    "linkedin_company_lookups": 118,
    "linkedin_company_resolved": 112,
    "linkedin_domain_resolved": 128
  }
}
wrote 47 new rows to /abs/path/to/leads.csv

linkedin_*_lookups / linkedin_*_resolved 计数器会报告 LinkedIn 作者域名补充链中的各个步骤（见下文）。linkedin_domain_resolved 是成功填充 domain 列的 LinkedIn 行数。linkedin_no_domain 是因该链无法解析出域名而被丢弃的 LinkedIn 行数——这些行无法安全地加入黑名单或去重。

如果 fetched_by_signal 的值全部为零，则可能是 Apify 任务尚未运行（请检查 Console），或者旁路任务注册表缺失。等待首次 Apify 运行完成后，再重新运行。

使用 --dry-run 查看即将追加的内容而不修改 CSV；在校准期间使用 --force 绕过每周检查。

Actor 路由

包含各信号 PICK 规则的完整目录见 references/actors.md。简要概述如下：

信号	全局默认项	区域性补充项
职位	bebity/linkedin-jobs-scraper、johnvc/google-jobs-scraper	Indeed（US/GB/IN/CA）、Stepstone（DE/AT/BE）、Seek（AU/NZ）、France Travail（FR）
融资	nexgendata/startup-funding-tracker、memo23/crunchbase-scraper、complex_intricate_networks/fundraising-and-startup-funding-scraper、signalbase/signalbase-api	Maddyness（FR）
LinkedIn 内容	harvestapi/linkedin-post-search（无需 Cookie，$2/1k 篇帖子）	深度抓取回退：curious_coder/linkedin-post-search-scraper（需要 Cookie）
LinkedIn 作者 → 公司域名（数据增强，由 aggregate.py 按需调用）	harvestapi/linkedin-profile-scraper（$4/1k 个个人资料）+ harvestapi/linkedin-company（按次查询）——两步调用	无——个人资料 URL 和公司 LinkedIn URL 是主键

setup_apify_tasks.py::pick_actors 中的路由逻辑与此表保持一致——编辑一处时，也要编辑另一处。

为什么 LinkedIn 数据增强按需运行而非定时运行——以及为何需要两步调用

harvestapi/linkedin-post-search 会返回作者姓名和职位头衔，但不会返回雇主的网站。没有域名，聚合器就无法检查黑名单，也无法将该信号与之前发现的公司去重——这意味着黑名单中的竞争对手可能通过 LinkedIn 帖子混入结果。

解析域名需要额外调用两次 Actor，并在 aggregate.py::enrich_linkedin_domains 中串联执行：

1. 对帖子行返回时没有域名的去重后作者个人资料 URL 集合调用 harvestapi/linkedin-profile-scraper。它会返回 currentPosition[0].companyLinkedinUrl 和 companyName，但不会返回公司网站。输入：{profileScraperMode: "Profile details no email ($4 per 1k)", urls: [...]}。
2. 对第 1 步返回的去重后公司 LinkedIn URL 集合调用 harvestapi/linkedin-company。它会返回 website。输入：{companies: [...]}。

这是聚合器中唯一同步调用 Actor 的路径——所有其他数据均来自预先安排的 Task 运行。之所以刻意例外，是因为只有在读取上一步的数据集之后，才能确定两步数据增强所需的输入（先是作者个人资料 URL，随后是公司 URL）。

成本。

• 个人资料抓取：每 1000 个个人资料 $4（选择了“无邮箱”档位——解析域名不需要查找邮箱）
• 公司抓取：harvestapi/linkedin-company 按事件付费
• 综合影响：对于一场包含 500 篇 LinkedIn 帖子的活动，若平均每位作者发 3 篇帖子，预计需要约 170 次个人资料查询和约 150 次公司查询（许多作者在同一家公司工作）

以下三个因素可控制成本：

• 规范化个人资料 URL 去重——同一作者发布 N 篇帖子只需查询一次个人资料（canonical_linkedin_profile_url 会去掉 ?miniProfileUrn=…，使样本帖子中同一作者合并为一个键）
• 在公司抓取环节按公司 URL 去重——同一家公司有 N 位作者时只需查询一次公司
• 如果所有 LinkedIn 行都已有域名，则完全跳过整个流程

调用 Actor——选择接口

setup_apify_tasks.py 直接使用 Apify REST API（不调用 Actor——它负责配置 Tasks 和 Schedules）。aggregate.py 使用 REST API 拉取每个任务最近一次成功运行的数据集条目。如果你想在排查问题时手动触发 Actor（例如第 4 步验证），可以使用以下任一方式：

选项 A：Apify CLI（推荐，便于移植）

每次调用都使用三个标志（--json、--user-agent、2>/dev/null）：

# 手动触发一个营销活动任务
apify tasks run <task-id> --wait 300 \
  --json \
  --user-agent apify-awesome-skills/apify-buying-signal-detection \
  2>/dev/null

# 列出为此营销活动配置的任务
apify tasks list --json 2>/dev/null | \
  jq '.[] | select(.name | startswith("<campaign-name>-"))'

# 查看某个任务的最新数据集
apify tasks last-run <task-id> --dataset --format json \
  --user-agent apify-awesome-skills/apify-buying-signal-detection 2>/dev/null

# 获取 Actor 的输入架构（用于决定是否将其添加到路由表）
apify actors info "<actor-id>" --input --json \
  --user-agent apify-awesome-skills/apify-buying-signal-detection 2>/dev/null

选项 B：Apify MCP 连接器

托管的 MCP 服务，地址为 mcp.apify.com。完整文档见 docs.apify.com/platform/integrations/mcp。

选项 C：自行选择的 MCP 客户端（例如 mcpc）

独立命令行客户端。请参阅 github.com/apify/mcpc。

故障排除

错误／症状	处理方法
APIFY_TOKEN not found	执行 export APIFY_TOKEN=$(cat ~/.apify_token)，或将其添加到 .env。请前往 console.apify.com/account/integrations 获取令牌。
no task registry for campaign '<name>'	你在运行 setup_apify_tasks.py 之前运行了 aggregate.py。请先运行设置脚本——它会写入聚合器所需的旁置文件。
合理的重新运行时提示 skipped: already run this week	传入 --force。保护机制会保留本周的条目，并在此基础上去重；不会覆盖条目。
任务在 Apify 上运行，但 aggregate.py 报告 "fetched_by_signal": {"jobs": 0}	Actor 已运行，但返回了零个条目。检查 Actor 的运行日志，查看是否有架构错误（关键词格式错误、不支持的国家/地区代码）。修复后，通过 apify tasks run 手动运行任务并重新聚合。
任务已配置，但始终没有数据写入	Apify Schedule 可能已停用。在 Console 中打开 Schedules → <campaign>-schedule，确认其已启用。还要检查计划任务的 cron 是否符合你的时区假设——除非设置了 timezone，否则计划任务均使用 UTC。
成本高于预期	请参阅 references/gotchas.md#cost-guardrails。最常见的原因是范围过宽的 LinkedIn 搜索词导致 harvestapi/linkedin-post-search 成本倍增。第二常见的原因是添加了所有地区的招聘 Actor，而 ICP 实际上只面向两个国家/地区销售。
转发内容使 LinkedIn 信号计数虚高	聚合器会在去重前剥除 trackingId 和 utm_* 查询参数，将 URL 规范化；但 LinkedIn 的 URL 方案会定期变化。如果发现同一篇帖子出现两次，请检查 URL 是否仅在剥除列表中未包含的参数上有所不同，并将该参数添加到 aggregate.py 中的 strip_tracking()。
公司更名后出现重复线索	去重优先依据 domain。如果公司更换域名，聚合器会将其视为新线索。只能手动协调——没有自动修复方法。
多台机器同时写入同一个 leads.csv	不支持。此方案假设只有一个写入方。请将 CSV 放在带锁的层后面（如 Google Sheets 导出、flock 等），或按机器分区。