endorphin-tabularium-docs/scripts/summarize-ai-income.py

"""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)
    lines = [
        '# AI 公司收入样本分布快照', '',
        f"截至 {DATA['asOf']}，7日窗口内有 {len(points)} 个产品的美元平台MRR可纳入分组描述。创意软件订阅组的样本均值较高、CV也较大；营销软件组与消费者个人应用组的CV接近。各组只有1至5个产品，尚不足以估计行业总体或造假概率。", '',
        f"这是已采集资料的固定快照。[动态工作台]({BASE})提供筛选与原始证据。窗口按最后同步日计算，不保证所有点同日，也不是{len(DATA['companies']):,}条公司/产品记录的收入普查。", '',
        '## 7日窗口的分组结果', '',
        'MRR单位为美元每月。CV为样本标准差除以均值；最大值剔除降幅表示移除最高收入产品后均值下降的比例。单个样本不计算离散度。', '',
        '| 业务及收费模式 | 平台 | 样本数 | 均值 | 中位数 | CV | 最大值剔除降幅 |',
        '| --- | --- | --- | --- | --- | --- | --- |',
    ]
    for key, values in sorted(groups(points).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 += ['',
        '在至少3个样本的组内，内容创意软件订阅的均值较高，但移除最大样本后的下降也最大。营销软件与个人应用的CV仅有微小差别，不据此宣称哪个领域更稳定。两个内容类单样本组虽有更高数值，不能作为典型收入或领域排名。', '',
        '均值和中位数接近只描述当前采集点，并不证明样本具有代表性。平台公开者、成立年、规模、获客渠道与收费方式均可能影响结果。', '',
        '## 样本与证据入口', '',
        '点击产品名称可打开公司明细，再查看平台记录及其限制。连接已过期的近期快照仍按所标日期保留，并不表示连接当前有效。', '',
        '| 产品 | 成立年 | 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, default=ROOT / '.artifacts/documents-migration.json')
    args = parser.parse_args()
    verify()
    content = render()
    if args.check:
        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, samples and report are consistent')
    else:
        TARGET.parent.mkdir(parents=True, exist_ok=True)
        TARGET.write_text(content)
        print(TARGET.relative_to(ROOT))


if __name__ == '__main__':
    main()