Epictetus
301 lines
12 KiB
Python
301 lines
12 KiB
Python
#!/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
|
|
def _get_half_life(item):
|
|
"""Return section/tier-specific half-life in hours, or None to fall back
|
|
to decay_index's half_life_h argument (default 18h news half-life)."""
|
|
# Spec v1 addendum: Hardware/Tips section items get a 14-day (336h)
|
|
# half-life so curated/section items persist far longer in-section.
|
|
ms = (item.get("manual_section") or "").upper()
|
|
if ms 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"]
|
|
# No special case -> let decay_index use its half_life_h parameter.
|
|
return None
|
|
|
|
|
|
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, falling back to the passed half_life_h
|
|
hl = _get_half_life(it)
|
|
if hl is None:
|
|
hl = half_life_h
|
|
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()
|