"""Trend engine (spec §144–145): terms extracted from event and news titles (1–3-grams, stopwords, company names removed, lowercase), aggregated per day into `trends(term, day, mentions, companies)` and only kept when ≥ `settings.trends_min_companies` distinct companies use them. Momentum (7/30/90 d) is normalised by the number of active companies so that coverage growth alone does not create trends. """ from __future__ import annotations import logging import re from collections import defaultdict from datetime import UTC, date, datetime, timedelta from typing import Any from companyatlas.config import settings from companyatlas.db import execute, fetch_all, fetch_val, jsonb, transaction from companyatlas.services.clustering import normalize_entity_key from companyatlas.services.periodic import periodic log = logging.getLogger(__name__) TRENDS_FORMULA_VERSION = "trends-v1" MAX_TERMS_PER_TITLE = 40 STOPWORDS = frozenset(["a", "an", "the", "and", "or", "of", "to", "in", "on", "at", "for", "by", "with", "from", "as", "is", "are", "was", "were", "be", "been", "being", "this", "that", "these", "those", "it", "its", "into", "over", "under", "about", "after", "before", "than", "then", "there", "their", "they", "them", "we", "our", "you", "your", "he", "she", "his", "her", "not", "no", "yes", "new", "now", "more", "most", "less", "least", "very", "up", "down", "out", "off", "via", "per", "vs", "detected", "listed", "observed", "monitored", "longer", "visible", "no", "page", "pages", "updated", "update", "updates", "changed", "change", "changes", "section", "sections", "block", "blocks", "content", "position", "positions", "job", "jobs", "listing", "listings", "careers", "career", "title", "plan", "plans", "price", "prices", "pricing", "tier", "tiers", "product", "products", "office", "offices", "location", "locations", "country", "presence", "leadership", "executive", "team", "news", "release", "blog", "post", "entry", "changelog", "investor", "earnings", "homepage", "website", "site", "material", "materially", "redesigned", "documentation", "docs", "api", "reference", "terms", "service", "privacy", "policy", "security", "added", "removed", "increase", "decrease", "signal", "one", "two", "three", "announces", "announced", "announce", "introduces", "introducing", "launches", "launched", "launch", "today", "year", "years", "month", "months", "week", "weeks", "day", "days", "q1", "q2", "q3", "q4", "fy", "inc", "corp", "ltd", "llc", "co", "company", "companies", "group", "plc", "ag", "sa", "nv", "se", "gmbh"]) _TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9\-\+\.]{1,}") def extract_terms(title: str, *, company_name: str | None = None) -> set[str]: """Lowercased 1–3-grams over alphabetic tokens, stopwords dropped at n-gram edges, company name removed first.""" text = (title or "").lower() if ":" in text and text.split(":", 1)[0].strip() in {"news release", "blog post", "changelog entry", "investor update", "earnings release"}: text = text.split(":", 1)[1] if company_name: norm = normalize_entity_key(company_name) if norm: text = re.sub(r"(?= 3 and not t.isdigit()] terms: set[str] = set() for n in (1, 2, 3): for i in range(len(tokens) - n + 1): gram = tokens[i:i + n] if gram[0] in STOPWORDS or gram[-1] in STOPWORDS: continue if n == 1 and (len(gram[0]) < 4 or gram[0] in STOPWORDS): continue if all(g in STOPWORDS for g in gram): continue terms.add(" ".join(gram)) if len(terms) >= MAX_TERMS_PER_TITLE: return terms return terms async def compute_trends(day: date | None = None) -> dict[str, Any]: """Aggregate terms for one UTC day (default: today) into `trends`. Idempotent.""" day = day or datetime.now(UTC).date() d0 = datetime(day.year, day.month, day.day, tzinfo=UTC) d1 = d0 + timedelta(days=1) mentions: dict[str, int] = defaultdict(int) companies: dict[str, set[str]] = defaultdict(set) async with transaction() as conn: rows = await fetch_all(conn, """ select e.title, e.company_id, co.display_name from events e join companies co on co.id = e.company_id where e.detected_at >= :d0 and e.detected_at < :d1 and e.status in ('active', 'review') union all select n.title, n.company_id, co.display_name from news_items n join companies co on co.id = n.company_id where n.first_seen_at >= :d0 and n.first_seen_at < :d1""", d0=d0, d1=d1) for r in rows: for term in extract_terms(r["title"], company_name=r["display_name"]): mentions[term] += 1 companies[term].add(r["company_id"]) kept = {t: (mentions[t], len(companies[t])) for t in mentions if len(companies[t]) >= settings.trends_min_companies} await execute(conn, "delete from trends where day = :d", d=day) for term, (m, c) in kept.items(): await execute(conn, "insert into trends (term, day, mentions, companies) values (:t, :d, :m, :c) on conflict (term, day) do update set mentions = excluded.mentions, companies = excluded.companies", t=term[:120], d=day, m=m, c=c) return {"day": day.isoformat(), "titles": len(rows), "terms_seen": len(mentions), "terms_kept": len(kept)} async def compute_trends_range(days: int) -> list[dict[str, Any]]: today = datetime.now(UTC).date() return [await compute_trends(today - timedelta(days=i)) for i in range(days - 1, -1, -1)] async def trend_momentum(window_days: int = 7, *, limit: int = 30) -> list[dict[str, Any]]: """Terms ranked by coverage-normalised momentum: rate = mentions / active companies; momentum = (rate_now − rate_prev) / max(rate_prev, ε).""" today = datetime.now(UTC).date() cur_from, prev_from = today - timedelta(days=window_days - 1), today - timedelta(days=2 * window_days - 1) async with transaction() as conn: rows = await fetch_all(conn, """ select term, day, mentions, companies from trends where day >= :prev_from order by term, day""", prev_from=prev_from) cov_cur = await fetch_val(conn, "select coalesce(avg(companies_active), 0) from global_daily where day >= :f and day <= :t", f=cur_from, t=today) cov_prev = await fetch_val(conn, "select coalesce(avg(companies_active), 0) from global_daily where day >= :f and day < :t", f=prev_from, t=cur_from) cov_cur = float(cov_cur or 0) or 1.0 cov_prev = float(cov_prev or 0) or cov_cur per: dict[str, dict[str, Any]] = {} for r in rows: t = per.setdefault(r["term"], {"cur": 0, "prev": 0, "companies": set(), "series": {}}) if r["day"] >= cur_from: t["cur"] += r["mentions"] t["companies"].add(r["companies"]) else: t["prev"] += r["mentions"] t["series"][r["day"]] = r["mentions"] out = [] eps = 0.5 / cov_prev for term, t in per.items(): if t["cur"] == 0: continue rate_now, rate_prev = t["cur"] / cov_cur, t["prev"] / cov_prev momentum = (rate_now - rate_prev) / max(rate_prev, eps) series = [t["series"].get(cur_from + timedelta(days=i), 0) for i in range(window_days)] out.append({"term": term, "mentions": t["cur"], "companies": max(t["companies"]) if t["companies"] else 0, "momentum": round(momentum, 3), "series": series, "window_days": window_days, "formula_version": TRENDS_FORMULA_VERSION}) out.sort(key=lambda x: (x["momentum"], x["mentions"]), reverse=True) return out[:limit] async def store_momentum_snapshots() -> None: for w in (7, 30, 90): items = await trend_momentum(w, limit=50) async with transaction() as conn: await execute(conn, "insert into settings_kv (key, value, updated_at) values (:k, cast(:v as jsonb), now()) on conflict (key) do update set value = excluded.value, updated_at = now()", k=f"trends:momentum:{w}d", v=jsonb({"computed_at": datetime.now(UTC).isoformat(), "items": items})) @periodic("trends", cron="25 * * * *") async def trends_task() -> None: today = datetime.now(UTC).date() stats = await compute_trends(today) if datetime.now(UTC).hour == 0: await compute_trends(today - timedelta(days=1)) await store_momentum_snapshots() log.info("trends", extra=stats) __all__ = ["STOPWORDS", "TRENDS_FORMULA_VERSION", "compute_trends", "compute_trends_range", "extract_terms", "store_momentum_snapshots", "trend_momentum"]