athena-oracle/clickability.py

#!/usr/bin/env python3
"""Clickability Index for Athena entries.

CORRECTED for the REAL schema (verified 2026-07-10):
  - Table is `entries`, not `items`.
  - Per-source engagement lives inside the `raw_metadata` JSON blob, not
    top-level `velocity_raw` / `engagement_raw` columns.
  - `content_type` is COMPUTED, not stored.

This module is read-only against the DB (SELECT only). It does not
modify oracle.db.

The MULTIPLIERS and formula match the approved plan exactly.
"""
import sqlite3, json, math, re, os, time
from datetime import datetime, timezone
from collections import defaultdict

DB_PATH = os.path.join(os.path.dirname(__file__), "oracle.db")

MULTIPLIERS = {}  # category multipliers removed: clickability is now virality-driven, not category-driven

# Virality weights (clickability = how viral/spreadable an item is right now)
VEL_W = 0.50
ENG_W = 0.50
SIG_W = 0.0  # signal_score no longer in the clickability blend (pure virality)

NOW = None  # set in main/fetch for age math


def get_connection():
    return sqlite3.connect(DB_PATH)


def _classify(src, title, summary):
    t = (title + " " + (summary or "")).lower()
    # Show HN — check first (builder posts)
    if re.search(r"\bshow\s+hn\b", t) or (src == "hackernews" and re.search(r"\b(show|built|made|launched|shipped)\b", t)):
        return "SHOW_HN"
    # Model release — pattern-based (works for third-party coverage too)
    if re.search(r"\b(gpt-|gpt5|gpt-5|deepseek|glm-|llama|qwen|claude|gemini|mistral|flux|stable-diffusion|sora|kimi|grok)\b", t) \
            and re.search(r"\b(releases?|released|v\d|launch|unveil|model|new\s+model|update|version)\b", t):
        return "MODEL_RELEASE"
    if re.search(r"\b(releases?|released|launches?|unveils?|announces?|debut|new\s+model|gpt-5|deepseek-v|glm-5)\b", t) \
            and re.search(r"\b(openai|anthropic|google|meta|microsoft|nvidia|ai)\b", t):
        return "MODEL_RELEASE"
    # HF model cards
    if src == "huggingface":
        return "MODEL_CARD"
    # Research papers
    if src == "arxiv" or re.search(r"\b(paper|study|benchmark|arxiv|proposes|learns?|novel|framework\s+for|towards)\b", t):
        return "RESEARCH"
    # Business/legal
    if re.search(r"\b(sues|lawsuit|funding|raises|acqui|ipo|valued|stealing|trade secret|layoff|hire[ds]?|exec|ceo)\b", t) \
            and not re.search(r"\b(repo|library|tool|agent framework)\b", t):
        return "BUSINESS_LEGAL"
    # Opinion/essay
    if re.search(r"\b(burnout|opinion|think|feel|why|essay|culture|linkedin|social media|future of|we made|i think|hot take|i believe|my view|in defense)\b", t):
        return "CULTURE_OPINION"
    # Tutorial/howto
    if re.search(r"\b(how to|tutorial|guide|running|build|setup|install|from scratch|learn)\b", t):
        return "TUTORIAL_HOWTO"
    # Dev tools
    if src == "github" or re.search(r"\b(repo|library|framework|tool|agent|sdk|cli|extension|plugin|app|engine)\b", t):
        return "DEV_TOOL_DRAMA"
    return "OTHER"


def _extract(src, md):
    """Return (velocity_raw, engagement_raw, age_hours)."""
    if src == "hackernews":
        pts = md.get("score", 0) or 0
        cmts = md.get("descendants", 0) or 0
        age_h = None
        if md.get("time"):
            try:
                age_h = max((NOW - md["time"]) / 3600.0, 0.1)
            except Exception:
                age_h = None
        vel = (pts / age_h) if age_h else pts
        return vel, (pts + 2 * cmts), age_h
    if src == "reddit":
        ups = md.get("ups", 0) or 0
        cmts = md.get("num_comments", 0) or 0
        return ups, (ups + 2 * cmts), None
    if src == "huggingface":
        likes = md.get("likes", 0) or 0
        return likes, likes, None
    if src == "github":
        spd = md.get("stars_per_day", 0) or 0
        stars = md.get("stars", 0) or 0
        return spd, stars, None
    if src == "arxiv":
        return 0.0, 0.0, None
    return 0.0, 0.0, None


def fetch_items(conn):
    global NOW
    NOW = __import__("time").time()
    cur = conn.cursor()
    cur.execute("SELECT id, title, url, source, summary, signal_score, raw_metadata, first_seen, "
                "curated_by, manual_section, manual_tier "
                "FROM entries")
    cols = [d[0] for d in cur.description]
    out = []
    for row in cur.fetchall():
        d = dict(zip(cols, row))
        try:
            md = json.loads(d.get("raw_metadata") or "{}")
        except Exception:
            md = {}
        vel, eng, age = _extract(d["source"], md)
        ct = _classify(d["source"], d.get("title") or "", d.get("summary") or "")
        created_at = md.get("createdAt") if d["source"] == "huggingface" else None
        out.append({
            "id": d["id"],
            "title": d.get("title") or "",
            "url": d.get("url") or "",
            "source": d["source"],
            "summary": d.get("summary") or "",
            "signal_score": d.get("signal_score") or 0,
            "velocity_raw": vel,
            "engagement_raw": eng,
            "content_type": ct,
            "first_seen": d.get("first_seen") or "",
            "created_at": created_at or "",
            "age_hours": 0.0,
            "curated_by": d.get("curated_by") or "",
            "manual_section": d.get("manual_section") or "",
            "manual_tier": d.get("manual_tier") or "",
        })
    return out


def log1p_norm(values):
    log_vals = [math.log1p(max(v, 0)) for v in values]
    if not log_vals:
        return []
    min_v, max_v = min(log_vals), max(log_vals)
    if max_v == min_v:
        return [0.0] * len(values)
    return [(v - min_v) / (max_v - min_v) for v in log_vals]


def compute_index(items):
    velocities = [it.get("velocity_raw", 0) or 0 for it in items]
    engagements = [it.get("engagement_raw", 0) or 0 for it in items]
    signals = [it.get("signal_score", 0) or 0 for it in items]

    vel_norm = log1p_norm(velocities)
    eng_norm = log1p_norm(engagements)
    sig_norm = log1p_norm(signals)

    for i, item in enumerate(items):
        raw = vel_norm[i] * VEL_W + eng_norm[i] * ENG_W + sig_norm[i] * SIG_W
        # Items with 0 engagement (arXiv, RSS, Reddit no-data) get a small base score
        # from signal_score so they can decay naturally instead of being stuck forever.
        # Floor: 0.05 * signal_score_norm — enough to rank, low enough to sink fast.
        if raw == 0 and sig_norm[i] > 0:
            raw = 0.05 * sig_norm[i]
        item["clickability"] = round(raw, 4)
        item["section"] = ""  # sections removed; flat ranked feed
    return items


def _age_hours(item):
    """Effective news-age in hours.

    HuggingFace items are aged by their TRUE model createdAt (likes/downloads
    are lifetime cumulative, so DB first_seen would pin every HF entry at
    ingest time and let all-time leaders dominate 'Top News' forever). All
    other sources are aged by DB first_seen.
    """
    if item.get("source") == "huggingface" and item.get("created_at"):
        s = item["created_at"]
    else:
        s = item.get("first_seen") or ""
    if not s:
        return 0.0
    try:
        ts = datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S").replace(
            tzinfo=timezone.utc).timestamp()
        return max((time.time() - ts) / 3600.0, 0.0)
    except Exception:
        return 0.0


# Category-specific half-lives (hours) — controls how long each type stays competitive.
# Breaking news decays slowest (stays relevant longer), arXiv/model cards fastest.
CATEGORY_HALF_LIVES = {
    "breaking": 36.0,       # Red — truly groundbreaking, double the standard
    "update": 24.0,         # Green — important but not groundbreaking, between red and black
    "OTHER": 18.0,          # Black — standard decay rate
}

# Map content_type to half-life, with tier override for breaking/update
def _get_half_life(item):
    """Return half-life in hours based on tier and content_type."""
    # Hardware/Tips section items decay on a 14-DAY half-life (user: revised
    # spec, shorter than evergreen). Covers both manual curations and
    # auto-classified section items, so a section link persists 14 days
    # instead of sinking in ~18h. Beyond this window the item is routed to
    # Archive (generator).
    ms = (item.get("manual_section") or "").upper()
    if ms in ("HARDWARE", "TIPS"):
        return 336.0
    sec = (item.get("computed_section") or "").upper()
    if sec in ("HARDWARE", "TIPS"):
        return 336.0
    tier = item.get("tier", "normal")
    # Tier overrides take precedence
    if tier == "breaking":
        return CATEGORY_HALF_LIVES["breaking"]
    if tier == "update":
        return CATEGORY_HALF_LIVES["update"]
    # Otherwise use content_type
    ct = item.get("content_type", "OTHER")
    return CATEGORY_HALF_LIVES.get(ct, CATEGORY_HALF_LIVES["OTHER"])


def decay_index(items, half_life_h=18.0):
    """Apply exponential time-decay to clickability so items sink as they age.

    Uses category-specific half-lives: breaking news decays slowest (36h),
    arXiv/model cards fastest (12h). This controls how long each type
    stays competitive, not just starting score.

    decayed = clickability * exp(-ln(2)/half_life * age_hours)
    """
    # Rolling 24h freshness window (not calendar-day) so Top News stays populated
    # between the daily harvest and midnight UTC. Decay still sinks old items.
    cutoff = datetime.now(timezone.utc).timestamp() - 24 * 3600
    for it in items:
        age = _age_hours(it)
        it["age_hours"] = round(age, 1)
        base = it.get("clickability", 0) or 0
        # Category-specific half-life
        hl = _get_half_life(it)
        k = math.log(2) / hl
        it["clickability_decayed"] = round(base * math.exp(-k * age), 4)
        it["effective_half_life"] = hl
        # Freshness flag: ingested within the last 24h -> eligible for Top News.
        fs = it.get("first_seen") or ""
        try:
            ts = datetime.fromisoformat(fs.replace("Z", "+00:00")).timestamp()
        except ValueError:
            ts = 0
        it["fresh"] = ts >= cutoff
    return items


def _pearson(xs, ys):
    n = len(xs)
    if n < 3:
        return None
    mx, my = sum(xs) / n, sum(ys) / n
    num = sum((x - mx) * (y - my) for x, y in zip(xs, ys))
    den = math.sqrt(sum((x - mx) ** 2 for x in xs) * sum((y - my) ** 2 for y in ys))
    return num / den if den else None


def main():
    conn = get_connection()
    items = fetch_items(conn)
    conn.close()
    if not items:
        print("No items found.")
        return

    computed = compute_index(items)
    computed.sort(key=lambda x: x["clickability"], reverse=True)

    print(f"=== TOP 20 BY CLICKABILITY INDEX (n={len(items)} items) ===\n")
    for i, item in enumerate(computed[:20], 1):
        print(f"{i:2}. [{item['clickability']:.4f}] {item['source']:11} | {item['title'][:58]}")
        print(f"    section={item['section']} | type={item['content_type']} | "
              f"vel={item['velocity_raw']:.1f} eng={item['engagement_raw']:.1f} sig={item['signal_score']:.2f}")

    # Backtest: Clickability Index vs ACTUAL HN engagement
    hn = [it for it in computed if it["source"] == "hackernews" and it["engagement_raw"] > 0]
    if hn:
        r_full = _pearson([it["engagement_raw"] for it in hn],
                          [it["clickability"] for it in hn])
        # Honest baseline: signal_score alone vs HN engagement (legacy prior)
        r_sig = _pearson([it["engagement_raw"] for it in hn],
                         [it["signal_score"] for it in hn])
        print(f"\n--- BACKTEST (HN, n={len(hn)}) ---")
        print(f"ClickabilityIndex vs actual HN engagement : r = {r_full:.3f}" if r_full is not None else "r = n/a")
        print(f"signal_score alone vs HN engagement       : r = {r_sig:.3f}" if r_sig is not None else "r = n/a")
        print("NOTE: engagement_raw is a 40% component of the index, so the full-index")
        print("      r is structurally high. The meaningful comparison is whether the")
        print("      index RANKS high-engagement items above low-engagement ones vs the")
        print("      legacy signal_score prior (r_sig above).")


if __name__ == "__main__":
    main()