179 lines
10 KiB
Python
179 lines
10 KiB
Python
"""Render a fixed research snapshot; --check performs data analysis only, no build."""
|
||
import argparse
|
||
import datetime as dt
|
||
import json
|
||
from pathlib import Path
|
||
import statistics as st
|
||
import sys
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
DATA = json.loads((ROOT / 'content/research/ai-companies.seed.json').read_text())
|
||
REFERENCE = json.loads((ROOT / 'content/research/ai-income-analysis.json').read_text())
|
||
TARGET = ROOT / '.artifacts/reports/01-income-snapshot.md'
|
||
SOURCE_SLUG = 'ai-company-research/01-income-snapshot'
|
||
BASE = '/docs/other-research/ai-company-research/overview'
|
||
COMPANIES = {c['id']: c for c in DATA['companies']}
|
||
CATEGORIES = {c['id']: c['name'] for c in DATA['categories']}
|
||
END = dt.date.fromisoformat(DATA['asOf'])
|
||
|
||
|
||
def sample(days):
|
||
latest, excluded = {}, []
|
||
for o in DATA['observations']:
|
||
if o.get('selection') != 'platform' or o.get('metric') != 'mrr':
|
||
continue
|
||
c = COMPANIES[o['company']]
|
||
day = o.get('observedAt')
|
||
try:
|
||
parsed = dt.date.fromisoformat(day)
|
||
valid_date = parsed.isoformat() == day
|
||
except (TypeError, ValueError):
|
||
valid_date = False
|
||
valid = (c['year'] in (2025, 2026) and o.get('origin') == 'payment-platform'
|
||
and o['unit'] == 'USD' and o['bound'] == '='
|
||
and o.get('comparisonGroup') == 'trustmrr-usd-mrr'
|
||
and o.get('provider') and isinstance(o['value'], (int, float))
|
||
and 0 <= o['value'] < float('inf') and valid_date
|
||
and 0 <= (END - parsed).days <= days)
|
||
if not valid:
|
||
excluded.append(o)
|
||
continue
|
||
prior = latest.get(c['id'])
|
||
if prior is None or day > prior['observedAt']:
|
||
latest[c['id']] = o
|
||
return latest, excluded
|
||
|
||
|
||
def groups(points):
|
||
result = {}
|
||
for cid, o in points.items():
|
||
c = COMPANIES[cid]
|
||
key = (c['cat'], o['provider'], c.get('businessModel', '收费模式未核验'))
|
||
result.setdefault(key, []).append(o['value'])
|
||
return result
|
||
|
||
|
||
def stats(values):
|
||
mean = st.mean(values)
|
||
sd = st.stdev(values) if len(values) > 1 else None
|
||
reduced = st.mean(sorted(values)[:-1]) if len(values) > 1 else None
|
||
return dict(n=len(values), mean=mean, median=st.median(values),
|
||
sampleDeviation=sd, cv=sd / mean if sd is not None and mean else None,
|
||
meanWithoutLargest=reduced,
|
||
largestRemovalDrop=(mean - reduced) / mean if reduced is not None and mean else None)
|
||
|
||
|
||
def verify():
|
||
assert REFERENCE['sourceVersion'] == DATA['version'], 'Reference revision mismatch'
|
||
for window in REFERENCE['windows']:
|
||
points, _ = sample(window['days'])
|
||
assert set(points) == set(window['products']), 'Sample selection mismatch'
|
||
computed = groups(points)
|
||
assert len(computed) == len(window['groups'])
|
||
for group in window['groups']:
|
||
values = computed[(group['category'], group['provider'], group['model'])]
|
||
for key, value in stats(values).items():
|
||
expected = group[key]
|
||
assert (value is None and expected is None) or (
|
||
value is not None and expected is not None
|
||
and abs(value - expected) <= 1e-8 * max(1, abs(expected))), key
|
||
|
||
|
||
def money(value):
|
||
return f'{value:,.0f}' if value is not None else '不计算'
|
||
|
||
|
||
def link(cid):
|
||
return f"[{COMPANIES[cid]['name']}]({BASE}?company={cid}#research-company-detail)"
|
||
|
||
|
||
def render():
|
||
points, excluded = sample(7)
|
||
grouped = groups(points)
|
||
described = [(key, stats(values)) for key, values in sorted(grouped.items())]
|
||
sizes = [value['n'] for _, value in described]
|
||
eligible = [(key, value) for key, value in described if value['n'] >= 3]
|
||
interpretation = '各组均不足3个样本，暂不比较分组离散度或收入高低。'
|
||
if eligible:
|
||
highest_key, highest = max(eligible, key=lambda item: item[1]['mean'])
|
||
sensitive_key, sensitive = max(eligible, key=lambda item: item[1]['largestRemovalDrop'] or 0)
|
||
cvs = [value['cv'] for _, value in eligible if value['cv'] is not None]
|
||
cv_text = f"CV范围为{min(cvs):.3f}—{max(cvs):.3f}。" if cvs else '各组CV均不可计算。'
|
||
interpretation = (
|
||
f"在至少3个样本的组内，当前均值最高的是{CATEGORIES[highest_key[0]]} · {highest_key[2]}"
|
||
f"（{highest_key[1]}，{money(highest['mean'])}美元每月）。{cv_text}"
|
||
f"剔除最大值后均值下降最多的是{CATEGORIES[sensitive_key[0]]} · {sensitive_key[2]}"
|
||
f"（{sensitive_key[1]}，下降{sensitive['largestRemovalDrop'] or 0:.1%}）。"
|
||
'这些只是当前采集点的描述，不据微小CV差异宣称某领域更稳定。'
|
||
)
|
||
size_text = f"各组目前只有{min(sizes)}至{max(sizes)}个产品" if sizes else '当前没有可比较的产品'
|
||
lines = [
|
||
'# AI 公司收入样本分布快照', '',
|
||
f"截至 {DATA['asOf']}，7日窗口内有 {len(points)} 个产品的美元平台MRR可纳入分组描述。按业务、支付平台和收费模式分别计算均值、中位数与离散度。{size_text}，尚不足以估计行业总体或造假概率。", '',
|
||
f"这是已采集资料的固定快照。[动态工作台]({BASE})提供筛选与原始证据。窗口按最后同步日计算，不保证所有点同日，也不是{len(DATA['companies']):,}条公司/产品记录的收入普查。", '',
|
||
'## 7日窗口的分组结果', '',
|
||
'MRR单位为美元每月。CV为样本标准差除以均值；最大值剔除降幅表示移除最高收入产品后均值下降的比例。单个样本不计算离散度。', '',
|
||
'| 业务及收费模式 | 平台 | 样本数 | 均值 | 中位数 | CV | 最大值剔除降幅 |',
|
||
'| --- | --- | --- | --- | --- | --- | --- |',
|
||
]
|
||
for key, values in sorted(grouped.items()):
|
||
cat, provider, model = key
|
||
v = stats(values)
|
||
cv = f"{v['cv']:.3f}" if v['cv'] is not None else '不计算'
|
||
drop = f"{v['largestRemovalDrop']:.1%}" if v['largestRemovalDrop'] is not None else '不计算'
|
||
lines.append(f"| {CATEGORIES[cat]} · {model} | {provider} | {v['n']} | {money(v['mean'])} | {money(v['median'])} | {cv} | {drop} |")
|
||
lines += ['',
|
||
interpretation, '',
|
||
'单个样本不计算离散度，也不能作为典型收入或领域排名；不同收费模式和支付平台的分组保留各自含义。', '',
|
||
'均值和中位数接近只描述当前采集点，并不证明样本具有代表性。平台公开者、成立年、规模、获客渠道与收费方式均可能影响结果。', '',
|
||
'## 样本与证据入口', '',
|
||
'点击产品名称可打开公司明细，再查看平台记录及其限制。连接已过期的近期快照仍按所标日期保留，并不表示连接当前有效。', '',
|
||
'| 产品 | 成立年 | MRR美元每月 | 最后同步 | 平台 | 连接状态 |',
|
||
'| --- | --- | --- | --- | --- | --- |',
|
||
]
|
||
for cid, o in sorted(points.items()):
|
||
status = '已过期的日期快照' if o.get('connectionExpired') else '采集时未标过期'
|
||
lines.append(f"| {link(cid)} | {COMPANIES[cid]['year']} | {money(o['value'])} | {o['observedAt']} | {o['provider']} | {status} |")
|
||
fresh = stats(groups(points)[('C06', 'Stripe', '软件订阅')])
|
||
broad, _ = sample(90)
|
||
older = stats(groups(broad)[('C06', 'Stripe', '软件订阅')])
|
||
lines += ['', '## 扩大窗口后的敏感性', '',
|
||
f"营销软件订阅组从7日窗口的{fresh['n']}个产品扩大到90日窗口的{older['n']}个产品，均值从{money(fresh['mean'])}升至{money(older['mean'])}美元每月，中位数从{money(fresh['median'])}变为{money(older['median'])}，CV从{fresh['cv']:.3f}升至{older['cv']:.3f}。", '',
|
||
f"新增点来自{link('AI016')}的8月历史快照，连接已过期。移除该最大值后，均值回到{money(older['meanWithoutLargest'])}美元，下降{older['largestRemovalDrop']:.1%}。这是日期与样本构成的敏感性，不能解释为行业风险变化。", '',
|
||
'## 默认窗口排除的历史记录', '',
|
||
'| 产品 | MRR美元每月 | 最后同步 | 排除原因 |',
|
||
'| --- | --- | --- | --- |',
|
||
]
|
||
for o in excluded:
|
||
lines.append(f"| {link(o['company'])} | {money(o['value'])} | {o.get('observedAt', '未知')} | 超过7日窗口，保留历史观测 |")
|
||
lines += ['', '## 对 ARR 与商业模式的含义', '',
|
||
'MRR乘12只是参考年化。公司声明的ARR、合同管线、近30日收款、一次性额度和服务交付收入仍需分开；支付平台重建指标也不等于财务审计。方法与平台限制见[TrustMRR FAQ](https://trustmrr.com/faq)。', '',
|
||
'当前更适合选择访谈对象和设计收费试验，不能据最高均值直接决定业务方向。软件订阅优先核留存与推理成本，服务交付优先核人工工时和返工，消费者应用优先核渠道成本、退款和平台抽成。', '',
|
||
f"行业ARR比较仍需补齐同一业务截止日、币种、经常性定义及业务边界。收入日期不完整或同页数值冲突的声明，继续保留在[核查工作台]({BASE})，不补成零值或精确数字。", '',
|
||
]
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('--check', action='store_true')
|
||
parser.add_argument('--document-export', type=Path)
|
||
args = parser.parse_args()
|
||
verify()
|
||
content = render()
|
||
if args.check:
|
||
if args.document_export is not None:
|
||
bundle = json.loads(args.document_export.read_text())
|
||
stored = next((d['content'] for d in bundle['documents'] if d['sourceSlug'] == SOURCE_SLUG), None)
|
||
if stored != content:
|
||
sys.exit('Income snapshot differs; regenerate after reviewing data changes')
|
||
print('Income reference and samples verified; supplied document export compared when requested')
|
||
else:
|
||
TARGET.parent.mkdir(parents=True, exist_ok=True)
|
||
TARGET.write_text(content)
|
||
print(TARGET.relative_to(ROOT))
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|