"""Cross-signal detection (spec §92–93): hiring surge / freeze, launch build-up, international expansion, pricing migration, developer push, enterprise repositioning, AI acceleration, abnormal activity. Signals are *labelled as signals*, carry a strength (0–1), a confidence, an explanation and the evidence (event ids, metric values) that produced them, and expire. They never assert facts. Company-scope signals are recomputed hourly for companies with recent activity; industry/country aggregates are derived from them. """ from __future__ import annotations import logging from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta from typing import Any from companyatlas.config import settings from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction from companyatlas.ids import new_id from companyatlas.services.metrics import saturate from companyatlas.services.periodic import periodic from companyatlas.taxonomy import SIGNAL_KINDS, SIGNALS_FORMULA_VERSION, EventType, Metric log = logging.getLogger(__name__) SCOPE_MIN_COMPANIES = 3 LAUNCH_WINDOW_DAYS = 14 SURGE_MOMENTUM_PCT = 30.0 FREEZE_MOMENTUM_PCT = -30.0 MIN_OPEN_FOR_MOMENTUM = 10 ENTERPRISE_RE = ("enterprise", "contact sales", "sso", "saml", "audit log", "dedicated", "custom pricing", "procurement", "soc 2", "soc2") @dataclass(slots=True) class SignalDraft: kind: str strength: float confidence: float title: str explanation: str evidence: dict[str, Any] = field(default_factory=dict) window_days: int = 30 @dataclass(slots=True) class SignalInputs: """Everything the detectors look at for one company (pure, testable).""" metrics: dict[str, float] events: list[dict[str, Any]] # {id, event_type, event_subtype, importance, tags, detected_at, title} jobs: dict[str, Any] # open_now, new_30d, ai_new_30d, new_countries_90d plans_contact_sales_new: int = 0 locations_new_countries: list[str] = field(default_factory=list) now: datetime = field(default_factory=lambda: datetime.now(UTC)) def _age(now: datetime, at: datetime) -> float: if at.tzinfo is None: at = at.replace(tzinfo=UTC) return (now - at).total_seconds() / 86400.0 def _ids(events: list[dict[str, Any]]) -> list[str]: return [e["id"] for e in events[:25]] def detect(inp: SignalInputs) -> list[SignalDraft]: m, ev, j, now = inp.metrics, inp.events, inp.jobs, inp.now out: list[SignalDraft] = [] window = settings.signals_window_days recent = [e for e in ev if _age(now, e["detected_at"]) <= window] open_now = int(j.get("open_now") or 0) mom30 = m.get(Metric.HIRING_MOMENTUM_30D) # hiring surge -------------------------------------------------------------------------------------------------- surge_events = [e for e in recent if e["event_subtype"] in ("HIRING_SURGE", "JOB_COUNT_INCREASE")] 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): strength = saturate(max(mom30 or 0, 30.0) / 50.0, 1.5) out.append(SignalDraft("hiring_surge", round(strength, 3), 0.7, "Hiring surge signal", f"Open listings observed up {mom30:.0f}% over 30 days ({open_now} open)." if mom30 is not None else f"{len(surge_events)} hiring increase events detected in the last {window} days.", {"event_ids": _ids(surge_events), "hiring_momentum_30d": mom30, "open_jobs": open_now})) # hiring freeze — careful wording: listings no longer visible -------------------------------------------------- freeze_events = [e for e in recent if e["event_subtype"] in ("HIRING_FREEZE_SIGNAL", "JOB_COUNT_DECREASE")] 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): strength = saturate(abs(min(mom30 or -30.0, -30.0)) / 50.0, 1.5) out.append(SignalDraft("hiring_freeze", round(strength, 3), 0.6, "Hiring slowdown signal: listings no longer visible", (f"Monitored open listings are down {abs(mom30):.0f}% over 30 days with {int(j.get('new_30d') or 0)} new listings; " "listings no longer visible on public pages are not evidence of workforce decisions.") if mom30 is not None else f"{len(freeze_events)} events of listings no longer visible in the last {window} days.", {"event_ids": _ids(freeze_events), "hiring_momentum_30d": mom30, "jobs_new_30d": j.get("new_30d")})) # launch build-up ----------------------------------------------------------------------------------------------- win14 = [e for e in ev if _age(now, e["detected_at"]) <= LAUNCH_WINDOW_DAYS] cats = { "product": [e for e in win14 if e["event_type"] == EventType.PRODUCT], "docs": [e for e in win14 if e["event_subtype"] in ("DOC_CHANGE", "DOCUMENTATION_CHANGE", "API_CHANGE", "API_LAUNCH", "SDK_RELEASE")], "changelog": [e for e in win14 if e["event_subtype"] == "CHANGELOG_ENTRY"], "careers": [e for e in win14 if e["event_type"] == EventType.HIRING and e["event_subtype"] != "JOB_COUNT_DECREASE"], "messaging": [e for e in win14 if e["event_subtype"] in ("MESSAGING_CHANGE", "HOMEPAGE_REDESIGN", "WEBSITE_CHANGE")], } present = {k: v for k, v in cats.items() if v} if len(present) >= 3 and cats["product"] or len(present) >= 4: strength = min(1.0, 0.3 + 0.15 * len(present) + 0.05 * min(6, sum(len(v) for v in present.values()))) out.append(SignalDraft("launch_buildup", round(strength, 3), round(0.45 + 0.08 * len(present), 2), "Possible launch preparation signal", "Within 14 days: " + ", ".join(f"{len(v)} {k}" for k, v in present.items()) + " events detected on public pages. " "This pattern often precedes announcements; it is a probabilistic signal, not a confirmation.", {"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)) # expansion ----------------------------------------------------------------------------------------------------- geo_events = [e for e in recent if e["event_subtype"] in ("COUNTRY_EXPANSION", "NEW_LOCATION", "NEW_OFFICE")] new_countries = sorted(set(inp.locations_new_countries) | set(j.get("new_countries_90d") or [])) 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)): strength = min(1.0, 0.4 + 0.2 * len(new_countries) + 0.05 * len(geo_events)) out.append(SignalDraft("expansion", round(strength, 3), 0.65, "International expansion signal", (f"New country presence listed: {', '.join(new_countries[:5])}. " if new_countries else "") + 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 "."), {"event_ids": _ids(geo_events), "new_countries": new_countries, "geo_expansion": m.get(Metric.GEO_EXPANSION)})) # pricing migration --------------------------------------------------------------------------------------------- pricing = [e for e in recent if e["event_type"] == EventType.PRICING] subtypes = {e["event_subtype"] for e in pricing} if len(pricing) >= 2 or ({"NEW_PRICING_TIER", "PRICING_TIER_REMOVED"} <= subtypes): strength = min(1.0, 0.3 + 0.15 * len(pricing) + (0.2 if {"NEW_PRICING_TIER", "PRICING_TIER_REMOVED"} <= subtypes else 0)) out.append(SignalDraft("pricing_migration", round(strength, 3), 0.7, "Pricing migration signal", f"{len(pricing)} pricing events in {window} days: " + ", ".join(sorted(subtypes)).lower().replace("_", " ") + ".", {"event_ids": _ids(pricing), "subtypes": sorted(subtypes), "pricing_activity": m.get(Metric.PRICING_ACTIVITY)})) # developer push ------------------------------------------------------------------------------------------------ dev = [e for e in recent if e["event_type"] == EventType.DEVELOPER] dev_m = m.get(Metric.DEVELOPER_MOMENTUM) if len(dev) >= 3 or (dev_m is not None and dev_m >= 60): strength = saturate(len(dev) + (dev_m or 0) / 40.0, 4.0) out.append(SignalDraft("developer_push", round(strength, 3), 0.7, "Developer push signal", 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 "."), {"event_ids": _ids(dev), "developer_momentum": dev_m})) # enterprise repositioning -------------------------------------------------------------------------------------- ent_events = [e for e in recent if e["event_subtype"] == "ENTERPRISE_REPOSITIONING" or "enterprise" in (e.get("tags") or []) or any(k in (e.get("title") or "").lower() for k in ENTERPRISE_RE)] if inp.plans_contact_sales_new or len(ent_events) >= 2: strength = min(1.0, 0.35 + 0.25 * min(2, inp.plans_contact_sales_new) + 0.1 * len(ent_events)) out.append(SignalDraft("enterprise_repositioning", round(strength, 3), 0.55, "Enterprise repositioning signal", (f"{inp.plans_contact_sales_new} new contact-sales / enterprise pricing tier(s) listed. " if inp.plans_contact_sales_new else "") + (f"{len(ent_events)} events mention enterprise features or messaging." if ent_events else ""), {"event_ids": _ids(ent_events), "contact_sales_tiers_new": inp.plans_contact_sales_new})) # AI acceleration ----------------------------------------------------------------------------------------------- ai_events = [e for e in recent if e["event_subtype"] in ("AI_HIRING", "AI_LAUNCH") or "ai" in (e.get("tags") or [])] ai_new = int(j.get("ai_new_30d") or 0) new30 = int(j.get("new_30d") or 0) ai_share = (ai_new / new30) if new30 else 0.0 if len(ai_events) >= 2 or (new30 >= 4 and ai_share >= 0.25): strength = min(1.0, 0.3 + 0.1 * len(ai_events) + ai_share) out.append(SignalDraft("ai_acceleration", round(strength, 3), 0.6, "AI acceleration signal", 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 ".") + " Based on public job titles, product names and announcements only.", {"event_ids": _ids(ai_events), "ai_new_30d": ai_new, "jobs_new_30d": new30, "ai_adoption": m.get(Metric.AI_ADOPTION)})) # abnormal activity --------------------------------------------------------------------------------------------- z = m.get(Metric.ANOMALY_SCORE) if z is not None and z >= settings.anomaly_z: out.append(SignalDraft("abnormal_activity", round(min(1.0, z / (settings.anomaly_z * 2)), 3), 0.7, "Unusual activity signal", f"This week's meaningful changes are {z:.1f} standard deviations above this company's baseline.", {"anomaly_z": z, "event_ids": _ids(recent[:10])}, 7)) return out # ================================================================================================================ persistence async def _inputs(conn, company_id: str, now: datetime) -> SignalInputs: # type: ignore[no-untyped-def] 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)} events = await fetch_all(conn, """select id, event_type, event_subtype, importance, tags, detected_at, title from events where company_id = :c and detected_at >= :since and status in ('active', 'review') order by detected_at desc limit 500""", c=company_id, since=now - timedelta(days=90)) jobs = await fetch_one(conn, """select count(*) filter (where status = 'open') as open_now, count(*) filter (where first_seen_at > :d30) as new_30d, count(*) filter (where first_seen_at > :d30 and is_ai) as ai_new_30d from jobs where company_id = :c""", c=company_id, d30=now - timedelta(days=30)) or {} 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", c=company_id, s=now - timedelta(days=90))] 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 {} 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", c=company_id, s=now - timedelta(days=90))] 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), locations_new_countries=loc_countries, now=now) async def upsert_company_signals(conn, company_id: str, drafts: list[SignalDraft], now: datetime) -> dict[str, int]: # type: ignore[no-untyped-def] stats = {"inserted": 0, "updated": 0, "expired": 0} 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)} seen: set[str] = set() for d in drafts: seen.add(d.kind) evidence = {**d.evidence, "formula_version": SIGNALS_FORMULA_VERSION} expires = now + timedelta(days=settings.signals_ttl_days) if d.kind in active: await execute(conn, """update signals set strength = :s, confidence = :c, title = :t, explanation = :x, evidence = cast(:e as jsonb), window_days = :w, 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, exp=expires, id=active[d.kind]["id"]) stats["updated"] += 1 else: 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) values (:id, :c, 'company', null, :k, :s, :conf, :t, :x, cast(:e as jsonb), :w, :now, :exp, 'active')""", 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, now=now, exp=expires) stats["inserted"] += 1 for kind, row in active.items(): if kind not in seen: await execute(conn, "update signals set status = 'expired', expires_at = least(coalesce(expires_at, :now), :now) where id = :id", now=now, id=row["id"]) stats["expired"] += 1 return stats async def compute_scope_signals(conn, now: datetime) -> int: # type: ignore[no-untyped-def] """Industry / country aggregates: ≥ SCOPE_MIN_COMPANIES companies sharing an active signal kind.""" rows = await fetch_all(conn, """ select s.kind, co.country, coalesce(co.industry_primary, co.industries[1]) as industry, co.slug, s.strength 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) groups: dict[tuple[str, str, str], list[dict[str, Any]]] = {} for r in rows: if r["country"]: groups.setdefault(("country", str(r["country"]), r["kind"]), []).append(r) if r["industry"]: groups.setdefault(("industry", str(r["industry"]), r["kind"]), []).append(r) n = 0 kept: list[str] = [] await execute(conn, "update signals set status = 'expired' where scope in ('industry', 'country') and status = 'active' and expires_at <= :now", now=now) for (scope, key, kind), items in groups.items(): if len(items) < SCOPE_MIN_COMPANIES: continue kept.append(f"{scope}:{key}:{kind}") label = kind.replace("_", " ") strength = round(min(1.0, saturate(len(items), 6.0) * (sum(float(i["strength"]) for i in items) / len(items) + 0.3)), 3) title = f"{label.capitalize()} signal across {len(items)} companies in {key}" evidence = {"companies": [i["slug"] for i in items[:50]], "count": len(items), "formula_version": SIGNALS_FORMULA_VERSION} 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) expires = now + timedelta(days=settings.signals_ttl_days) if existing: await execute(conn, "update signals set strength = :st, title = :t, evidence = cast(:e as jsonb), expires_at = :exp where id = :id", st=strength, t=title, e=jsonb(evidence), exp=expires, id=existing["id"]) else: await execute(conn, """insert into signals (id, company_id, scope, scope_key, kind, strength, confidence, title, explanation, evidence, window_days, detected_at, expires_at) values (:id, null, :s, :k, :kind, :st, 0.6, :t, :x, cast(:e as jsonb), :w, :now, :exp)""", 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, x=f"{len(items)} monitored companies in this {scope} currently carry an active {label} signal.") n += 1 # aggregates whose supporting companies dropped below the threshold are no longer signals await execute(conn, """update signals set status = 'expired' where scope in ('industry', 'country') and status = 'active' and (scope || ':' || scope_key || ':' || kind) <> all(cast(:kept as text[]))""", kept=kept) return n async def compute_signals(company_ids: list[str] | None = None, *, now: datetime | None = None) -> dict[str, int]: now = now or datetime.now(UTC) stats = {"companies": 0, "inserted": 0, "updated": 0, "expired": 0, "scope": 0} async with transaction() as conn: if company_ids is None: rows = await fetch_all(conn, """select distinct company_id as id from events where detected_at >= :since and status in ('active', 'review') union select company_id from signals where scope = 'company' and status = 'active' union select company_id from metrics_current where metric = 'anomaly_score' and value >= :z""", since=now - timedelta(days=90), z=settings.anomaly_z) company_ids = [r["id"] for r in rows] for cid in company_ids: try: async with transaction() as conn: drafts = detect(await _inputs(conn, cid, now)) s = await upsert_company_signals(conn, cid, drafts, now) stats["companies"] += 1 for k in ("inserted", "updated", "expired"): stats[k] += s[k] except Exception: log.exception("signals failed", extra={"company_id": cid}) async with transaction() as conn: stats["scope"] = await compute_scope_signals(conn, now) return stats @periodic("signals", every_s=3600, initial_delay_s=120) async def signals_task() -> None: log.info("signals", extra=await compute_signals()) __all__ = ["SIGNAL_KINDS", "SignalDraft", "SignalInputs", "compute_scope_signals", "compute_signals", "detect", "upsert_company_signals"]