spb/company-atlas

1"""Signals: pure detectors (wording, strength, evidence) and the DB upsert/expiry/scope aggregation flow."""2from __future__ import annotations34from datetime import UTC, datetime, timedelta56import pytest7from factories import intel_db  # noqa: F401 — registers the fixture89from companyatlas.services.signals import SignalInputs, detect10from companyatlas.taxonomy import FORBIDDEN_WORDING, Metric1112NOW = datetime(2026, 9, 12, 12, tzinfo=UTC)131415def _ev(subtype: str, *, days_ago: float = 2, event_type: str | None = None, tags: list[str] | None = None, title: str = "") -> dict:16    from companyatlas.taxonomy import EVENT_SUBTYPES, EventType1718    return {"id": f"evt_{subtype}_{days_ago}", "event_type": event_type or str(EVENT_SUBTYPES.get(subtype, (EventType.OTHER, 0))[0]), "event_subtype": subtype,19            "importance": 0.6, "tags": tags or [], "detected_at": NOW - timedelta(days=days_ago), "title": title}202122def _inputs(**kw) -> SignalInputs:  # type: ignore[no-untyped-def]23    base = {"metrics": {}, "events": [], "jobs": {"open_now": 0, "new_30d": 0, "ai_new_30d": 0, "new_countries_90d": []}, "now": NOW}24    base.update(kw)25    return SignalInputs(**base)262728def _kinds(drafts):  # type: ignore[no-untyped-def]29    return {d.kind: d for d in drafts}303132def test_hiring_surge_and_freeze_wording():33    surge = _kinds(detect(_inputs(metrics={Metric.HIRING_MOMENTUM_30D: 45.0}, jobs={"open_now": 40, "new_30d": 15, "ai_new_30d": 0, "new_countries_90d": []})))34    assert "hiring_surge" in surge and surge["hiring_surge"].evidence["hiring_momentum_30d"] == 45.0 and 0 < surge["hiring_surge"].strength <= 135    freeze = _kinds(detect(_inputs(metrics={Metric.HIRING_MOMENTUM_30D: -55.0}, jobs={"open_now": 9, "new_30d": 0, "ai_new_30d": 0, "new_countries_90d": []})))36    assert "hiring_freeze" in freeze37    text = (freeze["hiring_freeze"].title + freeze["hiring_freeze"].explanation).lower()38    assert "no longer visible" in text and not any(b in text for b in FORBIDDEN_WORDING)39    assert "hiring_surge" not in freeze404142def test_launch_buildup_requires_multiple_categories_within_14_days():43    partial = detect(_inputs(events=[_ev("NEW_PRODUCT"), _ev("DOC_CHANGE")]))44    assert "launch_buildup" not in _kinds(partial)45    full = _kinds(detect(_inputs(events=[_ev("NEW_PRODUCT"), _ev("DOC_CHANGE"), _ev("CHANGELOG_ENTRY"), _ev("JOB_COUNT_INCREASE")])))46    assert "launch_buildup" in full47    d = full["launch_buildup"]48    assert d.window_days == 14 and d.title.startswith("Possible launch preparation signal") and "probabilistic" in d.explanation49    assert set(d.evidence["categories"]) == {"product", "docs", "changelog", "careers"} and len(d.evidence["event_ids"]) == 450    stale = detect(_inputs(events=[_ev("NEW_PRODUCT", days_ago=20), _ev("DOC_CHANGE", days_ago=20), _ev("CHANGELOG_ENTRY", days_ago=20), _ev("JOB_COUNT_INCREASE", days_ago=20)]))51    assert "launch_buildup" not in _kinds(stale)525354def test_expansion_pricing_developer_enterprise_ai_abnormal():55    drafts = _kinds(detect(_inputs(56        metrics={Metric.DEVELOPER_MOMENTUM: 70.0, Metric.ANOMALY_SCORE: 3.1},57        events=[_ev("COUNTRY_EXPANSION"), _ev("NEW_PRICING_TIER"), _ev("PRICING_TIER_REMOVED"), _ev("API_CHANGE"), _ev("DOC_CHANGE"), _ev("CHANGELOG_ENTRY"),58                _ev("AI_HIRING"), _ev("NEWS_RELEASE", tags=["ai"]), _ev("MESSAGING_CHANGE", title="Now with SSO and audit logs for enterprise")],59        jobs={"open_now": 20, "new_30d": 8, "ai_new_30d": 4, "new_countries_90d": ["JP"]}, plans_contact_sales_new=1, locations_new_countries=["JP"])))60    assert {"expansion", "pricing_migration", "developer_push", "enterprise_repositioning", "ai_acceleration", "abnormal_activity"} <= set(drafts)61    assert drafts["expansion"].evidence["new_countries"] == ["JP"]62    assert drafts["pricing_migration"].evidence["subtypes"] == ["NEW_PRICING_TIER", "PRICING_TIER_REMOVED"]63    assert drafts["ai_acceleration"].evidence["ai_new_30d"] == 4 and "public job titles" in drafts["ai_acceleration"].explanation64    assert drafts["abnormal_activity"].window_days == 7 and drafts["abnormal_activity"].evidence["anomaly_z"] == 3.165    for d in drafts.values():66        assert 0 < d.strength <= 1 and 0 < d.confidence <= 1 and "signal" in d.title.lower()676869def test_sparse_inputs_produce_nothing():70    assert detect(_inputs()) == []71    assert detect(_inputs(metrics={Metric.ANOMALY_SCORE: 1.0}, events=[_ev("BLOG_POST")])) == []727374@pytest.mark.usefixtures("intel_db")75async def test_signal_upsert_expiry_and_scope_aggregate():76    from factories import cleanup, make_company, make_event7778    from companyatlas.db import execute, fetch_all, transaction79    from companyatlas.services.signals import compute_signals8081    try:82        async with transaction() as conn:83            companies = [await make_company(conn, country="CA", industries=["fintech"]) for _ in range(3)]84            for co in companies:85                for st in ("NEW_PRICING_TIER", "PRICING_TIER_REMOVED"):86                    await make_event(conn, co, subtype=st, days_ago=1)87        ids = [c["id"] for c in companies]88        first = await compute_signals(ids)89        assert first["inserted"] == 3 and first["scope"] >= 190        second = await compute_signals(ids)91        assert second["inserted"] == 0 and second["updated"] == 3           # idempotent: same signal updated, not duplicated92        async with transaction() as conn:93            rows = await fetch_all(conn, "select scope, scope_key, kind, status, evidence from signals where company_id = any(cast(:ids as text[])) or (scope <> 'company' and scope_key = 'CA' and kind = 'pricing_migration')", ids=ids)94            company_rows = [r for r in rows if r["scope"] == "company"]95            assert len(company_rows) == 3 and all(r["kind"] == "pricing_migration" and r["status"] == "active" for r in company_rows)96            scope_rows = [r for r in rows if r["scope"] == "country"]97            assert scope_rows and scope_rows[0]["evidence"]["count"] >= 398            # evidence disappears → signal expires99            await execute(conn, "update events set status = 'retracted' where company_id = :c", c=ids[0])100        third = await compute_signals([ids[0]])101        assert third["expired"] == 1102        async with transaction() as conn:103            st = await fetch_all(conn, "select status from signals where company_id = :c", c=ids[0])104            assert [r["status"] for r in st] == ["expired"]105            await execute(conn, "delete from signals where scope = 'country' and scope_key = 'CA' and kind = 'pricing_migration'")106    finally:107        await cleanup()108