基于 Python 的 X 平台 KOL 账号挖掘：候选发现、画像补全与打分排序

原创

用户12869588

发布于 2026-11-02 22:35:32

540

一、问题：怎样找到"真正有影响力"的账号

无论是做海外产品推广、开发者生态运营，还是做行业研究，都会遇到同一个问题：在某个细分领域里，哪些 X（Twitter）账号说话有人听？

最直观的做法是按粉丝数排序，但这个指标有明显的缺陷：

粉丝可能是"死粉"：早年积累的大量粉丝早已不活跃，发帖几乎没有互动；
领域不匹配：一个百万粉丝的娱乐账号偶尔聊一次 AI，并不代表它在 AI 领域有影响力；
账号已停更：粉丝数很高，但最近半年没有发过帖子；
数据造假：刷粉、互刷互动的账号在任何平台都存在。

因此需要一套多维度的评估方法。本文把这个问题拆成四步：发现候选 → 补全画像 → 计算指标 → 打分排序，并给出一套可直接修改权重的 Python 实现。

二、整体流程

领域关键词 → 搜索热门推文 → 提取作者作为候选池（记录命中次数）
         → 批量获取用户资料 → 拉取每个候选的近期推文
         → 计算四类指标 → 规则过滤 → 归一化加权打分 → 输出排序名单

四类指标及其含义如下：

指标	计算方式	衡量什么
粉丝规模	log(粉丝数)	潜在覆盖面
互动率	近期推文互动中位数 / 粉丝数	粉丝是否真实、是否活跃
发帖频率	近 30 天日均原创发帖数	账号是否持续活跃
话题相关度	近期推文中命中领域关键词的比例 + 简介是否相关	是否真的属于这个领域

三、准备工作

本文示例使用第三方接口 TwexAPI 作为数据来源，用到它的三个接口：

POST /twitter/advanced_search/page：高级搜索，用于发现候选账号；
POST /twitter/users：请求体是用户名数组，批量获取用户资料；
POST /twitter/{screen_name}/timeline/page：分页获取某个用户的时间线，count 取值为 1～100。

# config.py  示例代码
import os

BASE_URL = "https://api.twexapi.io"
HEADERS = {"Authorization": f"Bearer {os.environ['DATA_API_KEY']}"}

# 领域关键词：既用于搜索，也用于计算话题相关度
DOMAIN_KEYWORDS = ["LLM", "RAG", "AI agent", "MCP", "fine-tuning", "大模型", "智能体"]

关键词的选择直接决定候选池的质量，建议同时包含英文术语、中文说法和领域内的常见缩写。

四、第一步：通过关键词发现候选账号

思路是：搜索领域关键词下的热门推文，把作者收集起来。一个账号在多个关键词下、多条热门推文中反复出现，本身就说明它在该领域比较活跃。

# discover.py  示例代码
import time
from collections import Counter
import requests
from config import BASE_URL, HEADERS, DOMAIN_KEYWORDS


def search(query: str, pages: int = 3) -> list[dict]:
    out, cursor = [], ""
    for _ in range(pages):
        r = requests.post(f"{BASE_URL}/twitter/advanced_search/page", headers=HEADERS,
                          json={"searchTerms": [query], "sortBy": "Top", "next_cursor": cursor},
                          timeout=30)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", 10)))
            continue
        r.raise_for_status()
        page = r.json()
        out += [t for t in page.get("data") or [] if t]
        if not page.get("has_next_page"):
            break
        cursor = page.get("next_cursor") or ""
    return out


def discover(min_faves: int = 50) -> Counter:
    hits = Counter()
    for kw in DOMAIN_KEYWORDS:
        for t in search(f'"{kw}" min_faves:{min_faves} -filter:retweets'):
            name = (t.get("user") or {}).get("screen_name")
            if name:
                hits[name.lower()] += 1
        time.sleep(1)
    return hits

hits 记录了每个账号在热门结果中出现的次数，后面会作为相关度的一个辅助信号。为了控制后续的数据量，可以只保留出现两次及以上的账号，或者取出现次数最多的前 200 个。

除了搜索，还有一种扩展候选池的方法：从已知的种子 KOL 出发，查看他们的关注列表。领域专家关注的人，往往也是同领域的专家。本文为简洁起见只使用搜索方式。

五、第二步：补全用户画像

搜索结果里虽然带有作者信息，但不同推文中的作者信息可能不完整。用批量接口一次获取多个用户的完整资料更可靠。根据接口文档，返回的是用户资料列表，查不到的用户对应 null。

# profiles.py  示例代码
import requests
from config import BASE_URL, HEADERS


def fetch_profiles(names: list[str], batch: int = 50) -> list[dict]:
    users = []
    for i in range(0, len(names), batch):
        r = requests.post(f"{BASE_URL}/twitter/users", headers=HEADERS,
                          json=names[i:i + batch], timeout=60)
        r.raise_for_status()
        body = r.json()
        # 兼容两种返回形态：直接返回列表，或包在 data 字段中
        items = body.get("data", []) if isinstance(body, dict) else body
        users += [u for u in items if u]           # 过滤查不到的 null
    return users

返回的用户资料中，本文会用到这些字段：screen_name、name、description（简介）、followers_count、following_count、statuses_count、listed_count（被加入列表的次数）、created_at、is_blue_verified 和 protected。

六、第三步：拉取近期推文

互动率、发帖频率和话题相关度都需要基于账号的近期推文来计算。时间线接口返回的每条推文包含正文、发布时间和互动数（favorite_count、retweet_count、reply_count、quote_count、view_count 等）。

# timeline.py  示例代码
import time
import requests
from config import BASE_URL, HEADERS


def recent_tweets(screen_name: str, limit: int = 100) -> list[dict]:
    out, cursor = [], ""
    while len(out) < limit:
        r = requests.post(f"{BASE_URL}/twitter/{screen_name}/timeline/page", headers=HEADERS,
                          json={"next_cursor": cursor, "count": min(100, limit - len(out))},
                          timeout=30)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", 10)))
            continue
        r.raise_for_status()
        page = r.json()
        out += page.get("data") or []
        if not page.get("has_next_page"):
            break
        cursor = page.get("next_cursor") or ""
    return out

每个账号取最近 100 条左右就足够评估了。候选数量较多时，这一步是调用量的大头，建议把结果按账号缓存成 JSON 文件，调整打分逻辑时就不用重新拉取。

七、第四步：计算指标

# metrics.py  示例代码
import math
import statistics
import pandas as pd
from config import DOMAIN_KEYWORDS

KW = [k.lower() for k in DOMAIN_KEYWORDS]


def is_relevant(text: str) -> bool:
    text = (text or "").lower()
    return any(k in text for k in KW)


def compute(user: dict, tweets: list[dict], hit_count: int) -> dict:
    df = pd.DataFrame(tweets)
    followers = int(user.get("followers_count") or 0)
    row = {
        "screen_name": user.get("screen_name"),
        "name": user.get("name"),
        "followers": followers,
        "following": int(user.get("following_count") or 0),
        "listed": int(user.get("listed_count") or 0),
        "blue": bool(user.get("is_blue_verified")),
        "protected": bool(user.get("protected")),
        "search_hits": hit_count,
        "bio_relevant": is_relevant(user.get("description")),
    }
    if df.empty:
        return {**row, "eng_rate": 0.0, "posts_per_day": 0.0, "topic_ratio": 0.0, "sample": 0}

    # 时间解析：使用 errors="coerce"，无法解析的时间记为 NaT
    df["ts"] = pd.to_datetime(df.get("created_at"), errors="coerce", utc=True)
    cutoff = pd.Timestamp.now(tz="UTC") - pd.Timedelta(days=30)
    recent = df[df["ts"] >= cutoff]

    for c in ("favorite_count", "retweet_count", "reply_count", "quote_count"):
        df[c] = pd.to_numeric(df.get(c), errors="coerce").fillna(0)
    df["eng"] = df[["favorite_count", "retweet_count", "reply_count", "quote_count"]].sum(axis=1)

    # 用中位数而不是平均数：避免一条爆款推文拉高整体
    median_eng = statistics.median(df["eng"]) if len(df) else 0
    row["eng_rate"] = median_eng / max(followers, 1)
    row["posts_per_day"] = len(recent) / 30
    row["topic_ratio"] = df["full_text"].map(is_relevant).mean() if "full_text" in df else 0.0
    row["sample"] = len(df)
    return row

几个细节需要说明：

互动率使用中位数。KOL 偶尔会有一条爆款推文，平均数会被极端值拉高，中位数更能代表日常水平；
发帖频率只统计近 30 天。这样能识别出"历史上很活跃、最近已停更"的账号；
话题相关度用近期推文中命中领域关键词的比例来衡量，再加上简介是否相关作为补充。关键词匹配比较粗糙，进阶做法是用文本向量计算推文与领域描述的语义相似度；
时间字段的格式在不同数据源之间可能不同，所以使用 errors="coerce" 容错解析；如果解析结果大量为空，需要先检查实际返回的格式。

八、第五步：规则过滤与加权打分

先用硬性规则排除明显不合适的账号，再对剩下的账号打分：

# score.py  示例代码
import numpy as np
import pandas as pd

WEIGHTS = {"reach": 0.30, "engagement": 0.30, "activity": 0.15, "relevance": 0.25}


def filter_rules(df: pd.DataFrame) -> pd.DataFrame:
    mask = (
        ~df["protected"]                               # 受保护账号无法评估
        & (df["followers"] >= 1000)                    # 规模下限
        & (df["posts_per_day"] >= 1 / 7)               # 近 30 天至少每周 1 条
        & (df["topic_ratio"] >= 0.1)                   # 至少一成内容与领域相关
        & (df["following"] < df["followers"] * 5)      # 关注数远大于粉丝数，疑似互粉号
    )
    return df[mask].copy()


def pct_rank(s: pd.Series) -> pd.Series:
    return s.rank(pct=True)


def score(df: pd.DataFrame) -> pd.DataFrame:
    df = filter_rules(df)
    df["reach"] = pct_rank(np.log1p(df["followers"]))
    df["engagement"] = pct_rank(df["eng_rate"])
    df["activity"] = pct_rank(df["posts_per_day"].clip(upper=10))   # 超过 10 条/天不再加分
    df["relevance"] = pct_rank(0.7 * df["topic_ratio"]
                               + 0.2 * df["bio_relevant"].astype(float)
                               + 0.1 * np.log1p(df["search_hits"]) / np.log1p(df["search_hits"].max()))
    df["score"] = sum(df[k] * w for k, w in WEIGHTS.items()) * 100
    return df.sort_values("score", ascending=False)

为什么用分位数排名而不是 Min-Max 归一化？因为粉丝数、互动率这类指标都是典型的长尾分布，少数头部账号会把 Min-Max 的结果压缩到接近 0。分位数排名只关心相对位置，对极端值不敏感，四个维度的量纲也自然统一到了 0～1。

发帖频率设置了上限，是为了避免"刷屏型"账号（例如每天发几十条的自动转载号）在活跃度上得分过高。

九、输出与人工复核

# main.py  示例代码
import pandas as pd
from discover import discover
from profiles import fetch_profiles
from timeline import recent_tweets
from metrics import compute
from score import score

hits = discover()
candidates = [n for n, c in hits.most_common(200)]
rows = []
for u in fetch_profiles(candidates):
    name = u["screen_name"]
    rows.append(compute(u, recent_tweets(name), hits.get(name.lower(), 0)))

ranked = score(pd.DataFrame(rows))
ranked["profile"] = "https://x.com/" + ranked["screen_name"]
cols = ["screen_name", "name", "score", "followers", "eng_rate",
        "posts_per_day", "topic_ratio", "listed", "profile"]
ranked[cols].to_csv("kol_ranking.csv", index=False, encoding="utf-8-sig")
print(ranked[cols].head(20).to_string(index=False))

输出的 CSV 用 utf-8-sig 编码，直接用 Excel 打开不会出现中文乱码。名单中保留了各项分指标和主页链接，便于复核时判断"这个账号为什么排在这里"。

自动打分只能完成初筛，最终决策前建议人工复核前 20～50 个账号：浏览主页和最近的推文，确认内容质量、立场和语气是否符合合作需求。

十、权重调优与常见问题

权重怎么定：没有标准答案。品牌曝光类需求可以提高 reach 的权重；技术布道类需求更看重 relevance 和 engagement。可以先由业务方给出一份"理想 KOL"的样例名单，再调整权重，使样例账号尽量排在前面；
小号互动率虚高：粉丝数很少的账号，互动率天然偏高。规模下限和 reach 维度可以起到平衡作用，也可以把互动率乘以一个随粉丝数增长的置信系数；
互刷检测：如果某个账号的互动主要来自少数几个固定账号，就可能存在互刷。进阶做法是抽样分析回复者，计算回复者的集中度；
listed_count 是一个容易被忽视的信号：被其他用户加入列表的次数，往往反映了账号在专业圈子中的认可度，可以作为额外维度加入打分；
定期更新：KOL 的活跃度会变化，建议每月重新运行一次，对比排名变化。

十一、合规提示

本文只使用公开的账号资料和公开推文，目的是识别领域内的活跃创作者。实际使用时请注意：遵守数据来源平台的服务条款和当地隐私法规；不收集与评估无关的个人信息；不对名单中的账号进行骚扰式的批量私信；对外合作时以账号公开披露的联系方式为准。

十二、小结

只按粉丝数排序容易被"死粉""跨领域大号"误导，需要综合规模、互动、活跃度和相关度四个维度；
通过关键词搜索发现候选，用批量接口补全画像，再用时间线数据计算行为指标；
互动率使用中位数，频率只看近期，归一化采用分位数排名，都是为了应对社交数据的长尾特性；
规则过滤加人工复核，才能让自动化结果真正可用。

说明：文中代码为示例代码，字段名和返回结构请以所用数据接口的文档为准，权重和阈值需要根据具体业务调整。

参考资料

示例数据接口文档：https://docs.twexapi.io

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

数据分析

数据挖掘

python