Blueskyのリンクは30日で何%死ぬか — firehoseで前向きに測るlink rot

import asyncio, json, datetime
import websockets, tldextract

JETSTREAM = "wss://jetstream2.us-east.host.bsky.network/subscribe?wantedCollections=app.bsky.feed.post"

def classify(host):
    if host.endswith("github.com"):
        return "github"
    if any(n in host for n in ("nytimes", "bbc", "cnn", "reuters", "guardian")):
        return "news"
    return "blog_other"

async def capture(on_record, cursor=None):
    backoff = 1
    while True:
        url = JETSTREAM + (f"&cursor={cursor}" if cursor else "")
        try:
            async with websockets.connect(url, max_size=None) as ws:
                backoff = 1
                async for raw in ws:  # pagination via Jetstream time-cursor
                    ev = json.loads(raw)
                    cursor = ev.get("time_us", cursor)
                    rec = ev.get("commit", {}).get("record", {})
                    for facet in rec.get("facets", []):
                        for f in facet.get("features", []):
                            link = f.get("uri", "")
                            if f.get("$type", "").endswith("link") and link.startswith("http"):
                                host = tldextract.extract(link).registered_domain
                                on_record(
                                    uri=ev.get("commit", {}).get("rkey"),
                                    did=ev.get("did"),
                                    captured_at=datetime.datetime.utcnow().isoformat(),
                                    external_url=link,
                                    domain=host,
                                    domain_class=classify(host),
                                )
        except Exception:
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)

import numpy as np, pandas as pd

rng = np.random.default_rng(42)
N = 2_400_000
BASE = 0.087

# --- Synthetic corpus (replace with real link-rot table) ---
domains = ["bit.ly", "t.co", "wikipedia.org", "nytimes.com", "reddit.com", "cdn.example.com"]
dweight = [.08, .15, .12, .05, .30, .30]
drate = {"bit.ly": .22, "t.co": .20, "wikipedia.org": .02,
         "nytimes.com": .03, "reddit.com": .09, "cdn.example.com": .21}
dom = rng.choice(domains, N, p=dweight)
share = rng.lognormal(0.5, 1.0, N)          # distinct-post count
eng = rng.lognormal(2.0, 1.5, N)            # reposts+likes
p_dead = np.clip([drate[d] for d in dom] - 0.02 * np.log1p(share) - 0.01 * np.log1p(eng), .005, .95)
dead30 = rng.random(N) < p_dead
# day-of-death for dead links: heavy early mass (ephemeral)
death_day = np.where(
    dead30,
    np.minimum(30, rng.choice([1, 2, 3, 5, 7, 14, 21, 30], N, p=[.22, .14, .10, .09, .07, .13, .12, .13])),
    -1,
)
df = pd.DataFrame({"domain": dom, "share": share, "eng": eng, "dead30": dead30, "death_day": death_day})

# H1: domain concentration vs 8.7% baseline (min 500 links)
g = df.groupby("domain").agg(n=("dead30", "size"), rate=("dead30", "mean"))
g = g[g.n >= 500].assign(ratio=lambda x: x.rate / BASE).sort_values("rate", ascending=False)
print("H1 per-domain 30d dead rate (baseline %.3f):\n%s\n" % (BASE, g.round(4)))

# H2: survival timing — fraction of deaths occurring by day 7
dd = df.loc[df.dead30, "death_day"]
by7 = (dd <= 7).mean()
print("H2 share of 30d deaths by day7: %.3f  -> front-loaded" % by7)

git clone https://example.com/your-org/bluesky-linkrot.git
cd bluesky-linkrot
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt   # websockets, tldextract, numpy, pandas, scipy, statsmodels
python fetch.py    # Jetstreamを購読しURLコーパスを構築
python analyze.py  # ドメイン集計・生存タイミング・回帰を実行