#!/usr/bin/env python3 """Reproduce the Show HN title-signal analysis using only Python stdlib.""" from __future__ import annotations import argparse import collections import csv import datetime as dt import gzip import hashlib import html import json import math import pathlib import re import statistics import urllib.parse from zoneinfo import ZoneInfo SIGNALS = { "open_source": ("Open sourceを明記", re.compile(r"\bopen[ -]?source\b|\bOSS\b", re.I)), "ai_llm": ("AI / LLMを明記", re.compile(r"\bAI\b|\bLLMs?\b|\bGPT\b|\btransformer", re.I)), "local_privacy": ("local / privacy系", re.compile(r"\blocal(?:ly)?\b|\boffline\b|\bprivacy\b|\bprivate\b|self-host", re.I)), "browser_wasm": ("browser / WASM系", re.compile(r"\bbrowser\b|\bWASM\b|WebAssembly|WebGPU", re.I)), "i_built": ("I built / madeで開始", re.compile(r"^I\s+(?:built|made|created|wrote)\b", re.I)), "has_number": ("数字を含む", re.compile(r"\d")), "question": ("疑問符を含む", re.compile(r"\?")), "free": ("freeを明記", re.compile(r"\bfree\b", re.I)), } LENGTH_BINS = [ ("under_30", "29文字以下", 0, 30), ("30_49", "30〜49文字", 30, 50), ("50_69", "50〜69文字", 50, 70), ("70_plus", "70文字以上", 70, 10_000), ] def load_rows(file: pathlib.Path) -> list[dict]: opener = gzip.open if file.suffix == ".gz" else open with opener(file, "rt", encoding="utf-8") as stream: return [json.loads(line) for line in stream if line.strip()] def clean_title(raw: str) -> str: return re.sub(r"^\s*Show\s+HN\s*:\s*", "", raw or "", flags=re.I).strip() def median(values: list[int]) -> float: return round(float(statistics.median(values)), 2) if values else 0.0 def wilson(successes: int, total: int, z: float = 1.959963984540054) -> tuple[float, float]: if total == 0: return 0.0, 0.0 rate = successes / total denominator = 1 + z * z / total center = (rate + z * z / (2 * total)) / denominator margin = z * math.sqrt(rate * (1 - rate) / total + z * z / (4 * total * total)) / denominator return center - margin, center + margin def two_proportion_p(a_success: int, a_total: int, b_success: int, b_total: int) -> float: if not a_total or not b_total: return 1.0 pooled = (a_success + b_success) / (a_total + b_total) variance = pooled * (1 - pooled) * (1 / a_total + 1 / b_total) if variance <= 0: return 1.0 z = (a_success / a_total - b_success / b_total) / math.sqrt(variance) return math.erfc(abs(z) / math.sqrt(2)) def fdr_bh(items: list[dict]) -> None: ordered = sorted(enumerate(items), key=lambda pair: pair[1]["pValue"]) adjusted = [1.0] * len(items) running = 1.0 for rank_from_end in range(len(ordered) - 1, -1, -1): original_index, item = ordered[rank_from_end] rank = rank_from_end + 1 running = min(running, item["pValue"] * len(items) / rank) adjusted[original_index] = min(1.0, running) for item, value in zip(items, adjusted): item["pFdr"] = round(value, 6) def url_type(value: str | None) -> str: if not value: return "no_url" host = (urllib.parse.urlparse(value).hostname or "").lower() if host in {"github.com", "gitlab.com", "codeberg.org"}: return "code_host" return "custom_domain" def group_stats(rows: list[dict], predicate) -> dict: subset = [row for row in rows if predicate(row)] successes = sum(row["success10"] for row in subset) low, high = wilson(successes, len(subset)) return { "n": len(subset), "medianPoints": median([row["points"] for row in subset]), "success10Rate": round(successes / len(subset), 6) if subset else 0.0, "success10CiLow": round(low, 6), "success10CiHigh": round(high, 6), "success50Rate": round(sum(row["success50"] for row in subset) / len(subset), 6) if subset else 0.0, } def stratified_risk_difference(rows: list[dict], predicate) -> float: strata: dict[tuple, list[dict]] = collections.defaultdict(list) for row in rows: strata[(row["month"], row["ptWeekday"], row["ptHour"] // 3)].append(row) numerator = 0.0 denominator = 0.0 for values in strata.values(): exposed = [row for row in values if predicate(row)] control = [row for row in values if not predicate(row)] if not exposed or not control: continue weight = len(exposed) * len(control) / len(values) difference = sum(row["success10"] for row in exposed) / len(exposed) - sum(row["success10"] for row in control) / len(control) numerator += weight * difference denominator += weight return numerator / denominator if denominator else 0.0 def esc(value) -> str: return html.escape(str(value), quote=True) def svg_header(title: str, subtitle: str, width: int, height: int) -> list[str]: return [ f'', f'{esc(title)}', f'{esc(subtitle)}', '', '', f'{esc(title)}', f'{esc(subtitle)}', ] def write_length_svg(groups: list[dict], output: pathlib.Path) -> None: width, height = 1200, 700 left, right, top, bottom = 110, 60, 145, 120 plot_w, plot_h = width - left - right, height - top - bottom max_rate = max(group["success10CiHigh"] for group in groups) * 1.18 max_rate = max(max_rate, 0.1) lines = svg_header("Show HN: タイトル長と10点到達率", "2025-07-01〜2026-06-30の全投稿。棒は比率、線は95% Wilson信頼区間。", width, height) for tick in range(6): rate = max_rate * tick / 5 y = top + plot_h - plot_h * rate / max_rate lines += [f'', f'{rate*100:.0f}%'] slot = plot_w / len(groups) bar_w = slot * 0.56 for index, group in enumerate(groups): x = left + slot * index + (slot - bar_w) / 2 rate = group["success10Rate"] y = top + plot_h - plot_h * rate / max_rate ci_y1 = top + plot_h - plot_h * group["success10CiHigh"] / max_rate ci_y2 = top + plot_h - plot_h * group["success10CiLow"] / max_rate center_x = x + bar_w / 2 lines += [ f'', f'', f'', f'', f'{rate*100:.1f}%', f'{esc(group["label"])}', f'n={group["n"]:,}', ] lines += [f'', '出典: Algolia HN Search API。到達率は因果効果を示しません。Trendiumo Lab。', ''] output.parent.mkdir(parents=True, exist_ok=True) output.write_text("\n".join(lines), encoding="utf-8") def write_signal_svg(signals: list[dict], output: pathlib.Path) -> None: rows = sorted(signals, key=lambda item: item["adjustedDifferencePp"], reverse=True) width, height = 1200, 820 left, right, top, bottom = 305, 90, 145, 80 plot_w = width - left - right max_abs = max(abs(item["adjustedDifferencePp"]) for item in rows) * 1.2 or 1 center = left + plot_w / 2 scale = (plot_w / 2) / max_abs lines = svg_header("タイトル表現と10点到達率の調整差", "月・曜日・PT 3時間帯で層別したリスク差。探索的分析、因果推論ではありません。", width, height) lines += [f'', f'0 pt'] row_h = (height - top - bottom) / len(rows) for index, item in enumerate(rows): y = top + row_h * index + row_h / 2 value = item["adjustedDifferencePp"] bar_x = center if value >= 0 else center + value * scale bar_w = abs(value * scale) color = "#087f8c" if value >= 0 else "#c64b45" label_x = center + value * scale + (10 if value >= 0 else -10) anchor = "start" if value >= 0 else "end" lines += [ f'{esc(item["label"])}', f'', f'{value:+.1f} pt', f'n={item["n"]:,}', ] lines += ['調整差は層内比較の加重平均。FDR補正値は results.json に収録。Trendiumo Lab。', ''] output.parent.mkdir(parents=True, exist_ok=True) output.write_text("\n".join(lines), encoding="utf-8") def write_csv(file: pathlib.Path, rows: list[dict]) -> None: file.parent.mkdir(parents=True, exist_ok=True) with file.open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0].keys())) writer.writeheader() writer.writerows(rows) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--input", default="evidence/data/show-hn-2025-07-01_2026-06-30.jsonl.gz") parser.add_argument("--output", default="evidence/results.json") parser.add_argument("--figures", default="figures") args = parser.parse_args() input_file = pathlib.Path(args.input) raw_rows = load_rows(input_file) pt = ZoneInfo("America/Los_Angeles") rows = [] for raw in raw_rows: title = clean_title(str(raw.get("title") or "")) if not title: continue created = dt.datetime.fromtimestamp(int(raw["created_at_i"]), tz=dt.timezone.utc) created_pt = created.astimezone(pt) points = max(0, int(raw.get("points") or 0)) rows.append({ "objectId": str(raw.get("objectID") or ""), "title": title, "titleLength": len(title), "points": points, "success10": int(points >= 10), "success50": int(points >= 50), "month": created.strftime("%Y-%m"), "ptWeekday": created_pt.weekday(), "ptHour": created_pt.hour, "urlType": url_type(raw.get("url")), }) total_success10 = sum(row["success10"] for row in rows) total_success50 = sum(row["success50"] for row in rows) title_groups = [] for group_id, label, low, high in LENGTH_BINS: predicate = lambda row, low=low, high=high: low <= row["titleLength"] < high selected = [row for row in rows if predicate(row)] control = [row for row in rows if not predicate(row)] stats = group_stats(rows, predicate) selected_success = sum(row["success10"] for row in selected) control_success = sum(row["success10"] for row in control) control_rate = control_success / len(control) if control else 0.0 title_groups.append({ "id": group_id, "label": label, **stats, "controlSuccess10Rate": round(control_rate, 6), "rawDifferencePp": round((stats["success10Rate"] - control_rate) * 100, 3), "adjustedDifferencePp": round(stratified_risk_difference(rows, predicate) * 100, 3), "pValue": round(two_proportion_p(selected_success, len(selected), control_success, len(control)), 8), }) fdr_bh(title_groups) signal_rows = [] for signal_id, (label, pattern) in SIGNALS.items(): predicate = lambda row, pattern=pattern: bool(pattern.search(row["title"])) selected = [row for row in rows if predicate(row)] control = [row for row in rows if not predicate(row)] stats = group_stats(rows, predicate) selected_success = sum(row["success10"] for row in selected) control_success = sum(row["success10"] for row in control) control_rate = control_success / len(control) if control else 0.0 signal_rows.append({ "id": signal_id, "label": label, **stats, "controlSuccess10Rate": round(control_rate, 6), "rawDifferencePp": round((stats["success10Rate"] - control_rate) * 100, 3), "adjustedDifferencePp": round(stratified_risk_difference(rows, predicate) * 100, 3), "pValue": round(two_proportion_p(selected_success, len(selected), control_success, len(control)), 8), }) fdr_bh(signal_rows) url_labels = {"code_host": "GitHub等コードホスト", "custom_domain": "独自・外部ドメイン", "no_url": "URLなし"} url_groups = [] for group_id, label in url_labels.items(): predicate = lambda row, group_id=group_id: row["urlType"] == group_id selected = [row for row in rows if predicate(row)] control = [row for row in rows if not predicate(row)] stats = group_stats(rows, predicate) selected_success = sum(row["success10"] for row in selected) control_success = sum(row["success10"] for row in control) control_rate = control_success / len(control) if control else 0.0 url_groups.append({ "id": group_id, "label": label, **stats, "controlSuccess10Rate": round(control_rate, 6), "rawDifferencePp": round((stats["success10Rate"] - control_rate) * 100, 3), "adjustedDifferencePp": round(stratified_risk_difference(rows, predicate) * 100, 3), "pValue": round(two_proportion_p(selected_success, len(selected), control_success, len(control)), 8), }) fdr_bh(url_groups) best_length = max(title_groups, key=lambda item: item["success10Rate"]) strongest_positive = max(signal_rows, key=lambda item: item["adjustedDifferencePp"]) strongest_negative = min(signal_rows, key=lambda item: item["adjustedDifferencePp"]) generated_at = dt.datetime.now(dt.timezone.utc).isoformat() results = { "schemaVersion": 1, "generatedAt": generated_at, "methodology": { "design": "Predefined exploratory cohort analysis", "primaryOutcome": "Algolia points >= 10 at snapshot time", "titleLengthBins": [item[1] for item in LENGTH_BINS], "multipleTesting": "Two-proportion z tests with Benjamini-Hochberg FDR adjustment for eight predefined title signals", "adjustment": "Weighted within-stratum risk difference; strata are calendar month, PT weekday, and PT three-hour block", "causalClaim": False, }, "sample": { "totalPosts": len(rows), "periodStart": "2025-07-01T00:00:00Z", "periodEndExclusive": "2026-07-01T00:00:00Z", "snapshotAt": generated_at, "medianPoints": median([row["points"] for row in rows]), "success10Count": total_success10, "success10Rate": round(total_success10 / len(rows), 6), "success50Count": total_success50, "success50Rate": round(total_success50 / len(rows), 6), "datasetSha256": hashlib.sha256(input_file.read_bytes()).hexdigest(), }, "titleLength": title_groups, "signals": signal_rows, "urlType": url_groups, "headline": { "bestLengthBinId": best_length["id"], "bestLengthBinLabel": best_length["label"], "bestLengthBinRate": best_length["success10Rate"], "strongestPositiveSignalId": strongest_positive["id"], "strongestPositiveAdjustedDifferencePp": strongest_positive["adjustedDifferencePp"], "strongestNegativeSignalId": strongest_negative["id"], "strongestNegativeAdjustedDifferencePp": strongest_negative["adjustedDifferencePp"], }, } output = pathlib.Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(results, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") write_csv(output.parent / "title-length.csv", title_groups) write_csv(output.parent / "signals.csv", signal_rows) write_csv(output.parent / "url-types.csv", url_groups) figures = pathlib.Path(args.figures) write_length_svg(title_groups, figures / "title-length-success.svg") write_signal_svg(signal_rows, figures / "signal-adjusted-lift.svg") print(json.dumps({"rows": len(rows), "results": str(output), "headline": results["headline"]}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()