spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Proprietary metrics (spec §29–37, §91, §146–147, §177): reproducible, coverage-normalised, never fabricated.23Every metric row carries `formula_version`, `inputs` (the numbers the formula saw) and `computed_at`. A metric without inputs is not4written (no row ≠ 0). Formulas and parameters: `taxonomy.METRIC_PARAMS`, `taxonomy.CCI_WEIGHTS`, documented in docs/SCORING.md.56 compute_company_metrics(ids | None) hourly (@periodic metrics-hourly): companies active in the last 90 d; nightly: all7 compute_daily(day) daily aggregates: company_daily, global_daily (activity_index, baseline 100), baselines8 compute_daily_catch_up() fills missed days since the last computed day9"""10from __future__ import annotations1112import logging13import math14import statistics15from dataclasses import dataclass, field16from datetime import UTC, date, datetime, timedelta17from typing import Any1819from companyatlas.config import settings20from companyatlas.db import execute, fetch_all, fetch_one, fetch_val, jsonb, transaction21from companyatlas.services.periodic import periodic22from companyatlas.taxonomy import (23 AI_KEYWORDS,24 CCI_FORMULA_VERSION,25 CCI_WEIGHTS,26 METRIC_PARAMS,27 METRICS_FORMULA_VERSION,28 ChangeKind,29 EventType,30 Metric,31 Surface,32)3334log = logging.getLogger(__name__)3536P = METRIC_PARAMS37MEANINGFUL_KINDS = (ChangeKind.MEANINGFUL, ChangeKind.MAJOR, ChangeKind.CRITICAL)38CHANGE_WEIGHT = {ChangeKind.MEANINGFUL: P["change_weight_meaningful"], ChangeKind.MAJOR: P["change_weight_major"], ChangeKind.CRITICAL: P["change_weight_critical"]}39PRODUCT_SUBTYPES = {"PRODUCT_LAUNCH", "NEW_PRODUCT", "PRODUCT_REMOVED", "PRODUCT_RENAME", "PRODUCT_UPDATE", "FEATURE_LAUNCH", "CHANGELOG_ENTRY", "DOC_CHANGE",40 "DOCUMENTATION_CHANGE", "API_CHANGE", "API_LAUNCH", "SDK_RELEASE", "AI_LAUNCH"}41AI_SUBTYPES = {"AI_HIRING", "AI_LAUNCH"}42PRODUCT_SURFACES = {Surface.PRODUCTS, Surface.CHANGELOG, Surface.DOCS, Surface.API, Surface.DEVELOPER, Surface.SERVICES, Surface.SOLUTIONS}43DEVELOPER_SURFACES = {Surface.DOCS, Surface.API, Surface.DEVELOPER, Surface.CHANGELOG}44COMM_SURFACES = {Surface.NEWSROOM, Surface.BLOG, Surface.FEED, Surface.INVESTOR_RELATIONS, Surface.RESEARCH}45GEO_SURFACES = {Surface.LOCATIONS, Surface.CONTACT, Surface.ABOUT, Surface.CAREERS, Surface.JOBS_BOARD}46HIRING_SURFACES = {Surface.CAREERS, Surface.JOBS_BOARD}47BASELINE_METRICS = ("meaningful_changes_weekly", "jobs_new_weekly", "news_weekly")484950# ================================================================================================================ pure formulas515253def saturate(x: float, k: float) -> float:54 """0–1 saturation: 1 − exp(−x/k). x=k → 0.63, x=3k → 0.95."""55 return 0.0 if x <= 0 else 1.0 - math.exp(-x / k)565758def decay(age_days: float, tau: float) -> float:59 return math.exp(-max(0.0, age_days) / tau)606162@dataclass(slots=True)63class MetricValue:64 metric: str65 value: float66 confidence: float67 inputs: dict[str, Any] = field(default_factory=dict)68 formula_version: str = METRICS_FORMULA_VERSION697071def activity_score(changes: list[tuple[float, str]], events: list[tuple[float, float]], active_sensors: int) -> MetricValue | None:72 """changes = [(age_days, kind)], events = [(age_days, importance)] within the 30-day window.73 raw = Σ w(kind)·e^(−age/τ) + Σ importance·e^(−age/τ); density = raw / sensors^0.5; score = 100·log1p(density)/log1p(D_MAX)."""74 if active_sensors <= 0 and not changes and not events:75 return None76 tau = P["activity_decay_tau_days"]77 raw_changes = sum(CHANGE_WEIGHT.get(k, 1.0) * decay(a, tau) for a, k in changes)78 raw_events = sum(float(i) * decay(a, tau) for a, i in events)79 raw = raw_changes + raw_events80 denom = max(1.0, float(active_sensors)) ** P["activity_coverage_exp"]81 density = raw / denom82 score = min(100.0, 100.0 * math.log1p(density) / math.log1p(P["activity_density_max"]))83 conf = min(0.95, 0.5 + 0.05 * active_sensors)84 return MetricValue(Metric.ACTIVITY_SCORE, round(score, 2), round(conf, 2),85 {"changes_30d": len(changes), "events_30d": len(events), "raw_changes": round(raw_changes, 4), "raw_events": round(raw_events, 4),86 "active_sensors": active_sensors, "density": round(density, 4), "tau_days": tau, "coverage_exp": P["activity_coverage_exp"],87 "density_max": P["activity_density_max"]})888990def hiring_momentum(open_now: int, open_then: int, *, window: int, extra: dict[str, Any] | None = None, confidence: float = 0.8) -> MetricValue | None:91 """% change of open listings vs the reconstructed count `window` days ago. Requires ≥ `hiring_min_listings` at the reference point."""92 if open_then < P["hiring_min_listings"]:93 return None94 pct = (open_now - open_then) / open_then * 100.095 metric = {7: Metric.HIRING_MOMENTUM_7D, 30: Metric.HIRING_MOMENTUM_30D, 90: Metric.HIRING_MOMENTUM_90D}[window]96 return MetricValue(metric, round(pct, 2), confidence, {"open_now": open_now, "open_then": open_then, "window_days": window, **(extra or {})})979899def ai_adoption(*, ai_open: int | None, open_jobs: int | None, ai_events_90d: int, keyword_hits: int, has_text_inputs: bool) -> MetricValue | None:100 """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).101 Weights are renormalised over the inputs that exist."""102 comps: dict[str, tuple[float, float]] = {}103 if open_jobs is not None and open_jobs > 0:104 comps["jobs"] = (P["ai_jobs_weight"], min(1.0, (ai_open or 0) / open_jobs * 2)) # 50 % AI roles → saturates105 if ai_events_90d or has_text_inputs or open_jobs:106 comps["events"] = (P["ai_events_weight"], saturate(ai_events_90d, P["ai_events_saturation"]))107 if has_text_inputs:108 comps["keywords"] = (P["ai_keywords_weight"], saturate(keyword_hits, P["ai_keywords_saturation"]))109 if not comps:110 return None111 total_w = sum(w for w, _ in comps.values())112 score = 100.0 * sum(w * v for w, v in comps.values()) / total_w113 return MetricValue(Metric.AI_ADOPTION, round(score, 2), 0.6, {"ai_open": ai_open, "open_jobs": open_jobs, "ai_events_90d": ai_events_90d,114 "keyword_hits": keyword_hits, "components": {k: round(v, 4) for k, (_, v) in comps.items()},115 "weights_used": {k: w for k, (w, _) in comps.items()}})116117118def saturating_score(metric: str, weighted_sum: float, k: float, inputs: dict[str, Any], confidence: float = 0.7) -> MetricValue:119 return MetricValue(metric, round(100.0 * saturate(weighted_sum, k), 2), confidence, {**inputs, "weighted_sum": round(weighted_sum, 4), "saturation_k": k})120121122def corporate_change_index(values: dict[str, float]) -> MetricValue | None:123 """Σ w·component over available components, renormalised. Hiring momentum (%) is mapped to 0–100 as 50 + clamp(m, −100, 100)/2."""124 comps: dict[str, float] = {}125 for metric in CCI_WEIGHTS:126 if metric not in values:127 continue128 v = values[metric]129 if metric == Metric.HIRING_MOMENTUM_30D:130 v = 50.0 + max(-100.0, min(100.0, v)) / 2.0131 comps[str(metric)] = max(0.0, min(100.0, v))132 if not comps:133 return None134 total_w = sum(CCI_WEIGHTS[m] for m in CCI_WEIGHTS if str(m) in comps)135 value = sum(CCI_WEIGHTS[m] * comps[str(m)] for m in CCI_WEIGHTS if str(m) in comps) / total_w136 return MetricValue(Metric.CORPORATE_CHANGE_INDEX, round(value, 2), round(0.4 + 0.6 * total_w, 2),137 {"components": comps, "weights": {str(m): CCI_WEIGHTS[m] for m in CCI_WEIGHTS if str(m) in comps}, "weight_coverage": round(total_w, 3)},138 formula_version=CCI_FORMULA_VERSION)139140141def anomaly_score(this_week: int, mean: float, stddev: float, samples: int) -> MetricValue | None:142 if samples < P["anomaly_min_samples"]:143 return None144 sd = max(float(stddev), 0.5) # floor avoids infinite z on flat baselines145 z = (this_week - mean) / sd146 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,147 "stddev_floor": 0.5})148149150def historical_coverage(*, observed: int, expected: float, days_with_obs: int, days_since_first: int, surfaces: int) -> MetricValue | None:151 if expected <= 0 or days_since_first <= 0:152 return None153 obs_cov = min(1.0, observed / expected)154 continuity = min(1.0, days_with_obs / max(1, days_since_first))155 source_cov = min(1.0, surfaces / P["coverage_expected_surfaces"])156 score = 100.0 * (P["coverage_obs_weight"] * obs_cov + P["coverage_continuity_weight"] * continuity + P["coverage_sources_weight"] * source_cov)157 return MetricValue(Metric.HISTORICAL_COVERAGE, round(score, 2), 0.8, {"observed": observed, "expected": round(expected, 1), "days_with_obs": days_with_obs,158 "days_since_first": days_since_first, "surfaces": surfaces})159160161# ================================================================================================================ per-company loader162163164@dataclass(slots=True)165class CompanyContext:166 company: dict[str, Any]167 sensors: list[dict[str, Any]]168 changes_30d: list[dict[str, Any]]169 events_90d: list[dict[str, Any]]170 jobs: dict[str, Any]171 products: dict[str, Any]172 news: dict[str, Any]173 locations: dict[str, Any]174 titles_ai_hits: int175 baselines: dict[str, dict[str, Any]]176 observations: dict[str, Any]177 now: datetime178179180async def load_context(conn, company_id: str, now: datetime) -> CompanyContext | None: # type: ignore[no-untyped-def]181 company = await fetch_one(conn, "select id, slug, first_observed_at, industries, country from companies where id = :id", id=company_id)182 if company is None:183 return None184 d30, d90 = now - timedelta(days=30), now - timedelta(days=90)185 sensors = await fetch_all(conn, """select id, surface, status, base_interval_s, current_interval_s, created_at, observation_count, last_success_at, meaningful_change_count186 from sensors where company_id = :c and retired_at is null""", c=company_id)187 changes = await fetch_all(conn, """select detected_at, kind, surface from changes where company_id = :c and detected_at >= :since188 and kind in ('meaningful', 'major', 'critical')""", c=company_id, since=d30)189 events = await fetch_all(conn, """select detected_at, importance, event_type, event_subtype, tags, surface from events where company_id = :c190 and detected_at >= :since and status in ('active', 'review') order by detected_at desc limit 3000""", c=company_id, since=d90)191 jobs_row = await fetch_one(conn, """192 select count(*) filter (where status = 'open') as open_now,193 count(*) filter (where status = 'open' and is_ai) as ai_open,194 count(*) filter (where status = 'open' and remote) as remote_open,195 count(*) filter (where first_seen_at <= :t7 and (removed_at is null or removed_at > :t7)) as open_7,196 count(*) filter (where first_seen_at <= :t30 and (removed_at is null or removed_at > :t30)) as open_30,197 count(*) filter (where first_seen_at <= :t90 and (removed_at is null or removed_at > :t90)) as open_90,198 count(*) filter (where first_seen_at > :t30 and not baseline) as new_30d,199 count(*) filter (where removed_at > :t30) as removed_30d,200 count(*) filter (where first_seen_at > :t30 and is_ai and not baseline) as ai_new_30d,201 count(*) as total202 from jobs where company_id = :c""", c=company_id, t7=now - timedelta(days=7), t30=d30, t90=d90)203 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)204 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)205 new_job_countries = await fetch_all(conn, """206 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)207 products = await fetch_one(conn, """select count(*) filter (where status = 'listed') as listed, count(*) as total,208 array_agg(name) filter (where status = 'listed') as names from products where company_id = :c""", c=company_id)209 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,210 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_90d211 from news_items where company_id = :c""", c=company_id, d30=d30, d90=d90)212 locations = await fetch_one(conn, """213 select count(*) filter (where status = 'listed') as listed, count(*) filter (where first_seen_at > :since and not baseline) as new_90d, count(*) as total,214 count(distinct country) filter (where status = 'listed' and country is not null) as countries215 from locations where company_id = :c""", c=company_id, since=d90)216 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)",217 c=company_id, since=d90)218 titles = await fetch_all(conn, """select distinct on (s.sensor_id) s.title from snapshots s join sensors se on se.id = s.sensor_id219 where se.company_id = :c order by s.sensor_id, s.fetched_at desc""", c=company_id)220 baselines = await fetch_all(conn, "select metric, mean, stddev, samples from baselines where company_id = :c", c=company_id)221 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,222 count(distinct sensor_id) as sensors_observed from observations where company_id = :c""", c=company_id)223 this_week = await fetch_val(conn, "select count(*) from changes where company_id = :c and detected_at >= :since and kind in ('meaningful', 'major', 'critical')",224 c=company_id, since=now - timedelta(days=7))225 jobs = {**(jobs_row or {}), "by_country": [{"country": r["country"], "n": r["n"]} for r in by_country],226 "by_department": [{"department": r["department"], "n": r["n"]} for r in by_department], "new_countries_90d": [r["country"] for r in new_job_countries]}227 return CompanyContext(228 company=company, sensors=sensors, changes_30d=changes, events_90d=events, jobs=jobs,229 products={**(products or {}), "names": list((products or {}).get("names") or [])},230 news={**(news or {}), "titles_90d": list((news or {}).get("titles_90d") or [])},231 locations={**(locations or {}), "new_countries_90d": [r["country"] for r in new_loc_countries]},232 titles_ai_hits=sum(1 for t in titles if _ai_hit(t.get("title"))),233 baselines={r["metric"]: r for r in baselines}, observations={**(obs or {}), "this_week_meaningful": int(this_week or 0)}, now=now)234235236def _ai_hit(text: str | None) -> bool:237 if not text:238 return False239 hay = f" {text.lower()} "240 return any(k in hay for k in AI_KEYWORDS)241242243def _age(now: datetime, at: datetime) -> float:244 if at.tzinfo is None:245 at = at.replace(tzinfo=UTC)246 return max(0.0, (now - at).total_seconds() / 86400.0)247248249# ================================================================================================================ compute250251252def compute_from_context(ctx: CompanyContext) -> list[MetricValue]:253 now = ctx.now254 out: list[MetricValue] = []255 active_sensors = [s for s in ctx.sensors if s["status"] in ("active", "failing", "stale")]256 surfaces = {s["surface"] for s in ctx.sensors}257 n_active = len(active_sensors)258 events_30 = [e for e in ctx.events_90d if _age(now, e["detected_at"]) <= P["activity_window_days"]]259260 # activity ----------------------------------------------------------------------------------------------------261 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)262 if act and (ctx.changes_30d or events_30 or n_active):263 out.append(act)264265 # hiring ------------------------------------------------------------------------------------------------------266 j = ctx.jobs267 has_jobs_source = bool(surfaces & HIRING_SURFACES) or int(j.get("total") or 0) > 0268 jobs_conf = 0.9 if Surface.JOBS_BOARD in surfaces else 0.75269 if has_jobs_source:270 open_now = int(j.get("open_now") or 0)271 remote_ratio = round(int(j.get("remote_open") or 0) / open_now, 4) if open_now else None272 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,273 "by_country": j.get("by_country"), "by_department": j.get("by_department"), "ai_open": int(j.get("ai_open") or 0)}274 out.append(MetricValue(Metric.OPEN_JOBS, float(open_now), jobs_conf, extra))275 for window, key in ((7, "open_7"), (30, "open_30"), (90, "open_90")):276 m = hiring_momentum(open_now, int(j.get(key) or 0), window=window, extra=extra if window == 30 else None, confidence=jobs_conf)277 if m:278 out.append(m)279280 # AI adoption -------------------------------------------------------------------------------------------------281 ai_events = sum(1 for e in ctx.events_90d if e["event_subtype"] in AI_SUBTYPES or "ai" in (e.get("tags") or []))282 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_hits283 has_text = bool(ctx.products["names"] or ctx.news["titles_90d"] or ctx.sensors)284 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,285 ai_events_90d=ai_events, keyword_hits=kw_hits, has_text_inputs=has_text)286 if ai:287 out.append(ai)288289 # product velocity / developer / communication / pricing / leadership ------------------------------------------290 tau = P["activity_decay_tau_days"] * 3 # 90-day windows decay slower291 if surfaces & PRODUCT_SURFACES or any(e["event_type"] == EventType.PRODUCT for e in ctx.events_90d):292 prod_events = [e for e in ctx.events_90d if e["event_subtype"] in PRODUCT_SUBTYPES]293 s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in prod_events)294 out.append(saturating_score(Metric.PRODUCT_VELOCITY, s, P["velocity_saturation"], {"events_90d": len(prod_events), "surfaces": sorted(surfaces & PRODUCT_SURFACES)}))295 if surfaces & DEVELOPER_SURFACES or any(e["event_type"] == EventType.DEVELOPER for e in ctx.events_90d):296 dev_events = [e for e in ctx.events_90d if e["event_type"] == EventType.DEVELOPER]297 dev_changes = [c for c in ctx.changes_30d if c["surface"] in DEVELOPER_SURFACES]298 s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in dev_events) + 0.5 * len(dev_changes)299 out.append(saturating_score(Metric.DEVELOPER_MOMENTUM, s, P["developer_saturation"], {"events_90d": len(dev_events), "meaningful_changes_30d": len(dev_changes),300 "surfaces": sorted(surfaces & DEVELOPER_SURFACES)}))301 if surfaces & COMM_SURFACES or int(ctx.news.get("total") or 0) > 0:302 comm_events = [e for e in events_30 if e["event_type"] in (EventType.COMMUNICATION, EventType.INVESTOR_RELATIONS)]303 n_news = int(ctx.news.get("n_30d") or 0)304 s = float(max(n_news, len(comm_events))) + 0.5 * min(n_news, len(comm_events))305 out.append(saturating_score(Metric.COMMUNICATION_ACTIVITY, s, P["communication_saturation"], {"news_items_30d": n_news, "events_30d": len(comm_events)}))306 if Surface.PRICING in surfaces or any(e["event_type"] == EventType.PRICING for e in ctx.events_90d):307 pr_events = [e for e in ctx.events_90d if e["event_type"] == EventType.PRICING]308 pr_changes = [c for c in ctx.changes_30d if c["surface"] == Surface.PRICING]309 s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in pr_events) + 0.25 * len(pr_changes)310 out.append(saturating_score(Metric.PRICING_ACTIVITY, s, P["pricing_saturation"], {"events_90d": len(pr_events), "meaningful_changes_30d": len(pr_changes)}))311 if Surface.LEADERSHIP in surfaces or any(e["event_type"] == EventType.LEADERSHIP for e in ctx.events_90d):312 ld_events = [e for e in ctx.events_90d if e["event_type"] == EventType.LEADERSHIP]313 s = sum(float(e["importance"]) * decay(_age(now, e["detected_at"]), tau) for e in ld_events)314 out.append(saturating_score(Metric.LEADERSHIP_ACTIVITY, s, P["leadership_saturation"], {"events_90d": len(ld_events)}))315316 # geographic expansion ----------------------------------------------------------------------------------------317 loc = ctx.locations318 if surfaces & GEO_SURFACES or int(loc.get("total") or 0) > 0 or has_jobs_source:319 new_countries = set(loc.get("new_countries_90d") or []) | set(j.get("new_countries_90d") or [])320 geo_events = [e for e in ctx.events_90d if e["event_type"] == EventType.LOCATION]321 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")322 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),323 "countries_listed": int(loc.get("countries") or 0), "location_events_90d": len(geo_events)}))324325 # CCI ---------------------------------------------------------------------------------------------------------326 cci = corporate_change_index({m.metric: m.value for m in out})327 if cci:328 out.append(cci)329330 # anomaly -----------------------------------------------------------------------------------------------------331 b = ctx.baselines.get("meaningful_changes_weekly")332 if b:333 an = anomaly_score(int(ctx.observations.get("this_week_meaningful") or 0), float(b["mean"]), float(b["stddev"]), int(b["samples"]))334 if an:335 out.append(an)336337 # historical coverage -----------------------------------------------------------------------------------------338 first = ctx.company.get("first_observed_at")339 if first and ctx.sensors:340 days_since_first = max(1, int(_age(now, first)))341 expected = 0.0342 for s in ctx.sensors:343 created = s.get("created_at") or first344 span_s = max(0.0, (now - (created if created.tzinfo else created.replace(tzinfo=UTC))).total_seconds())345 expected += span_s / max(900, int(s.get("current_interval_s") or s.get("base_interval_s") or 86400))346 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),347 days_since_first=days_since_first, surfaces=len(surfaces))348 if hc:349 out.append(hc)350 return out351352353async def write_metrics(conn, company_id: str, values: list[MetricValue], now: datetime) -> None: # type: ignore[no-untyped-def]354 day = now.astimezone(UTC).date()355 for m in values:356 await execute(conn, """357 insert into metrics_current (company_id, metric, value, confidence, inputs, formula_version, computed_at)358 values (:c, :m, :v, :conf, cast(:inputs as jsonb), :fv, :at)359 on conflict (company_id, metric) do update set value = excluded.value, confidence = excluded.confidence, inputs = excluded.inputs,360 formula_version = excluded.formula_version, computed_at = excluded.computed_at""",361 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)362 await execute(conn, """363 insert into metric_series (company_id, metric, day, value, confidence, formula_version) values (:c, :m, :d, :v, :conf, :fv)364 on conflict (company_id, metric, day) do update set value = excluded.value, confidence = excluded.confidence, formula_version = excluded.formula_version""",365 c=company_id, m=str(m.metric), d=day, v=float(m.value), conf=float(m.confidence), fv=m.formula_version)366367368async def compute_company_metrics(company_ids: list[str] | None = None, *, all_companies: bool = False, now: datetime | None = None) -> dict[str, int]:369 """Compute and store metrics. `company_ids=None` → companies with activity in the last `metrics_active_window_days` (or all)."""370 now = now or datetime.now(UTC)371 stats = {"companies": 0, "metrics": 0, "skipped": 0}372 async with transaction() as conn:373 if company_ids is None:374 if all_companies:375 rows = await fetch_all(conn, "select id from companies where onboarding_status <> 'no_website' order by importance desc")376 else:377 rows = await fetch_all(conn, """378 select id from companies where greatest(coalesce(last_observed_at, 'epoch'), coalesce(last_change_at, 'epoch'), coalesce(last_event_at, 'epoch')) >= :since379 or id in (select distinct company_id from changes where detected_at >= :since)380 or id in (select distinct company_id from events where detected_at >= :since)381 order by importance desc""", since=now - timedelta(days=settings.metrics_active_window_days))382 company_ids = [r["id"] for r in rows]383 for cid in company_ids:384 try:385 async with transaction() as conn:386 ctx = await load_context(conn, cid, now)387 if ctx is None:388 stats["skipped"] += 1389 continue390 values = compute_from_context(ctx)391 if not values:392 stats["skipped"] += 1393 continue394 await write_metrics(conn, cid, values, now)395 stats["companies"] += 1396 stats["metrics"] += len(values)397 except Exception:398 log.exception("metrics failed", extra={"company_id": cid})399 return stats400401402# ================================================================================================================ daily aggregates403404405def _day_bounds(day: date) -> tuple[datetime, datetime]:406 d0 = datetime(day.year, day.month, day.day, tzinfo=UTC)407 return d0, d0 + timedelta(days=1)408409410async def compute_daily(day: date) -> dict[str, Any]:411 """company_daily + global_daily (+ baselines) for one UTC day. Idempotent (upserts)."""412 d0, d1 = _day_bounds(day)413 async with transaction() as conn:414 per: dict[str, dict[str, Any]] = {}415416 def bucket(cid: str) -> dict[str, Any]:417 return per.setdefault(cid, {"observations": 0, "changes": 0, "meaningful_changes": 0, "events": 0, "events_by_type": {}, "jobs_open": None,418 "jobs_new": 0, "jobs_removed": 0, "jobs_ai_open": None, "news_items": 0, "sensors_active": None})419420 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):421 bucket(r["company_id"])["observations"] = int(r["n"])422 for r in await fetch_all(conn, """select company_id, count(*) as n, count(*) filter (where kind in ('meaningful','major','critical')) as m423 from changes where detected_at >= :d0 and detected_at < :d1 group by 1""", d0=d0, d1=d1):424 b = bucket(r["company_id"])425 b["changes"], b["meaningful_changes"] = int(r["n"]), int(r["m"])426 for r in await fetch_all(conn, """select company_id, event_type, count(*) as n from events where detected_at >= :d0 and detected_at < :d1427 and status in ('active', 'review') group by 1, 2""", d0=d0, d1=d1):428 b = bucket(r["company_id"])429 b["events"] += int(r["n"])430 b["events_by_type"][r["event_type"]] = int(r["n"])431 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,432 count(*) filter (where removed_at >= :d0 and removed_at < :d1) as rem_n,433 count(*) filter (where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1)) as open_n,434 count(*) filter (where first_seen_at < :d1 and (removed_at is null or removed_at >= :d1) and is_ai) as ai_n435 from jobs group by 1""", d0=d0, d1=d1):436 if not (r["new_n"] or r["rem_n"] or r["open_n"]):437 continue438 b = bucket(r["company_id"])439 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"])440 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):441 bucket(r["company_id"])["news_items"] = int(r["n"])442 if per:443 ids = list(per)444 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')445 and created_at < :d1 group by 1""", ids=ids, d1=d1):446 per[r["company_id"]]["sensors_active"] = int(r["n"])447 for cid, b in per.items():448 await execute(conn, """449 insert into company_daily (company_id, day, observations, changes, meaningful_changes, events, events_by_type, jobs_open, jobs_new, jobs_removed,450 jobs_ai_open, news_items, sensors_active)451 values (:c, :d, :o, :ch, :m, :e, cast(:ebt as jsonb), :jo, :jn, :jr, :jai, :news, :sa)452 on conflict (company_id, day) do update set observations = excluded.observations, changes = excluded.changes,453 meaningful_changes = excluded.meaningful_changes, events = excluded.events, events_by_type = excluded.events_by_type,454 jobs_open = excluded.jobs_open, jobs_new = excluded.jobs_new, jobs_removed = excluded.jobs_removed, jobs_ai_open = excluded.jobs_ai_open,455 news_items = excluded.news_items, sensors_active = excluded.sensors_active""",456 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"],457 jn=b["jobs_new"], jr=b["jobs_removed"], jai=b["jobs_ai_open"], news=b["news_items"], sa=b["sensors_active"])458459 # global -----------------------------------------------------------------------------------------------------460 g = await fetch_one(conn, """select count(distinct company_id) as companies_active, count(distinct sensor_id) as sensors_active, count(*) as observations461 from observations where fetched_at >= :d0 and fetched_at < :d1""", d0=d0, d1=d1) or {}462 changes_total = sum(b["changes"] for b in per.values())463 meaningful = sum(b["meaningful_changes"] for b in per.values())464 events_total = sum(b["events"] for b in per.values())465 ebt: dict[str, int] = {}466 for b in per.values():467 for k, v in b["events_by_type"].items():468 ebt[k] = ebt.get(k, 0) + v469 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)470 sensors_active = int(g.get("sensors_active") or 0)471 companies_active = int(g.get("companies_active") or 0)472 if sensors_active == 0 and per: # no observation rows (e.g. backfilled changes) → fall back to active sensors473 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)474 companies_active = len(per)475 ratio = meaningful / sensors_active if sensors_active else None476 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""",477 d=day, d_from=day - timedelta(days=int(P["index_trailing_days"])))478 baseline_ratios = [r["meaningful_changes"] / r["sensors_active"] for r in baseline_rows if r["sensors_active"]]479 baseline = statistics.fmean(baseline_ratios) if baseline_ratios else None480 activity_index = round(ratio / baseline * 100.0, 2) if (ratio is not None and baseline) else None481 by_country, by_industry = await _breakdowns(conn, per)482 await execute(conn, """483 insert into global_daily (day, companies_active, sensors_active, observations, changes, meaningful_changes, events, events_by_type, jobs_open, jobs_new,484 jobs_removed, activity_index, by_country, by_industry, computed_at)485 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())486 on conflict (day) do update set companies_active = excluded.companies_active, sensors_active = excluded.sensors_active, observations = excluded.observations,487 changes = excluded.changes, meaningful_changes = excluded.meaningful_changes, events = excluded.events, events_by_type = excluded.events_by_type,488 jobs_open = excluded.jobs_open, jobs_new = excluded.jobs_new, jobs_removed = excluded.jobs_removed, activity_index = excluded.activity_index,489 by_country = excluded.by_country, by_industry = excluded.by_industry, computed_at = now()""",490 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),491 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,492 bc=jsonb(by_country), bi=jsonb(by_industry))493 baselines_n = await _compute_baselines(conn, day)494 return {"day": day.isoformat(), "companies": len(per), "meaningful_changes": meaningful, "events": events_total, "sensors_active": sensors_active,495 "activity_index": activity_index, "baseline_days": len(baseline_ratios), "baselines": baselines_n}496497498async def _breakdowns(conn, per: dict[str, dict[str, Any]]) -> tuple[dict[str, Any], dict[str, Any]]: # type: ignore[no-untyped-def]499 if not per:500 return {}, {}501 rows = await fetch_all(conn, "select id, country, industry_primary, industries from companies where id = any(cast(:ids as text[]))", ids=list(per))502 by_country: dict[str, dict[str, int]] = {}503 by_industry: dict[str, dict[str, int]] = {}504 for r in rows:505 b = per[r["id"]]506 if r["country"]:507 c = by_country.setdefault(str(r["country"]), {"companies": 0, "meaningful_changes": 0, "events": 0})508 c["companies"] += 1509 c["meaningful_changes"] += b["meaningful_changes"]510 c["events"] += b["events"]511 ind = r["industry_primary"] or (r["industries"][0] if r["industries"] else None)512 if ind:513 i = by_industry.setdefault(str(ind), {"companies": 0, "meaningful_changes": 0, "events": 0})514 i["companies"] += 1515 i["meaningful_changes"] += b["meaningful_changes"]516 i["events"] += b["events"]517 return by_country, by_industry518519520async def _compute_baselines(conn, day: date) -> int: # type: ignore[no-untyped-def]521 """Per-company mean/stddev of weekly meaningful changes, new jobs and news items over `settings.baseline_window_days` ending at `day`."""522 window = settings.baseline_window_days523 start = day - timedelta(days=window)524 rows = await fetch_all(conn, """525 select company_id, floor((day - cast(:start as date)) / 7.0)::int as week,526 sum(meaningful_changes) as m, sum(jobs_new) as jn, sum(news_items) as news527 from company_daily where day > :start and day <= :day group by 1, 2""", start=start, day=day)528 per: dict[str, dict[int, dict[str, int]]] = {}529 for r in rows:530 per.setdefault(r["company_id"], {})[int(r["week"])] = {"m": int(r["m"]), "jn": int(r["jn"]), "news": int(r["news"])}531 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",532 ids=list(per))} if per else {}533 n = 0534 weeks_total = max(1, window // 7)535 for cid, weeks in per.items():536 first = first_days.get(cid)537 if first is None:538 continue539 observed_weeks = min(weeks_total, max(1, ((day - max(first, start)).days // 7) + 1))540 if observed_weeks < 2:541 continue542 for metric, key in zip(BASELINE_METRICS, ("m", "jn", "news"), strict=True):543 series = [weeks.get(w, {}).get(key, 0) for w in range(weeks_total - observed_weeks, weeks_total)]544 mean = statistics.fmean(series)545 sd = statistics.pstdev(series) if len(series) > 1 else 0.0546 await execute(conn, """547 insert into baselines (company_id, metric, mean, stddev, samples, window_days, computed_at) values (:c, :m, :mean, :sd, :n, :w, now())548 on conflict (company_id, metric) do update set mean = excluded.mean, stddev = excluded.stddev, samples = excluded.samples,549 window_days = excluded.window_days, computed_at = now()""", c=cid, m=metric, mean=mean, sd=sd, n=len(series), w=window)550 n += 1551 return n552553554async def compute_daily_catch_up(*, include_today: bool = False, max_days: int = 400) -> list[dict[str, Any]]:555 """Compute every missing day between the last computed day (or the dataset start) and yesterday."""556 today = datetime.now(UTC).date()557 end = today if include_today else today - timedelta(days=1)558 async with transaction() as conn:559 last = await fetch_val(conn, "select max(day) from global_daily")560 first = await fetch_val(conn, """select least(coalesce((select min(date(fetched_at)) from observations), cast(:t as date)),561 coalesce((select min(date(detected_at)) from changes), cast(:t as date)),562 coalesce((select min(date(detected_at)) from events), cast(:t as date)))""", t=today)563 start = (last + timedelta(days=1)) if last else (first or end)564 results: list[dict[str, Any]] = []565 day = max(start, end - timedelta(days=max_days))566 while day <= end:567 results.append(await compute_daily(day))568 day += timedelta(days=1)569 return results570571572# ================================================================================================================ periodic573574575@periodic("metrics-hourly", cron=settings.metrics_cron)576async def metrics_hourly_task() -> None:577 stats = await compute_company_metrics()578 log.info("metrics-hourly", extra=stats)579580581@periodic("daily-aggregates", cron=settings.daily_cron)582async def daily_task() -> None:583 results = await compute_daily_catch_up()584 log.info("daily-aggregates", extra={"days": len(results), "last": results[-1] if results else None})585 stats = await compute_company_metrics(all_companies=True)586 log.info("metrics-nightly", extra=stats)587588589__all__ = ["CompanyContext", "MetricValue", "activity_score", "ai_adoption", "anomaly_score", "compute_company_metrics", "compute_daily",590 "compute_daily_catch_up", "compute_from_context", "corporate_change_index", "hiring_momentum", "historical_coverage", "load_context",591 "saturate", "saturating_score", "write_metrics"]592