"""Aggregate producers shared by several routers (`/pulse` composes them): industry & country rows, rankings, trends, map buckets, global stats and the activity index. Every producer degrades to empty lists / `null` values on an empty database — nothing is invented.""" from __future__ import annotations import asyncio from collections import defaultdict from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy.ext.asyncio import AsyncConnection from companyatlas import archive from companyatlas.api import queries as q from companyatlas.api import serializers as ser from companyatlas.api.common import cache, cached from companyatlas.db import connection, fetch_all, fetch_one from companyatlas.ids import slugify from companyatlas.taxonomy import METRICS_FORMULA_VERSION, Metric # ------------------------------------------------------------------------------------------------ industries / countries _ROW_METRICS = (Metric.ACTIVITY_SCORE.value, Metric.HIRING_MOMENTUM_30D.value, Metric.AI_ADOPTION.value) def _avg_map(rows: list[dict[str, Any]], key: str) -> dict[str, dict[str, float]]: out: dict[str, dict[str, float]] = defaultdict(dict) for r in rows: if r[key] is not None and r["v"] is not None: out[r[key]][r["metric"]] = float(r["v"]) return out async def industry_rows(conn: AsyncConnection, *, country: str | None = None) -> list[dict[str, Any]]: scope = " and c.country = cast(:country as char(2))" if country else "" params: dict[str, Any] = {"country": country} if country else {} d7, d30 = q.days_ago(7), q.days_ago(30) taxonomy = await fetch_all(conn, "select slug, name, parent_slug, description, sort_order from industries order by sort_order, name") companies = await fetch_all(conn, f"select ind, count(*) as n from companies c, unnest(c.industries) ind where c.status = 'ACTIVE'{scope} " "group by ind", **params) events = await fetch_all(conn, "select ind, e.event_type, count(*) filter (where e.detected_at >= :d7) as n7, count(*) as n30 " "from events e join companies c on c.id = e.company_id, unnest(c.industries) ind " f"where e.status = 'active' and e.detected_at >= :d30{scope} group by ind, e.event_type", d7=d7, d30=d30, **params) metrics = await fetch_all(conn, "select ind, m.metric, avg(m.value) as v from metrics_current m join companies c on c.id = m.company_id, " f"unnest(c.industries) ind where m.metric = any(cast(:ms as text[])){scope} group by ind, m.metric", ms=list(_ROW_METRICS), **params) n_companies = {r["ind"]: int(r["n"]) for r in companies} ev7: dict[str, int] = defaultdict(int) ev30: dict[str, int] = defaultdict(int) types: dict[str, dict[str, int]] = defaultdict(dict) for r in events: ev7[r["ind"]] += int(r["n7"]) ev30[r["ind"]] += int(r["n30"]) types[r["ind"]][r["event_type"]] = int(r["n30"]) avg = _avg_map(metrics, "ind") names = {t["slug"]: t for t in taxonomy} slugs = list(names) + [s for s in n_companies if s not in names] rows = [] for slug in slugs: t = names.get(slug, {}) m = avg.get(slug, {}) rows.append({"slug": slug, "name": t.get("name") or slug.replace("-", " ").title(), "parent_slug": t.get("parent_slug"), "description": t.get("description"), "companies": n_companies.get(slug, 0), "events_7d": ev7.get(slug, 0), "events_30d": ev30.get(slug, 0), "hiring_momentum_30d": ser.metric_value("hiring_momentum_30d", m.get("hiring_momentum_30d")), "activity_score": ser.metric_value("activity_score", m.get("activity_score")), "ai_adoption": ser.metric_value("ai_adoption", m.get("ai_adoption")), "top_event_types": [k for k, _ in sorted(types.get(slug, {}).items(), key=lambda kv: (-kv[1], kv[0]))[:3]]}) rows.sort(key=lambda r: (-r["companies"], -(r["activity_score"] or 0), r["name"])) return rows async def country_rows(conn: AsyncConnection, *, industry: str | None = None) -> list[dict[str, Any]]: scope = " and cast(:industry as text) = any(c.industries)" if industry else "" params: dict[str, Any] = {"industry": industry} if industry else {} d7, d30 = q.days_ago(7), q.days_ago(30) ref = await fetch_all(conn, "select code, name, region, subregion, lat, lon from countries order by name") companies = await fetch_all(conn, f"select c.country as code, count(*) as n from companies c where c.status = 'ACTIVE' and c.country is not null{scope} " "group by c.country", **params) events = await fetch_all(conn, "select c.country as code, count(*) filter (where e.detected_at >= :d7) as n7, count(*) as n30 " "from events e join companies c on c.id = e.company_id " f"where e.status = 'active' and e.detected_at >= :d30 and c.country is not null{scope} group by c.country", d7=d7, d30=d30, **params) metrics = await fetch_all(conn, "select c.country as code, m.metric, avg(m.value) as v from metrics_current m join companies c on c.id = m.company_id " f"where m.metric = any(cast(:ms as text[])) and c.country is not null{scope} group by c.country, m.metric", ms=list(_ROW_METRICS), **params) mix = await fetch_all(conn, "select c.country as code, ind, count(*) as n from companies c, unnest(c.industries) ind " f"where c.status = 'ACTIVE' and c.country is not null{scope} group by c.country, ind", **params) n_companies = {r["code"]: int(r["n"]) for r in companies} ev = {r["code"]: (int(r["n7"]), int(r["n30"])) for r in events} avg = _avg_map(metrics, "code") mixes: dict[str, list[tuple[str, int]]] = defaultdict(list) for r in mix: mixes[r["code"]].append((r["ind"], int(r["n"]))) names = {r["code"]: r for r in ref} codes = list(names) + [c for c in n_companies if c not in names] rows = [] for code in codes: c = names.get(code, {}) m = avg.get(code, {}) name = c.get("name") or code rows.append({"code": code, "slug": slugify(name), "name": name, "region": c.get("region"), "subregion": c.get("subregion"), "companies": n_companies.get(code, 0), "events_7d": ev.get(code, (0, 0))[0], "events_30d": ev.get(code, (0, 0))[1], "hiring_momentum_30d": ser.metric_value("hiring_momentum_30d", m.get("hiring_momentum_30d")), "activity_score": ser.metric_value("activity_score", m.get("activity_score")), "ai_adoption": ser.metric_value("ai_adoption", m.get("ai_adoption")), "industry_mix": [{"industry": i, "companies": n} for i, n in sorted(mixes.get(code, []), key=lambda x: -x[1])[:6]], "lat": c.get("lat"), "lon": c.get("lon")}) rows.sort(key=lambda r: (-r["companies"], -(r["activity_score"] or 0), r["name"])) return rows async def cached_industry_rows(country: str | None = None) -> list[dict[str, Any]]: async def produce() -> list[dict[str, Any]]: async with connection() as conn: return await industry_rows(conn, country=country) return await cached(f"industries:{country or ''}", 300, produce) async def cached_country_rows(industry: str | None = None) -> list[dict[str, Any]]: async def produce() -> list[dict[str, Any]]: async with connection() as conn: return await country_rows(conn, industry=industry) return await cached(f"countries:{industry or ''}", 300, produce) async def resolve_country(conn: AsyncConnection, key: str) -> dict[str, Any] | None: """Accept an ISO-2 code (`CA`) or a name slug (`canada`).""" key = (key or "").strip() if not key: return None if len(key) == 2: row = await fetch_one(conn, "select code, name, region, subregion, lat, lon from countries where code = cast(:c as char(2))", c=key.upper()) if row: return row rows = await fetch_all(conn, "select code, name, region, subregion, lat, lon from countries") k = slugify(key) for r in rows: if slugify(r["name"]) == k or r["code"].lower() == key.lower(): return r return None # ------------------------------------------------------------------------------------------------ rankings RANKING_KINDS: dict[str, dict[str, Any]] = { "most_active": {"metric": Metric.ACTIVITY_SCORE.value}, "hiring_growth": {"metric": "hiring_momentum_{w}", "min": 0.0}, "hiring_decline": {"metric": "hiring_momentum_{w}", "asc": True, "max": 0.0}, "product_velocity": {"metric": Metric.PRODUCT_VELOCITY.value}, "ai_active": {"metric": Metric.AI_ADOPTION.value}, "geo_expansion": {"metric": Metric.GEO_EXPANSION.value}, "developer_momentum": {"metric": Metric.DEVELOPER_MOMENTUM.value}, "pricing_changes": {"events": "PRICING"}, "unusual_activity": {"metric": Metric.ANOMALY_SCORE.value}, } _HIRING_WINDOW = {"24h": "7d", "7d": "7d", "30d": "30d", "90d": "90d", "1y": "90d"} async def ranking(conn: AsyncConnection, kind: str, window: str, *, country: str | None = None, industry: str | None = None, limit: int = 50) -> list[dict[str, Any]]: """[{company_id, value, delta}] for a ranking kind. `delta` = value − series value at the start of the window (null if unknown).""" spec = RANKING_KINDS[kind] scope, params = [], {} if country: scope.append("c.country = cast(:country as char(2))") params["country"] = country.upper()[:2] if industry: scope.append("cast(:industry as text) = any(c.industries)") params["industry"] = industry[:80] scope_sql = (" and " + " and ".join(scope)) if scope else "" since = q.window_start(window) if "events" in spec: rows = await fetch_all(conn, "select e.company_id, count(*) as value from events e join companies c on c.id = e.company_id " f"where e.status = 'active' and e.event_type = :et and e.detected_at >= :since and c.status = 'ACTIVE'{scope_sql} " "group by e.company_id order by value desc, e.company_id limit :limit", et=spec["events"], since=since, limit=limit, **params) return [{"company_id": r["company_id"], "value": float(r["value"]), "delta": None} for r in rows] metric = spec["metric"].format(w=_HIRING_WINDOW.get(window, "30d")) bounds = "" if "min" in spec: bounds += " and m.value > :vmin" params["vmin"] = spec["min"] if "max" in spec: bounds += " and m.value < :vmax" params["vmax"] = spec["max"] order = "m.value asc" if spec.get("asc") else "m.value desc" rows = await fetch_all(conn, "select m.company_id, m.value from metrics_current m join companies c on c.id = m.company_id " f"where m.metric = :metric and c.status = 'ACTIVE'{scope_sql}{bounds} order by {order}, m.company_id limit :limit", metric=metric, limit=limit, **params) ids = [r["company_id"] for r in rows] past = await q.metric_values_at(conn, ids, metric, since.date()) out = [] for r in rows: v = float(r["value"]) p = past.get(r["company_id"]) out.append({"company_id": r["company_id"], "value": v, "delta": round(v - p, 2) if p is not None else None}) return out async def ranking_cards(conn: AsyncConnection, kind: str, window: str, *, country: str | None = None, industry: str | None = None, limit: int = 50, sparkline: bool = False) -> list[dict[str, Any]]: items = await ranking(conn, kind, window, country=country, industry=industry, limit=limit) cards = await q.fetch_cards_by_ids(conn, [i["company_id"] for i in items], sparkline=sparkline) by_id = {c["id"]: c for c in cards} out = [] for rank, it in enumerate(items, start=1): row = by_id.get(it["company_id"]) if row is None: continue card = ser.company_card(row) metric = RANKING_KINDS[kind].get("metric", "").format(w=_HIRING_WINDOW.get(window, "30d")) card.update({"rank": rank, "value": ser.metric_value(metric, it["value"]) if metric else int(it["value"]), "delta": it["delta"]}) out.append(card) return out # ------------------------------------------------------------------------------------------------ trends / signals async def trend_rows(conn: AsyncConnection, window_days: int, limit: int) -> list[dict[str, Any]]: d0 = q.days_ago(window_days).date() top = await fetch_all(conn, "select term, sum(mentions) as mentions, max(companies) as companies from trends where day >= :d group by term " "order by mentions desc, term limit :limit", d=d0, limit=limit) if not top: return [] terms = [t["term"] for t in top] series = await fetch_all(conn, "select term, day, mentions from trends where day >= :d and term = any(cast(:terms as text[])) order by term, day", d=d0, terms=terms) by_term: dict[str, list[tuple[Any, int]]] = defaultdict(list) for r in series: by_term[r["term"]].append((r["day"], int(r["mentions"]))) mid = q.days_ago(window_days // 2 or 1).date() out = [] for t in top: pts = by_term.get(t["term"], []) first = sum(m for d, m in pts if d < mid) second = sum(m for d, m in pts if d >= mid) momentum = round((second - first) / first * 100, 1) if first > 0 else None out.append({"term": t["term"], "mentions": int(t["mentions"]), "companies": int(t["companies"] or 0), "momentum": momentum, "series": [m for _, m in pts]}) return out # ------------------------------------------------------------------------------------------------ map MAP_MAX_BUCKETS = 600 async def map_buckets(conn: AsyncConnection, metric: str = "events_30d") -> list[dict[str, Any]]: d30 = q.days_ago(30) ev_by_company = {r["company_id"]: int(r["n"]) for r in await fetch_all( conn, "select company_id, count(*) as n from events where status = 'active' and detected_at >= :d group by company_id", d=d30)} jobs_by_company = {r["company_id"]: int(r["n"]) for r in await fetch_all( conn, "select company_id, count(*) as n from jobs where status = 'open' group by company_id")} countries = await fetch_all(conn, "select k.code, k.name, k.lat, k.lon, c.id, c.slug, c.display_name, c.importance from countries k " "join companies c on c.country = k.code and c.status = 'ACTIVE' where k.lat is not null order by c.importance desc") cities = await fetch_all(conn, "select l.country, l.city, avg(l.lat) as lat, avg(l.lon) as lon, " "array_agg(distinct l.company_id) as company_ids from locations l join companies c on c.id = l.company_id " "where l.status = 'listed' and l.lat is not null and l.lon is not null and l.city is not null " "group by l.country, l.city order by count(distinct l.company_id) desc limit :lim", lim=MAP_MAX_BUCKETS) by_country: dict[str, dict[str, Any]] = {} for r in countries: b = by_country.setdefault(r["code"], {"lat": r["lat"], "lon": r["lon"], "country": r["code"], "city": None, "companies": 0, "events_30d": 0, "jobs_open": 0, "top": [], "_ids": []}) b["companies"] += 1 b["events_30d"] += ev_by_company.get(r["id"], 0) b["jobs_open"] += jobs_by_company.get(r["id"], 0) if len(b["top"]) < 3: b["top"].append({"slug": r["slug"], "display_name": r["display_name"]}) names: dict[str, tuple[str, str]] = {} if cities: ids = sorted({cid for r in cities for cid in (r["company_ids"] or [])}) names = {r["id"]: (r["slug"], r["display_name"]) for r in await fetch_all( conn, "select id, slug, display_name from companies where id = any(cast(:ids as text[])) order by importance desc", ids=ids[:5000])} buckets = list(by_country.values()) for r in cities: ids = [cid for cid in (r["company_ids"] or []) if cid in names] buckets.append({"lat": round(float(r["lat"]), 4), "lon": round(float(r["lon"]), 4), "country": r["country"], "city": r["city"], "companies": len(ids), "events_30d": sum(ev_by_company.get(i, 0) for i in ids), "jobs_open": sum(jobs_by_company.get(i, 0) for i in ids), "top": [{"slug": names[i][0], "display_name": names[i][1]} for i in ids[:3]]}) key = {"companies": "companies", "hiring": "jobs_open"}.get(metric, "events_30d") buckets.sort(key=lambda b: (-b[key], -b["companies"])) for b in buckets: b.pop("_ids", None) return buckets[:MAP_MAX_BUCKETS] # ------------------------------------------------------------------------------------------------ global stats / index async def archive_stats() -> dict[str, int]: async def produce() -> dict[str, int]: async with connection() as conn: kv = await q.settings_value(conn, "archive:stats") if isinstance(kv, dict) and "objects" in kv: return {"objects": int(kv.get("objects") or 0), "bytes": int(kv.get("bytes") or 0)} return await asyncio.to_thread(archive.store_stats) return await cached("archive:stats", 600, produce) async def global_stats(conn: AsyncConnection) -> dict[str, Any]: today = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) row = await fetch_one(conn, """ select (select count(*) from companies) as companies, (select count(*) from companies where status = 'ACTIVE' and onboarding_status = 'active') as companies_active, (select count(*) from sensors where status <> 'retired') as sensors, (select count(*) from sensors where status = 'active') as sensors_active, (select coalesce(sum(observation_count), 0) from sensors) as observations, (select coalesce(sum(snapshot_count), 0) from sensors) as snapshots, (select count(*) from changes) as changes, (select count(*) from changes where kind in ('meaningful', 'major', 'critical')) as meaningful_changes, (select count(*) from events where status = 'active') as events, (select count(*) from jobs where status = 'open') as jobs_open, (select count(distinct country) from companies where country is not null and status = 'ACTIVE') as countries, (select count(distinct ind) from companies c, unnest(c.industries) ind where c.status = 'ACTIVE') as industries, (select count(*) from observations where fetched_at >= :today) as observations_today, (select count(*) from changes where detected_at >= :today) as changes_today, (select count(*) from events where status = 'active' and detected_at >= :today) as events_today, (select min(first_observed_at) from companies) as oldest_observation_at, (select max(fetched_at) from observations) as last_observation_at """, today=today) row = row or {} started = await q.settings_value(conn, "dataset_started_at") started_dt = q.parse_iso(started) if isinstance(started, str) else None now = datetime.now(UTC) oldest = row.get("oldest_observation_at") return {"companies": int(row.get("companies") or 0), "companies_active": int(row.get("companies_active") or 0), "sensors": int(row.get("sensors") or 0), "sensors_active": int(row.get("sensors_active") or 0), "observations": int(row.get("observations") or 0), "snapshots": int(row.get("snapshots") or 0), "changes": int(row.get("changes") or 0), "meaningful_changes": int(row.get("meaningful_changes") or 0), "events": int(row.get("events") or 0), "jobs_open": int(row.get("jobs_open") or 0), "countries": int(row.get("countries") or 0), "industries": int(row.get("industries") or 0), "observations_today": int(row.get("observations_today") or 0), "changes_today": int(row.get("changes_today") or 0), "events_today": int(row.get("events_today") or 0), "dataset_started_at": started_dt, "dataset_age_days": (now - started_dt).days if started_dt else None, "oldest_history_days": (now - oldest).days if oldest else None, "last_observation_at": row.get("last_observation_at"), "archive": await archive_stats()} async def cached_global_stats() -> dict[str, Any]: async def produce() -> dict[str, Any]: async with connection() as conn: return await global_stats(conn) return await cached("stats", 60, produce) async def global_daily_rows(conn: AsyncConnection, days: int) -> list[dict[str, Any]]: rows = await fetch_all(conn, "select * from global_daily where day >= :d order by day asc limit :lim", d=q.days_ago(days).date(), lim=days + 1) return [{"day": r["day"], "companies_active": r["companies_active"], "sensors_active": r["sensors_active"], "observations": r["observations"], "changes": r["changes"], "meaningful_changes": r["meaningful_changes"], "events": r["events"], "events_by_type": ser._dict(r["events_by_type"]), "jobs_open": r["jobs_open"], "jobs_new": r["jobs_new"], "jobs_removed": r["jobs_removed"], "activity_index": ser._float(r["activity_index"], 2), "by_country": ser._dict(r["by_country"]), "by_industry": ser._dict(r["by_industry"])} for r in rows] def _index_at(rows: list[dict[str, Any]], days_back: int) -> float | None: if not rows: return None target = rows[-1]["day"] - timedelta(days=days_back) candidates = [r for r in rows if r["day"] <= target and r["activity_index"] is not None] return candidates[-1]["activity_index"] if candidates else None async def activity_index(conn: AsyncConnection, days: int = 365) -> dict[str, Any]: rows = await global_daily_rows(conn, days) with_value = [r for r in rows if r["activity_index"] is not None] latest = with_value[-1] if with_value else None value = latest["activity_index"] if latest else None v7, v30 = _index_at(with_value, 7), _index_at(with_value, 30) formula = await q.settings_value(conn, "index:formula_version") return {"value": value, "baseline": 100, "delta_7d": round(value - v7, 2) if value is not None and v7 is not None else None, "delta_30d": round(value - v30, 2) if value is not None and v30 is not None else None, "series": [{"day": r["day"], "value": r["activity_index"], "confidence": None} for r in with_value], "by_type": (latest or {}).get("events_by_type", {}), "by_country": [{"key": k, "value": v} for k, v in sorted((latest or {}).get("by_country", {}).items(), key=lambda kv: -float(kv[1] or 0))[:50]], "by_industry": [{"key": k, "value": v} for k, v in sorted((latest or {}).get("by_industry", {}).items(), key=lambda kv: -float(kv[1] or 0))[:50]], "formula_version": formula if isinstance(formula, str) else METRICS_FORMULA_VERSION, "computed_at": (latest or {}).get("day")} async def system_health(conn: AsyncConnection) -> dict[str, Any]: today = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) row = await fetch_one(conn, """ select (select count(*) from sensors where status = 'active') as sensors_online, (select count(*) from sensors where status in ('failing', 'stale', 'blocked')) as sensors_failing, (select count(*) from observations where fetched_at >= :today) as observations_today, (select count(*) from events where status = 'active' and detected_at >= :today) as events_today, (select count(distinct country) from companies where country is not null and status = 'ACTIVE') as countries_covered, (select extract(epoch from (now() - min(run_at))) from queue_jobs where status = 'pending' and run_at <= now()) as queue_lag_s, (select count(*) from observations where fetched_at >= now() - interval '10 minutes') as obs_10m, (select count(*) from observations where fetched_at >= now() - interval '24 hours') as obs_24h, (select count(*) from observations where fetched_at >= now() - interval '24 hours' and failure_class is null) as ok_24h """, today=today) or {} hb = await q.settings_value(conn, "scheduler:heartbeat") tick = None if isinstance(hb, dict): tick = hb.get("at") or hb.get("ts") or hb.get("time") or hb.get("last_tick_at") elif isinstance(hb, str): tick = hb obs_24h = int(row.get("obs_24h") or 0) return {"sensors_online": int(row.get("sensors_online") or 0), "sensors_failing": int(row.get("sensors_failing") or 0), "observations_today": int(row.get("observations_today") or 0), "events_today": int(row.get("events_today") or 0), "countries_covered": int(row.get("countries_covered") or 0), "queue_lag_s": round(float(row["queue_lag_s"]), 1) if row.get("queue_lag_s") is not None else 0.0, "scheduler_last_tick_at": tick, "fetch_per_min": round(int(row.get("obs_10m") or 0) / 10.0, 2), "success_rate_24h": round(int(row.get("ok_24h") or 0) / obs_24h, 4) if obs_24h else None} async def live_events(conn: AsyncConnection, limit: int, **filters: Any) -> list[dict[str, Any]]: where, params = q.event_filters(**filters) return [ser.event(r) for r in await q.fetch_events(conn, where, params, sort="recent", limit=limit)] def clear_aggregate_cache() -> None: cache.clear() __all__ = ["MAP_MAX_BUCKETS", "RANKING_KINDS", "activity_index", "archive_stats", "cached_country_rows", "cached_global_stats", "cached_industry_rows", "clear_aggregate_cache", "country_rows", "global_daily_rows", "global_stats", "industry_rows", "live_events", "map_buckets", "ranking", "ranking_cards", "resolve_country", "system_health", "trend_rows"]