SPB Git forge
28commits 1branches 0releases
7.7 MBsize
maindefault branch
10 days agolast push
Python 66.3% TypeScript 22.7% JavaScript 8.6% HTML 1.4% CSS 0.7%
25.9 KB · 418 lines python
Raw Blame History
1"""Aggregate producers shared by several routers (`/pulse` composes them): industry & country rows, rankings, trends, map buckets,2global stats and the activity index. Every producer degrades to empty lists / `null` values on an empty database — nothing is invented."""3from __future__ import annotations45import asyncio6from collections import defaultdict7from datetime import UTC, datetime, timedelta8from typing import Any910from sqlalchemy.ext.asyncio import AsyncConnection1112from companyatlas import archive13from companyatlas.api import queries as q14from companyatlas.api import serializers as ser15from companyatlas.api.common import cache, cached16from companyatlas.db import connection, fetch_all, fetch_one17from companyatlas.ids import slugify18from companyatlas.taxonomy import METRICS_FORMULA_VERSION, Metric1920# ------------------------------------------------------------------------------------------------ industries / countries2122_ROW_METRICS = (Metric.ACTIVITY_SCORE.value, Metric.HIRING_MOMENTUM_30D.value, Metric.AI_ADOPTION.value)232425def _avg_map(rows: list[dict[str, Any]], key: str) -> dict[str, dict[str, float]]:26    out: dict[str, dict[str, float]] = defaultdict(dict)27    for r in rows:28        if r[key] is not None and r["v"] is not None:29            out[r[key]][r["metric"]] = float(r["v"])30    return out313233async def industry_rows(conn: AsyncConnection, *, country: str | None = None) -> list[dict[str, Any]]:34    scope = " and c.country = cast(:country as char(2))" if country else ""35    params: dict[str, Any] = {"country": country} if country else {}36    d7, d30 = q.days_ago(7), q.days_ago(30)37    taxonomy = await fetch_all(conn, "select slug, name, parent_slug, description, sort_order from industries order by sort_order, name")38    companies = await fetch_all(conn, f"select ind, count(*) as n from companies c, unnest(c.industries) ind where c.status = 'ACTIVE'{scope} "39                                      "group by ind", **params)40    events = await fetch_all(conn, "select ind, e.event_type, count(*) filter (where e.detected_at >= :d7) as n7, count(*) as n30 "41                                   "from events e join companies c on c.id = e.company_id, unnest(c.industries) ind "42                                   f"where e.status = 'active' and e.detected_at >= :d30{scope} group by ind, e.event_type", d7=d7, d30=d30, **params)43    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, "44                                    f"unnest(c.industries) ind where m.metric = any(cast(:ms as text[])){scope} group by ind, m.metric",45                              ms=list(_ROW_METRICS), **params)46    n_companies = {r["ind"]: int(r["n"]) for r in companies}47    ev7: dict[str, int] = defaultdict(int)48    ev30: dict[str, int] = defaultdict(int)49    types: dict[str, dict[str, int]] = defaultdict(dict)50    for r in events:51        ev7[r["ind"]] += int(r["n7"])52        ev30[r["ind"]] += int(r["n30"])53        types[r["ind"]][r["event_type"]] = int(r["n30"])54    avg = _avg_map(metrics, "ind")55    names = {t["slug"]: t for t in taxonomy}56    slugs = list(names) + [s for s in n_companies if s not in names]57    rows = []58    for slug in slugs:59        t = names.get(slug, {})60        m = avg.get(slug, {})61        rows.append({"slug": slug, "name": t.get("name") or slug.replace("-", " ").title(), "parent_slug": t.get("parent_slug"),62                     "description": t.get("description"), "companies": n_companies.get(slug, 0), "events_7d": ev7.get(slug, 0),63                     "events_30d": ev30.get(slug, 0), "hiring_momentum_30d": ser.metric_value("hiring_momentum_30d", m.get("hiring_momentum_30d")),64                     "activity_score": ser.metric_value("activity_score", m.get("activity_score")),65                     "ai_adoption": ser.metric_value("ai_adoption", m.get("ai_adoption")),66                     "top_event_types": [k for k, _ in sorted(types.get(slug, {}).items(), key=lambda kv: (-kv[1], kv[0]))[:3]]})67    rows.sort(key=lambda r: (-r["companies"], -(r["activity_score"] or 0), r["name"]))68    return rows697071async def country_rows(conn: AsyncConnection, *, industry: str | None = None) -> list[dict[str, Any]]:72    scope = " and cast(:industry as text) = any(c.industries)" if industry else ""73    params: dict[str, Any] = {"industry": industry} if industry else {}74    d7, d30 = q.days_ago(7), q.days_ago(30)75    ref = await fetch_all(conn, "select code, name, region, subregion, lat, lon from countries order by name")76    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} "77                                      "group by c.country", **params)78    events = await fetch_all(conn, "select c.country as code, count(*) filter (where e.detected_at >= :d7) as n7, count(*) as n30 "79                                   "from events e join companies c on c.id = e.company_id "80                                   f"where e.status = 'active' and e.detected_at >= :d30 and c.country is not null{scope} group by c.country",81                             d7=d7, d30=d30, **params)82    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 "83                                    f"where m.metric = any(cast(:ms as text[])) and c.country is not null{scope} group by c.country, m.metric",84                              ms=list(_ROW_METRICS), **params)85    mix = await fetch_all(conn, "select c.country as code, ind, count(*) as n from companies c, unnest(c.industries) ind "86                                f"where c.status = 'ACTIVE' and c.country is not null{scope} group by c.country, ind", **params)87    n_companies = {r["code"]: int(r["n"]) for r in companies}88    ev = {r["code"]: (int(r["n7"]), int(r["n30"])) for r in events}89    avg = _avg_map(metrics, "code")90    mixes: dict[str, list[tuple[str, int]]] = defaultdict(list)91    for r in mix:92        mixes[r["code"]].append((r["ind"], int(r["n"])))93    names = {r["code"]: r for r in ref}94    codes = list(names) + [c for c in n_companies if c not in names]95    rows = []96    for code in codes:97        c = names.get(code, {})98        m = avg.get(code, {})99        name = c.get("name") or code100        rows.append({"code": code, "slug": slugify(name), "name": name, "region": c.get("region"), "subregion": c.get("subregion"),101                     "companies": n_companies.get(code, 0), "events_7d": ev.get(code, (0, 0))[0], "events_30d": ev.get(code, (0, 0))[1],102                     "hiring_momentum_30d": ser.metric_value("hiring_momentum_30d", m.get("hiring_momentum_30d")),103                     "activity_score": ser.metric_value("activity_score", m.get("activity_score")),104                     "ai_adoption": ser.metric_value("ai_adoption", m.get("ai_adoption")),105                     "industry_mix": [{"industry": i, "companies": n} for i, n in sorted(mixes.get(code, []), key=lambda x: -x[1])[:6]],106                     "lat": c.get("lat"), "lon": c.get("lon")})107    rows.sort(key=lambda r: (-r["companies"], -(r["activity_score"] or 0), r["name"]))108    return rows109110111async def cached_industry_rows(country: str | None = None) -> list[dict[str, Any]]:112    async def produce() -> list[dict[str, Any]]:113        async with connection() as conn:114            return await industry_rows(conn, country=country)115    return await cached(f"industries:{country or ''}", 300, produce)116117118async def cached_country_rows(industry: str | None = None) -> list[dict[str, Any]]:119    async def produce() -> list[dict[str, Any]]:120        async with connection() as conn:121            return await country_rows(conn, industry=industry)122    return await cached(f"countries:{industry or ''}", 300, produce)123124125async def resolve_country(conn: AsyncConnection, key: str) -> dict[str, Any] | None:126    """Accept an ISO-2 code (`CA`) or a name slug (`canada`)."""127    key = (key or "").strip()128    if not key:129        return None130    if len(key) == 2:131        row = await fetch_one(conn, "select code, name, region, subregion, lat, lon from countries where code = cast(:c as char(2))", c=key.upper())132        if row:133            return row134    rows = await fetch_all(conn, "select code, name, region, subregion, lat, lon from countries")135    k = slugify(key)136    for r in rows:137        if slugify(r["name"]) == k or r["code"].lower() == key.lower():138            return r139    return None140141142# ------------------------------------------------------------------------------------------------ rankings143144RANKING_KINDS: dict[str, dict[str, Any]] = {145    "most_active": {"metric": Metric.ACTIVITY_SCORE.value},146    "hiring_growth": {"metric": "hiring_momentum_{w}", "min": 0.0},147    "hiring_decline": {"metric": "hiring_momentum_{w}", "asc": True, "max": 0.0},148    "product_velocity": {"metric": Metric.PRODUCT_VELOCITY.value},149    "ai_active": {"metric": Metric.AI_ADOPTION.value},150    "geo_expansion": {"metric": Metric.GEO_EXPANSION.value},151    "developer_momentum": {"metric": Metric.DEVELOPER_MOMENTUM.value},152    "pricing_changes": {"events": "PRICING"},153    "unusual_activity": {"metric": Metric.ANOMALY_SCORE.value},154}155_HIRING_WINDOW = {"24h": "7d", "7d": "7d", "30d": "30d", "90d": "90d", "1y": "90d"}156157158async def ranking(conn: AsyncConnection, kind: str, window: str, *, country: str | None = None, industry: str | None = None,159                  limit: int = 50) -> list[dict[str, Any]]:160    """[{company_id, value, delta}] for a ranking kind. `delta` = value − series value at the start of the window (null if unknown)."""161    spec = RANKING_KINDS[kind]162    scope, params = [], {}163    if country:164        scope.append("c.country = cast(:country as char(2))")165        params["country"] = country.upper()[:2]166    if industry:167        scope.append("cast(:industry as text) = any(c.industries)")168        params["industry"] = industry[:80]169    scope_sql = (" and " + " and ".join(scope)) if scope else ""170    since = q.window_start(window)171    if "events" in spec:172        rows = await fetch_all(conn, "select e.company_id, count(*) as value from events e join companies c on c.id = e.company_id "173                                     f"where e.status = 'active' and e.event_type = :et and e.detected_at >= :since and c.status = 'ACTIVE'{scope_sql} "174                                     "group by e.company_id order by value desc, e.company_id limit :limit", et=spec["events"], since=since,175                               limit=limit, **params)176        return [{"company_id": r["company_id"], "value": float(r["value"]), "delta": None} for r in rows]177    metric = spec["metric"].format(w=_HIRING_WINDOW.get(window, "30d"))178    bounds = ""179    if "min" in spec:180        bounds += " and m.value > :vmin"181        params["vmin"] = spec["min"]182    if "max" in spec:183        bounds += " and m.value < :vmax"184        params["vmax"] = spec["max"]185    order = "m.value asc" if spec.get("asc") else "m.value desc"186    rows = await fetch_all(conn, "select m.company_id, m.value from metrics_current m join companies c on c.id = m.company_id "187                                 f"where m.metric = :metric and c.status = 'ACTIVE'{scope_sql}{bounds} order by {order}, m.company_id limit :limit",188                           metric=metric, limit=limit, **params)189    ids = [r["company_id"] for r in rows]190    past = await q.metric_values_at(conn, ids, metric, since.date())191    out = []192    for r in rows:193        v = float(r["value"])194        p = past.get(r["company_id"])195        out.append({"company_id": r["company_id"], "value": v, "delta": round(v - p, 2) if p is not None else None})196    return out197198199async def ranking_cards(conn: AsyncConnection, kind: str, window: str, *, country: str | None = None, industry: str | None = None,200                        limit: int = 50, sparkline: bool = False) -> list[dict[str, Any]]:201    items = await ranking(conn, kind, window, country=country, industry=industry, limit=limit)202    cards = await q.fetch_cards_by_ids(conn, [i["company_id"] for i in items], sparkline=sparkline)203    by_id = {c["id"]: c for c in cards}204    out = []205    for rank, it in enumerate(items, start=1):206        row = by_id.get(it["company_id"])207        if row is None:208            continue209        card = ser.company_card(row)210        metric = RANKING_KINDS[kind].get("metric", "").format(w=_HIRING_WINDOW.get(window, "30d"))211        card.update({"rank": rank, "value": ser.metric_value(metric, it["value"]) if metric else int(it["value"]), "delta": it["delta"]})212        out.append(card)213    return out214215216# ------------------------------------------------------------------------------------------------ trends / signals217218219async def trend_rows(conn: AsyncConnection, window_days: int, limit: int) -> list[dict[str, Any]]:220    d0 = q.days_ago(window_days).date()221    top = await fetch_all(conn, "select term, sum(mentions) as mentions, max(companies) as companies from trends where day >= :d group by term "222                                "order by mentions desc, term limit :limit", d=d0, limit=limit)223    if not top:224        return []225    terms = [t["term"] for t in top]226    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",227                             d=d0, terms=terms)228    by_term: dict[str, list[tuple[Any, int]]] = defaultdict(list)229    for r in series:230        by_term[r["term"]].append((r["day"], int(r["mentions"])))231    mid = q.days_ago(window_days // 2 or 1).date()232    out = []233    for t in top:234        pts = by_term.get(t["term"], [])235        first = sum(m for d, m in pts if d < mid)236        second = sum(m for d, m in pts if d >= mid)237        momentum = round((second - first) / first * 100, 1) if first > 0 else None238        out.append({"term": t["term"], "mentions": int(t["mentions"]), "companies": int(t["companies"] or 0), "momentum": momentum,239                    "series": [m for _, m in pts]})240    return out241242243# ------------------------------------------------------------------------------------------------ map244245MAP_MAX_BUCKETS = 600246247248async def map_buckets(conn: AsyncConnection, metric: str = "events_30d") -> list[dict[str, Any]]:249    d30 = q.days_ago(30)250    ev_by_company = {r["company_id"]: int(r["n"]) for r in await fetch_all(251        conn, "select company_id, count(*) as n from events where status = 'active' and detected_at >= :d group by company_id", d=d30)}252    jobs_by_company = {r["company_id"]: int(r["n"]) for r in await fetch_all(253        conn, "select company_id, count(*) as n from jobs where status = 'open' group by company_id")}254    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 "255                                      "join companies c on c.country = k.code and c.status = 'ACTIVE' where k.lat is not null order by c.importance desc")256    cities = await fetch_all(conn, "select l.country, l.city, avg(l.lat) as lat, avg(l.lon) as lon, "257                                   "array_agg(distinct l.company_id) as company_ids from locations l join companies c on c.id = l.company_id "258                                   "where l.status = 'listed' and l.lat is not null and l.lon is not null and l.city is not null "259                                   "group by l.country, l.city order by count(distinct l.company_id) desc limit :lim", lim=MAP_MAX_BUCKETS)260    by_country: dict[str, dict[str, Any]] = {}261    for r in countries:262        b = by_country.setdefault(r["code"], {"lat": r["lat"], "lon": r["lon"], "country": r["code"], "city": None, "companies": 0, "events_30d": 0,263                                              "jobs_open": 0, "top": [], "_ids": []})264        b["companies"] += 1265        b["events_30d"] += ev_by_company.get(r["id"], 0)266        b["jobs_open"] += jobs_by_company.get(r["id"], 0)267        if len(b["top"]) < 3:268            b["top"].append({"slug": r["slug"], "display_name": r["display_name"]})269    names: dict[str, tuple[str, str]] = {}270    if cities:271        ids = sorted({cid for r in cities for cid in (r["company_ids"] or [])})272        names = {r["id"]: (r["slug"], r["display_name"]) for r in await fetch_all(273            conn, "select id, slug, display_name from companies where id = any(cast(:ids as text[])) order by importance desc", ids=ids[:5000])}274    buckets = list(by_country.values())275    for r in cities:276        ids = [cid for cid in (r["company_ids"] or []) if cid in names]277        buckets.append({"lat": round(float(r["lat"]), 4), "lon": round(float(r["lon"]), 4), "country": r["country"], "city": r["city"], "companies": len(ids),278                        "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),279                        "top": [{"slug": names[i][0], "display_name": names[i][1]} for i in ids[:3]]})280    key = {"companies": "companies", "hiring": "jobs_open"}.get(metric, "events_30d")281    buckets.sort(key=lambda b: (-b[key], -b["companies"]))282    for b in buckets:283        b.pop("_ids", None)284    return buckets[:MAP_MAX_BUCKETS]285286287# ------------------------------------------------------------------------------------------------ global stats / index288289290async def archive_stats() -> dict[str, int]:291    async def produce() -> dict[str, int]:292        async with connection() as conn:293            kv = await q.settings_value(conn, "archive:stats")294        if isinstance(kv, dict) and "objects" in kv:295            return {"objects": int(kv.get("objects") or 0), "bytes": int(kv.get("bytes") or 0)}296        return await asyncio.to_thread(archive.store_stats)297    return await cached("archive:stats", 600, produce)298299300async def global_stats(conn: AsyncConnection) -> dict[str, Any]:301    today = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)302    row = await fetch_one(conn, """303        select (select count(*) from companies) as companies,304               (select count(*) from companies where status = 'ACTIVE' and onboarding_status = 'active') as companies_active,305               (select count(*) from sensors where status <> 'retired') as sensors,306               (select count(*) from sensors where status = 'active') as sensors_active,307               (select coalesce(sum(observation_count), 0) from sensors) as observations,308               (select coalesce(sum(snapshot_count), 0) from sensors) as snapshots,309               (select count(*) from changes) as changes,310               (select count(*) from changes where kind in ('meaningful', 'major', 'critical')) as meaningful_changes,311               (select count(*) from events where status = 'active') as events,312               (select count(*) from jobs where status = 'open') as jobs_open,313               (select count(distinct country) from companies where country is not null and status = 'ACTIVE') as countries,314               (select count(distinct ind) from companies c, unnest(c.industries) ind where c.status = 'ACTIVE') as industries,315               (select count(*) from observations where fetched_at >= :today) as observations_today,316               (select count(*) from changes where detected_at >= :today) as changes_today,317               (select count(*) from events where status = 'active' and detected_at >= :today) as events_today,318               (select min(first_observed_at) from companies) as oldest_observation_at,319               (select max(fetched_at) from observations) as last_observation_at320    """, today=today)321    row = row or {}322    started = await q.settings_value(conn, "dataset_started_at")323    started_dt = q.parse_iso(started) if isinstance(started, str) else None324    now = datetime.now(UTC)325    oldest = row.get("oldest_observation_at")326    return {"companies": int(row.get("companies") or 0), "companies_active": int(row.get("companies_active") or 0),327            "sensors": int(row.get("sensors") or 0), "sensors_active": int(row.get("sensors_active") or 0),328            "observations": int(row.get("observations") or 0), "snapshots": int(row.get("snapshots") or 0), "changes": int(row.get("changes") or 0),329            "meaningful_changes": int(row.get("meaningful_changes") or 0), "events": int(row.get("events") or 0),330            "jobs_open": int(row.get("jobs_open") or 0), "countries": int(row.get("countries") or 0), "industries": int(row.get("industries") or 0),331            "observations_today": int(row.get("observations_today") or 0), "changes_today": int(row.get("changes_today") or 0),332            "events_today": int(row.get("events_today") or 0), "dataset_started_at": started_dt,333            "dataset_age_days": (now - started_dt).days if started_dt else None,334            "oldest_history_days": (now - oldest).days if oldest else None, "last_observation_at": row.get("last_observation_at"),335            "archive": await archive_stats()}336337338async def cached_global_stats() -> dict[str, Any]:339    async def produce() -> dict[str, Any]:340        async with connection() as conn:341            return await global_stats(conn)342    return await cached("stats", 60, produce)343344345async def global_daily_rows(conn: AsyncConnection, days: int) -> list[dict[str, Any]]:346    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)347    return [{"day": r["day"], "companies_active": r["companies_active"], "sensors_active": r["sensors_active"], "observations": r["observations"],348             "changes": r["changes"], "meaningful_changes": r["meaningful_changes"], "events": r["events"], "events_by_type": ser._dict(r["events_by_type"]),349             "jobs_open": r["jobs_open"], "jobs_new": r["jobs_new"], "jobs_removed": r["jobs_removed"],350             "activity_index": ser._float(r["activity_index"], 2), "by_country": ser._dict(r["by_country"]), "by_industry": ser._dict(r["by_industry"])}351            for r in rows]352353354def _index_at(rows: list[dict[str, Any]], days_back: int) -> float | None:355    if not rows:356        return None357    target = rows[-1]["day"] - timedelta(days=days_back)358    candidates = [r for r in rows if r["day"] <= target and r["activity_index"] is not None]359    return candidates[-1]["activity_index"] if candidates else None360361362async def activity_index(conn: AsyncConnection, days: int = 365) -> dict[str, Any]:363    rows = await global_daily_rows(conn, days)364    with_value = [r for r in rows if r["activity_index"] is not None]365    latest = with_value[-1] if with_value else None366    value = latest["activity_index"] if latest else None367    v7, v30 = _index_at(with_value, 7), _index_at(with_value, 30)368    formula = await q.settings_value(conn, "index:formula_version")369    return {"value": value, "baseline": 100, "delta_7d": round(value - v7, 2) if value is not None and v7 is not None else None,370            "delta_30d": round(value - v30, 2) if value is not None and v30 is not None else None,371            "series": [{"day": r["day"], "value": r["activity_index"], "confidence": None} for r in with_value],372            "by_type": (latest or {}).get("events_by_type", {}),373            "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]],374            "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]],375            "formula_version": formula if isinstance(formula, str) else METRICS_FORMULA_VERSION, "computed_at": (latest or {}).get("day")}376377378async def system_health(conn: AsyncConnection) -> dict[str, Any]:379    today = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)380    row = await fetch_one(conn, """381        select (select count(*) from sensors where status = 'active') as sensors_online,382               (select count(*) from sensors where status in ('failing', 'stale', 'blocked')) as sensors_failing,383               (select count(*) from observations where fetched_at >= :today) as observations_today,384               (select count(*) from events where status = 'active' and detected_at >= :today) as events_today,385               (select count(distinct country) from companies where country is not null and status = 'ACTIVE') as countries_covered,386               (select extract(epoch from (now() - min(run_at))) from queue_jobs where status = 'pending' and run_at <= now()) as queue_lag_s,387               (select count(*) from observations where fetched_at >= now() - interval '10 minutes') as obs_10m,388               (select count(*) from observations where fetched_at >= now() - interval '24 hours') as obs_24h,389               (select count(*) from observations where fetched_at >= now() - interval '24 hours' and failure_class is null) as ok_24h390    """, today=today) or {}391    hb = await q.settings_value(conn, "scheduler:heartbeat")392    tick = None393    if isinstance(hb, dict):394        tick = hb.get("at") or hb.get("ts") or hb.get("time") or hb.get("last_tick_at")395    elif isinstance(hb, str):396        tick = hb397    obs_24h = int(row.get("obs_24h") or 0)398    return {"sensors_online": int(row.get("sensors_online") or 0), "sensors_failing": int(row.get("sensors_failing") or 0),399            "observations_today": int(row.get("observations_today") or 0), "events_today": int(row.get("events_today") or 0),400            "countries_covered": int(row.get("countries_covered") or 0),401            "queue_lag_s": round(float(row["queue_lag_s"]), 1) if row.get("queue_lag_s") is not None else 0.0,402            "scheduler_last_tick_at": tick, "fetch_per_min": round(int(row.get("obs_10m") or 0) / 10.0, 2),403            "success_rate_24h": round(int(row.get("ok_24h") or 0) / obs_24h, 4) if obs_24h else None}404405406async def live_events(conn: AsyncConnection, limit: int, **filters: Any) -> list[dict[str, Any]]:407    where, params = q.event_filters(**filters)408    return [ser.event(r) for r in await q.fetch_events(conn, where, params, sort="recent", limit=limit)]409410411def clear_aggregate_cache() -> None:412    cache.clear()413414415__all__ = ["MAP_MAX_BUCKETS", "RANKING_KINDS", "activity_index", "archive_stats", "cached_country_rows", "cached_global_stats", "cached_industry_rows",416           "clear_aggregate_cache", "country_rows", "global_daily_rows", "global_stats", "industry_rows", "live_events", "map_buckets", "ranking",417           "ranking_cards", "resolve_country", "system_health", "trend_rows"]418