feat(skill): gsc_signal_delta helper + tests + code notes

custom-skills/35-seo-signal-validation/code/scripts/gsc_signal_delta.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""Day-normalized GSC query delta + mover ranking for signal validation.
+
+Reads two GSC query exports (recent, prior) — JSON list or TSV with a header row
+containing query / clicks / impressions / position — and reports day-normalized
+site totals, top gainers/decliners, and whether a claimed term is a real mover.
+This is the deterministic L1/L4 core of the 35-seo-signal-validation skill.
+"""
+from __future__ import annotations
+import argparse
+import json
+import sys
+from pathlib import Path
+
+
+def _norm_row(r: dict) -> dict:
+    def num(*keys, default=0.0):
+        for k in keys:
+            if k in r and r[k] not in (None, ""):
+                try:
+                    return float(str(r[k]).replace(",", ""))
+                except ValueError:
+                    pass
+        return default
+    query = (r.get("query") or r.get("term") or "")
+    if isinstance(r.get("keys"), list) and r["keys"]:
+        query = str(r["keys"][0])
+    return {
+        "query": str(query).strip(),
+        "clicks": num("clicks"),
+        "impressions": num("impressions", "impr"),
+        "position": num("position", "pos", default=0.0),
+    }
+
+
+def load_gsc(path: str) -> list[dict]:
+    """Parse a GSC export (JSON list/{rows:[...]} or TSV-with-header)."""
+    text = Path(path).read_text(encoding="utf-8").strip()
+    if not text:
+        return []
+    if text[0] in "[{":
+        data = json.loads(text)
+        if isinstance(data, dict):
+            data = data.get("rows", [])
+        return [_norm_row(r) for r in data]
+    lines = text.splitlines()
+    header = [h.strip().lower() for h in lines[0].split("\t")]
+    rows = []
+    for line in lines[1:]:
+        if line.strip():
+            rows.append(_norm_row(dict(zip(header, line.split("\t")))))
+    return rows
+
+
+def _by_query(rows: list[dict]) -> dict:
+    return {r["query"]: r for r in rows if r["query"]}
+
+
+def compute_delta(recent, prior, recent_days, prior_days,
+                  claim_term=None, top_n=10) -> dict:
+    if recent_days <= 0 or prior_days <= 0:
+        raise ValueError("recent_days and prior_days must be positive")
+    r_by, p_by = _by_query(recent), _by_query(prior)
+
+    def totals(rows):
+        return {"clicks": sum(r["clicks"] for r in rows),
+                "impressions": sum(r["impressions"] for r in rows)}
+    rt, pt = totals(recent), totals(prior)
+
+    def per_day(total, days):
+        return round(total / days, 2)
+
+    def pct(new, old):
+        return round((new - old) / old * 100, 1) if old else None
+
+    r_cpd, p_cpd = per_day(rt["clicks"], recent_days), per_day(pt["clicks"], prior_days)
+    r_ipd, p_ipd = per_day(rt["impressions"], recent_days), per_day(pt["impressions"], prior_days)
+
+    deltas = []
+    for q in set(r_by) | set(p_by):
+        rc = r_by.get(q, {}).get("clicks", 0.0)
+        pc = p_by.get(q, {}).get("clicks", 0.0)
+        deltas.append({"query": q, "delta_clicks": rc - pc,
+                       "recent_clicks": rc, "prior_clicks": pc})
+    deltas.sort(key=lambda d: d["delta_clicks"], reverse=True)
+    gainers = [d for d in deltas if d["delta_clicks"] > 0][:top_n]
+    decliners = sorted([d for d in deltas if d["delta_clicks"] < 0],
+                       key=lambda d: d["delta_clicks"])[:top_n]
+
+    out = {
+        "site_totals": {
+            "recent": {**rt, "clicks_per_day": r_cpd,
+                       "impressions_per_day": r_ipd, "days": recent_days},
+            "prior": {**pt, "clicks_per_day": p_cpd,
+                      "impressions_per_day": p_ipd, "days": prior_days},
+            "clicks_per_day_pct": pct(r_cpd, p_cpd),
+            "impressions_per_day_pct": pct(r_ipd, p_ipd),
+        },
+        "top_gainers": gainers,
+        "top_decliners": decliners,
+        "claim_term": None,
+        "verdict_hint": None,
+    }
+
+    if claim_term:
+        gainer_terms = {g["query"] for g in gainers}
+        rc, pc = r_by.get(claim_term, {}), p_by.get(claim_term, {})
+        in_movers = claim_term in gainer_terms
+        share = (rc.get("clicks", 0.0) / rt["clicks"] * 100) if rt["clicks"] else 0.0
+        out["claim_term"] = {
+            "term": claim_term, "found": bool(rc or pc),
+            "recent": {"clicks": rc.get("clicks", 0.0),
+                       "impressions": rc.get("impressions", 0.0),
+                       "position": rc.get("position")},
+            "prior": {"clicks": pc.get("clicks", 0.0),
+                      "impressions": pc.get("impressions", 0.0),
+                      "position": pc.get("position")},
+            "in_top_movers": in_movers,
+            "click_share_pct": round(share, 2),
+        }
+        if not in_movers and share < 1.0:
+            out["verdict_hint"] = (
+                f"'{claim_term}' contributes {share:.2f}% of recent clicks and is "
+                f"absent from top movers -> claimed impact likely ARTIFACT; real "
+                f"movement is elsewhere (see top_gainers).")
+        elif in_movers:
+            out["verdict_hint"] = (
+                f"'{claim_term}' is among top movers -> claim plausibly CONFIRMED/"
+                f"PARTIAL; corroborate with live SERP + entity layer.")
+        else:
+            out["verdict_hint"] = (
+                f"'{claim_term}' has non-trivial share ({share:.2f}%) but is not a "
+                f"top mover -> PARTIAL; inspect attribution.")
+    return out
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description="GSC signal delta for signal validation")
+    ap.add_argument("--recent", required=True)
+    ap.add_argument("--prior", required=True)
+    ap.add_argument("--recent-days", type=int, required=True)
+    ap.add_argument("--prior-days", type=int, required=True)
+    ap.add_argument("--claim-term", default=None)
+    ap.add_argument("--top-n", type=int, default=10)
+    a = ap.parse_args(argv)
+    out = compute_delta(load_gsc(a.recent), load_gsc(a.prior),
+                        a.recent_days, a.prior_days, a.claim_term, a.top_n)
+    json.dump(out, sys.stdout, ensure_ascii=False, indent=2)
+    sys.stdout.write("\n")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())