spb/company-atlas

1"""Trend engine (spec §144–145): terms extracted from event and news titles (1–3-grams, stopwords, company names removed, lowercase),2aggregated per day into `trends(term, day, mentions, companies)` and only kept when ≥ `settings.trends_min_companies` distinct companies3use them. Momentum (7/30/90 d) is normalised by the number of active companies so that coverage growth alone does not create trends.4"""5from __future__ import annotations67import logging8import re9from collections import defaultdict10from datetime import UTC, date, datetime, timedelta11from typing import Any1213from companyatlas.config import settings14from companyatlas.db import execute, fetch_all, fetch_val, jsonb, transaction15from companyatlas.services.clustering import normalize_entity_key16from companyatlas.services.periodic import periodic1718log = logging.getLogger(__name__)1920TRENDS_FORMULA_VERSION = "trends-v1"21MAX_TERMS_PER_TITLE = 4022STOPWORDS = 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"])23_TOKEN_RE = re.compile(r"[a-z0-9][a-z0-9\-\+\.]{1,}")242526def extract_terms(title: str, *, company_name: str | None = None) -> set[str]:27    """Lowercased 1–3-grams over alphabetic tokens, stopwords dropped at n-gram edges, company name removed first."""28    text = (title or "").lower()29    if ":" in text and text.split(":", 1)[0].strip() in {"news release", "blog post", "changelog entry", "investor update", "earnings release"}:30        text = text.split(":", 1)[1]31    if company_name:32        norm = normalize_entity_key(company_name)33        if norm:34            text = re.sub(r"(?<![a-z0-9])" + re.escape(norm) + r"(?![a-z0-9])", " ", normalize_entity_key(text) if norm in normalize_entity_key(text) else text)35    tokens = [t.strip(".-+") for t in _TOKEN_RE.findall(text)]36    tokens = [t for t in tokens if len(t) >= 3 and not t.isdigit()]37    terms: set[str] = set()38    for n in (1, 2, 3):39        for i in range(len(tokens) - n + 1):40            gram = tokens[i:i + n]41            if gram[0] in STOPWORDS or gram[-1] in STOPWORDS:42                continue43            if n == 1 and (len(gram[0]) < 4 or gram[0] in STOPWORDS):44                continue45            if all(g in STOPWORDS for g in gram):46                continue47            terms.add(" ".join(gram))48            if len(terms) >= MAX_TERMS_PER_TITLE:49                return terms50    return terms515253async def compute_trends(day: date | None = None) -> dict[str, Any]:54    """Aggregate terms for one UTC day (default: today) into `trends`. Idempotent."""55    day = day or datetime.now(UTC).date()56    d0 = datetime(day.year, day.month, day.day, tzinfo=UTC)57    d1 = d0 + timedelta(days=1)58    mentions: dict[str, int] = defaultdict(int)59    companies: dict[str, set[str]] = defaultdict(set)60    async with transaction() as conn:61        rows = await fetch_all(conn, """62            select e.title, e.company_id, co.display_name from events e join companies co on co.id = e.company_id63            where e.detected_at >= :d0 and e.detected_at < :d1 and e.status in ('active', 'review')64            union all65            select n.title, n.company_id, co.display_name from news_items n join companies co on co.id = n.company_id66            where n.first_seen_at >= :d0 and n.first_seen_at < :d1""", d0=d0, d1=d1)67        for r in rows:68            for term in extract_terms(r["title"], company_name=r["display_name"]):69                mentions[term] += 170                companies[term].add(r["company_id"])71        kept = {t: (mentions[t], len(companies[t])) for t in mentions if len(companies[t]) >= settings.trends_min_companies}72        await execute(conn, "delete from trends where day = :d", d=day)73        for term, (m, c) in kept.items():74            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",75                          t=term[:120], d=day, m=m, c=c)76    return {"day": day.isoformat(), "titles": len(rows), "terms_seen": len(mentions), "terms_kept": len(kept)}777879async def compute_trends_range(days: int) -> list[dict[str, Any]]:80    today = datetime.now(UTC).date()81    return [await compute_trends(today - timedelta(days=i)) for i in range(days - 1, -1, -1)]828384async def trend_momentum(window_days: int = 7, *, limit: int = 30) -> list[dict[str, Any]]:85    """Terms ranked by coverage-normalised momentum: rate = mentions / active companies; momentum = (rate_now − rate_prev) / max(rate_prev, ε)."""86    today = datetime.now(UTC).date()87    cur_from, prev_from = today - timedelta(days=window_days - 1), today - timedelta(days=2 * window_days - 1)88    async with transaction() as conn:89        rows = await fetch_all(conn, """90            select term, day, mentions, companies from trends where day >= :prev_from order by term, day""", prev_from=prev_from)91        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)92        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)93    cov_cur = float(cov_cur or 0) or 1.094    cov_prev = float(cov_prev or 0) or cov_cur95    per: dict[str, dict[str, Any]] = {}96    for r in rows:97        t = per.setdefault(r["term"], {"cur": 0, "prev": 0, "companies": set(), "series": {}})98        if r["day"] >= cur_from:99            t["cur"] += r["mentions"]100            t["companies"].add(r["companies"])101        else:102            t["prev"] += r["mentions"]103        t["series"][r["day"]] = r["mentions"]104    out = []105    eps = 0.5 / cov_prev106    for term, t in per.items():107        if t["cur"] == 0:108            continue109        rate_now, rate_prev = t["cur"] / cov_cur, t["prev"] / cov_prev110        momentum = (rate_now - rate_prev) / max(rate_prev, eps)111        series = [t["series"].get(cur_from + timedelta(days=i), 0) for i in range(window_days)]112        out.append({"term": term, "mentions": t["cur"], "companies": max(t["companies"]) if t["companies"] else 0, "momentum": round(momentum, 3), "series": series,113                    "window_days": window_days, "formula_version": TRENDS_FORMULA_VERSION})114    out.sort(key=lambda x: (x["momentum"], x["mentions"]), reverse=True)115    return out[:limit]116117118async def store_momentum_snapshots() -> None:119    for w in (7, 30, 90):120        items = await trend_momentum(w, limit=50)121        async with transaction() as conn:122            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()",123                          k=f"trends:momentum:{w}d", v=jsonb({"computed_at": datetime.now(UTC).isoformat(), "items": items}))124125126@periodic("trends", cron="25 * * * *")127async def trends_task() -> None:128    today = datetime.now(UTC).date()129    stats = await compute_trends(today)130    if datetime.now(UTC).hour == 0:131        await compute_trends(today - timedelta(days=1))132    await store_momentum_snapshots()133    log.info("trends", extra=stats)134135136__all__ = ["STOPWORDS", "TRENDS_FORMULA_VERSION", "compute_trends", "compute_trends_range", "extract_terms", "store_momentum_snapshots", "trend_momentum"]137