spb/company-atlas
Public
Python 66.3%
TypeScript 22.7%
JavaScript 8.6%
HTML 1.4%
CSS 0.7%
1"""Cross-signal detection (spec §92–93): hiring surge / freeze, launch build-up, international expansion, pricing migration, developer2push, enterprise repositioning, AI acceleration, abnormal activity. Signals are *labelled as signals*, carry a strength (0–1), a3confidence, an explanation and the evidence (event ids, metric values) that produced them, and expire. They never assert facts.45Company-scope signals are recomputed hourly for companies with recent activity; industry/country aggregates are derived from them.6"""7from __future__ import annotations89import logging10from dataclasses import dataclass, field11from datetime import UTC, datetime, timedelta12from typing import Any1314from companyatlas.config import settings15from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction16from companyatlas.ids import new_id17from companyatlas.services.metrics import saturate18from companyatlas.services.periodic import periodic19from companyatlas.taxonomy import SIGNAL_KINDS, SIGNALS_FORMULA_VERSION, EventType, Metric2021log = logging.getLogger(__name__)2223SCOPE_MIN_COMPANIES = 324LAUNCH_WINDOW_DAYS = 1425SURGE_MOMENTUM_PCT = 30.026FREEZE_MOMENTUM_PCT = -30.027MIN_OPEN_FOR_MOMENTUM = 1028ENTERPRISE_RE = ("enterprise", "contact sales", "sso", "saml", "audit log", "dedicated", "custom pricing", "procurement", "soc 2", "soc2")293031@dataclass(slots=True)32class SignalDraft:33 kind: str34 strength: float35 confidence: float36 title: str37 explanation: str38 evidence: dict[str, Any] = field(default_factory=dict)39 window_days: int = 30404142@dataclass(slots=True)43class SignalInputs:44 """Everything the detectors look at for one company (pure, testable)."""45 metrics: dict[str, float]46 events: list[dict[str, Any]] # {id, event_type, event_subtype, importance, tags, detected_at, title}47 jobs: dict[str, Any] # open_now, new_30d, ai_new_30d, new_countries_90d48 plans_contact_sales_new: int = 049 locations_new_countries: list[str] = field(default_factory=list)50 now: datetime = field(default_factory=lambda: datetime.now(UTC))515253def _age(now: datetime, at: datetime) -> float:54 if at.tzinfo is None:55 at = at.replace(tzinfo=UTC)56 return (now - at).total_seconds() / 86400.0575859def _ids(events: list[dict[str, Any]]) -> list[str]:60 return [e["id"] for e in events[:25]]616263def detect(inp: SignalInputs) -> list[SignalDraft]:64 m, ev, j, now = inp.metrics, inp.events, inp.jobs, inp.now65 out: list[SignalDraft] = []66 window = settings.signals_window_days67 recent = [e for e in ev if _age(now, e["detected_at"]) <= window]68 open_now = int(j.get("open_now") or 0)69 mom30 = m.get(Metric.HIRING_MOMENTUM_30D)7071 # hiring surge --------------------------------------------------------------------------------------------------72 surge_events = [e for e in recent if e["event_subtype"] in ("HIRING_SURGE", "JOB_COUNT_INCREASE")]73 if (mom30 is not None and mom30 >= SURGE_MOMENTUM_PCT and open_now >= MIN_OPEN_FOR_MOMENTUM) or any(e["event_subtype"] == "HIRING_SURGE" for e in recent):74 strength = saturate(max(mom30 or 0, 30.0) / 50.0, 1.5)75 out.append(SignalDraft("hiring_surge", round(strength, 3), 0.7, "Hiring surge signal",76 f"Open listings observed up {mom30:.0f}% over 30 days ({open_now} open)." if mom30 is not None else77 f"{len(surge_events)} hiring increase events detected in the last {window} days.",78 {"event_ids": _ids(surge_events), "hiring_momentum_30d": mom30, "open_jobs": open_now}))79 # hiring freeze — careful wording: listings no longer visible --------------------------------------------------80 freeze_events = [e for e in recent if e["event_subtype"] in ("HIRING_FREEZE_SIGNAL", "JOB_COUNT_DECREASE")]81 if (mom30 is not None and mom30 <= FREEZE_MOMENTUM_PCT and int(j.get("new_30d") or 0) <= 1) or any(e["event_subtype"] == "HIRING_FREEZE_SIGNAL" for e in recent):82 strength = saturate(abs(min(mom30 or -30.0, -30.0)) / 50.0, 1.5)83 out.append(SignalDraft("hiring_freeze", round(strength, 3), 0.6, "Hiring slowdown signal: listings no longer visible",84 (f"Monitored open listings are down {abs(mom30):.0f}% over 30 days with {int(j.get('new_30d') or 0)} new listings; "85 "listings no longer visible on public pages are not evidence of workforce decisions.") if mom30 is not None else86 f"{len(freeze_events)} events of listings no longer visible in the last {window} days.",87 {"event_ids": _ids(freeze_events), "hiring_momentum_30d": mom30, "jobs_new_30d": j.get("new_30d")}))88 # launch build-up -----------------------------------------------------------------------------------------------89 win14 = [e for e in ev if _age(now, e["detected_at"]) <= LAUNCH_WINDOW_DAYS]90 cats = {91 "product": [e for e in win14 if e["event_type"] == EventType.PRODUCT],92 "docs": [e for e in win14 if e["event_subtype"] in ("DOC_CHANGE", "DOCUMENTATION_CHANGE", "API_CHANGE", "API_LAUNCH", "SDK_RELEASE")],93 "changelog": [e for e in win14 if e["event_subtype"] == "CHANGELOG_ENTRY"],94 "careers": [e for e in win14 if e["event_type"] == EventType.HIRING and e["event_subtype"] != "JOB_COUNT_DECREASE"],95 "messaging": [e for e in win14 if e["event_subtype"] in ("MESSAGING_CHANGE", "HOMEPAGE_REDESIGN", "WEBSITE_CHANGE")],96 }97 present = {k: v for k, v in cats.items() if v}98 if len(present) >= 3 and cats["product"] or len(present) >= 4:99 strength = min(1.0, 0.3 + 0.15 * len(present) + 0.05 * min(6, sum(len(v) for v in present.values())))100 out.append(SignalDraft("launch_buildup", round(strength, 3), round(0.45 + 0.08 * len(present), 2), "Possible launch preparation signal",101 "Within 14 days: " + ", ".join(f"{len(v)} {k}" for k, v in present.items()) + " events detected on public pages. "102 "This pattern often precedes announcements; it is a probabilistic signal, not a confirmation.",103 {"event_ids": _ids([e for v in present.values() for e in v]), "categories": {k: len(v) for k, v in present.items()}}, LAUNCH_WINDOW_DAYS))104 # expansion -----------------------------------------------------------------------------------------------------105 geo_events = [e for e in recent if e["event_subtype"] in ("COUNTRY_EXPANSION", "NEW_LOCATION", "NEW_OFFICE")]106 new_countries = sorted(set(inp.locations_new_countries) | set(j.get("new_countries_90d") or []))107 if any(e["event_subtype"] == "COUNTRY_EXPANSION" for e in geo_events) or (new_countries and (geo_events or int(j.get("new_30d") or 0) > 0)):108 strength = min(1.0, 0.4 + 0.2 * len(new_countries) + 0.05 * len(geo_events))109 out.append(SignalDraft("expansion", round(strength, 3), 0.65, "International expansion signal",110 (f"New country presence listed: {', '.join(new_countries[:5])}. " if new_countries else "") +111 f"{len(geo_events)} location events detected in the last {window} days" + (", with job listings in new countries." if j.get("new_countries_90d") else "."),112 {"event_ids": _ids(geo_events), "new_countries": new_countries, "geo_expansion": m.get(Metric.GEO_EXPANSION)}))113 # pricing migration ---------------------------------------------------------------------------------------------114 pricing = [e for e in recent if e["event_type"] == EventType.PRICING]115 subtypes = {e["event_subtype"] for e in pricing}116 if len(pricing) >= 2 or ({"NEW_PRICING_TIER", "PRICING_TIER_REMOVED"} <= subtypes):117 strength = min(1.0, 0.3 + 0.15 * len(pricing) + (0.2 if {"NEW_PRICING_TIER", "PRICING_TIER_REMOVED"} <= subtypes else 0))118 out.append(SignalDraft("pricing_migration", round(strength, 3), 0.7, "Pricing migration signal",119 f"{len(pricing)} pricing events in {window} days: " + ", ".join(sorted(subtypes)).lower().replace("_", " ") + ".",120 {"event_ids": _ids(pricing), "subtypes": sorted(subtypes), "pricing_activity": m.get(Metric.PRICING_ACTIVITY)}))121 # developer push ------------------------------------------------------------------------------------------------122 dev = [e for e in recent if e["event_type"] == EventType.DEVELOPER]123 dev_m = m.get(Metric.DEVELOPER_MOMENTUM)124 if len(dev) >= 3 or (dev_m is not None and dev_m >= 60):125 strength = saturate(len(dev) + (dev_m or 0) / 40.0, 4.0)126 out.append(SignalDraft("developer_push", round(strength, 3), 0.7, "Developer push signal",127 f"{len(dev)} developer-surface events (docs, API, changelog, SDK) in {window} days" + (f"; developer momentum {dev_m:.0f}/100." if dev_m is not None else "."),128 {"event_ids": _ids(dev), "developer_momentum": dev_m}))129 # enterprise repositioning --------------------------------------------------------------------------------------130 ent_events = [e for e in recent if e["event_subtype"] == "ENTERPRISE_REPOSITIONING" or "enterprise" in (e.get("tags") or [])131 or any(k in (e.get("title") or "").lower() for k in ENTERPRISE_RE)]132 if inp.plans_contact_sales_new or len(ent_events) >= 2:133 strength = min(1.0, 0.35 + 0.25 * min(2, inp.plans_contact_sales_new) + 0.1 * len(ent_events))134 out.append(SignalDraft("enterprise_repositioning", round(strength, 3), 0.55, "Enterprise repositioning signal",135 (f"{inp.plans_contact_sales_new} new contact-sales / enterprise pricing tier(s) listed. " if inp.plans_contact_sales_new else "") +136 (f"{len(ent_events)} events mention enterprise features or messaging." if ent_events else ""),137 {"event_ids": _ids(ent_events), "contact_sales_tiers_new": inp.plans_contact_sales_new}))138 # AI acceleration -----------------------------------------------------------------------------------------------139 ai_events = [e for e in recent if e["event_subtype"] in ("AI_HIRING", "AI_LAUNCH") or "ai" in (e.get("tags") or [])]140 ai_new = int(j.get("ai_new_30d") or 0)141 new30 = int(j.get("new_30d") or 0)142 ai_share = (ai_new / new30) if new30 else 0.0143 if len(ai_events) >= 2 or (new30 >= 4 and ai_share >= 0.25):144 strength = min(1.0, 0.3 + 0.1 * len(ai_events) + ai_share)145 out.append(SignalDraft("ai_acceleration", round(strength, 3), 0.6, "AI acceleration signal",146 f"{len(ai_events)} AI-related events in {window} days" + (f"; {ai_new} of {new30} new listings are AI-related ({ai_share:.0%})." if new30 else ".") +147 " Based on public job titles, product names and announcements only.",148 {"event_ids": _ids(ai_events), "ai_new_30d": ai_new, "jobs_new_30d": new30, "ai_adoption": m.get(Metric.AI_ADOPTION)}))149 # abnormal activity ---------------------------------------------------------------------------------------------150 z = m.get(Metric.ANOMALY_SCORE)151 if z is not None and z >= settings.anomaly_z:152 out.append(SignalDraft("abnormal_activity", round(min(1.0, z / (settings.anomaly_z * 2)), 3), 0.7, "Unusual activity signal",153 f"This week's meaningful changes are {z:.1f} standard deviations above this company's baseline.",154 {"anomaly_z": z, "event_ids": _ids(recent[:10])}, 7))155 return out156157158# ================================================================================================================ persistence159160161async def _inputs(conn, company_id: str, now: datetime) -> SignalInputs: # type: ignore[no-untyped-def]162 metrics = {r["metric"]: float(r["value"]) for r in await fetch_all(conn, "select metric, value from metrics_current where company_id = :c", c=company_id)}163 events = await fetch_all(conn, """select id, event_type, event_subtype, importance, tags, detected_at, title from events where company_id = :c164 and detected_at >= :since and status in ('active', 'review') order by detected_at desc limit 500""",165 c=company_id, since=now - timedelta(days=90))166 jobs = await fetch_one(conn, """select count(*) filter (where status = 'open') as open_now, count(*) filter (where first_seen_at > :d30) as new_30d,167 count(*) filter (where first_seen_at > :d30 and is_ai) as ai_new_30d from jobs where company_id = :c""",168 c=company_id, d30=now - timedelta(days=30)) or {}169 new_job_countries = [r["country"] for r in 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) > :s",170 c=company_id, s=now - timedelta(days=90))]171 plans_new = await fetch_one(conn, "select count(*) as n from pricing_plans where company_id = :c and contact_sales and first_seen_at > :s", c=company_id, s=now - timedelta(days=30)) or {}172 loc_countries = [r["country"] for r in 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) > :s",173 c=company_id, s=now - timedelta(days=90))]174 return SignalInputs(metrics=metrics, events=events, jobs={**jobs, "new_countries_90d": new_job_countries}, plans_contact_sales_new=int(plans_new.get("n") or 0),175 locations_new_countries=loc_countries, now=now)176177178async def upsert_company_signals(conn, company_id: str, drafts: list[SignalDraft], now: datetime) -> dict[str, int]: # type: ignore[no-untyped-def]179 stats = {"inserted": 0, "updated": 0, "expired": 0}180 active = {r["kind"]: r for r in await fetch_all(conn, "select id, kind from signals where company_id = :c and scope = 'company' and status = 'active'", c=company_id)}181 seen: set[str] = set()182 for d in drafts:183 seen.add(d.kind)184 evidence = {**d.evidence, "formula_version": SIGNALS_FORMULA_VERSION}185 expires = now + timedelta(days=settings.signals_ttl_days)186 if d.kind in active:187 await execute(conn, """update signals set strength = :s, confidence = :c, title = :t, explanation = :x, evidence = cast(:e as jsonb), window_days = :w,188 expires_at = :exp where id = :id""", s=d.strength, c=d.confidence, t=d.title, x=d.explanation, e=jsonb(evidence), w=d.window_days,189 exp=expires, id=active[d.kind]["id"])190 stats["updated"] += 1191 else:192 await execute(conn, """insert into signals (id, company_id, scope, scope_key, kind, strength, confidence, title, explanation, evidence, window_days, detected_at, expires_at, status)193 values (:id, :c, 'company', null, :k, :s, :conf, :t, :x, cast(:e as jsonb), :w, :now, :exp, 'active')""",194 id=new_id("signal"), c=company_id, k=d.kind, s=d.strength, conf=d.confidence, t=d.title, x=d.explanation, e=jsonb(evidence), w=d.window_days,195 now=now, exp=expires)196 stats["inserted"] += 1197 for kind, row in active.items():198 if kind not in seen:199 await execute(conn, "update signals set status = 'expired', expires_at = least(coalesce(expires_at, :now), :now) where id = :id", now=now, id=row["id"])200 stats["expired"] += 1201 return stats202203204async def compute_scope_signals(conn, now: datetime) -> int: # type: ignore[no-untyped-def]205 """Industry / country aggregates: ≥ SCOPE_MIN_COMPANIES companies sharing an active signal kind."""206 rows = await fetch_all(conn, """207 select s.kind, co.country, coalesce(co.industry_primary, co.industries[1]) as industry, co.slug, s.strength208 from signals s join companies co on co.id = s.company_id where s.scope = 'company' and s.status = 'active' and s.expires_at > :now""", now=now)209 groups: dict[tuple[str, str, str], list[dict[str, Any]]] = {}210 for r in rows:211 if r["country"]:212 groups.setdefault(("country", str(r["country"]), r["kind"]), []).append(r)213 if r["industry"]:214 groups.setdefault(("industry", str(r["industry"]), r["kind"]), []).append(r)215 n = 0216 kept: list[str] = []217 await execute(conn, "update signals set status = 'expired' where scope in ('industry', 'country') and status = 'active' and expires_at <= :now", now=now)218 for (scope, key, kind), items in groups.items():219 if len(items) < SCOPE_MIN_COMPANIES:220 continue221 kept.append(f"{scope}:{key}:{kind}")222 label = kind.replace("_", " ")223 strength = round(min(1.0, saturate(len(items), 6.0) * (sum(float(i["strength"]) for i in items) / len(items) + 0.3)), 3)224 title = f"{label.capitalize()} signal across {len(items)} companies in {key}"225 evidence = {"companies": [i["slug"] for i in items[:50]], "count": len(items), "formula_version": SIGNALS_FORMULA_VERSION}226 existing = await fetch_one(conn, "select id from signals where scope = :s and scope_key = :k and kind = :kind and status = 'active'", s=scope, k=key, kind=kind)227 expires = now + timedelta(days=settings.signals_ttl_days)228 if existing:229 await execute(conn, "update signals set strength = :st, title = :t, evidence = cast(:e as jsonb), expires_at = :exp where id = :id",230 st=strength, t=title, e=jsonb(evidence), exp=expires, id=existing["id"])231 else:232 await execute(conn, """insert into signals (id, company_id, scope, scope_key, kind, strength, confidence, title, explanation, evidence, window_days, detected_at, expires_at)233 values (:id, null, :s, :k, :kind, :st, 0.6, :t, :x, cast(:e as jsonb), :w, :now, :exp)""",234 id=new_id("signal"), s=scope, k=key, kind=kind, st=strength, t=title, e=jsonb(evidence), w=settings.signals_window_days, now=now, exp=expires,235 x=f"{len(items)} monitored companies in this {scope} currently carry an active {label} signal.")236 n += 1237 # aggregates whose supporting companies dropped below the threshold are no longer signals238 await execute(conn, """update signals set status = 'expired' where scope in ('industry', 'country') and status = 'active'239 and (scope || ':' || scope_key || ':' || kind) <> all(cast(:kept as text[]))""", kept=kept)240 return n241242243async def compute_signals(company_ids: list[str] | None = None, *, now: datetime | None = None) -> dict[str, int]:244 now = now or datetime.now(UTC)245 stats = {"companies": 0, "inserted": 0, "updated": 0, "expired": 0, "scope": 0}246 async with transaction() as conn:247 if company_ids is None:248 rows = await fetch_all(conn, """select distinct company_id as id from events where detected_at >= :since and status in ('active', 'review')249 union select company_id from signals where scope = 'company' and status = 'active'250 union select company_id from metrics_current where metric = 'anomaly_score' and value >= :z""",251 since=now - timedelta(days=90), z=settings.anomaly_z)252 company_ids = [r["id"] for r in rows]253 for cid in company_ids:254 try:255 async with transaction() as conn:256 drafts = detect(await _inputs(conn, cid, now))257 s = await upsert_company_signals(conn, cid, drafts, now)258 stats["companies"] += 1259 for k in ("inserted", "updated", "expired"):260 stats[k] += s[k]261 except Exception:262 log.exception("signals failed", extra={"company_id": cid})263 async with transaction() as conn:264 stats["scope"] = await compute_scope_signals(conn, now)265 return stats266267268@periodic("signals", every_s=3600, initial_delay_s=120)269async def signals_task() -> None:270 log.info("signals", extra=await compute_signals())271272273__all__ = ["SIGNAL_KINDS", "SignalDraft", "SignalInputs", "compute_scope_signals", "compute_signals", "detect", "upsert_company_signals"]274