JST Peak Bias in the Bluesky Firehose: Japanese Posts Cluster 2× More Tightly Around Prime Time Than English

import asyncio, time
from datetime import datetime, timezone, timedelta
import websockets, dag_cbor, fasttext, pandas as pd

JST   = timedelta(hours=9)
model = fasttext.load_model("lid.176.bin")
rows  = []

async def collect(seconds: int = 172_800):  # 48 時間
    uri = "wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos"
    async with websockets.connect(uri, max_size=2**23) as ws:
        t0 = time.monotonic()
        while time.monotonic() - t0 < seconds:
            msg = await ws.recv()
            try:
                # Firehose フレームは CBOR header + CBOR body の 2 部構成
                body = dag_cbor.decode(msg)          # 簡易実装：本来は分割が必要
                for op in body.get("ops", []):
                    if op["action"] != "create":
                        continue
                    if "/app.bsky.feed.post/" not in op.get("path", ""):
                        continue
                    record   = body.get("blocks", {}).get(str(op["cid"]), {})
                    text     = record.get("text", "")
                    if not text:
                        continue
                    labels, confs = model.predict(text.replace("\n", " "), k=1)
                    lang = labels[0].replace("__label__", "")
                    if lang not in ("ja", "en") or confs[0] < 0.80:
                        continue
                    now_jst = datetime.now(timezone.utc) + JST
                    rows.append({
                        "timestamp_utc": datetime.now(timezone.utc).isoformat(),
                        "hour_jst":      now_jst.hour,
                        "lang":          lang,
                        "did":           body.get("repo", ""),
                        "text_length":   len(text),
                    })
            except Exception:
                pass

asyncio.run(collect())
pd.DataFrame(rows).to_parquet("firehose_sample.parquet")

import pandas as pd
import numpy as np
from scipy.stats import median_abs_deviation

df  = pd.read_parquet("firehose_sample.parquet")
agg = df.groupby(["lang", "hour_jst"]).size().unstack("lang").fillna(0)
rel = agg / agg.mean()                        # 相対インデックス

# ピーク対トラフ比
ptr = rel.max() / rel.min()
print("PTR:\n", ptr)

# Robust z-score（MAD ベース）
def robust_z(s: pd.Series) -> pd.Series:
    mad = median_abs_deviation(s, scale="normal")
    return (s - s.median()) / mad if mad > 0 else s * 0.0

rel["ja_z"] = robust_z(rel["ja"])
rel["en_z"] = robust_z(rel["en"])
surge_ja = rel[rel["ja_z"] > 2].index.tolist()
print("サージ時間帯 (ja):", surge_ja)

# JST 20–23 集中係数
c_ja = agg.loc[agg.index.isin([20,21,22,23]), "ja"].sum() / agg["ja"].sum()
c_en = agg.loc[agg.index.isin([20,21,22,23]), "en"].sum() / agg["en"].sum()
print(f"C_ja={c_ja:.3f}  C_en={c_en:.3f}  ratio={c_ja/c_en:.2f}")

# ブートストラップ 95% CI（1000 回）
days    = df["timestamp_utc"].str[:10].unique()
windows = [df[df["timestamp_utc"].str[:10] == d] for d in days]
ptrs    = []
rng     = np.random.default_rng(42)
for _ in range(1000):
    idx    = rng.integers(0, len(windows), size=len(windows))
    sample = pd.concat([windows[i] for i in idx])
    a      = sample.groupby(["lang","hour_jst"]).size().unstack("lang").fillna(0)
    r      = a / a.mean()
    ptrs.append((r["ja"].max() / r["ja"].min(), r["en"].max() / r["en"].min()))
arr = np.array(ptrs)
print("PTR 95% CI ja:", np.percentile(arr[:,0], [2.5, 97.5]))
print("PTR 95% CI en:", np.percentile(arr[:,1], [2.5, 97.5]))

# 交差相関（lag −12〜+12 時間）
x  = rel["ja"].values - rel["ja"].mean()
y  = rel["en"].values - rel["en"].mean()
n  = len(x)
cc = np.real(np.fft.ifft(np.fft.fft(x, 2*n) * np.conj(np.fft.fft(y, 2*n))))[:n]
print(f"最大交差相関ラグ: {int(np.argmax(cc))} 時間")

# 1. 依存パッケージのインストール
pip install websockets dag-cbor fasttext pandas pyarrow scipy numpy

# 2. fastText 言語識別モデルのダウンロード（約 917 MB）
curl -sLO https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.bin

# 3. 48 時間収集（バックグラウンド実行）
nohup python fetch.py > fetch.log 2>&1 &
echo "PID: $!"

# 4. 接続断時の再開（seq は fetch.log から取得）
# SEQ=$(grep -oP '"seq":\K[0-9]+' fetch.log | tail -1)
# 再接続 URL: wss://bsky.network/xrpc/com.atproto.sync.subscribeRepos?cursor=$SEQ

# 5. 解析・グラフ生成
python analyze.py

# メモリ目安: ピーク帯に 2–5 GB RSS を使用する。
# 48 時間分の Parquet は概ね 200–800 MB（投稿量による）。