#!/usr/bin/env python3 """ cadence-by-size.py — does posting cadence still go with view rate once channel size is held roughly constant? stats.json -> posting_cadence reports a Spearman rho of about -0.54 between posts per day and a channel's median view-to-subscriber ratio, and prints the median subscriber count beside each cadence quartile because the busy channels are also the big ones. This script asks what is left of that association inside each size band. Reads channels.csv and posts.csv (both produced by analyse.py). Touches no network, uses only the standard library, and is deterministic: run it twice and the output is byte-identical. python3 cadence-by-size.py -> cadence-by-size-stats.json Four checks, each one closing a different way the raw figure could mislead: 1. Within size band. The same rank correlation, computed separately for the small (<1k), mid (1k-50k) and large (50k+) bands, plus cadence-tercile medians inside each band. 2. Partial rank correlation on subscriber count. Ranks of cadence and ratio, with the linear effect of the rank of subscriber count removed from both. Does not depend on where the band boundaries were drawn. 3. A burst-proof cadence. posts_per_day is posts over the visible span, so twenty posts in ten minutes scores in the thousands. The median gap in hours between consecutive posts (from posts.csv timestamps) is not moved by one burst. Checked with the same within-band correlation. 4. Age-restricted ratio. A busy channel's visible posts are all young, and young posts have not finished collecting views (stats.json -> age_curve_longitudinal), so a busy channel would show a lower ratio even if nothing else differed. Here each channel's ratio is recomputed from its posts aged 2 to 30 days only (the same window as size_bands -> post_ratio_age_2_to_30_days), on channels with at least 3 such posts. The channels that drop out are reported: in the large band they are the busiest, in the small and mid bands the quietest. """ import csv import json import os import statistics from datetime import datetime HERE = os.path.dirname(os.path.abspath(__file__)) BANDS = ["small (<1k)", "mid (1k-50k)", "large (50k+)"] MIN_AGED_POSTS = 3 def ranks(v): order = sorted(range(len(v)), key=lambda i: v[i]) r = [0.0] * len(v) i = 0 while i < len(order): j = i while j + 1 < len(order) and v[order[j + 1]] == v[order[i]]: j += 1 avg = (i + j) / 2.0 + 1 for k in range(i, j + 1): r[order[k]] = avg i = j + 1 return r def pearson(a, b): ma, mb = statistics.fmean(a), statistics.fmean(b) num = sum((x - ma) * (y - mb) for x, y in zip(a, b)) den = (sum((x - ma) ** 2 for x in a) * sum((y - mb) ** 2 for y in b)) ** 0.5 return num / den if den else None def spearman(x, y): if len(x) < 3: return None r = pearson(ranks(x), ranks(y)) return round(r, 4) if r is not None else None def partial_spearman(x, y, z): """Rank correlation of x and y with the rank of z partialled out.""" rx, ry, rz = ranks(x), ranks(y), ranks(z) rxy, rxz, ryz = pearson(rx, ry), pearson(rx, rz), pearson(ry, rz) den = ((1 - rxz ** 2) * (1 - ryz ** 2)) ** 0.5 return round((rxy - rxz * ryz) / den, 4) if den else None def med(v): return round(statistics.median(v), 4) if v else None def terciles(rows, key): rows = sorted(rows, key=lambda r: r[key]) n = len(rows) out = [] for t in range(3): chunk = rows[int(n * t / 3):int(n * (t + 1) / 3)] out.append({ "tercile": t + 1, "channels": len(chunk), key + "_min": round(chunk[0][key], 3), key + "_max": round(chunk[-1][key], 3), "median_ratio": med([c["median_ratio"] for c in chunk]), "median_subscribers": statistics.median([c["subscribers"] for c in chunk]), }) return out def parse_ts(s): return datetime.fromisoformat(s.replace("Z", "+00:00")) GAP_BUCKETS = [("under 10 minutes", 0, 10), ("10 to 60 minutes", 10, 60), ("1 to 6 hours", 60, 360), ("over 6 hours", 360, None)] def bursts(posts, direction="previous"): """direction="previous": gap since the post before; tests the later posts of a quick run. direction="next": gap until the post after; tests the earlier posts, the ones a run might push down the chat.""" by_bucket = {name: [] for name, _, _ in GAP_BUCKETS} per_chan = {name: {} for name, _, _ in GAP_BUCKETS} used = 0 for ch in sorted(posts): ps = sorted(posts[ch], key=lambda p: parse_ts(p["posted_utc"])) ts = [parse_ts(p["posted_utc"]) for p in ps] idx = range(1, len(ps)) if direction == "previous" else range(0, len(ps) - 1) win = [i for i in idx if 168.0 <= float(ps[i]["age_hours_at_capture"]) <= 1440.0] if len(win) < 5: continue norm = statistics.median(int(ps[i]["views"]) for i in win) if norm <= 0: continue used += 1 for i in win: gap = ((ts[i] - ts[i - 1]) if direction == "previous" else (ts[i + 1] - ts[i])).total_seconds() / 60.0 rel = int(ps[i]["views"]) / norm for name, lo, hi in GAP_BUCKETS: if gap >= lo and (hi is None or gap < hi): by_bucket[name].append(rel) per_chan[name].setdefault(ch, []).append(rel) return { "what_it_is": "posts 7-60 days old; views as a multiple of the median views of the " "same channel's posts in that window, by minutes %s" % ("since the previous post" if direction == "previous" else "until the next post"), "channels_used": used, "buckets": [{ ("gap_since_previous_post" if direction == "previous" else "gap_until_next_post"): name, "posts": len(by_bucket[name]), "channels": len(per_chan[name]), "median_relative_views_over_posts": med(by_bucket[name]), "median_relative_views_over_channels": med( [statistics.median(v) for v in per_chan[name].values()]), "share_of_posts_below_0_75": round( sum(1 for x in by_bucket[name] if x < 0.75) / len(by_bucket[name]), 4) if by_bucket[name] else None, } for name, _, _ in GAP_BUCKETS], } def main(): with open(os.path.join(HERE, "channels.csv"), newline="") as f: chans = [] for r in csv.DictReader(f): try: ppd = float(r["posts_per_day"]) except ValueError: continue chans.append({ "channel": r["channel"], "subscribers": int(r["subscribers"]), "size_band": r["size_band"], "median_ratio": float(r["median_ratio"]), "posts_per_day": ppd, }) posts = {} with open(os.path.join(HERE, "posts.csv"), newline="") as f: for r in csv.DictReader(f): posts.setdefault(r["channel"], []).append(r) for c in chans: ps = sorted(posts.get(c["channel"], []), key=lambda p: parse_ts(p["posted_utc"])) ts = [parse_ts(p["posted_utc"]) for p in ps] gaps = [(b - a).total_seconds() / 3600.0 for a, b in zip(ts, ts[1:])] c["median_gap_hours"] = statistics.median(gaps) if gaps else None aged = [float(p["view_to_subscriber_ratio"]) for p in ps if 48.0 <= float(p["age_hours_at_capture"]) <= 720.0] c["aged_posts"] = len(aged) c["ratio_2_to_30_days"] = statistics.median(aged) if len(aged) >= MIN_AGED_POSTS else None def null95(n): return round(1.96 / (n - 1) ** 0.5, 3) if n > 1 else None def block(rows): x = [r["posts_per_day"] for r in rows] y = [r["median_ratio"] for r in rows] g = [r for r in rows if r["median_gap_hours"] is not None] a = [r for r in rows if r["ratio_2_to_30_days"] is not None] dropped = [r for r in rows if r["ratio_2_to_30_days"] is None] return { "channels": len(rows), "median_subscribers": statistics.median([r["subscribers"] for r in rows]), "median_posts_per_day": med(x), "median_gap_hours_between_posts": med([r["median_gap_hours"] for r in g]), "spearman_posts_per_day_vs_ratio": spearman(x, y), "null_band_95": null95(len(rows)), "spearman_posts_per_day_vs_subscribers": spearman(x, [r["subscribers"] for r in rows]), "spearman_median_gap_vs_ratio": spearman( [r["median_gap_hours"] for r in g], [r["median_ratio"] for r in g]), "by_cadence_tercile": terciles(rows, "posts_per_day"), "age_restricted": { "what_it_is": "channel ratio recomputed from posts aged 48h to 30 days only; " "channels with fewer than %d such posts are left out" % MIN_AGED_POSTS, "channels_with_enough_aged_posts": len(a), "channels_left_out": len(dropped), "median_posts_per_day_left_out": med([r["posts_per_day"] for r in dropped]), "median_posts_per_day_kept": med([r["posts_per_day"] for r in a]), "spearman_posts_per_day_vs_ratio_2_to_30_days": spearman( [r["posts_per_day"] for r in a], [r["ratio_2_to_30_days"] for r in a]), "null_band_95": null95(len(a)), }, } out = { "question": "Is posting cadence associated with view-to-subscriber ratio once " "channel size is held roughly constant?", "inputs": ["channels.csv", "posts.csv"], "caution": "Observational, one night, recent posts only. A negative rho means " "busier channels show lower ratios; it does not show that posting less " "would raise anyone's ratio. Each rho carries null_band_95, an approximate " "95% band for zero correlation at that n.", "all_channels": block(chans), "partial_spearman_posts_per_day_vs_ratio_controlling_subscribers": partial_spearman( [c["posts_per_day"] for c in chans], [c["median_ratio"] for c in chans], [c["subscribers"] for c in chans]), "spearman_subscribers_vs_ratio": spearman( [c["subscribers"] for c in chans], [c["median_ratio"] for c in chans]), "by_size_band": {b: block([c for c in chans if c["size_band"] == b]) for b in BANDS}, "bursts_within_channel": bursts(posts), "bursts_within_channel_by_gap_to_next_post": bursts(posts, "next"), } with open(os.path.join(HERE, "cadence-by-size-stats.json"), "w") as f: json.dump(out, f, indent=1, ensure_ascii=False) f.write("\n") print(json.dumps(out, indent=1, ensure_ascii=False)) if __name__ == "__main__": main()