"""Proprietary metrics (spec §29–37, §91, §146–147, §177): reproducible, coverage-normalised, never fabricated. Every metric row carries `formula_version`, `inputs` (the numbers the formula saw) and `computed_at`. A metric without inputs is not written (no row ≠ 0). Formulas and parameters: `taxonomy.METRIC_PARAMS`, `taxonomy.CCI_WEIGHTS`, documented in docs/SCORING.md. compute_company_metrics(ids | None) hourly (@periodic metrics-hourly): companies active in the last 90 d; nightly: all compute_daily(day) daily aggregates: company_daily, global_daily (activity_index, baseline 100), baselines compute_daily_catch_up() fills missed days since the last computed day """ from __future__ import annotations import logging import math import statistics from dataclasses import dataclass, field from datetime import UTC, date, datetime, timedelta from typing import Any from companyatlas.config import settings from companyatlas.db import execute, fetch_all, fetch_one, fetch_val, jsonb, transaction from companyatlas.services.periodic import periodic from companyatlas.taxonomy import ( AI_KEYWORDS, CCI_FORMULA_VERSION, CCI_WEIGHTS, METRIC_PARAMS, METRICS_FORMULA_VERSION, ChangeKind, EventType, Metric, Surface, ) log = logging.getLogger(__name__) P = METRIC_PARAMS MEANINGFUL_KINDS = (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL) CHANGE_WEIGHT = {ChangeKind.MEANINGFUL: P["change_weight_meaningful"], ChangeKind.MAJOR: P["change_weight_major"], ChangeKind.CRITICAL: P["change_weight_critical"]} PRODUCT_SUBTYPES = {"PRODUCT_LAUNCH", "NEW_PRODUCT", "PRODUCT_REMOVED", "PRODUCT_RENAME", "PRODUCT_UPDATE", "FEATURE_LAUNCH", "CHANGELOG_ENTRY", "DOC_CHANGE", "DOCUMENTATION_CHANGE", "API_CHANGE", "API_LAUNCH", "SDK_RELEASE", "AI_LAUNCH"} AI_SUBTYPES = {"AI_HIRING", "AI_LAUNCH"} PRODUCT_SURFACES = {Surface.PRODUCTS, Surface.CHANGELOG, Surface.DOCS, Surface.API, Surface.DEVELOPER, Surface.SERVICES, Surface.SOLUTIONS} DEVELOPER_SURFACES = {Surface.DOCS, Surface.API, Surface.DEVELOPER, Surface.CHANGELOG} COMM_SURFACES = {Surface.NEWSROOM, Surface.BLOG, Surface.FEED, Surface.INVESTOR_RELATIONS, Surface.RESEARCH} GEO_SURFACES = {Surface.LOCATIONS, Surface.CONTACT, Surface.ABOUT, Surface.CAREERS, Surface.JOBS_BOARD} HIRING_SURFACES = {Surface.CAREERS, Surface.JOBS_BOARD} BASELINE_METRICS = ("meaningful_changes_weekly", "jobs_new_weekly", "news_weekly") # ================================================================================================================ pure formulas def saturate(x: float, k: float) -> float: """0–1 saturation: 1 − exp(−x/k). x=k → 0.63, x=3k → 0.95.""" return 0.0 if x <= 0 else 1.0 - math.exp(-x / k) def decay(age_days: float, tau: float) -> float: return math.exp(-max(0.0, age_days) / tau) @dataclass(slots=True) class MetricValue: metric: str value: float confidence: float inputs: dict[str, Any] = field(default_factory=dict) formula_version: str = METRICS_FORMULA_VERSION def activity_score(changes: list[tuple[float, str]], events: list[tuple[float, float]], active_sensors: int) -> MetricValue | None: """changes = [(age_days, kind)], events = [(age_days, importance)] within the 30-day window. raw = Σ w(kind)·e^(−age/τ) + Σ importance·e^(−age/τ); density = raw / sensors^0.5; score = 100·log1p(density)/log1p(D_MAX).""" if active_sensors <= 0 and not changes and not events: return None tau = P["activity_decay_tau_days"] raw_changes = sum(CHANGE_WEIGHT.get(k, 1.0) * decay(a, tau) for a, k in changes) raw_events = sum(float(i) * decay(a, tau) for a, i in events) raw = raw_changes + raw_events denom = max(1.0, float(active_sensors)) ** P["activity_coverage_exp"] density = raw / denom score = min(100.0, 100.0 * math.log1p(density) / math.log1p(P["activity_density_max"])) conf = min(0.95, 0.5 + 0.05 * active_sensors) return MetricValue(Metric.ACTIVITY_SCORE, round(score, 2), round(conf, 2), {"changes_30d": len(changes), "events_30d": len(events), "raw_changes": round(raw_changes, 4), "raw_events": round(raw_events, 4), "active_sensors": active_sensors, "density": round(density, 4), "tau_days": tau, "coverage_exp": P["activity_coverage_exp"], "density_max": P["activity_density_max"]}) def hiring_momentum(open_now: int, open_then: int, *, window: int, extra: dict[str, Any] | None = None, confidence: float = 0.8) -> MetricValue | None: """% change of open listings vs the reconstructed count `window` days ago. Requires ≥ `hiring_min_listings` at the reference point.""" if open_then < P["hiring_min_listings"]: return None pct = (open_now - open_then) / open_then * 100.0 metric = {7: Metric.HIRING_MOMENTUM_7D, 30: Metric.HIRING_MOMENTUM_30D, 90: Metric.HIRING_MOMENTUM_90D}[window] return MetricValue(metric, round(pct, 2), confidence, {"open_now": open_now, "open_then": open_then, "window_days": window, **(extra or {})}) def ai_adoption(*, ai_open: int | None, open_jobs: int | None, ai_events_90d: int, keyword_hits: int, has_text_inputs: bool) -> MetricValue | None: """Observable signals only: share of AI jobs (w 0.5), AI events (w 0.25), AI keyword hits in product names / news / page titles (w 0.25). Weights are renormalised over the inputs that exist.""" comps: dict[str, tuple[float, float]] = {} if open_jobs is not None and open_jobs > 0: comps["jobs"] = (P["ai_jobs_weight"], min(1.0, (ai_open or 0) / open_jobs * 2)) # 50 % AI roles → saturates if ai_events_90d or has_text_inputs or open_jobs: comps["events"] = (P["ai_events_weight"], saturate(ai_events_90d, P["ai_events_saturation"])) if has_text_inputs: comps["keywords"] = (P["ai_keywords_weight"], saturate(keyword_hits, P["ai_keywords_saturation"])) if not comps: return None total_w = sum(w for w, _ in comps.values()) score = 100.0 * sum(w * v for w, v in comps.values()) / total_w return MetricValue(Metric.AI_ADOPTION, round(score, 2), 0.6, {"ai_open": ai_open, "open_jobs": open_jobs, "ai_events_90d": ai_events_90d, "keyword_hits": keyword_hits, "components": {k: round(v, 4) for k, (_, v) in comps.items()}, "weights_used": {k: w for k, (w, _) in comps.items()}}) def saturating_score(metric: str, weighted_sum: float, k: float, inputs: dict[str, Any], confidence: float = 0.7) -> MetricValue: return MetricValue(metric, round(100.0 * saturate(weighted_sum, k), 2), confidence, {**inputs, "weighted_sum": round(weighted_sum, 4), "saturation_k": k}) def corporate_change_index(values: dict[str, float]) -> MetricValue | None: """Σ w·component over available components, renormalised. Hiring momentum (%) is mapped to 0–100 as 50 + clamp(m, −100, 100)/2.""" comps: dict[str, float] = {} for metric in CCI_WEIGHTS: if metric not in values: continue v = values[metric] if metric == Metric.HIRING_MOMENTUM_30D: v = 50.0 + max(-100.0, min(100.0, v)) / 2.0 comps[str(metric)] = max(0.0, min(100.0, v)) if not comps: return None total_w = sum(CCI_WEIGHTS[m] for m in CCI_WEIGHTS if str(m) in comps) value = sum(CCI_WEIGHTS[m] * comps[str(m)] for m in CCI_WEIGHTS if str(m) in comps) / total_w return MetricValue(Metric.CORPORATE_CHANGE_INDEX, round(value, 2), round(0.4 + 0.6 * total_w, 2), {"components": comps, "weights": {str(m): CCI_WEIGHTS[m] for m in CCI_WEIGHTS if str(m) in comps}, "weight_coverage": round(total_w, 3)}, formula_version=CCI_FORMULA_VERSION) def anomaly_score(this_week: int, mean: float, stddev: float, samples: int) -> MetricValue | None: if samples < P["anomaly_min_samples"]: return None sd = max(float(stddev), 0.5) # floor avoids infinite z on flat baselines z = (this_week - mean) / sd return MetricValue(Metric.ANOMALY_SCORE, round(z, 3), min(0.9, 0.4 + 0.05 * samples), {"this_week": this_week, "mean": mean, "stddev": stddev, "samples": samples, "stddev_floor": 0.5}) def historical_coverage(*, observed: int, expected: float, days_with_obs: int, days_since_first: int, surfaces: int) -> MetricValue | None: if expected <= 0 or days_since_first <= 0: return None obs_cov = min(1.0, observed / expected) continuity = min(1.0, days_with_obs / max(1, days_since_first)) source_cov = min(1.0, surfaces / P["coverage_expected_surfaces"]) score = 100.0 * (P["coverage_obs_weight"] * obs_cov + P["coverage_continuity_weight"] * continuity + P["coverage_sources_weight"] * source_cov) return MetricValue(Metric.HISTORICAL_COVERAGE, round(score, 2), 0.8, {"observed": observed, "expected": round(expected, 1), "days_with_obs": days_with_obs, "days_since_first": days_since_first, "surfaces": surfaces}) # ================================================================================================================ per-company loader @dataclass(slots=True) class CompanyContext: company: dict[str, Any] sensors: list[dict[str, Any]] changes_30d: list[dict[str, Any]] events_90d: list[dict[str, Any]] jobs: dict[str, Any] products: dict[str, Any] news: dict[str, Any] locations: dict[str, Any] titles_ai_hits: int baselines: dict[str, dict[str, Any]] observations: dict[str, Any] now: datetime async def load_context(conn, company_id: str, now: datetime) -> CompanyContext | None: # type: ignore[no-untyped-def] company = await fetch_one(conn, "select id, slug, first_observed_at, industries, country from companies where id = :id", id=company_id) if company is None: return None d30, d90 = now - timedelta(days=30), now - timedelta(days=90) sensors = await fetch_all(conn, """select id, surface, status, base_interval_s, current_interval_s, created_at, observation_count, last_success_at, meaningful_change_count from sensors where company_id = :c and retired_at is null""", c=company_id) changes = await fetch_all(conn, """select detected_at, kind, surface from changes where company_id = :c and detected_at >= :since and kind in ('meaningful', 'major', 'critical')""", c=company_id, since=d30) events = await fetch_all(conn, """select detected_at, importance, event_type, event_subtype, tags, surface from events where company_id = :c and detected_at >= :since and status in ('active', 'review') order by detected_at desc limit 3000""", c=company_id, since=d90) jobs_row = await fetch_one(conn, """ select count(*) filter (where status = 'open') as open_now, count(*) filter (where status = 'open' and is_ai) as ai_open, count(*) filter (where status = 'open' and remote) as remote_open, count(*) filter (where first_seen_at <= :t7 and (removed_at is null or removed_at > :t7)) as open_7, count(*) filter (where first_seen_at <= :t30 and (removed_at is null or removed_at > :t30)) as open_30, count(*) filter (where first_seen_at <= :t90 and (removed_at is null or removed_at > :t90)) as open_90, count(*) filter (where first_seen_at > :t30 and not baseline) as new_30d, count(*) filter (where removed_at > :t30) as removed_30d, count(*) filter (where first_seen_at > :t30 and is_ai and not baseline) as ai_new_30d, count(*) as total from jobs where company_id = :c""", c=company_id, t7=now - timedelta(days=7), t30=d30, t90=d90) by_country = await fetch_all(conn, "select country, count(*) as n from jobs where company_id = :c and status = 'open' and country is not null group by country order by n desc limit 20", c=company_id) by_department = await fetch_all(conn, "select department, count(*) as n from jobs where company_id = :c and status = 'open' and department is not null group by department order by n desc limit 20", c=company_id) new_job_countries = await fetch_all(conn, """ select country from jobs where company_id = :c and country is not null group by country having min(first_seen_at) > :since and bool_and(not baseline)""", c=company_id, since=d90) products = await fetch_one(conn, """select count(*) filter (where status = 'listed') as listed, count(*) as total, array_agg(name) filter (where status = 'listed') as names from products where company_id = :c""", c=company_id) news = await fetch_one(conn, """select count(*) filter (where coalesce(published_at, first_seen_at) >= :d30 and not (baseline and published_at is null)) as n_30d, count(*) as total, array_agg(title) filter (where coalesce(published_at, first_seen_at) >= :d90 and not (baseline and published_at is null)) as titles_90d from news_items where company_id = :c""", c=company_id, d30=d30, d90=d90) locations = await fetch_one(conn, """ select count(*) filter (where status = 'listed') as listed, count(*) filter (where first_seen_at > :since and not baseline) as new_90d, count(*) as total, count(distinct country) filter (where status = 'listed' and country is not null) as countries from locations where company_id = :c""", c=company_id, since=d90) new_loc_countries = await fetch_all(conn, "select country from locations where company_id = :c and country is not null group by country having min(first_seen_at) > :since and bool_and(not baseline)", c=company_id, since=d90) titles = await fetch_all(conn, """select distinct on (s.sensor_id) s.title from snapshots s join sensors se on se.id = s.sensor_id where se.company_id = :c order by s.sensor_id, s.fetched_at desc""", c=company_id) baselines = await fetch_all(conn, "select metric, mean, stddev, samples from baselines where company_id = :c", c=company_id) obs = await fetch_one(conn, """select count(*) filter (where failure_class is null) as ok, count(*) as total, count(distinct date(fetched_at)) as days_with_obs, count(distinct sensor_id) as sensors_observed from observations where company_id = :c""", c=company_id) this_week = await fetch_val(conn, "select count(*) from changes where company_id = :c and detected_at >= :since and kind in ('meaningful', 'major', 'critical')", c=company_id, since=now - timedelta(days=7)) jobs = {**(jobs_row or {}), "by_country": [{"country": r["country"], "n": r["n"]} for r in by_country], "by_department": [{"department": r["department"], "n": r["n"]} for r in by_department], "new_countries_90d": [r["country"] for r in new_job_countries]} return CompanyContext( company=company, sensors=sensors, changes_30d=changes, events_90d=events, jobs=jobs, products={**(products or {}), "names": list((products or {}).get("names") or [])}, news={**(news or {}), "titles_90d": list((news or {}).get("titles_90d") or [])}, locations={**(locations or {}), "new_countries_90d": [r["country"] for r in new_loc_countries]}, titles_ai_hits=sum(1 for t in titles if _ai_hit(t.get("title"))), baselines={r["metric"]: r for r in baselines}, observations={**(obs or {}), "this_week_meaningful": int(this_week or 0)}, now=now) def _ai_hit(text: str | None) -> bool: if not text: return False hay = f" {text.lower()} " return any(k in hay for k in AI_KEYWORDS) def _age(now: datetime, at: datetime) -> float: if at.tzinfo is None: at = at.replace(tzinfo=UTC) return max(0.0, (now - at).total_seconds() / 86400.0) # ================================================================================================================ compute def compute_from_context(ctx: CompanyContext) -> list[MetricValue]: now = ctx.now out: list[MetricValue] = [] active_sensors = [s for s in ctx.sensors if s["status"] in ("active", "failing", "stale")] surfaces = {s["surface"] for s in ctx.sensors} n_active = len(active_sensors) events_30 = [e for e in ctx.events_90d if _age(now, e["detected_at"]) <= P["activity_window_days"]] # activity ---------------------------------------------------------------------------------------------------- act = activity_score([(_age(now, c["detected_at"]), str(c["kind"])) for c in ctx.changes_30d], [(_age(now, e["detected_at"]), float(e["importance"])) for e in events_30], n_active) if act and (ctx.changes_30d or events_30 or n_active): out.append(act) # hiring ------------------------------------------------------------------------------------------------------ j = ctx.jobs has_jobs_source = bool(surfaces & HIRING_SURFACES) or int(j.get("total") or 0) > 0 jobs_conf = 0.9 if Surface.JOBS_BOARD in surfaces else 0.75 if has_jobs_source: open_now = int(j.get("open_now") or 0) remote_ratio = round(int(j.get("remote_open") or 0) / open_now, 4) if open_now else None extra = {"jobs_new_30d": int(j.get("new_30d") or 0), "jobs_removed_30d": int(j.get("removed_30d") or 0), "remote_ratio": remote_ratio, "by_country": j.get("by_country"), "by_department": j.get("by_department"), "ai_open": int(j.get("ai_open") or 0)} out.append(MetricValue(Metric.OPEN_JOBS, float(open_now), jobs_conf, extra)) for window, key in ((7, "open_7"), (30, "open_30"), (90, "open_90")): m = hiring_momentum(open_now, int(j.get(key) or 0), window=window, extra=extra if window == 30 else None, confidence=jobs_conf) if m: out.append(m) # AI adoption ------------------------------------------------------------------------------------------------- ai_events = sum(1 for e in ctx.events_90d if e["event_subtype"] in AI_SUBTYPES or "ai" in (e.get("tags") or [])) kw_hits = sum(1 for n in ctx.products["names"] if _ai_hit(n)) + sum(1 for t in ctx.news["titles_90d"] if _ai_hit(t)) + ctx.titles_ai_hits has_text = bool(ctx.products["names"] or ctx.news["titles_90d"] or ctx.sensors) ai = ai_adoption(ai_open=int(j.get("ai_open") or 0) if has_jobs_source else None, open_jobs=int(j.get("open_now") or 0) if has_jobs_source else None, ai_events_90d=ai_events, keyword_hits=kw_hits, has_text_inputs=has_text) if ai: out.append(ai) # product velocity / developer / communication / pricing / leadership ------------------------------------------ tau = P["activity_decay_tau_days"] * 3 # 90-day windows decay slower if surfaces & PRODUCT_SURFACES or any(e["event_type"] == EventType.PRODUCT for e in ctx.events_90d): prod_events = [e for e in ctx.events_90d if e["event_subtype"] in PRODUCT_SUBTYPES] s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in prod_events) out.append(saturating_score(Metric.PRODUCT_VELOCITY, s, P["velocity_saturation"], {"events_90d": len(prod_events), "surfaces": sorted(surfaces & PRODUCT_SURFACES)})) if surfaces & DEVELOPER_SURFACES or any(e["event_type"] == EventType.DEVELOPER for e in ctx.events_90d): dev_events = [e for e in ctx.events_90d if e["event_type"] == EventType.DEVELOPER] dev_changes = [c for c in ctx.changes_30d if c["surface"] in DEVELOPER_SURFACES] s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in dev_events) + 0.5 * len(dev_changes) out.append(saturating_score(Metric.DEVELOPER_MOMENTUM, s, P["developer_saturation"], {"events_90d": len(dev_events), "meaningful_changes_30d": len(dev_changes), "surfaces": sorted(surfaces & DEVELOPER_SURFACES)})) if surfaces & COMM_SURFACES or int(ctx.news.get("total") or 0) > 0: comm_events = [e for e in events_30 if e["event_type"] in (EventType.COMMUNICATION, EventType.INVESTOR_RELATIONS)] n_news = int(ctx.news.get("n_30d") or 0) s = float(max(n_news, len(comm_events))) + 0.5 * min(n_news, len(comm_events)) out.append(saturating_score(Metric.COMMUNICATION_ACTIVITY, s, P["communication_saturation"], {"news_items_30d": n_news, "events_30d": len(comm_events)})) if Surface.PRICING in surfaces or any(e["event_type"] == EventType.PRICING for e in ctx.events_90d): pr_events = [e for e in ctx.events_90d if e["event_type"] == EventType.PRICING] pr_changes = [c for c in ctx.changes_30d if c["surface"] == Surface.PRICING] s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in pr_events) + 0.25 * len(pr_changes) out.append(saturating_score(Metric.PRICING_ACTIVITY, s, P["pricing_saturation"], {"events_90d": len(pr_events), "meaningful_changes_30d": len(pr_changes)})) if Surface.LEADERSHIP in surfaces or any(e["event_type"] == EventType.LEADERSHIP for e in ctx.events_90d): ld_events = [e for e in ctx.events_90d if e["event_type"] == EventType.LEADERSHIP] s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in ld_events) out.append(saturating_score(Metric.LEADERSHIP_ACTIVITY, s, P["leadership_saturation"], {"events_90d": len(ld_events)})) # geographic expansion ---------------------------------------------------------------------------------------- loc = ctx.locations if surfaces & GEO_SURFACES or int(loc.get("total") or 0) > 0 or has_jobs_source: new_countries = set(loc.get("new_countries_90d") or []) | set(j.get("new_countries_90d") or []) geo_events = [e for e in ctx.events_90d if e["event_type"] == EventType.LOCATION] s = float(int(loc.get("new_90d") or 0)) + P["geo_country_weight"] * len(new_countries) + sum(float(e["importance"]) for e in geo_events if e["event_subtype"] == "COUNTRY_EXPANSION") out.append(saturating_score(Metric.GEO_EXPANSION, s, P["geo_saturation"], {"new_locations_90d": int(loc.get("new_90d") or 0), "new_countries_90d": sorted(new_countries), "countries_listed": int(loc.get("countries") or 0), "location_events_90d": len(geo_events)})) # CCI --------------------------------------------------------------------------------------------------------- cci = corporate_change_index({m.metric: m.value for m in out}) if cci: out.append(cci) # anomaly ----------------------------------------------------------------------------------------------------- b = ctx.baselines.get("meaningful_changes_weekly") if b: an = anomaly_score(int(ctx.observations.get("this_week_meaningful") or 0), float(b["mean"]), float(b["stddev"]), int(b["samples"])) if an: out.append(an) # historical coverage ----------------------------------------------------------------------------------------- first = ctx.company.get("first_observed_at") if first and ctx.sensors: days_since_first = max(1, int(_age(now, first))) expected = 0.0 for s in ctx.sensors: created = s.get("created_at") or first span_s = max(0.0, (now - (created if created.tzinfo else created.replace(tzinfo=UTC))).total_seconds()) expected += span_s / max(900, int(s.get("current_interval_s") or s.get("base_interval_s") or 86400)) hc = historical_coverage(observed=int(ctx.observations.get("ok") or 0), expected=expected, days_with_obs=int(ctx.observations.get("days_with_obs") or 0), days_since_first=days_since_first, surfaces=len(surfaces)) if hc: out.append(hc) return out async def write_metrics(conn, company_id: str, values: list[MetricValue], now: datetime) -> None: # type: ignore[no-untyped-def] day = now.astimezone(UTC).date() for m in values: await execute(conn, """ insert into metrics_current (company_id, metric, value, confidence, inputs, formula_version, computed_at) values (:c, :m, :v, :conf, cast(:inputs as jsonb), :fv, :at) on conflict (company_id, metric) do update set value = excluded.value, confidence = excluded.confidence, inputs = excluded.inputs, formula_version = excluded.formula_version, computed_at = excluded.computed_at""", c=company_id, m=str(m.metric), v=float(m.value), conf=float(m.confidence), inputs=jsonb(m.inputs), fv=m.formula_version, at=now) await execute(conn, """ insert into metric_series (company_id, metric, day, value, confidence, formula_version) values (:c, :m, :d, :v, :conf, :fv) on conflict (company_id, metric, day) do update set value = excluded.value, confidence = excluded.confidence, formula_version = excluded.formula_version""", c=company_id, m=str(m.metric), d=day, v=float(m.value), conf=float(m.confidence), fv=m.formula_version) async def compute_company_metrics(company_ids: list[str] | None = None, *, all_companies: bool = False, now: datetime | None = None) -> dict[str, int]: """Compute and store metrics. `company_ids=None` → companies with activity in the last `metrics_active_window_days` (or all).""" now = now or datetime.now(UTC) stats = {"companies": 0, "metrics": 0, "skipped": 0} async with transaction() as conn: if company_ids is None: if all_companies: rows = await fetch_all(conn, "select id from companies where onboarding_status <> 'no_website' order by importance desc") else: rows = await fetch_all(conn, """ select id from companies where greatest(coalesce(last_observed_at, 'epoch'), coalesce(last_change_at, 'epoch'), coalesce(last_event_at, 'epoch')) >= :since or id in (select distinct company_id from changes where detected_at >= :since) or id in (select distinct company_id from events where detected_at >= :since) order by importance desc""", since=now - timedelta(days=settings.metrics_active_window_days)) company_ids = [r["id"] for r in rows] for cid in company_ids: try: async with transaction() as conn: ctx = await load_context(conn, cid, now) if ctx is None: stats["skipped"] += 1 continue values = compute_from_context(ctx) if not values: stats["skipped"] += 1 continue await write_metrics(conn, cid, values, now) stats["companies"] += 1 stats["metrics"] += len(values) except Exception: log.exception("metrics failed", extra={"company_id": cid}) return stats # ================================================================================================================ daily aggregates def _day_bounds(day: date) -> tuple[datetime, datetime]: d0 = datetime(day.year, day.month, day.day, tzinfo=UTC) return d0, d0 + timedelta(days=1) async def compute_daily(day: date) -> dict[str, Any]: """company_daily + global_daily (+ baselines) for one UTC day. Idempotent (upserts).""" d0, d1 = _day_bounds(day) async with transaction() as conn: per: dict[str, dict[str, Any]] = {} def bucket(cid: str) -> dict[str, Any]: return per.setdefault(cid, {"observations": 0, "changes": 0, "meaningful_changes": 0, "events": 0, "events_by_type": {}, "jobs_open": None, "jobs_new": 0, "jobs_removed": 0, "jobs_ai_open": None, "news_items": 0, "sensors_active": None}) for r in await fetch_all(conn, "select company_id, count(*) as n from observations where fetched_at >= :d0 and fetched_at < :d1 group by 1", d0=d0, d1=d1): bucket(r["company_id"])["observations"] = int(r["n"]) for r in await fetch_all(conn, """select company_id, count(*) as n, count(*) filter (where kind in ('meaningful','major','critical')) as m from changes where detected_at >= :d0 and detected_at < :d1 group by 1""", d0=d0, d1=d1): b = bucket(r["company_id"]) b["changes"], b["meaningful_changes"] = int(r["n"]), int(r["m"]) for r in await fetch_all(conn, """select company_id, event_type, count(*) as n from events where detected_at >= :d0 and detected_at < :d1 and status in ('active', 'review') group by 1, 2""", d0=d0, d1=d1): b = bucket(r["company_id"]) b["events"] += int(r["n"]) b["events_by_type"][r["event_type"]] = int(r["n"]) for r in await fetch_all(conn, """select company_id, count(*) filter (where first_seen_at >= :d0 and first_seen_at < :d1 and not baseline) as new_n, count(*) filter (where removed_at >= :d0 and removed_at < :d1) as rem_n, count(*) filter (where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1)) as open_n, count(*) filter (where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1) and is_ai) as ai_n from jobs group by 1""", d0=d0, d1=d1): if not (r["new_n"] or r["rem_n"] or r["open_n"]): continue b = bucket(r["company_id"]) b["jobs_new"], b["jobs_removed"], b["jobs_open"], b["jobs_ai_open"] = int(r["new_n"]), int(r["rem_n"]), int(r["open_n"]), int(r["ai_n"]) for r in await fetch_all(conn, "select company_id, count(*) as n from news_items where first_seen_at >= :d0 and first_seen_at < :d1 and not baseline group by 1", d0=d0, d1=d1): bucket(r["company_id"])["news_items"] = int(r["n"]) if per: ids = list(per) for r in await fetch_all(conn, """select company_id, count(*) as n from sensors where company_id = any(cast(:ids as text[])) and status in ('active','failing','stale') and created_at < :d1 group by 1""", ids=ids, d1=d1): per[r["company_id"]]["sensors_active"] = int(r["n"]) for cid, b in per.items(): await execute(conn, """ insert into company_daily (company_id, day, observations, changes, meaningful_changes, events, events_by_type, jobs_open, jobs_new, jobs_removed, jobs_ai_open, news_items, sensors_active) values (:c, :d, :o, :ch, :m, :e, cast(:ebt as jsonb), :jo, :jn, :jr, :jai, :news, :sa) on conflict (company_id, day) do update set observations = excluded.observations, changes = excluded.changes, meaningful_changes = excluded.meaningful_changes, events = excluded.events, events_by_type = excluded.events_by_type, jobs_open = excluded.jobs_open, jobs_new = excluded.jobs_new, jobs_removed = excluded.jobs_removed, jobs_ai_open = excluded.jobs_ai_open, news_items = excluded.news_items, sensors_active = excluded.sensors_active""", c=cid, d=day, o=b["observations"], ch=b["changes"], m=b["meaningful_changes"], e=b["events"], ebt=jsonb(b["events_by_type"]), jo=b["jobs_open"], jn=b["jobs_new"], jr=b["jobs_removed"], jai=b["jobs_ai_open"], news=b["news_items"], sa=b["sensors_active"]) # global ----------------------------------------------------------------------------------------------------- g = await fetch_one(conn, """select count(distinct company_id) as companies_active, count(distinct sensor_id) as sensors_active, count(*) as observations from observations where fetched_at >= :d0 and fetched_at < :d1""", d0=d0, d1=d1) or {} changes_total = sum(b["changes"] for b in per.values()) meaningful = sum(b["meaningful_changes"] for b in per.values()) events_total = sum(b["events"] for b in per.values()) ebt: dict[str, int] = {} for b in per.values(): for k, v in b["events_by_type"].items(): ebt[k] = ebt.get(k, 0) + v jobs_open = await fetch_val(conn, "select count(*) from jobs where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1)", d1=d1) sensors_active = int(g.get("sensors_active") or 0) companies_active = int(g.get("companies_active") or 0) if sensors_active == 0 and per: # no observation rows (e.g. backfilled changes) → fall back to active sensors sensors_active = int(await fetch_val(conn, "select count(*) from sensors where status in ('active','failing','stale') and created_at < :d1", d1=d1) or 0) companies_active = len(per) ratio = meaningful / sensors_active if sensors_active else None baseline_rows = await fetch_all(conn, """select meaningful_changes, sensors_active from global_daily where day < :d and day >= :d_from and sensors_active > 0""", d=day, d_from=day - timedelta(days=int(P["index_trailing_days"]))) baseline_ratios = [r["meaningful_changes"] / r["sensors_active"] for r in baseline_rows if r["sensors_active"]] baseline = statistics.fmean(baseline_ratios) if baseline_ratios else None activity_index = round(ratio / baseline * 100.0, 2) if (ratio is not None and baseline) else None by_country, by_industry = await _breakdowns(conn, per) await execute(conn, """ insert into global_daily (day, companies_active, sensors_active, observations, changes, meaningful_changes, events, events_by_type, jobs_open, jobs_new, jobs_removed, activity_index, by_country, by_industry, computed_at) values (:d, :ca, :sa, :o, :ch, :m, :e, cast(:ebt as jsonb), :jo, :jn, :jr, :ai, cast(:bc as jsonb), cast(:bi as jsonb), now()) on conflict (day) do update set companies_active = excluded.companies_active, sensors_active = excluded.sensors_active, observations = excluded.observations, changes = excluded.changes, meaningful_changes = excluded.meaningful_changes, events = excluded.events, events_by_type = excluded.events_by_type, jobs_open = excluded.jobs_open, jobs_new = excluded.jobs_new, jobs_removed = excluded.jobs_removed, activity_index = excluded.activity_index, by_country = excluded.by_country, by_industry = excluded.by_industry, computed_at = now()""", d=day, ca=companies_active, sa=sensors_active, o=int(g.get("observations") or 0), ch=changes_total, m=meaningful, e=events_total, ebt=jsonb(ebt), jo=int(jobs_open or 0), jn=sum(b["jobs_new"] for b in per.values()), jr=sum(b["jobs_removed"] for b in per.values()), ai=activity_index, bc=jsonb(by_country), bi=jsonb(by_industry)) baselines_n = await _compute_baselines(conn, day) return {"day": day.isoformat(), "companies": len(per), "meaningful_changes": meaningful, "events": events_total, "sensors_active": sensors_active, "activity_index": activity_index, "baseline_days": len(baseline_ratios), "baselines": baselines_n} async def _breakdowns(conn, per: dict[str, dict[str, Any]]) -> tuple[dict[str, Any], dict[str, Any]]: # type: ignore[no-untyped-def] if not per: return {}, {} rows = await fetch_all(conn, "select id, country, industry_primary, industries from companies where id = any(cast(:ids as text[]))", ids=list(per)) by_country: dict[str, dict[str, int]] = {} by_industry: dict[str, dict[str, int]] = {} for r in rows: b = per[r["id"]] if r["country"]: c = by_country.setdefault(str(r["country"]), {"companies": 0, "meaningful_changes": 0, "events": 0}) c["companies"] += 1 c["meaningful_changes"] += b["meaningful_changes"] c["events"] += b["events"] ind = r["industry_primary"] or (r["industries"][0] if r["industries"] else None) if ind: i = by_industry.setdefault(str(ind), {"companies": 0, "meaningful_changes": 0, "events": 0}) i["companies"] += 1 i["meaningful_changes"] += b["meaningful_changes"] i["events"] += b["events"] return by_country, by_industry async def _compute_baselines(conn, day: date) -> int: # type: ignore[no-untyped-def] """Per-company mean/stddev of weekly meaningful changes, new jobs and news items over `settings.baseline_window_days` ending at `day`.""" window = settings.baseline_window_days start = day - timedelta(days=window) rows = await fetch_all(conn, """ select company_id, floor((day - cast(:start as date)) / 7.0)::int as week, sum(meaningful_changes) as m, sum(jobs_new) as jn, sum(news_items) as news from company_daily where day > :start and day <= :day group by 1, 2""", start=start, day=day) per: dict[str, dict[int, dict[str, int]]] = {} for r in rows: per.setdefault(r["company_id"], {})[int(r["week"])] = {"m": int(r["m"]), "jn": int(r["jn"]), "news": int(r["news"])} first_days = {r["company_id"]: r["first"] for r in await fetch_all(conn, "select company_id, min(day) as first from company_daily where company_id = any(cast(:ids as text[])) group by 1", ids=list(per))} if per else {} n = 0 weeks_total = max(1, window // 7) for cid, weeks in per.items(): first = first_days.get(cid) if first is None: continue observed_weeks = min(weeks_total, max(1, ((day - max(first, start)).days // 7) + 1)) if observed_weeks < 2: continue for metric, key in zip(BASELINE_METRICS, ("m", "jn", "news"), strict=True): series = [weeks.get(w, {}).get(key, 0) for w in range(weeks_total - observed_weeks, weeks_total)] mean = statistics.fmean(series) sd = statistics.pstdev(series) if len(series) > 1 else 0.0 await execute(conn, """ insert into baselines (company_id, metric, mean, stddev, samples, window_days, computed_at) values (:c, :m, :mean, :sd, :n, :w, now()) on conflict (company_id, metric) do update set mean = excluded.mean, stddev = excluded.stddev, samples = excluded.samples, window_days = excluded.window_days, computed_at = now()""", c=cid, m=metric, mean=mean, sd=sd, n=len(series), w=window) n += 1 return n async def compute_daily_catch_up(*, include_today: bool = False, max_days: int = 400) -> list[dict[str, Any]]: """Compute every missing day between the last computed day (or the dataset start) and yesterday.""" today = datetime.now(UTC).date() end = today if include_today else today - timedelta(days=1) async with transaction() as conn: last = await fetch_val(conn, "select max(day) from global_daily") first = await fetch_val(conn, """select least(coalesce((select min(date(fetched_at)) from observations), cast(:t as date)), coalesce((select min(date(detected_at)) from changes), cast(:t as date)), coalesce((select min(date(detected_at)) from events), cast(:t as date)))""", t=today) start = (last + timedelta(days=1)) if last else (first or end) results: list[dict[str, Any]] = [] day = max(start, end - timedelta(days=max_days)) while day <= end: results.append(await compute_daily(day)) day += timedelta(days=1) return results # ================================================================================================================ periodic @periodic("metrics-hourly", cron=settings.metrics_cron) async def metrics_hourly_task() -> None: stats = await compute_company_metrics() log.info("metrics-hourly", extra=stats) @periodic("daily-aggregates", cron=settings.daily_cron) async def daily_task() -> None: results = await compute_daily_catch_up() log.info("daily-aggregates", extra={"days": len(results), "last": results[-1] if results else None}) stats = await compute_company_metrics(all_companies=True) log.info("metrics-nightly", extra=stats) __all__ = ["CompanyContext", "MetricValue", "activity_score", "ai_adoption", "anomaly_score", "compute_company_metrics", "compute_daily", "compute_daily_catch_up", "compute_from_context", "corporate_change_index", "hiring_momentum", "historical_coverage", "load_context", "saturate", "saturating_score", "write_metrics"]