AIの可視性サンプルサイズ：実際に必要な実行回数はどれくらいか

export SCRAPELESS_API_KEY="your_scrapeless_api_key"
mkdir -p ai-visibility-variance && cd ai-visibility-variance
printf 'YourBrand\nRivalOne\nRivalTwo\n' > watchlist.txt

# capture.py — capture a fixed engine x prompt x repeat matrix
import hashlib
import json
import os
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor

API = "https://api.scrapeless.com/api/v2/scraper/execute"
KEY = os.environ["SCRAPELESS_API_KEY"]
OUT = "runs"
REPEATS = 5

PROMPTS = {
    "P1": "What are the best web scraping APIs for developers in 2026?",
    "P2": "Which services provide cloud browsers for automated data collection?",
    "P3": "What tools do developers use to collect Google search results programmatically?",
    "P4": "Recommend a proxy provider for large-scale public web data collection.",
}

# Same prompt and country everywhere. Only actor-required extras differ, and
# every one of them is pinned: an input that moves between repeats would be
# measured as engine variance.
ENGINES = {
    "perplexity": ("scraper.perplexity", {}),
    "grok":       ("scraper.grok",       {"mode": "MODEL_MODE_FAST"}),
    "gemini":     ("scraper.gemini",     {}),
    "aimode":     ("scraper.aimode",     {}),
    "chatgpt":    ("scraper.chatgpt",    {"web_search": True, "shopping": False}),
}


def execute(actor, payload, timeout=300):
    request = urllib.request.Request(
        API,
        data=json.dumps({"actor": actor, "input": payload}).encode(),
        headers={"Content-Type": "application/json", "x-api-token": KEY},
        method="POST",
    )
    started = time.time()
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return json.loads(response.read().decode()), round(time.time() - started, 1)


def capture(job):
    engine, prompt_id, repeat = job
    actor, extra = ENGINES[engine]
    payload = {"prompt": PROMPTS[prompt_id], "country": "US", **extra}
    # Hash the actor with the payload. Three of these engines take an identical
    # payload, so a payload-only hash would collapse them into one fingerprint
    # and stop proving that each cell repeated the same request.
    fingerprint = hashlib.sha256(
        json.dumps({"actor": actor, **payload}, sort_keys=True).encode()).hexdigest()[:16]
    try:
        body, elapsed = execute(actor, payload)
    except Exception as exc:
        # A cell that returns nothing is logged as missing, never dropped.
        # A silent hole would bias every statistic computed downstream.
        return {"ok": False, "note": f"{engine} {prompt_id} r{repeat}: {type(exc).__name__}"}
    with open(f"{OUT}/{engine}__{prompt_id}__r{repeat}.json", "w") as handle:
        json.dump({
            "engine": engine, "prompt_id": prompt_id, "repeat": repeat,
            "elapsed_s": elapsed, "input_fingerprint": fingerprint,
            "captured_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "response": body,
        }, handle)
    return {"ok": True, "fingerprint": fingerprint}


os.makedirs(OUT, exist_ok=True)
jobs = [(e, p, r) for e in ENGINES for p in PROMPTS for r in range(1, REPEATS + 1)]
print(f"{len(jobs)} captures = {len(ENGINES)} engines"
      f" x {len(PROMPTS)} prompts x {REPEATS} repeats")

with ThreadPoolExecutor(max_workers=12) as pool:
    results = list(pool.map(capture, jobs))

stored = [r for r in results if r["ok"]]
print(f"stored {len(stored)} captures in {OUT}/")
print(f"missing cells: {[r['note'] for r in results if not r['ok']] or 'none'}")
print(f"{len({r['fingerprint'] for r in stored})} distinct input fingerprints "
      f"across {len(stored)} captures")

# layers.py — reduce every capture to three comparable measurement layers
import json
import os
import re
from urllib.parse import urlparse

WATCHLIST = [line.strip() for line in open("watchlist.txt") if line.strip()]

CITATIONS = {
    "perplexity": lambda r: [c.get("url") for c in r.get("web_results") or []],
    "grok":       lambda r: [c.get("url") for c in (r.get("footnotes") or {}).values()],
    "gemini":     lambda r: [c.get("url") for c in r.get("citations") or []],
    "aimode":     lambda r: [c.get("url") for c in r.get("citations") or []],
    "chatgpt":    lambda r: [c.get("url") for c in r.get("search_result") or []],
}
ANSWER_KEY = {
    "perplexity": "result_text", "grok": "full_response", "gemini": "result_text",
    "aimode": "result_text", "chatgpt": "result_text",
}


def registrable(url):
    netloc = urlparse(url).netloc.lower()
    return netloc[4:] if netloc.startswith("www.") else netloc


def mentions(text):
    """Watchlist names present in the answer, keyed by first-mention offset."""
    lowered = text.lower()
    found = {}
    for name in WATCHLIST:
        hit = re.search(rf"(?<![a-z0-9]){re.escape(name.lower())}(?![a-z0-9])", lowered)
        if hit:
            found[name] = hit.start()
    return found


rows = []
for filename in sorted(os.listdir("runs")):
    record = json.load(open(f"runs/{filename}"))
    result = record["response"].get("task_result") or {}
    engine = record["engine"]
    answer = result.get(ANSWER_KEY[engine]) or ""
    found = mentions(answer)
    rows.append({
        "engine": engine,
        "prompt_id": record["prompt_id"],
        "repeat": record["repeat"],
        "cited_hosts": sorted({
            host for url in CITATIONS[engine](result)
            if url and (host := registrable(url))
        }),
        "answer_chars": len(answer),
        "answer_tokens": sorted(set(re.findall(r"[a-z0-9]+", answer.lower()))),
        # Ordered by first mention, so list position survives into stage 3.
        "brands_present": sorted(found, key=found.get),
    })

with open("layers.jsonl", "w") as handle:
    for row in rows:
        handle.write(json.dumps(row) + "\n")

cited = sum(len(r["cited_hosts"]) for r in rows)
print(f"reduced {len(rows)} captures -> layers.jsonl")
print(f"L1 citations: {cited} cited hosts, {cited / len(rows):.1f} per capture")
print(f"L2 prose:     {sum(r['answer_chars'] for r in rows) / len(rows):.0f} chars per answer")
print(f"L3 brands:    {sum(len(r['brands_present']) for r in rows)} watchlist hits total")

# stability.py — how far each layer moves inside a cell
import json
import statistics as stats
from collections import defaultdict
from itertools import combinations

cells = defaultdict(dict)
for line in open("layers.jsonl"):
    row = json.loads(line)
    cells[(row["engine"], row["prompt_id"])][row["repeat"]] = row


def jaccard(left, right):
    union = left | right
    return len(left & right) / len(union) if union else None


def mean_pairwise(sets):
    scores = [j for a, b in combinations(sets, 2) if (j := jaccard(a, b)) is not None]
    return stats.mean(scores) if scores else None


per_engine = defaultdict(lambda: defaultdict(list))
brands = defaultdict(lambda: {"stable": 0, "flipping": 0})

for (engine, _prompt), reps in sorted(cells.items()):
    keys = sorted(reps)
    hosts = [set(reps[k]["cited_hosts"]) for k in keys]
    tokens = [set(reps[k]["answer_tokens"]) for k in keys]
    lengths = [reps[k]["answer_chars"] for k in keys]

    per_engine[engine]["cite"].append(mean_pairwise(hosts))
    per_engine[engine]["prose"].append(mean_pairwise(tokens))
    per_engine[engine]["cv"].append(stats.pstdev(lengths) / stats.mean(lengths) * 100)
    per_engine[engine]["union"].append(len(set().union(*hosts)))
    per_engine[engine]["core"].append(len(set.intersection(*hosts)))

    for brand in set().union(*[set(reps[k]["brands_present"]) for k in keys]):
        seen = sum(1 for k in keys if brand in reps[k]["brands_present"])
        brands[engine]["stable" if seen == len(keys) else "flipping"] += 1

header = f"{'engine':<12}{'citeJaccard':>12}{'proseJaccard':>13}{'lengthCV%':>11}"
print(header + f"{'hostsEveryRun':>15}{'brandsStable%':>15}")
for engine, series in per_engine.items():
    core, union = sum(series["core"]), sum(series["union"])
    counts = brands[engine]
    stable_pct = counts["stable"] / (counts["stable"] + counts["flipping"]) * 100
    print(f"{engine:<12}{stats.mean(series['cite']):>12.3f}"
          f"{stats.mean(series['prose']):>13.3f}{stats.mean(series['cv']):>11.1f}"
          f"{f'{core}/{union}':>15}{stable_pct:>14.1f}%")

# samplesize.py — measured stability -> a required number of runs
import json
import math
import statistics as stats
from collections import Counter, defaultdict
from itertools import combinations

cells = defaultdict(dict)
for line in open("layers.jsonl"):
    row = json.loads(line)
    cells[(row["engine"], row["prompt_id"])][row["repeat"]] = row


def wilson(hits, runs, z=1.96):
    proportion = hits / runs
    denominator = 1 + z * z / runs
    centre = (proportion + z * z / (2 * runs)) / denominator
    half = z * math.sqrt(
        proportion * (1 - proportion) / runs + z * z / (4 * runs * runs)) / denominator
    return max(0.0, centre - half) * 100, min(1.0, centre + half) * 100


coverage = defaultdict(list)
for reps in cells.values():
    keys = sorted(reps)
    universe = set().union(*[set(reps[k]["cited_hosts"]) for k in keys])
    if not universe:
        continue
    for n in range(1, len(keys) + 1):
        coverage[n].append(stats.mean([
            len(set().union(*[set(reps[k]["cited_hosts"]) for k in combo])) / len(universe)
            for combo in combinations(keys, n)
        ]))

print("SET QUESTION — share of the 5-run cited-domain union that n runs surface")
for n in sorted(coverage):
    print(f"  n={n}: {stats.mean(coverage[n]) * 100:5.1f}%")

observed = Counter()
for reps in cells.values():
    keys = sorted(reps)
    for brand in set().union(*[set(reps[k]["brands_present"]) for k in keys]):
        observed[sum(1 for k in keys if brand in reps[k]["brands_present"])] += 1

total = sum(observed.values())
print(f"\nRATE QUESTION — {total} brand-by-cell observations at n=5")
for seen in sorted(observed):
    low, high = wilson(seen, 5)
    print(f"  seen {seen}/5 in {observed[seen]:>3} cases ({observed[seen] / total * 100:4.1f}%)"
          f"  95% interval {low:5.1f}%-{high:5.1f}%  width {high - low:4.1f}pt")

print("\nRATE QUESTION — interval width at a 40% appearance rate, by run count")
for n in (1, 5, 10, 20, 30, 50, 100, 200):
    low, high = wilson(round(0.4 * n), n)
    print(f"  n={n:<4} {low:5.1f}%-{high:5.1f}%  width {high - low:4.1f}pt")