"""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()