如何通过 API 分析 Twitter 竞争对手
本指南介绍使用 Sorsa API 分析 X(原 Twitter)竞争对手的完整工作流:资料对标、内容策略拆解、受众构成、公众情感和声量份额。各阶段对应具体端点,并可组合为定期运行的每周报告。 所有示例使用 Python 3.8+ 和requests。请将各代码片段中的 YOUR_API_KEY 替换为实际密钥。页面底部提供包含所有辅助函数的完整整合脚本。
免费开始: 本指南全部端点都可使用初始赠送的 100 次请求,一次性赠送,无需信用卡,永不过期。可先以较浅的分页深度试跑完整流程,再选择套餐。
注意: 更多策略背景和实例请参阅博客上的 Twitter 竞品分析:开发者指南。
无代码方案: 临时并排比较两个账号时,可使用资料对比工具。它返回粉丝数、互动率、每条推文平均点赞和转推数、发帖频率以及账号年龄。互动率计算器可计算单个账号的每条推文互动率。
环境设置
import requests
import time
import csv
from pathlib import Path
from datetime import date, datetime, timedelta, timezone
API_KEY = "YOUR_API_KEY"
BASE = "https://api.sorsa.io/v3"
HEADERS = {"ApiKey": API_KEY}
JSON_HEADERS = {**HEADERS, "Content-Type": "application/json"}
ApiKey 请求头验证身份,详情见身份验证。
第一阶段:资料对标
端点:GET /v3/info、GET /v3/info-batch
建立基线:粉丝数、发帖量、账号年龄、简介和认证状态。使用 /info-batch 一次获取最多 100 份资料,只扣除一次请求配额。
快照脚本
def get_profiles(usernames):
"""Fetch profiles for up to 100 accounts in a single API call."""
resp = requests.get(
f"{BASE}/info-batch",
headers=HEADERS,
params=[("usernames", u) for u in usernames],
timeout=30,
)
resp.raise_for_status()
return resp.json().get("users", [])
competitors = ["stripe", "wise", "revolutapp"]
profiles = get_profiles(competitors)
print(f"{'Handle':<18} {'Followers':>12} {'Tweets':>10} {'Following':>10} {'Verified':>10}")
print("-" * 64)
for p in profiles:
print(
f"@{p['username']:<17} "
f"{p['followers_count']:>12,} "
f"{p['tweets_count']:>10,} "
f"{p['followings_count']:>10,} "
f"{str(p.get('verified', False)):>10}"
)
追踪随时间的增长
一次快照只能提供基线,测量增长至少需要两个带日期的观察点。通过 cron、GitHub Actions 等按日或按周记录快照,并计算变化量:def log_snapshot(profiles, output_file="snapshots.csv"):
"""Append today's snapshot to a running CSV log."""
file_exists = Path(output_file).exists()
today = date.today().isoformat()
with open(output_file, "a", newline="") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["date", "username", "followers", "tweets", "following"])
for p in profiles:
writer.writerow([
today,
p["username"],
p["followers_count"],
p["tweets_count"],
p["followings_count"],
])
def compute_growth(csv_file, username, days=7):
with open(csv_file, encoding="utf-8") as f:
rows = [r for r in csv.DictReader(f) if r["username"].lower() == username.lower()]
if len(rows) < 2:
return None
rows.sort(key=lambda row: row["date"])
latest_row = rows[-1]
cutoff = date.fromisoformat(latest_row["date"]) - timedelta(days=days)
earlier_rows = [r for r in rows if date.fromisoformat(r["date"]) <= cutoff]
if not earlier_rows:
return None
latest = int(latest_row["followers"])
earlier = int(earlier_rows[-1]["followers"])
return ((latest - earlier) / earlier) * 100 if earlier else None
log_snapshot(profiles)
for handle in competitors:
g = compute_growth("snapshots.csv", handle, days=7)
if g is not None:
print(f"@{handle}: {g:+.2f}% weekly follower growth")
days;若需更贴近目标时间,应每天记录。生产历史中应同时保存用户 ID 和用户名,避免改名后将同一账号拆成多条记录。
增长公式:
Growth Rate % = ((Followers Today - Followers N Days Ago) / Followers N Days Ago) * 100
简介和定位变化
/info 返回 description、location、bio_urls 和 created_at。比较快照中的这些字段,可发现简介、链接目标等定位变化,无需额外调用。
第二阶段:内容策略
端点:POST /v3/user-tweets、POST /v3/search-tweets
获取竞争对手近期推文,拆解原创、回复、引用和转推的内容组合、平均互动和表现最佳的帖子。
获取近期推文
/user-tweets 每页最多返回 20 条。与官方 X API 时间线端点不同,它没有 3,200 条硬上限,可继续分页获取更早历史。时间跨度很大时,使用带 since:/until: 的 /search-tweets 更可靠,见下方说明。
def fetch_user_tweets(username, max_pages=10):
"""Pull a competitor's recent tweets via pagination."""
all_tweets = []
cursor = None
for _ in range(max_pages):
body = {"username": username}
if cursor:
body["next_cursor"] = cursor
resp = requests.post(
f"{BASE}/user-tweets",
headers=JSON_HEADERS,
json=body,
timeout=30,
)
resp.raise_for_status()
data = resp.json()
all_tweets.extend(data.get("tweets", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_tweets
拆解内容组合
def analyze_content(tweets, username):
if not tweets:
return None
total = len(tweets)
likes = [t.get("likes_count", 0) for t in tweets]
retweets = [t.get("retweet_count", 0) for t in tweets]
replies = [t.get("reply_count", 0) for t in tweets]
original = sum(1 for t in tweets if not t.get("is_reply") and not t.get("retweeted_status"))
reply_count = sum(1 for t in tweets if t.get("is_reply"))
quote_count = sum(1 for t in tweets if t.get("is_quote_status"))
with_media = sum(1 for t in tweets if t.get("entities"))
top_tweet = max(tweets, key=lambda t: t.get("likes_count", 0))
return {
"username": username,
"sample_size": total,
"avg_likes": sum(likes) / total,
"avg_retweets": sum(retweets) / total,
"avg_replies": sum(replies) / total,
"original_pct": original / total * 100,
"reply_pct": reply_count / total * 100,
"quote_pct": quote_count / total * 100,
"media_pct": with_media / total * 100,
"top_tweet_likes": top_tweet.get("likes_count", 0),
"top_tweet_text": top_tweet.get("full_text", "")[:200],
}
for handle in competitors:
tweets = fetch_user_tweets(handle, max_pages=10)
result = analyze_content(tweets, handle)
if result:
print(f"\n@{result['username']} (n={result['sample_size']})")
print(f" Avg likes/tweet: {result['avg_likes']:.1f}")
print(f" Avg retweets/tweet: {result['avg_retweets']:.1f}")
print(f" Content mix: {result['original_pct']:.0f}% original / "
f"{result['reply_pct']:.0f}% replies / {result['quote_pct']:.0f}% quotes / "
f"{result['media_pct']:.0f}% with media")
print(f" Top tweet: ({result['top_tweet_likes']} likes) {result['top_tweet_text']}")
历史对比
比较同一账号两个时间窗口,例如第一季度与第四季度时,应从/user-tweets 改用 /search-tweets,并使用 since: 和 until:。完整语法见搜索运算符,回填方式见历史数据。
def fetch_tweets_in_range(username, since_date, until_date):
query = f"from:{username} since:{since_date} until:{until_date}"
resp = requests.post(
f"{BASE}/search-tweets",
headers=JSON_HEADERS,
json={"query": query, "order": "latest"},
timeout=30,
)
resp.raise_for_status()
return resp.json().get("tweets", []) # First page only; paginate for a full period.
q1_tweets = fetch_tweets_in_range("stripe", "2026-01-01", "2026-04-01")
q4_tweets = fetch_tweets_in_range("stripe", "2025-10-01", "2026-01-01")
第三阶段:受众构成
端点:GET /v3/followers、GET /v3/verified-followers、GET /v3/followers-stats
竞争对手的粉丝列表揭示其受众。主要有两种成本不同的方式。
认证粉丝(低成本)
/verified-followers 只返回关注目标的认证账号。这是受众中信息价值较高的一部分,成本也远低于获取完整关系图谱。
def fetch_verified_followers(username, max_pages=10):
all_users = []
cursor = None
for _ in range(max_pages):
params = {"username": username}
if cursor:
params["next_cursor"] = cursor
resp = requests.get(
f"{BASE}/verified-followers",
headers=HEADERS,
params=params,
timeout=30,
)
resp.raise_for_status()
data = resp.json()
all_users.extend(data.get("users", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_users
for handle in competitors:
verified = fetch_verified_followers(handle, max_pages=5)
top = sorted(verified, key=lambda u: u.get("followers_count", 0), reverse=True)[:10]
print(f"\n@{handle}: {len(verified)} verified followers fetched")
for u in top:
print(f" @{u['username']:<25} {u['followers_count']:>10,} followers")
完整粉丝提取(高成本)
/followers 每页最多返回 200 份资料。100 万粉丝的账号完整提取约需 5,000 次请求。请据此规划,套餐上限见价格,批量和预算方式见优化 API 使用。
def fetch_all_followers(username, max_pages=200):
"""Pull all followers via pagination. Cost scales with account size."""
all_users = []
cursor = None
for _ in range(max_pages):
params = {"username": username}
if cursor:
params["next_cursor"] = cursor
resp = requests.get(
f"{BASE}/followers",
headers=HEADERS,
params=params,
timeout=30,
)
resp.raise_for_status()
data = resp.json()
all_users.extend(data.get("users", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_users
受众重叠
获得两个账号的粉丝列表后,按用户 ID 求集合交集:followers_a = {u["id"] for u in fetch_all_followers("competitor_a")}
followers_b = {u["id"] for u in fetch_all_followers("competitor_b")}
overlap = followers_a & followers_b
only_a = followers_a - followers_b
only_b = followers_b - followers_a
print(f"Shared audience: {len(overlap):,}")
print(f"Unique to @competitor_a: {len(only_a):,}")
print(f"Unique to @competitor_b: {len(only_b):,}")
denominator = min(len(followers_a), len(followers_b))
overlap_ratio = len(overlap) / denominator if denominator else 0
print(f"Overlap ratio: {overlap_ratio:.1%}")
加密货币与 Web3 粉丝分类
对于 Sorsa 加密货币数据库中的账号,/followers-stats 返回意见领袖、项目和 VC 分类。背景见 Sorsa Score 与加密货币分析。
def get_follower_breakdown(username):
resp = requests.get(
f"{BASE}/followers-stats",
headers=HEADERS,
params={"username": username},
timeout=30,
)
resp.raise_for_status()
return resp.json()
for handle in ["VitalikButerin", "saylor"]:
stats = get_follower_breakdown(handle)
print(f"\n@{handle}:")
print(f" Tracked followers: {stats['followers_count']}")
print(f" Influencers: {stats['influencers_count']}")
print(f" Projects: {stats['projects_count']}")
print(f" VCs: {stats['venture_capitals_count']}")
第四阶段:公众情感与提及
端点:POST /v3/mentions
/mentions 支持最低点赞、转推、回复数和日期范围筛选。使用 min_likes 可减少机器人回复、自动标记等低价值噪声。完整筛选项见追踪提及。
获取高互动提及
def fetch_mentions(handle, min_likes=10, since_date=None, until_date=None, max_pages=5):
all_mentions = []
cursor = None
for _ in range(max_pages):
body = {"query": handle, "order": "popular", "min_likes": min_likes}
if since_date:
body["since_date"] = since_date
if until_date:
body["until_date"] = until_date
if cursor:
body["next_cursor"] = cursor
resp = requests.post(
f"{BASE}/mentions",
headers=JSON_HEADERS,
json=body,
timeout=30,
)
resp.raise_for_status()
data = resp.json()
all_mentions.extend(data.get("tweets", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_mentions
使用 VADER 进行情感分类
VADER 是针对社交媒体文本优化的开源情感分析库。它在本地运行,无逐次调用费用,并能较好处理否定、程度词和表情符号。# pip install vaderSentiment
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
analyzer = SentimentIntensityAnalyzer()
def classify_sentiment(mentions):
results = {"positive": [], "negative": [], "neutral": []}
for m in mentions:
score = analyzer.polarity_scores(m["full_text"])["compound"]
if score >= 0.05:
results["positive"].append((score, m))
elif score <= -0.05:
results["negative"].append((score, m))
else:
results["neutral"].append((score, m))
return results
for handle in competitors:
mentions = fetch_mentions(handle, min_likes=10, max_pages=5)
s = classify_sentiment(mentions)
print(f"\n@{handle}: {len(mentions)} mentions analyzed")
print(f" Positive: {len(s['positive'])} Negative: {len(s['negative'])} Neutral: {len(s['neutral'])}")
if s["negative"]:
worst = min(s["negative"], key=lambda x: x[0])
text = worst[1]["full_text"][:150].replace("\n", " ")
print(f" Sharpest negative: {text}...")
full_text 输入 LLM API(如 OpenAI、Anthropic)。混合方案先用 VADER 筛选,仅对被标记或高互动提及使用 LLM,便于控制成本。
第五阶段:声量份额
声量份额(SOV)衡量某品牌的提及量占整个类别的比例,公式为:SOV = (your mentions in period) / (your mentions + sum of competitor mentions in period)
def count_mentions(handle, days=7, min_likes=0):
until = datetime.now(timezone.utc).date().isoformat()
since = (datetime.now(timezone.utc) - timedelta(days=days)).date().isoformat()
mentions = fetch_mentions(
handle,
min_likes=min_likes,
since_date=since,
until_date=until,
max_pages=20,
)
return len(mentions)
brand = "your_handle"
your_mentions = count_mentions(brand, days=7, min_likes=5)
competitor_mentions = {h: count_mentions(h, days=7, min_likes=5) for h in competitors}
total = your_mentions + sum(competitor_mentions.values())
print(f"\nShare of voice, last 7 days (min 5 likes):")
print(f" @{brand:<20} {your_mentions:>5} ({(your_mentions/total*100 if total else 0):.1f}%)")
for h, n in sorted(competitor_mentions.items(), key=lambda x: -x[1]):
print(f" @{h:<20} {n:>5} ({(n/total*100 if total else 0):.1f}%)")
- 设置最低互动门槛,例如
min_likes=5,以过滤机器人和垃圾内容噪声。 - 追踪周环比变化,而不只看绝对快照。类别层面的事件可能扭曲绝对数量,掩盖自身变化。
- 这些计数受
max_pages限制,只描述获取到的样本。若要可比,应使用相同日期和筛选,并确认每个品牌都已读取全部页面;否则标明报告基于抽样。 - 如果计算类别关键词的 SOV,例如“embedded finance”而非品牌提及,应将
/mentions替换为/search-tweets,并以该关键词查询作为分母。
整合的每周报告脚本
此脚本组合五个阶段,可放入 cron 任务、GitHub Actions 定时运行或其他任务执行器。import requests
import csv
import time
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
API_KEY = "YOUR_API_KEY"
BASE = "https://api.sorsa.io/v3"
HEADERS = {"ApiKey": API_KEY}
JSON_HEADERS = {**HEADERS, "Content-Type": "application/json"}
BRAND = "your_handle"
COMPETITORS = ["competitor1", "competitor2", "competitor3"]
SNAPSHOT_FILE = "snapshots.csv"
analyzer = SentimentIntensityAnalyzer()
def get_profiles(usernames):
resp = requests.get(
f"{BASE}/info-batch",
headers=HEADERS,
params=[("usernames", u) for u in usernames],
timeout=30,
)
resp.raise_for_status()
return resp.json().get("users", [])
def log_snapshot(profiles, output_file=SNAPSHOT_FILE):
file_exists = Path(output_file).exists()
today = date.today().isoformat()
with open(output_file, "a", newline="") as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(["date", "username", "followers", "tweets", "following"])
for p in profiles:
writer.writerow([
today, p["username"], p["followers_count"],
p["tweets_count"], p["followings_count"],
])
def compute_growth(csv_file, username, days=7):
with open(csv_file, encoding="utf-8") as f:
rows = [r for r in csv.DictReader(f) if r["username"].lower() == username.lower()]
if len(rows) < 2:
return None
rows.sort(key=lambda row: row["date"])
latest_row = rows[-1]
cutoff = date.fromisoformat(latest_row["date"]) - timedelta(days=days)
earlier_rows = [r for r in rows if date.fromisoformat(r["date"]) <= cutoff]
if not earlier_rows:
return None
latest = int(latest_row["followers"])
earlier = int(earlier_rows[-1]["followers"])
return ((latest - earlier) / earlier) * 100 if earlier else None
def fetch_user_tweets(username, max_pages=5):
all_tweets = []
cursor = None
for _ in range(max_pages):
body = {"username": username}
if cursor:
body["next_cursor"] = cursor
resp = requests.post(f"{BASE}/user-tweets", headers=JSON_HEADERS, json=body, timeout=30)
resp.raise_for_status()
data = resp.json()
all_tweets.extend(data.get("tweets", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_tweets
def analyze_content(tweets, username):
if not tweets:
return None
total = len(tweets)
likes = [t.get("likes_count", 0) for t in tweets]
original = sum(1 for t in tweets if not t.get("is_reply") and not t.get("retweeted_status"))
with_media = sum(1 for t in tweets if t.get("entities"))
return {
"username": username,
"sample_size": total,
"avg_likes": sum(likes) / total,
"original_pct": original / total * 100,
"media_pct": with_media / total * 100,
}
def fetch_verified_followers(username, max_pages=3):
all_users = []
cursor = None
for _ in range(max_pages):
params = {"username": username}
if cursor:
params["next_cursor"] = cursor
resp = requests.get(f"{BASE}/verified-followers", headers=HEADERS, params=params, timeout=30)
resp.raise_for_status()
data = resp.json()
all_users.extend(data.get("users", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_users
def fetch_mentions(handle, min_likes=10, since_date=None, until_date=None, max_pages=5):
all_mentions = []
cursor = None
for _ in range(max_pages):
body = {"query": handle, "order": "popular", "min_likes": min_likes}
if since_date:
body["since_date"] = since_date
if until_date:
body["until_date"] = until_date
if cursor:
body["next_cursor"] = cursor
resp = requests.post(f"{BASE}/mentions", headers=JSON_HEADERS, json=body, timeout=30)
resp.raise_for_status()
data = resp.json()
all_mentions.extend(data.get("tweets", []))
cursor = data.get("next_cursor")
if not cursor:
break
time.sleep(0.1)
return all_mentions
def classify_sentiment(mentions):
results = {"positive": [], "negative": [], "neutral": []}
for m in mentions:
score = analyzer.polarity_scores(m["full_text"])["compound"]
if score >= 0.05:
results["positive"].append((score, m))
elif score <= -0.05:
results["negative"].append((score, m))
else:
results["neutral"].append((score, m))
return results
def count_mentions(handle, days=7, min_likes=0):
until = datetime.now(timezone.utc).date().isoformat()
since = (datetime.now(timezone.utc) - timedelta(days=days)).date().isoformat()
return len(fetch_mentions(handle, min_likes=min_likes, since_date=since, until_date=until, max_pages=20))
def header(text):
line = "=" * 64
print(f"\n{line}\n{text}\n{line}")
def run_weekly_report():
header("PHASE 1: PROFILE BENCHMARKS")
profiles = get_profiles(COMPETITORS + [BRAND])
print(f"{'Handle':<18} {'Followers':>12} {'Tweets':>10} {'Verified':>10}")
for p in profiles:
print(f"@{p['username']:<17} {p['followers_count']:>12,} "
f"{p['tweets_count']:>10,} {str(p.get('verified', False)):>10}")
log_snapshot(profiles)
for h in COMPETITORS + [BRAND]:
g = compute_growth(SNAPSHOT_FILE, h, days=7)
if g is not None:
print(f" @{h}: {g:+.2f}% weekly follower growth")
header("PHASE 2: CONTENT STRATEGY")
for handle in COMPETITORS:
tweets = fetch_user_tweets(handle, max_pages=5)
result = analyze_content(tweets, handle)
if result:
print(f"@{result['username']}: avg {result['avg_likes']:.0f} likes/tweet, "
f"{result['original_pct']:.0f}% original, "
f"{result['media_pct']:.0f}% with media")
header("PHASE 3: VERIFIED FOLLOWERS")
for handle in COMPETITORS:
verified = fetch_verified_followers(handle, max_pages=3)
print(f"@{handle}: {len(verified)} verified followers in top pages")
header("PHASE 4: SENTIMENT")
for handle in COMPETITORS:
mentions = fetch_mentions(handle, min_likes=10, max_pages=3)
s = classify_sentiment(mentions)
print(f"@{handle}: {len(s['positive'])} pos / {len(s['negative'])} neg "
f"/ {len(s['neutral'])} neutral (n={len(mentions)})")
header("PHASE 5: SHARE OF VOICE (7d)")
your_n = count_mentions(BRAND, days=7, min_likes=5)
comp_n = {h: count_mentions(h, days=7, min_likes=5) for h in COMPETITORS}
total = your_n + sum(comp_n.values())
if total:
print(f" @{BRAND}: {your_n} ({your_n/total*100:.1f}%)")
for h, n in sorted(comp_n.items(), key=lambda x: -x[1]):
print(f" @{h}: {n} ({(n/total*100 if total else 0):.1f}%)")
if __name__ == "__main__":
run_weekly_report()