diff --git a/.agents/skills/pwc-cron-health/references/job-probes.md b/.agents/skills/pwc-cron-health/references/job-probes.md index 559b9572..ffc5c355 100644 --- a/.agents/skills/pwc-cron-health/references/job-probes.md +++ b/.agents/skills/pwc-cron-health/references/job-probes.md @@ -220,6 +220,13 @@ For every managed leaderboard sync, fetch the current upstream source read-only - **Independent proof:** Open one matched post from the JSON and confirm it is a paper author or the paper's own organization (business-verified account) introducing that paper, never an outside account; with apply on, confirm the paper page's X threads card shows it. - **Correctness:** `X search HTTP 401/403` means a bad or unentitled token and `429` an exhausted quota (the run stops at the first one). A zero `papers_scanned` means every recent candidate already has threads or used its three scans. +## `refresh_x_post_likes` + +- **Effect:** Hourly at :35, adds an `x_post_metrics` row for every `paper_x_threads` post from the last 7 days, then reads each due post (about 1, 2, 4, 8, 12, 24, 48, 72, 120 and 144 h after posting) and updates its likes-per-hour EMA (`like_velocity`). Syndication is tried first; posts it cannot serve use the paid X API (`X_API_BEARER_TOKEN`), capped at `PWC_X_LIKES_API_DAILY_CAP` (default 300) posts per UTC day, tracked in `x_post_likes_api_usage.json` under `PWC_CRON_LOCK_DIR`. Posts the API reports unavailable get `gone_at`. Report-only (reads through syndication and rolls back) unless `PWC_X_LIKES_APPLY=true`. +- **Freshness metrics** (dashboard Database Metrics): `last_write` = latest `x_post_metrics.last_read_at`, `tracked_posts` = posts with a pending reading. Report-only runs write nothing, so both stay empty until apply is on. +- **Independent proof:** Pick a recently read post (`ORDER BY last_read_at DESC`), open it on X, and confirm its like count is close to `like_count`; then confirm the paper's `x_trending_component` is set after the next `calculate_paper_trending_scores` run. +- **Correctness:** The JSON summary lists `syndication_reads`, `api_reads`, `api_reads_today`, `unread` and `api_error`. `api_error` with HTTP 401/402 means a bad token or exhausted X credits and exits 1; many `unread` with `api_reads_today` at the cap means syndication is failing and the cap is doing its job. The X part only affects rankings when `PWC_TRENDING_X_WEIGHT` in `/etc/paperswithcode/trending-cron.env` is above 0. + ## `link_google_scholar_profiles` - **Effect:** Checks Wikidata Scholar-ID candidates (P1960) for authors with 3+ papers: each new author once on Semantic Scholar (ORCID / S2 author ID on 15 sampled papers) into `author_google_scholar_candidates`, then spends the SerpAPI searches left this month (minus 10) reading Scholar profiles of S2-matched candidates, largest authors first, and upserts confirmed links into `author_google_scholar_profiles`; never deletes links. diff --git a/.github/workflows/scripts/README.md b/.github/workflows/scripts/README.md index 7e8b0c34..e4f8ddd6 100644 --- a/.github/workflows/scripts/README.md +++ b/.github/workflows/scripts/README.md @@ -36,6 +36,7 @@ Generated from `deploy/cron_schedules.py`; update this table when you change it. | `pwc-cron-link-google-scholar-profiles.timer` | `Wed *-*-* 05:20:00` | `link_google_scholar_profiles` | `backend/scripts/link_google_scholar_profiles.py` | | `pwc-cron-refresh-conference-deadlines.timer` | `*-*-* 07:40:00`
`*-*-* 19:40:00` | `refresh_conference_deadlines` | `backend/scripts/refresh_conference_deadlines.py` | | `pwc-cron-discover-x-threads.timer` | `*-*-* 08,20:22:00` | `discover_x_threads` | `backend/scripts/discover_x_threads.py` | +| `pwc-cron-refresh-x-post-likes.timer` | `*-*-* *:35:00` | `refresh_x_post_likes` | `backend/scripts/refresh_x_post_likes.py` | | `pwc-cron-post-trending-paper-to-x.timer` | `*-*-* 00/4:13:00` | `post_trending_paper_to_x` | `backend/scripts/post_trending_paper_to_x.py` | | `pwc-cron-post-x-insight.timer` | `*-*-* 09,17:43:00` | `post_x_insight` | `backend/scripts/post_x_insight.py` | | `pwc-cron-warm-paper-suggested-questions.timer` | `*-*-* *:33:00` | `warm_paper_suggested_questions` | `backend/scripts/warm_paper_suggested_questions.py` | @@ -182,6 +183,24 @@ history and stores them on repositories for the trending API/view pipeline. `update_citations.py` pulls citation counts from Semantic Scholar for ArXiv papers. Runs are bounded so hourly runs do not pile up under API throttling. +`refresh_x_post_likes.py` tracks how fast linked X posts gain likes, for the +X part of the paper trending score. Hourly at :35 it adds an `x_post_metrics` +row for each `paper_x_threads` post from the last week, then reads each post +about 1, 2, 4, 8, 12, 24, 48, 72, 120 and 144 hours after posting and keeps an +EMA of likes per hour. Reads use the free embed syndication endpoint first and +fall back to the paid X API (`X_API_BEARER_TOKEN`) for at most +`PWC_X_LIKES_API_DAILY_CAP` posts per UTC day (default 300). It only reports, +reading through syndication and rolling back, unless `PWC_X_LIKES_APPLY=true`. +`calculate_paper_trending_scores` ranks each paper's fastest post from the last +week against the others (posts under `PWC_TRENDING_X_MIN_LIKES_PER_HOUR`, +default 1, count for nothing) and weighs it by `PWC_TRENDING_X_WEIGHT` from +`/etc/paperswithcode/trending-cron.env` (default 0; GitHub and Hugging Face +share the rest equally). + +```bash +uv run --env-file backend/keys.env python -m backend.scripts.refresh_x_post_likes --max-posts 20 +``` + `discover_x_threads.py` finds X posts in which authors introduce recent papers and links them to the paper page's "X threads" card. Twice a day it takes up to 40 arXiv papers from the last 10 days that are trending or featured diff --git a/.github/workflows/scripts/hourly/calculate_paper_trending_scores.py b/.github/workflows/scripts/hourly/calculate_paper_trending_scores.py index 12c5ff5d..e7a16b50 100644 --- a/.github/workflows/scripts/hourly/calculate_paper_trending_scores.py +++ b/.github/workflows/scripts/hourly/calculate_paper_trending_scores.py @@ -3,14 +3,17 @@ Computes a paper-level trending score that blends the existing GitHub star velocity EMA (``repositories.trending_score``, stars/hour) with Hugging Face -artifact Hub trending scores (``hf_artifacts.trending_score``). The formula -lives in ``backend/utils/trending.py`` so it is unit-testable without a DB. +artifact Hub trending scores (``hf_artifacts.trending_score``) and, when +``PWC_TRENDING_X_WEIGHT`` is above 0, the like velocity of the paper's fastest +linked X post (``x_post_metrics.like_velocity``). The formula lives in +``backend/utils/trending.py`` so it is unit-testable without a DB. Pipeline: 1. Load the scoring population: latest-version papers published in the last - year, with their best official-repo velocity and linked HF artifact scores. - 2. Derive corpus normalization scales (95th percentiles) for GitHub velocity - and per-kind HF trending scores. + year, with their best official-repo velocity, linked HF artifact scores, + and fastest linked X post from the last week. + 2. Derive the quantile-rank corpora for GitHub velocity, per-kind HF + trending scores, and X like velocity. 3. Compute the composite score per paper and persist it to ``papers.trending_score`` (+ debug components). Papers with no live signal are cleared to NULL so they drop out of the composite feed. @@ -32,7 +35,10 @@ from backend.database.connection import db_manager from backend.utils.trending import ( + DEFAULT_X_MIN_LIKES_PER_HOUR, + DEFAULT_X_WEIGHT, KIND_WEIGHTS, + X_POST_WINDOW_DAYS, apply_decaying_pin, compute_paper_score, decaying_pin_floor, @@ -40,6 +46,7 @@ parse_trending_paper_id_pin_specs, parse_trending_pin_specs, sorted_positive, + x_corpus, ) # Score papers published within this window (bounds the work; older papers fall @@ -64,13 +71,25 @@ def log(message: str): print(f"{timestamp} - {message}", flush=True) +def x_settings() -> tuple[float, float]: + """Return the configured ``(x_weight, x_min_likes_per_hour)``.""" + weight = float(os.getenv("PWC_TRENDING_X_WEIGHT") or DEFAULT_X_WEIGHT) + if not 0.0 <= weight < 1.0: + raise ValueError("PWC_TRENDING_X_WEIGHT must be in [0, 1)") + min_likes = float( + os.getenv("PWC_TRENDING_X_MIN_LIKES_PER_HOUR") or DEFAULT_X_MIN_LIKES_PER_HOUR + ) + return weight, min_likes + + def load_population(session, cutoff): - """Return ``{paper_id: {"github_raw", "existing", "artifacts"}}``. + """Return ``{paper_id: {"github_raw", "existing", "artifacts", "x_raw"}}``. - ``github_raw``: best fresh official-repo stars/hour (0.0 if none). - ``existing``: whether the paper currently has a non-null trending_score. - ``artifacts``: list of ``(kind, trending_score)`` for linked HF artifacts, excluding aggregators (see ``is_hf_aggregator``). + - ``x_raw``: likes/hour of the fastest linked X post from the last week. """ log("Loading paper population (GitHub component)...") github_rows = session.execute( @@ -102,6 +121,7 @@ def load_population(session, cutoff): "github_raw": float(row.github_raw) if row.github_raw is not None else 0.0, "existing": bool(row.existing), "artifacts": [], + "x_raw": 0.0, } log(f"Population size: {len(population)} papers") @@ -143,6 +163,31 @@ def load_population(session, cutoff): log(f"HF artifact score links: {hf_links}") log(f"HF aggregator links ignored: {aggregator_links}") + log("Loading X like velocity component...") + x_rows = session.execute( + text(""" + SELECT t.paper_id, MAX(m.like_velocity) AS x_raw + FROM paper_x_threads t + JOIN x_post_metrics m ON m.tweet_id = t.tweet_id + JOIN papers p ON p.id = t.paper_id + WHERE p.is_latest_version = TRUE + AND p.date_published >= :cutoff + AND m.gone_at IS NULL + AND m.like_velocity > 0 + AND m.posted_at >= NOW() - (:window_days || ' days')::interval + GROUP BY t.paper_id + """), + {"cutoff": cutoff, "window_days": str(X_POST_WINDOW_DAYS)}, + ).fetchall() + x_papers = 0 + for row in x_rows: + entry = population.get(row.paper_id) + if entry is None: + continue + entry["x_raw"] = float(row.x_raw) + x_papers += 1 + log(f"Papers with a live X post: {x_papers}") + return population @@ -168,7 +213,16 @@ def _summ(values): return github_sorted, sorted_by_kind -def build_score_changes(population, github_sorted, sorted_by_kind, paper_ids=None): +def build_score_changes( + population, + github_sorted, + sorted_by_kind, + paper_ids=None, + *, + x_sorted=(), + x_weight=DEFAULT_X_WEIGHT, + x_min_likes_per_hour=DEFAULT_X_MIN_LIKES_PER_HOUR, +): """Compute updates/clears for all papers or an explicit target set.""" selected = sorted(population) if paper_ids is None else sorted(set(paper_ids)) updates = [] @@ -177,19 +231,25 @@ def build_score_changes(population, github_sorted, sorted_by_kind, paper_ids=Non entry = population.get(paper_id) if entry is None: continue - score, github_component, hf_component = compute_paper_score( + score, github_component, hf_component, x_component = compute_paper_score( entry["github_raw"], github_sorted, entry["artifacts"], sorted_by_kind, + x_raw=entry["x_raw"], + x_sorted=x_sorted, + x_weight=x_weight, + x_min_likes_per_hour=x_min_likes_per_hour, ) - if github_component > 0 or hf_component > 0: + # With X weighted at 0, an X-only paper scores 0 and stays unscored. + if score > 0: updates.append( { "id": paper_id, "score": round(score, 4), "g": round(github_component, 4), "h": round(hf_component, 4), + "x": round(x_component, 4), } ) elif entry["existing"]: @@ -239,11 +299,19 @@ def calculate_paper_scores(session, paper_ids=None) -> dict: return {"population": 0, "selected": 0, "written": 0, "cleared": 0} github_sorted, sorted_by_kind = derive_corpora(population) + x_weight, x_min_likes_per_hour = x_settings() + x_sorted = x_corpus( + (entry["x_raw"] for entry in population.values()), x_min_likes_per_hour + ) + log(f"X like velocity corpus: n={len(x_sorted)}, weight={x_weight}") updates, clears = build_score_changes( population, github_sorted, sorted_by_kind, paper_ids=paper_ids, + x_sorted=x_sorted, + x_weight=x_weight, + x_min_likes_per_hour=x_min_likes_per_hour, ) if paper_ids is None: # The population only covers the window, so aged-out scores would @@ -270,6 +338,7 @@ def write_scores(session, updates, clears): SET trending_score = :score, github_trending_component = :g, hf_trending_component = :h, + x_trending_component = :x, trending_score_updated_at = NOW(), updated_at = NOW() WHERE id = :id @@ -289,6 +358,7 @@ def write_scores(session, updates, clears): SET trending_score = NULL, github_trending_component = NULL, hf_trending_component = NULL, + x_trending_component = NULL, trending_score_updated_at = NOW(), updated_at = NOW() WHERE id = ANY(:ids) diff --git a/backend/database/migrations/add_x_post_metrics.sql b/backend/database/migrations/add_x_post_metrics.sql new file mode 100644 index 00000000..92986ec8 --- /dev/null +++ b/backend/database/migrations/add_x_post_metrics.sql @@ -0,0 +1,60 @@ +-- Migration: X post like velocity for the paper trending score +-- +-- `x_post_metrics` keeps one row per X post linked to a paper (a post can be +-- linked to several papers), with its latest like count and smoothed likes per +-- hour. The `refresh_x_post_likes` CRON reads each post on a decaying schedule +-- during its first week; `calculate_paper_trending_scores` reads the velocity. +-- `papers.x_trending_component` stores the normalized X part of the score. +-- +-- Run as the schema owner (postgres) for `public` before merging. +-- Idempotent: safe to re-run. + +BEGIN; + +-- Adding a nullable column is instant, but it needs a brief exclusive lock on +-- the busy papers table; fail fast instead of queueing behind long readers. +SET LOCAL lock_timeout = '10s'; + +CREATE TABLE IF NOT EXISTS public.x_post_metrics ( + tweet_id VARCHAR(32) PRIMARY KEY, + posted_at TIMESTAMPTZ NOT NULL, + like_count INTEGER, + like_velocity DOUBLE PRECISION, + last_read_at TIMESTAMPTZ, + last_source VARCHAR(16), + read_count INTEGER NOT NULL DEFAULT 0, + next_read_at TIMESTAMPTZ, + gone_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS ix_x_post_metrics_next_read_at + ON public.x_post_metrics (next_read_at) + WHERE next_read_at IS NOT NULL; + +ALTER TABLE public.papers + ADD COLUMN IF NOT EXISTS x_trending_component NUMERIC(12, 4); + +GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE public.x_post_metrics + TO paperswithcode_user; + +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'pwc_staging') THEN + GRANT SELECT ON TABLE public.x_post_metrics TO pwc_staging; + END IF; + -- Same grants deploy/configure_db_roles.py gives the isolated scoring role. + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'pwc_cron_trending') THEN + GRANT SELECT ON TABLE public.paper_x_threads, public.x_post_metrics + TO pwc_cron_trending; + GRANT UPDATE (x_trending_component) ON public.papers TO pwc_cron_trending; + END IF; +END +$$; + +COMMENT ON TABLE public.x_post_metrics IS +'Like counts and smoothed likes per hour of linked X posts, from the refresh_x_post_likes CRON.'; +COMMENT ON COLUMN public.papers.x_trending_component IS +'Quantile rank (0-1) of the paper''s fastest linked X post in the trending score.'; + +COMMIT; diff --git a/backend/models/database/__init__.py b/backend/models/database/__init__.py index de9b50a4..7102b1fb 100644 --- a/backend/models/database/__init__.py +++ b/backend/models/database/__init__.py @@ -175,6 +175,7 @@ PaperXTrending, XInsightPost, PaperXThreadScan, + XPostMetrics, PaperRepository, ) from backend.models.database.taxonomy import ( @@ -332,6 +333,7 @@ "PaperXTrending", "XInsightPost", "PaperXThreadScan", + "XPostMetrics", "PaperRepository", "Area", "Task", diff --git a/backend/models/database/papers.py b/backend/models/database/papers.py index f2b54c68..944d5a41 100644 --- a/backend/models/database/papers.py +++ b/backend/models/database/papers.py @@ -149,6 +149,9 @@ class Paper(Base): hf_trending_component: Mapped[Optional[float]] = mapped_column( Numeric(12, 4), nullable=True ) + x_trending_component: Mapped[Optional[float]] = mapped_column( + Numeric(12, 4), nullable=True + ) # When this paper was last scanned for linked Hugging Face artifacts. hf_artifacts_scanned_at: Mapped[Optional[datetime]] = mapped_column( DateTime(timezone=True), nullable=True diff --git a/backend/models/database/resources.py b/backend/models/database/resources.py index bf993839..6ca4b045 100644 --- a/backend/models/database/resources.py +++ b/backend/models/database/resources.py @@ -9,6 +9,7 @@ CheckConstraint, Date, DateTime, + Float, ForeignKey, Index, Integer, @@ -439,6 +440,28 @@ class PaperXThreadScan(Base): linked_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) +class XPostMetrics(Base): + """Like velocity of one linked X post, refreshed during its first week.""" + + __tablename__ = "x_post_metrics" + + tweet_id: Mapped[str] = mapped_column(String(32), primary_key=True) + posted_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + like_count: Mapped[Optional[int]] = mapped_column(Integer) + # Smoothed likes per hour (EMA over readings). + like_velocity: Mapped[Optional[float]] = mapped_column(Float) + last_read_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) + # "syndication" or "api". + last_source: Mapped[Optional[str]] = mapped_column(String(16)) + read_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + # NULL once the reading schedule is finished or the post is gone. + next_read_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) + gone_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), nullable=False + ) + + class PaperXPost(Base): """A post on the Papers with Code X account featuring one paper.""" diff --git a/backend/scripts/refresh_x_post_likes.py b/backend/scripts/refresh_x_post_likes.py new file mode 100644 index 00000000..5538e993 --- /dev/null +++ b/backend/scripts/refresh_x_post_likes.py @@ -0,0 +1,119 @@ +"""Production entrypoint for X post like refreshes. + +Read the like counts of linked X posts that are due on their first-week +schedule and update their likes-per-hour EMA in `x_post_metrics`, which feeds +the X part of the paper trending score. Syndication is tried first; posts it +cannot serve use the X API (X_API_BEARER_TOKEN), at most +PWC_X_LIKES_API_DAILY_CAP posts per UTC day (default 300). + +The job is report-only unless PWC_X_LIKES_APPLY=true (or --apply): it then reads +through syndication only and rolls back, so nothing is stored or billed. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import os +from dataclasses import asdict +from datetime import UTC, datetime +from pathlib import Path + +import httpx + +from backend.database.connection import get_database_session +from backend.services.x_post_likes_service import ( + make_api_reader, + make_syndication_reader, + refresh_likes, +) + +logger = logging.getLogger("refresh_x_post_likes") + +DEFAULT_API_DAILY_CAP = 300 + + +def api_usage_path() -> Path: + lock_dir = os.getenv( + "PWC_CRON_LOCK_DIR", + str(Path(__file__).resolve().parents[2] / ".github/workflows/scripts/data"), + ) + return Path(lock_dir) / "x_post_likes_api_usage.json" + + +def api_reads_today(path: Path, today: str) -> int: + if not path.exists(): + return 0 + usage = json.loads(path.read_text()) + return usage["reads"] if usage.get("day") == today else 0 + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--max-posts", type=int, default=500) + parser.add_argument( + "--apply", + action="store_true", + help="Store readings even when PWC_X_LIKES_APPLY is not true", + ) + args = parser.parse_args() + if not 1 <= args.max_posts <= 2000: + parser.error("--max-posts must be between 1 and 2000") + logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") + # One line per request would flood the journal with every hourly read. + logging.getLogger("httpx").setLevel(logging.WARNING) + apply = args.apply or os.environ.get("PWC_X_LIKES_APPLY") == "true" + + now = datetime.now(UTC) + today = now.date().isoformat() + usage_path = api_usage_path() + used = api_reads_today(usage_path, today) + cap = int(os.environ.get("PWC_X_LIKES_API_DAILY_CAP") or DEFAULT_API_DAILY_CAP) + token = os.environ.get("X_API_BEARER_TOKEN") + client = httpx.Client() + + session = get_database_session() + db = next(session) + try: + report = refresh_likes( + db, + now=now, + read_syndication=make_syndication_reader(client), + read_api=make_api_reader(client, token) if apply and token else None, + api_budget=max(cap - used, 0), + max_posts=args.max_posts, + ) + if apply: + db.commit() + if report.api_reads: + usage_path.parent.mkdir(parents=True, exist_ok=True) + usage_path.write_text( + json.dumps({"day": today, "reads": used + report.api_reads}) + ) + else: + db.rollback() + finally: + client.close() + try: + next(session) + except StopIteration: + pass + + print( + json.dumps( + { + "apply": apply, + "api_reads_today": used + report.api_reads, + "api_daily_cap": cap, + **asdict(report), + }, + sort_keys=True, + ) + ) + # A failing paid fallback (bad token, no credits) must show on the dashboard. + return 1 if report.api_error else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/services/autonomous_cron_operator.py b/backend/services/autonomous_cron_operator.py index c93fa006..b18dcf10 100644 --- a/backend/services/autonomous_cron_operator.py +++ b/backend/services/autonomous_cron_operator.py @@ -162,6 +162,7 @@ ), "scan_model_releases": ("backend/services/model_release_radar_service.py",), "discover_x_threads": ("backend/services/x_thread_discovery_service.py",), + "refresh_x_post_likes": ("backend/services/x_post_likes_service.py",), "post_trending_paper_to_x": ("backend/services/x_trending_post_service.py",), "warm_paper_suggested_questions": ( "backend/services/paper_suggested_questions_service.py", @@ -278,6 +279,7 @@ ), "scan_model_releases": ("backend/tests/test_model_release_radar_service.py",), "discover_x_threads": ("backend/tests/test_x_thread_discovery_service.py",), + "refresh_x_post_likes": ("backend/tests/test_x_post_likes_service.py",), "post_trending_paper_to_x": ("backend/tests/test_x_trending_post_service.py",), "warm_paper_suggested_questions": ( "backend/tests/test_paper_suggested_questions_warm_service.py", diff --git a/backend/services/cron_registry.py b/backend/services/cron_registry.py index 0128c251..bad7bd24 100644 --- a/backend/services/cron_registry.py +++ b/backend/services/cron_registry.py @@ -2071,6 +2071,36 @@ def validate_cron_jobs(jobs: Sequence[CronJobDefinition]) -> None: """, workflow_freshness_required=False, ), + CronJobDefinition( + id="refresh_x_post_likes", + scheduler="systemd", + schedule_policies=(SchedulePolicy.fixed(60, offset_minutes=35),), + schedule_grace_minutes=20, + systemd_timer="pwc-cron-refresh-x-post-likes.timer", + systemd_service="pwc-cron-refresh-x-post-likes.service", + systemd_extra_units=(), + output_evaluator="generic", + display_name="X Post Like Refresh", + category="hourly", + schedule="35 * * * * (hourly)", + script_path="backend/scripts/refresh_x_post_likes.py", + log_name="refresh_x_post_likes", + process_patterns=("backend.scripts.refresh_x_post_likes",), + lock_file=( + "/var/lib/paperswithcode/production/cron-locks/refresh_x_post_likes.lock" + ), + expected_runtime_minutes=3, + stale_after_minutes=180, + critical=False, + # Hours without a due post, and report-only runs, write nothing. + healthy_no_work_allowed=True, + db_freshness_query=""" + SELECT max(last_read_at) AS last_write, + count(*) FILTER (WHERE next_read_at IS NOT NULL) AS tracked_posts + FROM x_post_metrics + """, + workflow_freshness_required=False, + ), CronJobDefinition( id="post_trending_paper_to_x", scheduler="systemd", diff --git a/backend/services/paper_version_refresh_service.py b/backend/services/paper_version_refresh_service.py index 0554326e..c7160e24 100644 --- a/backend/services/paper_version_refresh_service.py +++ b/backend/services/paper_version_refresh_service.py @@ -149,6 +149,9 @@ async def refresh_arxiv_paper(self, arxiv_id: str) -> ArxivPaperRefreshResult: hf_trending_component=previous_latest.hf_trending_component if previous_latest else None, + x_trending_component=previous_latest.x_trending_component + if previous_latest + else None, hf_artifacts_scanned_at=previous_latest.hf_artifacts_scanned_at if previous_latest else None, diff --git a/backend/services/x_post_likes_service.py b/backend/services/x_post_likes_service.py new file mode 100644 index 00000000..03794ccb --- /dev/null +++ b/backend/services/x_post_likes_service.py @@ -0,0 +1,253 @@ +"""Refresh like counts of linked X posts during their first week. + +Each post linked to a paper (`paper_x_threads`) gets one `x_post_metrics` row, +read at about 1, 2, 4, 8, 12, 24, 48, 72, 120 and 144 hours after posting. Reads +use X's free embed syndication endpoint first; posts it cannot serve fall back +to paid X API lookups, which the caller caps per day. Each reading updates an +EMA of likes per hour that `calculate_paper_trending_scores` turns into the X +part of the paper trending score. Posts older than a week are never read. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Callable + +import httpx +from sqlalchemy import func, literal, select, update +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.orm import Session + +from backend.models.database import PaperXThread, XPostMetrics +from backend.services.x_thread_service import SYNDICATION_URL +from backend.utils.trending import X_POST_WINDOW_DAYS + +logger = logging.getLogger(__name__) + +READ_OFFSETS = tuple( + timedelta(hours=hours) for hours in (1, 2, 4, 8, 12, 24, 48, 72, 120, 144) +) +TRACKING_WINDOW = timedelta(days=X_POST_WINDOW_DAYS) +# Same smoothing as the GitHub star velocity EMA. +ALPHA = 0.75 +MIN_READING_HOURS = 0.25 +X_TWEETS_URL = "https://api.x.com/2/tweets" +API_BATCH_SIZE = 100 +SYNDICATION_INTERVAL_SECONDS = 0.25 + +# Returns the like count, or None when syndication cannot serve the post. +SyndicationRead = Callable[[str], "int | None"] +# Returns (like counts by id, ids X reports as unavailable). +ApiRead = Callable[[list[str]], "tuple[dict[str, int], set[str]]"] + + +class SyndicationRateLimited(Exception): + """Syndication answered HTTP 429; stop using it for this run.""" + + +@dataclass +class RefreshReport: + enrolled: int = 0 + due: int = 0 + syndication_reads: int = 0 + api_reads: int = 0 + gone: int = 0 + unread: int = 0 + expired: int = 0 + api_error: str | None = None + velocities: dict[str, float] = field(default_factory=dict) + + +def next_read_at(posted_at: datetime, now: datetime) -> datetime | None: + """The first scheduled reading after ``now``, or None once the schedule ends.""" + for offset in READ_OFFSETS: + if posted_at + offset > now: + return posted_at + offset + return None + + +def like_velocity(metrics: XPostMetrics, like_count: int, now: datetime) -> float: + """Smoothed likes per hour after a new reading. + + The first reading has nothing to compare against, so it uses the lifetime + average; later readings blend the rate since the previous one into the EMA. + """ + if metrics.like_count is None or metrics.last_read_at is None: + hours = max((now - metrics.posted_at).total_seconds() / 3600, 1.0) + return like_count / hours + hours = max((now - metrics.last_read_at).total_seconds() / 3600, MIN_READING_HOURS) + raw = max(like_count - metrics.like_count, 0) / hours + return ALPHA * raw + (1 - ALPHA) * (metrics.like_velocity or 0.0) + + +def enroll_posts(db: Session, now: datetime) -> int: + """Add a due row for every linked post from the last week that has none.""" + recent = ( + select( + PaperXThread.tweet_id, + func.min(PaperXThread.posted_at), + literal(now), + ) + .where(PaperXThread.posted_at >= now - TRACKING_WINDOW) + .group_by(PaperXThread.tweet_id) + ) + result = db.execute( + insert(XPostMetrics) + .from_select(["tweet_id", "posted_at", "next_read_at"], recent) + .on_conflict_do_nothing(index_elements=["tweet_id"]) + ) + return result.rowcount + + +def expire_posts(db: Session, now: datetime) -> int: + """Stop reading posts that left the window while a reading was pending.""" + result = db.execute( + update(XPostMetrics) + .where( + XPostMetrics.next_read_at.is_not(None), + XPostMetrics.posted_at < now - TRACKING_WINDOW, + ) + .values(next_read_at=None) + ) + return result.rowcount + + +def due_posts(db: Session, now: datetime, limit: int) -> list[XPostMetrics]: + return list( + db.scalars( + select(XPostMetrics) + .where( + XPostMetrics.next_read_at <= now, + XPostMetrics.gone_at.is_(None), + ) + .order_by(XPostMetrics.next_read_at, XPostMetrics.tweet_id) + .limit(limit) + ) + ) + + +def record_reading( + metrics: XPostMetrics, like_count: int, source: str, now: datetime +) -> None: + metrics.like_velocity = like_velocity(metrics, like_count, now) + metrics.like_count = like_count + metrics.last_read_at = now + metrics.last_source = source + metrics.read_count = (metrics.read_count or 0) + 1 + metrics.next_read_at = next_read_at(metrics.posted_at, now) + + +def refresh_likes( + db: Session, + *, + now: datetime, + read_syndication: SyndicationRead, + read_api: ApiRead | None, + api_budget: int, + max_posts: int, +) -> RefreshReport: + """Enroll new posts and read every due post; the caller commits or rolls back.""" + report = RefreshReport( + enrolled=enroll_posts(db, now), expired=expire_posts(db, now) + ) + due = due_posts(db, now, max_posts) + report.due = len(due) + + fallback: list[XPostMetrics] = [] + for index, metrics in enumerate(due): + try: + like_count = read_syndication(metrics.tweet_id) + except SyndicationRateLimited: + logger.warning("x_likes_syndication_rate_limited after=%d", index) + fallback.extend(due[index:]) + break + if like_count is None: + fallback.append(metrics) + continue + record_reading(metrics, like_count, "syndication", now) + report.syndication_reads += 1 + + paid = fallback[:api_budget] if read_api else [] + for start in range(0, len(paid), API_BATCH_SIZE): + batch = paid[start : start + API_BATCH_SIZE] + try: + counts, unavailable = read_api([metrics.tweet_id for metrics in batch]) + except httpx.HTTPError as error: + report.api_error = str(error) + logger.warning("x_likes_api_failed error=%s", error) + break + report.api_reads += len(batch) + for metrics in batch: + if metrics.tweet_id in counts: + record_reading(metrics, counts[metrics.tweet_id], "api", now) + elif metrics.tweet_id in unavailable: + metrics.gone_at = now + metrics.next_read_at = None + metrics.like_velocity = 0.0 + report.gone += 1 + + read = [metrics for metrics in due if metrics.last_read_at == now] + report.unread = len(due) - len(read) - report.gone + report.velocities = { + metrics.tweet_id: round(metrics.like_velocity, 3) for metrics in read + } + db.flush() + return report + + +def make_syndication_reader(client: httpx.Client) -> SyndicationRead: + """Read like counts from the public embed endpoint, pacing requests.""" + + def read(tweet_id: str) -> int | None: + time.sleep(SYNDICATION_INTERVAL_SECONDS) + try: + response = client.get( + SYNDICATION_URL, + params={"id": tweet_id, "lang": "en", "token": "a"}, + timeout=10, + ) + except httpx.HTTPError as error: + logger.info("x_likes_syndication_failed id=%s error=%s", tweet_id, error) + return None + if response.status_code == 429: + raise SyndicationRateLimited() + if response.status_code != 200 or not response.headers.get( + "content-type", "" + ).startswith("application/json"): + return None + post = response.json() + if post.get("__typename") != "Tweet" or post.get("favorite_count") is None: + return None + return int(post["favorite_count"]) + + return read + + +def make_api_reader(client: httpx.Client, bearer_token: str) -> ApiRead: + """Batch-read like counts with the X API (billed per post returned).""" + + def read(tweet_ids: list[str]) -> tuple[dict[str, int], set[str]]: + response = client.get( + X_TWEETS_URL, + params={"ids": ",".join(tweet_ids), "tweet.fields": "public_metrics"}, + headers={"Authorization": f"Bearer {bearer_token}"}, + timeout=30, + ) + response.raise_for_status() + payload = response.json() + counts = { + post["id"]: int(post["public_metrics"]["like_count"]) + for post in payload.get("data", []) + } + # Deleted, protected, or suspended posts come back as per-id errors. + unavailable = { + error["resource_id"] + for error in payload.get("errors", []) + if error.get("resource_id") + } - set(counts) + return counts, unavailable + + return read diff --git a/backend/tests/cron/fixtures/cron_effective_contracts.json b/backend/tests/cron/fixtures/cron_effective_contracts.json index 8de2726f..1c26dbf4 100644 --- a/backend/tests/cron/fixtures/cron_effective_contracts.json +++ b/backend/tests/cron/fixtures/cron_effective_contracts.json @@ -46,6 +46,7 @@ {"id": "link_google_scholar_profiles", "scheduler": "systemd", "schedule": "20 5 * * 3 (Wed 05:20 UTC)", "schedule_policies": [{"interval_minutes": null, "kind": "weekly", "minute_of_day": 320, "weekday": 2}], "schedule_grace_minutes": 180, "systemd_timer": "pwc-cron-link-google-scholar-profiles.timer", "systemd_service": "pwc-cron-link-google-scholar-profiles.service", "systemd_extra_units": [], "output_evaluator": "generic"}, {"id": "refresh_conference_deadlines", "scheduler": "systemd", "schedule": "40 7,19 * * * (collect twice daily, launch once daily)", "schedule_policies": [{"interval_minutes": 720, "kind": "fixed_interval", "minute_of_day": 460, "weekday": null}], "schedule_grace_minutes": 30, "systemd_timer": "pwc-cron-refresh-conference-deadlines.timer", "systemd_service": "pwc-cron-refresh-conference-deadlines.service", "systemd_extra_units": [], "output_evaluator": "conference_deadlines"}, {"id": "discover_x_threads", "scheduler": "systemd", "schedule": "22 8,20 * * * (twice daily)", "schedule_policies": [{"interval_minutes": 720, "kind": "fixed_interval", "minute_of_day": 502, "weekday": null}], "schedule_grace_minutes": 30, "systemd_timer": "pwc-cron-discover-x-threads.timer", "systemd_service": "pwc-cron-discover-x-threads.service", "systemd_extra_units": [], "output_evaluator": "generic"}, + {"id": "refresh_x_post_likes", "scheduler": "systemd", "schedule": "35 * * * * (hourly)", "schedule_policies": [{"interval_minutes": 60, "kind": "fixed_interval", "minute_of_day": 35, "weekday": null}], "schedule_grace_minutes": 20, "systemd_timer": "pwc-cron-refresh-x-post-likes.timer", "systemd_service": "pwc-cron-refresh-x-post-likes.service", "systemd_extra_units": [], "output_evaluator": "generic"}, {"id": "post_trending_paper_to_x", "scheduler": "systemd", "schedule": "13 */4 * * * (every four hours)", "schedule_policies": [{"interval_minutes": 240, "kind": "fixed_interval", "minute_of_day": 13, "weekday": null}], "schedule_grace_minutes": 30, "systemd_timer": "pwc-cron-post-trending-paper-to-x.timer", "systemd_service": "pwc-cron-post-trending-paper-to-x.service", "systemd_extra_units": [], "output_evaluator": "generic"}, {"id": "post_x_insight", "scheduler": "systemd", "schedule": "43 9,17 * * * (twice daily)", "schedule_policies": [{"interval_minutes": null, "kind": "daily", "minute_of_day": 583, "weekday": null}, {"interval_minutes": null, "kind": "daily", "minute_of_day": 1063, "weekday": null}], "schedule_grace_minutes": 30, "systemd_timer": "pwc-cron-post-x-insight.timer", "systemd_service": "pwc-cron-post-x-insight.service", "systemd_extra_units": [], "output_evaluator": "generic"}, {"id": "link_abstract_mentions", "scheduler": "systemd", "schedule": "51 * * * * (hourly)", "schedule_policies": [{"interval_minutes": 60, "kind": "fixed_interval", "minute_of_day": 51, "weekday": null}], "schedule_grace_minutes": 30, "systemd_timer": "pwc-cron-link-abstract-mentions.timer", "systemd_service": "pwc-cron-link-abstract-mentions.service", "systemd_extra_units": [], "output_evaluator": "generic"}, diff --git a/backend/tests/cron/test_registry_inventory.py b/backend/tests/cron/test_registry_inventory.py index 25572e3e..f5eea82f 100644 --- a/backend/tests/cron/test_registry_inventory.py +++ b/backend/tests/cron/test_registry_inventory.py @@ -27,8 +27,8 @@ def test_registry_covers_workflows_systemd_and_allowlist_exactly(): summary = _dashboard_job_summary( [{"category": job.category, "status": "healthy"} for job in jobs] ) - assert summary["total_jobs"] == 72 - assert summary["scheduled_jobs"] == 71 + assert summary["total_jobs"] == 73 + assert summary["scheduled_jobs"] == 72 assert summary["manual_jobs"] == 1 assert len(by_id) == len(jobs), "duplicate dashboard job IDs" diff --git a/backend/tests/deployment/test_cron_environment.py b/backend/tests/deployment/test_cron_environment.py index 38216d65..ba7fec9b 100644 --- a/backend/tests/deployment/test_cron_environment.py +++ b/backend/tests/deployment/test_cron_environment.py @@ -49,6 +49,7 @@ def test_restricted_trending_environment_drops_broad_credentials(): "PWC_DB_PASSWORD": "broad-secret", "PWC_TRENDING_DB_PASSWORD": "restricted-secret", "PWC_TRENDING_PIN_ARXIV_IDS": "2601.00001:72", + "PWC_TRENDING_X_WEIGHT": "0.3333", "PWC_GITHUB_TOKEN": "must-not-leak", "OPENAI_API_KEY": "must-not-leak", "PWC_CHAT_LANGFUSE_SECRET_KEY": "must-not-leak", @@ -62,6 +63,7 @@ def test_restricted_trending_environment_drops_broad_credentials(): assert environment["PWC_DB_USER"] == "pwc_cron_trending" assert environment["PWC_DB_PASSWORD"] == "restricted-secret" assert environment["PWC_TRENDING_PIN_ARXIV_IDS"] == "2601.00001:72" + assert environment["PWC_TRENDING_X_WEIGHT"] == "0.3333" assert "PWC_TRENDING_DB_PASSWORD" not in environment assert "PWC_GITHUB_TOKEN" not in environment assert "OPENAI_API_KEY" not in environment @@ -101,7 +103,8 @@ def test_trending_database_role_has_only_scoring_capabilities(): assert "REVOKE CREATE ON SCHEMA public" in sql assert ( "GRANT SELECT ON papers, paper_repositories, repositories, " - "repository_star_history, paper_hf_artifacts, hf_artifacts" + "repository_star_history, paper_hf_artifacts, hf_artifacts, " + "paper_x_threads, x_post_metrics" ) in sql assert ( "GRANT UPDATE (trending_score, trending_score_updated_at, updated_at) " @@ -109,7 +112,8 @@ def test_trending_database_role_has_only_scoring_capabilities(): ) in sql assert ( "GRANT UPDATE (trending_score, github_trending_component, " - "hf_trending_component, trending_score_updated_at, updated_at) ON papers" + "hf_trending_component, x_trending_component, trending_score_updated_at, " + "updated_at) ON papers" ) in sql for taxonomy_table in ("tasks", "methods", "datasets", "metrics", "conferences"): assert f"'{taxonomy_table}'" in sql @@ -121,6 +125,7 @@ def test_trending_environment_rotation_preserves_existing_role_password(): current = { "PWC_TRENDING_DB_PASSWORD": "PWC_TRENDING_DB_PASSWORD=stable-secret", "PWC_TRENDING_PIN_ARXIV_IDS": "PWC_TRENDING_PIN_ARXIV_IDS=2601.1:72", + "PWC_TRENDING_X_WEIGHT": "PWC_TRENDING_X_WEIGHT=0.3333", } password = configure_host_env.existing_or_random( current, "PWC_TRENDING_DB_PASSWORD" @@ -138,6 +143,7 @@ def test_trending_environment_rotation_preserves_existing_role_password(): assert values["PWC_TRENDING_DB_PASSWORD"] == "stable-secret" assert values["PWC_TRENDING_PIN_ARXIV_IDS"] == "2601.1:72" + assert values["PWC_TRENDING_X_WEIGHT"] == "0.3333" @pytest.mark.parametrize( diff --git a/backend/tests/deployment/test_host_env_credentials.py b/backend/tests/deployment/test_host_env_credentials.py index 135d055a..094b35d0 100644 --- a/backend/tests/deployment/test_host_env_credentials.py +++ b/backend/tests/deployment/test_host_env_credentials.py @@ -715,3 +715,14 @@ def test_chat_executor_units_cannot_read_environment_files(): for line in unit.splitlines(): if "chgrp" in line: assert line.startswith("ExecStartPost=+"), line + + +def test_x_post_likes_settings_survive_cron_overlay_rewrites(): + keys = {"PWC_X_LIKES_APPLY", "PWC_X_LIKES_API_DAILY_CAP"} + current = {key: f"{key}=existing" for key in keys} + + cron = configure_host_env.cron_passthrough_credentials( + {"PWC_GITHUB_TOKEN": "PWC_GITHUB_TOKEN=github"}, current + ) + + assert {key: cron[key] for key in keys} == current diff --git a/backend/tests/paper/test_trending_scores.py b/backend/tests/paper/test_trending_scores.py index e9bfd4a6..3dde8bcb 100644 --- a/backend/tests/paper/test_trending_scores.py +++ b/backend/tests/paper/test_trending_scores.py @@ -6,6 +6,8 @@ from types import SimpleNamespace from unittest.mock import Mock +import pytest + REPO_ROOT = Path(__file__).resolve().parents[3] SCRIPT = ( @@ -57,12 +59,15 @@ def test_population_ignores_hf_aggregator_links(): session.execute.side_effect = [ Mock(fetchall=Mock(return_value=github_rows)), Mock(fetchall=Mock(return_value=hf_rows)), + Mock(fetchall=Mock(return_value=[SimpleNamespace(paper_id=2, x_raw=12.5)])), ] population = cron.load_population(session, cutoff=None) assert population[1]["artifacts"] == [("model", 5.0)] assert population[2]["artifacts"] == [] + assert population[1]["x_raw"] == 0.0 + assert population[2]["x_raw"] == 12.5 # Paper 2 only had an aggregator, so its stored score is cleared. github_corpus, hf_corpora = cron.derive_corpora(population) updates, clears = cron.build_score_changes(population, github_corpus, hf_corpora) @@ -73,15 +78,15 @@ def test_population_ignores_hf_aggregator_links(): def test_targeted_compute_only_changes_selected_papers(monkeypatch): cron = _load_module() population = { - 1: {"github_raw": 10.0, "existing": False, "artifacts": []}, - 2: {"github_raw": 2.0, "existing": True, "artifacts": []}, - 3: {"github_raw": 0.0, "existing": True, "artifacts": []}, + 1: {"github_raw": 10.0, "existing": False, "artifacts": [], "x_raw": 0.0}, + 2: {"github_raw": 2.0, "existing": True, "artifacts": [], "x_raw": 0.0}, + 3: {"github_raw": 0.0, "existing": True, "artifacts": [], "x_raw": 0.0}, } calls = [] - def fake_compute(github_raw, github_corpus, artifacts, artifact_corpora): + def fake_compute(github_raw, github_corpus, artifacts, artifact_corpora, **_x): calls.append((github_raw, list(github_corpus))) - return github_raw, github_raw, 0.0 + return github_raw, github_raw, 0.0, 0.0 monkeypatch.setattr(cron, "compute_paper_score", fake_compute) github_corpus, hf_corpora = cron.derive_corpora(population) @@ -99,8 +104,8 @@ def fake_compute(github_raw, github_corpus, artifacts, artifact_corpora): def test_full_and_targeted_compute_agree_for_same_paper(monkeypatch): cron = _load_module() population = { - 1: {"github_raw": 10.0, "existing": False, "artifacts": []}, - 2: {"github_raw": 2.0, "existing": False, "artifacts": []}, + 1: {"github_raw": 10.0, "existing": False, "artifacts": [], "x_raw": 0.0}, + 2: {"github_raw": 2.0, "existing": False, "artifacts": [], "x_raw": 0.0}, } github_corpus, hf_corpora = cron.derive_corpora(population) full, _ = cron.build_score_changes(population, github_corpus, hf_corpora) @@ -150,7 +155,9 @@ def test_decaying_pins_share_state_for_arxiv_and_internal_paper_ids( def test_full_run_clears_scores_of_papers_that_left_the_window(monkeypatch): cron = _load_module() - population = {1: {"github_raw": 10.0, "existing": False, "artifacts": []}} + population = { + 1: {"github_raw": 10.0, "existing": False, "artifacts": [], "x_raw": 0.0} + } monkeypatch.setattr(cron, "load_population", lambda _session, _cutoff: population) monkeypatch.setattr( cron, "get_aged_out_scored_paper_ids", lambda _session, _cutoff: [7, 8] @@ -171,7 +178,9 @@ def test_full_run_clears_scores_of_papers_that_left_the_window(monkeypatch): def test_targeted_run_leaves_aged_out_scores_to_the_full_run(monkeypatch): cron = _load_module() - population = {1: {"github_raw": 10.0, "existing": False, "artifacts": []}} + population = { + 1: {"github_raw": 10.0, "existing": False, "artifacts": [], "x_raw": 0.0} + } monkeypatch.setattr(cron, "load_population", lambda _session, _cutoff: population) aged_out = Mock(return_value=[7]) monkeypatch.setattr(cron, "get_aged_out_scored_paper_ids", aged_out) @@ -184,3 +193,37 @@ def test_targeted_run_leaves_aged_out_scores_to_the_full_run(monkeypatch): cron.calculate_paper_scores(Mock(), paper_ids=[1]) aged_out.assert_not_called() + + +def test_x_only_paper_stays_unscored_while_x_weight_is_zero(): + cron = _load_module() + population = { + 1: {"github_raw": 10.0, "existing": False, "artifacts": [], "x_raw": 0.0}, + 2: {"github_raw": 0.0, "existing": True, "artifacts": [], "x_raw": 50.0}, + } + github_corpus, hf_corpora = cron.derive_corpora(population) + + updates, clears = cron.build_score_changes( + population, github_corpus, hf_corpora, x_sorted=[50.0], x_weight=0.0 + ) + assert [update["id"] for update in updates] == [1] + assert clears == [2] + + updates, clears = cron.build_score_changes( + population, github_corpus, hf_corpora, x_sorted=[50.0], x_weight=1 / 3 + ) + by_id = {update["id"]: update for update in updates} + assert by_id[2]["x"] == 1.0 + assert by_id[2]["score"] == pytest.approx(100 / 3, abs=1e-3) + assert by_id[1]["score"] == pytest.approx(100 / 3, abs=1e-3) + assert clears == [] + + +def test_x_settings_reject_a_weight_that_would_silence_github_and_hf(monkeypatch): + cron = _load_module() + monkeypatch.setenv("PWC_TRENDING_X_WEIGHT", "1") + with pytest.raises(ValueError, match="PWC_TRENDING_X_WEIGHT"): + cron.x_settings() + monkeypatch.setenv("PWC_TRENDING_X_WEIGHT", "0.3333") + monkeypatch.setenv("PWC_TRENDING_X_MIN_LIKES_PER_HOUR", "2") + assert cron.x_settings() == (0.3333, 2.0) diff --git a/backend/tests/test_trending_formula.py b/backend/tests/test_trending_formula.py index 619615ec..c3830d03 100644 --- a/backend/tests/test_trending_formula.py +++ b/backend/tests/test_trending_formula.py @@ -16,6 +16,8 @@ apply_decaying_pin, compute_paper_score, composite_score, + x_component, + x_corpus, decaying_pin_floor, github_component, hf_component, @@ -99,7 +101,7 @@ def test_deepseek_like_paper_trends_on_hf_alone(): models = sorted_positive([1, 4, 46, 102]) by_kind = {"model": models, "dataset": [], "space": []} - deepseek_score, g, h = compute_paper_score( + deepseek_score, g, h, _ = compute_paper_score( github_raw=0.0, github_sorted=github_sorted, artifact_scores=[("model", 102), ("model", 46), ("model", 4), ("model", 1)], @@ -109,9 +111,9 @@ def test_deepseek_like_paper_trends_on_hf_alone(): assert h == 1.0 assert deepseek_score == SCORE_SCALE * HF_WEIGHT # 50 - mid_github, _, _ = compute_paper_score(2.0, github_sorted, [], by_kind) - top_github, _, _ = compute_paper_score(20.0, github_sorted, [], by_kind) - both, _, _ = compute_paper_score(20.0, github_sorted, [("model", 102)], by_kind) + mid_github, *_ = compute_paper_score(2.0, github_sorted, [], by_kind) + top_github, *_ = compute_paper_score(20.0, github_sorted, [], by_kind) + both, *_ = compute_paper_score(20.0, github_sorted, [("model", 102)], by_kind) # HF-only DeepSeek paper beats a mid GitHub paper and ties the top one. assert deepseek_score > mid_github @@ -123,7 +125,7 @@ def test_deepseek_like_paper_trends_on_hf_alone(): def test_no_signal_paper_scores_zero(): - score, g, h = compute_paper_score(0.0, [1.0, 2.0], [], {"model": [1.0]}) + score, g, h, _ = compute_paper_score(0.0, [1.0, 2.0], [], {"model": [1.0]}) assert (score, g, h) == (0.0, 0.0, 0.0) @@ -168,3 +170,29 @@ def test_hf_aggregators_are_excluded_by_fan_out_and_denylist(): assert is_hf_aggregator("space", "tardellirs/paper-pulse", 1) # The denylist is per kind: a same-named repo of another kind still counts. assert not is_hf_aggregator("model", "tardellirs/paper-pulse", 1) + + +def test_x_weight_takes_its_share_equally_from_github_and_hf(): + assert composite_score(1.0, 1.0, 1.0, x_weight=1 / 3) == pytest.approx(SCORE_SCALE) + assert composite_score(1.0, 0.0, 0.0, x_weight=1 / 3) == pytest.approx( + SCORE_SCALE / 3 + ) + assert composite_score(0.0, 0.0, 1.0, x_weight=1 / 3) == pytest.approx( + SCORE_SCALE / 3 + ) + # Weighted at 0, X never changes a score. + assert composite_score(0.4, 0.6, 1.0) == composite_score(0.4, 0.6) + + +def test_slow_x_posts_count_for_nothing_and_stay_out_of_the_corpus(): + corpus = x_corpus([0.2, 0.9, 1.0, 5.0, 40.0], min_likes_per_hour=1.0) + assert corpus == [1.0, 5.0, 40.0] + assert x_component(0.9, corpus, 1.0) == 0.0 + assert x_component(5.0, corpus, 1.0) == pytest.approx(2 / 3) + assert x_component(40.0, corpus, 1.0) == 1.0 + + score, g, h, x = compute_paper_score( + 0.0, [], [], {}, x_raw=40.0, x_sorted=corpus, x_weight=1 / 3 + ) + assert (g, h, x) == (0.0, 0.0, 1.0) + assert score == pytest.approx(SCORE_SCALE / 3) diff --git a/backend/tests/test_x_post_likes_service.py b/backend/tests/test_x_post_likes_service.py new file mode 100644 index 00000000..06916038 --- /dev/null +++ b/backend/tests/test_x_post_likes_service.py @@ -0,0 +1,207 @@ +"""X post like refreshes: reading schedule, velocity EMA, and source fallback.""" + +from datetime import UTC, datetime, timedelta +from unittest.mock import Mock + +import httpx +import pytest + +from backend.models.database import XPostMetrics +from backend.services import x_post_likes_service as service + +NOW = datetime(2026, 11, 3, 12, 0, tzinfo=UTC) + + +def _metrics(tweet_id="1", *, age_hours=3.0, **values): + return XPostMetrics( + tweet_id=tweet_id, + posted_at=NOW - timedelta(hours=age_hours), + read_count=0, + **values, + ) + + +def _refresh(monkeypatch, due, *, read_syndication, read_api=None, api_budget=300): + monkeypatch.setattr(service, "enroll_posts", lambda _db, _now: 0) + monkeypatch.setattr(service, "expire_posts", lambda _db, _now: 0) + monkeypatch.setattr(service, "due_posts", lambda _db, _now, _limit: due) + return service.refresh_likes( + Mock(), + now=NOW, + read_syndication=read_syndication, + read_api=read_api, + api_budget=api_budget, + max_posts=500, + ) + + +def test_schedule_reads_densely_at_first_and_stops_before_the_week_ends(): + posted = NOW - timedelta(minutes=30) + assert service.next_read_at(posted, NOW) == posted + timedelta(hours=1) + assert service.next_read_at(posted, posted + timedelta(hours=13)) == ( + posted + timedelta(hours=24) + ) + assert service.next_read_at(posted, posted + timedelta(hours=144)) is None + + +def test_first_reading_uses_the_lifetime_average(): + metrics = _metrics(age_hours=4.0) + assert service.like_velocity(metrics, 200, NOW) == 50.0 + + +def test_first_reading_of_a_brand_new_post_counts_at_least_an_hour(): + metrics = _metrics(age_hours=0.1) + assert service.like_velocity(metrics, 30, NOW) == 30.0 + + +def test_later_readings_blend_the_new_rate_into_the_ema(): + metrics = _metrics( + age_hours=10.0, + like_count=100, + like_velocity=20.0, + last_read_at=NOW - timedelta(hours=2), + ) + # 40 new likes over 2 hours = 20/h; EMA 0.75 * 20 + 0.25 * 20. + assert service.like_velocity(metrics, 140, NOW) == pytest.approx(20.0) + # Like counts can drop (unlikes, removed accounts); the rate never goes negative. + assert service.like_velocity(metrics, 90, NOW) == pytest.approx(5.0) + + +def test_syndication_readings_are_stored_and_rescheduled(monkeypatch): + post = _metrics(age_hours=3.0) + report = _refresh(monkeypatch, [post], read_syndication=lambda _id: 60) + + assert report.syndication_reads == 1 + assert post.like_count == 60 + assert post.like_velocity == 20.0 + assert post.last_source == "syndication" + assert post.read_count == 1 + assert post.next_read_at == post.posted_at + timedelta(hours=4) + assert report.velocities == {"1": 20.0} + + +def test_posts_syndication_cannot_serve_fall_back_to_the_capped_api(monkeypatch): + posts = [_metrics(str(n)) for n in range(4)] + requested = [] + + def read_api(ids): + requested.append(ids) + return {"1": 90}, {"2"} + + report = _refresh( + monkeypatch, + posts, + read_syndication=lambda tweet_id: 30 if tweet_id == "0" else None, + read_api=read_api, + api_budget=2, + ) + + # Only two of the three fallback posts fit the remaining daily budget. + assert requested == [["1", "2"]] + assert report.api_reads == 2 + assert posts[1].last_source == "api" and posts[1].like_count == 90 + assert posts[2].gone_at == NOW and posts[2].next_read_at is None + assert report.gone == 1 + # The post over budget stays due for the next run. + assert posts[3].last_read_at is None + assert report.unread == 1 + + +def test_rate_limited_syndication_sends_the_rest_to_the_api(monkeypatch): + posts = [_metrics(str(n)) for n in range(3)] + + def read_syndication(tweet_id): + if tweet_id == "1": + raise service.SyndicationRateLimited() + return 10 + + report = _refresh( + monkeypatch, + posts, + read_syndication=read_syndication, + read_api=lambda ids: ({tweet_id: 5 for tweet_id in ids}, set()), + ) + + assert report.syndication_reads == 1 + assert report.api_reads == 2 + assert [post.last_source for post in posts] == ["syndication", "api", "api"] + + +def test_without_an_api_reader_fallback_posts_wait(monkeypatch): + post = _metrics() + report = _refresh(monkeypatch, [post], read_syndication=lambda _id: None) + + assert report.api_reads == 0 + assert report.unread == 1 + assert post.read_count == 0 + + +def test_api_failure_is_reported_without_losing_syndication_reads(monkeypatch): + posts = [_metrics("0"), _metrics("1")] + + def read_api(_ids): + raise httpx.HTTPStatusError( + "402 Payment Required", request=Mock(), response=Mock() + ) + + report = _refresh( + monkeypatch, + posts, + read_syndication=lambda tweet_id: 10 if tweet_id == "0" else None, + read_api=read_api, + ) + + assert report.syndication_reads == 1 + assert "402" in report.api_error + assert posts[0].like_count == 10 + assert posts[1].read_count == 0 + + +def _client(handler): + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def test_syndication_reader_returns_favorite_count(monkeypatch): + monkeypatch.setattr(service.time, "sleep", lambda _seconds: None) + client = _client( + lambda request: httpx.Response( + 200, json={"__typename": "Tweet", "favorite_count": 42} + ) + ) + assert service.make_syndication_reader(client)("123") == 42 + + +def test_syndication_reader_treats_tombstones_as_unavailable(monkeypatch): + monkeypatch.setattr(service.time, "sleep", lambda _seconds: None) + client = _client( + lambda request: httpx.Response(200, json={"__typename": "TweetTombstone"}) + ) + assert service.make_syndication_reader(client)("123") is None + + +def test_syndication_reader_raises_on_rate_limit(monkeypatch): + monkeypatch.setattr(service.time, "sleep", lambda _seconds: None) + client = _client(lambda request: httpx.Response(429)) + with pytest.raises(service.SyndicationRateLimited): + service.make_syndication_reader(client)("123") + + +def test_api_reader_batches_ids_and_reports_unavailable_posts(): + seen = {} + + def handler(request): + seen["ids"] = request.url.params["ids"] + seen["auth"] = request.headers["Authorization"] + return httpx.Response( + 200, + json={ + "data": [{"id": "1", "public_metrics": {"like_count": 7}}], + "errors": [{"resource_id": "2", "title": "Not Found Error"}], + }, + ) + + counts, unavailable = service.make_api_reader(_client(handler), "token")(["1", "2"]) + + assert seen == {"ids": "1,2", "auth": "Bearer token"} + assert counts == {"1": 7} + assert unavailable == {"2"} diff --git a/backend/utils/trending.py b/backend/utils/trending.py index 7ca2a6fb..6f6517b1 100644 --- a/backend/utils/trending.py +++ b/backend/utils/trending.py @@ -17,9 +17,18 @@ a_i = kind_weight_i * quantile_rank(hf_trending_score_i, all live scores of that kind) H = 1 - prod(1 - a_i) # weighted soft-OR: diminishing returns, stays in [0, 1] - paper_trending_score = SCORE_SCALE * (GITHUB_WEIGHT * G + HF_WEIGHT * H) + X = quantile_rank(fastest linked X post likes/hour, all live post velocities) -Consequences: + paper_trending_score = SCORE_SCALE * (w_g * G + w_h * H + w_x * X) + +The X weight ``w_x`` is configurable (``PWC_TRENDING_X_WEIGHT``, default 0) and +takes its share equally from GitHub and HF, so ``w_g = w_h = (1 - w_x) / 2`` and +the weights always sum to 1. Posts slower than +``PWC_TRENDING_X_MIN_LIKES_PER_HOUR`` (default 1) count for nothing and stay out +of the X corpus: the corpus is small and pre-selected for notable papers, so a +slow post would otherwise rank mid-table. + +Consequences (with X off): - A paper at the very top of GitHub velocity (G=1) and a paper with a top-of-Hub model (H=1) both reach SCORE_SCALE * 0.5 -> they rank together at the top, so a paper can trend purely on its Hugging Face artifacts (the DeepSeek-V4 case). @@ -42,11 +51,18 @@ import bisect from typing import Dict, Iterable, List, Sequence, Tuple -# Blend weights between the GitHub and Hugging Face components. Equal weights so -# the top of each signal reaches the same single-signal ceiling. +# Blend weights between the GitHub and Hugging Face components, both scaled by +# (1 - X weight). Equal weights so the top of each signal reaches the same +# single-signal ceiling. GITHUB_WEIGHT = 0.5 HF_WEIGHT = 0.5 +# Default X weight (off) and the slowest X post that still counts, in likes/hour. +DEFAULT_X_WEIGHT = 0.0 +DEFAULT_X_MIN_LIKES_PER_HOUR = 1.0 +# Only posts from the last week count; refresh_x_post_likes stops reading them. +X_POST_WINDOW_DAYS = 7 + # Per-kind importance of HF artifacts (applied to the quantile rank). KIND_WEIGHTS: Dict[str, float] = { "model": 1.0, @@ -200,9 +216,33 @@ def hf_component( return 1.0 - product -def composite_score(g_component: float, h_component: float) -> float: +def x_corpus(velocities: Iterable[float], min_likes_per_hour: float) -> List[float]: + """Sorted X velocities fast enough to count (the X rank corpus).""" + return sorted_positive(v for v in velocities if v >= min_likes_per_hour) + + +def x_component( + x_raw: float, x_sorted: Sequence[float], min_likes_per_hour: float +) -> float: + """Quantile-rank X component for a paper's fastest linked post.""" + if x_raw < min_likes_per_hour: + return 0.0 + return quantile_rank(x_sorted, x_raw) + + +def composite_score( + g_component: float, + h_component: float, + x_component: float = 0.0, + x_weight: float = DEFAULT_X_WEIGHT, +) -> float: """Combine the normalized components into the final paper score.""" - return SCORE_SCALE * (GITHUB_WEIGHT * g_component + HF_WEIGHT * h_component) + rest = 1.0 - x_weight + return SCORE_SCALE * ( + GITHUB_WEIGHT * rest * g_component + + HF_WEIGHT * rest * h_component + + x_weight * x_component + ) def compute_paper_score( @@ -211,8 +251,14 @@ def compute_paper_score( artifact_scores: Iterable[Tuple[str, float]], sorted_by_kind: Dict[str, Sequence[float]], kind_weights: Dict[str, float] = KIND_WEIGHTS, -) -> Tuple[float, float, float]: - """Compute ``(score, g_component, h_component)`` for one paper.""" + *, + x_raw: float = 0.0, + x_sorted: Sequence[float] = (), + x_weight: float = DEFAULT_X_WEIGHT, + x_min_likes_per_hour: float = DEFAULT_X_MIN_LIKES_PER_HOUR, +) -> Tuple[float, float, float, float]: + """Compute ``(score, g_component, h_component, x_component)`` for one paper.""" g = github_component(github_raw, github_sorted) h = hf_component(artifact_scores, sorted_by_kind, kind_weights) - return composite_score(g, h), g, h + x = x_component(x_raw, x_sorted, x_min_likes_per_hour) + return composite_score(g, h, x, x_weight), g, h, x diff --git a/deploy/configure_db_roles.py b/deploy/configure_db_roles.py index 194dbff5..cbd25ca2 100644 --- a/deploy/configure_db_roles.py +++ b/deploy/configure_db_roles.py @@ -79,9 +79,9 @@ def trending_role_sql(*, database: str, password: str) -> str: REVOKE CREATE ON SCHEMA public FROM pwc_cron_trending; GRANT CONNECT ON DATABASE {database} TO pwc_cron_trending; GRANT USAGE ON SCHEMA public TO pwc_cron_trending; -GRANT SELECT ON papers, paper_repositories, repositories, repository_star_history, paper_hf_artifacts, hf_artifacts TO pwc_cron_trending; +GRANT SELECT ON papers, paper_repositories, repositories, repository_star_history, paper_hf_artifacts, hf_artifacts, paper_x_threads, x_post_metrics TO pwc_cron_trending; GRANT UPDATE (trending_score, trending_score_updated_at, updated_at) ON repositories TO pwc_cron_trending; -GRANT UPDATE (trending_score, github_trending_component, hf_trending_component, trending_score_updated_at, updated_at) ON papers TO pwc_cron_trending; +GRANT UPDATE (trending_score, github_trending_component, hf_trending_component, x_trending_component, trending_score_updated_at, updated_at) ON papers TO pwc_cron_trending; DO $$ DECLARE @@ -148,8 +148,8 @@ def trending_role_sql(*, database: str, password: str) -> str: relation.relname = 'papers' AND attribute.attname IN ( 'trending_score', 'github_trending_component', - 'hf_trending_component', 'trending_score_updated_at', - 'updated_at' + 'hf_trending_component', 'x_trending_component', + 'trending_score_updated_at', 'updated_at' ) ) ) THEN diff --git a/deploy/configure_host_env.py b/deploy/configure_host_env.py index c74c081c..d2a6d8e6 100644 --- a/deploy/configure_host_env.py +++ b/deploy/configure_host_env.py @@ -69,6 +69,8 @@ "PWC_CONFERENCE_DEADLINES_HF_TOKEN", "PWC_GITHUB_TOKEN_HOT", "PWC_SEMANTIC_SCHOLAR_API_KEY_2", + "PWC_X_LIKES_API_DAILY_CAP", + "PWC_X_LIKES_APPLY", "PWC_X_THREAD_DISCOVERY_APPLY", "X_API_BEARER_TOKEN", "PWC_X_POST_MAX_CHARS", @@ -213,6 +215,12 @@ def existing_or_random(lines: dict[str, str], key: str) -> str: return decoded_value(line) if line else secrets.token_urlsafe(48) +# Operator-tuned trending score settings kept across trending-cron.env rewrites. +TRENDING_SCORE_SETTINGS = frozenset( + {"PWC_TRENDING_X_WEIGHT", "PWC_TRENDING_X_MIN_LIKES_PER_HOUR"} +) + + def trending_cron_values( legacy: dict[str, str], current: dict[str, str], @@ -228,7 +236,7 @@ def trending_cron_values( **{ key: decoded_value(line) for key, line in current.items() - if key.startswith("PWC_TRENDING_PIN_") + if key.startswith("PWC_TRENDING_PIN_") or key in TRENDING_SCORE_SETTINGS }, } diff --git a/deploy/cron_launcher.py b/deploy/cron_launcher.py index aafba6e0..78a6620f 100755 --- a/deploy/cron_launcher.py +++ b/deploy/cron_launcher.py @@ -41,6 +41,8 @@ "PWC_DB_PORT", "PWC_TRENDING_PIN_ARXIV_IDS", "PWC_TRENDING_PIN_PAPER_IDS", + "PWC_TRENDING_X_MIN_LIKES_PER_HOUR", + "PWC_TRENDING_X_WEIGHT", "TZ", } ) @@ -590,6 +592,15 @@ def lock_name(job: str, spec: Job, extra: list[str]) -> str: "section" ), ), + "refresh_x_post_likes": job( + "-m", + "backend.scripts.refresh_x_post_likes", + timeout="30m", + idempotency=( + "read each due post once per run and write only x_post_metrics; " + "report-only rolls back unless PWC_X_LIKES_APPLY=true" + ), + ), "post_trending_paper_to_x": job( "-m", "backend.scripts.post_trending_paper_to_x", diff --git a/deploy/cron_schedules.py b/deploy/cron_schedules.py index 39a29ab4..baaa7895 100644 --- a/deploy/cron_schedules.py +++ b/deploy/cron_schedules.py @@ -171,6 +171,11 @@ def invocations(self) -> tuple[tuple[str, ...], ...]: on_calendar=("*-*-* 08,20:22:00",), description="Discover author X threads for trending papers twice daily", ), + TimerSpec( + job="refresh_x_post_likes", + on_calendar=("*-*-* *:35:00",), + description="Refresh like counts of linked X posts in their first week", + ), TimerSpec( job="post_trending_paper_to_x", on_calendar=("*-*-* 00/4:13:00",), diff --git a/deploy/systemd/cron/pwc-cron-refresh-x-post-likes.service b/deploy/systemd/cron/pwc-cron-refresh-x-post-likes.service new file mode 100644 index 00000000..01ef6d4e --- /dev/null +++ b/deploy/systemd/cron/pwc-cron-refresh-x-post-likes.service @@ -0,0 +1,13 @@ +[Unit] +Description=Refresh like counts of linked X posts in their first week +After=network-online.target postgresql@16-main.service +Wants=network-online.target + +[Service] +Type=oneshot +ExecStart=/usr/local/sbin/pwc-cron-throttled refresh_x_post_likes +NoNewPrivileges=true +PrivateTmp=true +ProtectHome=true +ProtectSystem=strict +ReadWritePaths=/run/paperswithcode/cron-slots diff --git a/deploy/systemd/cron/pwc-cron-refresh-x-post-likes.timer b/deploy/systemd/cron/pwc-cron-refresh-x-post-likes.timer new file mode 100644 index 00000000..f9239a0f --- /dev/null +++ b/deploy/systemd/cron/pwc-cron-refresh-x-post-likes.timer @@ -0,0 +1,12 @@ +[Unit] +Description=Timer: Refresh like counts of linked X posts in their first week + +[Timer] +OnCalendar=*-*-* *:35:00 +RandomizedDelaySec=90 +AccuracySec=1m +Persistent=true +Unit=pwc-cron-refresh-x-post-likes.service + +[Install] +WantedBy=timers.target