spb/company-atlas

1"""Metrics: pure formulas (reproducible, coverage-normalised) and the DB flow (metrics_current / metric_series / company_daily / global_daily)."""2from __future__ import annotations34from datetime import UTC, date, datetime, timedelta56import pytest7from factories import intel_db  # noqa: F401 — registers the fixture89from companyatlas.services.metrics import (10    activity_score,11    ai_adoption,12    anomaly_score,13    corporate_change_index,14    hiring_momentum,15    historical_coverage,16    saturate,17)18from companyatlas.taxonomy import CCI_FORMULA_VERSION, METRICS_FORMULA_VERSION, Metric1920# ------------------------------------------------------------------------------------------------------------ pure212223def test_activity_score_coverage_normalisation():24    changes = [(1.0, "meaningful")] * 1025    small = activity_score(changes, [], active_sensors=2)26    big = activity_score(changes, [], active_sensors=32)27    assert small and big28    assert small.value > big.value                                  # same raw activity spread over 16× more sensors → lower density29    assert small.inputs["active_sensors"] == 2 and small.formula_version == METRICS_FORMULA_VERSION30    assert 0 < big.value < small.value <= 100313233def test_activity_score_decay_and_weights():34    fresh = activity_score([(0.0, "critical")], [], 1)35    old = activity_score([(30.0, "critical")], [], 1)36    minor_kind = activity_score([(0.0, "meaningful")], [], 1)37    assert fresh.value > old.value > 038    assert fresh.value > minor_kind.value39    assert activity_score([], [], 0) is None                         # no sensors, no activity → no row404142def test_hiring_momentum_requires_listings():43    assert hiring_momentum(10, 2, window=30) is None                 # < 3 listings at the reference point44    m = hiring_momentum(12, 8, window=30, extra={"jobs_new_30d": 5})45    assert m.metric == Metric.HIRING_MOMENTUM_30D and m.value == 50.0 and m.inputs["jobs_new_30d"] == 546    assert hiring_momentum(5, 10, window=7).value == -50.0474849def test_ai_adoption_renormalises_over_available_inputs():50    jobs_only = ai_adoption(ai_open=5, open_jobs=10, ai_events_90d=0, keyword_hits=0, has_text_inputs=False)51    assert jobs_only.value == pytest.approx(100 * (0.5 * 1.0 + 0.25 * 0.0) / 0.75, abs=0.01)52    none = ai_adoption(ai_open=None, open_jobs=None, ai_events_90d=0, keyword_hits=0, has_text_inputs=False)53    assert none is None54    text = ai_adoption(ai_open=None, open_jobs=None, ai_events_90d=3, keyword_hits=5, has_text_inputs=True)55    assert 0 < text.value < 100 and set(text.inputs["weights_used"]) == {"events", "keywords"}565758def test_cci_renormalises_and_maps_momentum():59    full = corporate_change_index({Metric.HIRING_MOMENTUM_30D: 50.0, Metric.PRODUCT_VELOCITY: 80.0, Metric.GEO_EXPANSION: 20.0, Metric.LEADERSHIP_ACTIVITY: 0.0,60                                   Metric.DEVELOPER_MOMENTUM: 40.0, Metric.COMMUNICATION_ACTIVITY: 60.0, Metric.PRICING_ACTIVITY: 10.0})61    expected = 0.25 * 75 + 0.20 * 80 + 0.15 * 20 + 0.15 * 0 + 0.10 * 40 + 0.10 * 60 + 0.05 * 1062    assert full.value == pytest.approx(expected, abs=0.01) and full.formula_version == CCI_FORMULA_VERSION63    partial = corporate_change_index({Metric.PRODUCT_VELOCITY: 80.0, Metric.PRICING_ACTIVITY: 20.0})64    assert partial.value == pytest.approx((0.20 * 80 + 0.05 * 20) / 0.25, abs=0.01)65    assert partial.inputs["weight_coverage"] == 0.25 and partial.confidence < full.confidence66    assert corporate_change_index({}) is None676869def test_anomaly_and_coverage():70    assert anomaly_score(10, 2.0, 1.0, samples=2) is None71    z = anomaly_score(10, 2.0, 1.0, samples=8)72    assert z.value == 8.073    flat = anomaly_score(3, 0.0, 0.0, samples=8)74    assert flat.value == 6.0                                          # stddev floor 0.575    cov = historical_coverage(observed=50, expected=100.0, days_with_obs=15, days_since_first=30, surfaces=4)76    assert cov.value == pytest.approx(100 * (0.5 * 0.5 + 0.3 * 0.5 + 0.2 * 0.5), abs=0.01)77    assert historical_coverage(observed=0, expected=0.0, days_with_obs=0, days_since_first=0, surfaces=0) is None78    assert saturate(6.0, 6.0) == pytest.approx(0.632, abs=0.001)798081# ------------------------------------------------------------------------------------------------------------ database828384@pytest.mark.usefixtures("intel_db")85async def test_company_metrics_end_to_end():86    from factories import (87        CONNECTOR_ATS,88        cleanup,89        make_change,90        make_company,91        make_event,92        make_job,93        make_location,94        make_observation,95        make_sensor,96    )9798    from companyatlas.db import fetch_all, transaction99    from companyatlas.services.metrics import compute_company_metrics100101    try:102        async with transaction() as conn:103            co = await make_company(conn, first_observed_days_ago=40)104            board = await make_sensor(conn, co, "jobs_board", connector_id=CONNECTOR_ATS, created_days_ago=40)105            await make_sensor(conn, co, "pricing", created_days_ago=40)106            await make_sensor(conn, co, "docs", created_days_ago=40)107            await make_sensor(conn, co, "locations", created_days_ago=40)108            # jobs: 8 open now, 4 of them existed 30 days ago (+ 1 removed since) → momentum_30d = (8 − 5) / 5 = +60 %109            for i in range(4):110                await make_job(conn, co, title=f"Engineer {i}", first_seen_days_ago=45, country=None, sensor_id=board["id"])111            await make_job(conn, co, title="Old role", first_seen_days_ago=45, removed_days_ago=10, country=None, sensor_id=board["id"])112            for i in range(4):113                await make_job(conn, co, title=f"ML Engineer {i}", first_seen_days_ago=5, is_ai=True, country="JP", sensor_id=board["id"])114            await make_location(conn, co, name="Tokyo office", country="JP", city="Tokyo", first_seen_days_ago=3)115            await make_event(conn, co, subtype="NEW_PRODUCT", importance=0.7, days_ago=2)116            await make_event(conn, co, subtype="DOC_CHANGE", importance=0.4, days_ago=3)117            await make_event(conn, co, subtype="PRICE_INCREASE", importance=0.8, days_ago=4)118            await make_event(conn, co, subtype="COUNTRY_EXPANSION", importance=0.8, days_ago=3)119            await make_event(conn, co, subtype="AI_HIRING", importance=0.5, days_ago=5, tags=["hiring", "ai"])120            await make_change(conn, board, significance=0.6, detected_at=datetime.now(UTC) - timedelta(days=1), status="processed")121            for d in range(0, 40, 2):122                await make_observation(conn, board, days_ago=d)123        stats = await compute_company_metrics([co["id"]])124        assert stats["companies"] == 1125        async with transaction() as conn:126            rows = {r["metric"]: r for r in await fetch_all(conn, "select * from metrics_current where company_id = :c", c=co["id"])}127            series = await fetch_all(conn, "select metric from metric_series where company_id = :c and day = current_date", c=co["id"])128        assert set(series and [r["metric"] for r in series]) == set(rows)129        assert rows["open_jobs"]["value"] == 8130        assert rows["hiring_momentum_30d"]["value"] == pytest.approx(60.0)131        assert rows["hiring_momentum_30d"]["inputs"]["jobs_new_30d"] == 4 and rows["hiring_momentum_30d"]["confidence"] == pytest.approx(0.9, abs=0.01)132        assert rows["hiring_momentum_7d"]["value"] == pytest.approx(100.0)  # 4 open 7 days ago (removed role already gone), 4 new since133        assert "hiring_momentum_90d" not in rows                          # nothing was open 90 days ago → no fabricated momentum134        assert rows["ai_adoption"]["value"] > 40 and rows["ai_adoption"]["inputs"]["ai_open"] == 4135        assert rows["geo_expansion"]["inputs"]["new_countries_90d"] == ["JP"] and rows["geo_expansion"]["value"] > 0136        assert rows["product_velocity"]["value"] > 0 and rows["pricing_activity"]["value"] > 0 and rows["developer_momentum"]["value"] > 0137        assert "leadership_activity" not in rows                          # no leadership sensor and no leadership events → no row138        cci = rows["corporate_change_index"]139        assert cci["formula_version"] == CCI_FORMULA_VERSION and 0 < cci["value"] <= 100 and "leadership_activity" not in cci["inputs"]["components"]140        assert rows["activity_score"]["value"] > 0 and rows["activity_score"]["inputs"]["active_sensors"] == 4141        assert 0 < rows["historical_coverage"]["value"] <= 100142        assert "anomaly_score" not in rows                                # no baseline yet143    finally:144        await cleanup()145146147@pytest.mark.usefixtures("intel_db")148async def test_daily_aggregates_and_activity_index():149    from factories import cleanup, make_change, make_company, make_event, make_observation, make_sensor150151    from companyatlas.db import fetch_all, fetch_one, transaction152    from companyatlas.services.metrics import compute_daily153154    day0 = date(2001, 1, 10)                                          # far in the past: never collides with live data, cleaned by factories155    try:156        async with transaction() as conn:157            a = await make_company(conn, country="CA", industries=["fintech"])158            b = await make_company(conn, country="US", industries=["retail"])159            sa = await make_sensor(conn, a, "homepage")160            sb = await make_sensor(conn, b, "homepage")161            for i in range(7):162                d = datetime(2001, 1, 4 + i, 12, tzinfo=UTC)163                for s in (sa, sb):164                    await make_observation(conn, s, days_ago=(datetime.now(UTC) - d).total_seconds() / 86400)165                if i < 6:166                    await make_change(conn, sa, significance=0.6, detected_at=d, status="processed")167            # day0: burst on company a168            for _ in range(5):169                await make_change(conn, sa, significance=0.7, detected_at=datetime(2001, 1, 10, 13, tzinfo=UTC), status="processed")170            await make_event(conn, a, subtype="PRICE_INCREASE", days_ago=(datetime.now(UTC) - datetime(2001, 1, 10, 14, tzinfo=UTC)).total_seconds() / 86400)171        for i in range(6):172            await compute_daily(date(2001, 1, 4 + i))173        result = await compute_daily(day0)174        assert result["companies"] == 2 and result["meaningful_changes"] == 5 and result["events"] == 1175        assert result["baseline_days"] == 6176        assert result["activity_index"] == pytest.approx(500.0)     # 5 changes / 2 sensors vs baseline 1 change / 2 sensors → ×100177        async with transaction() as conn:178            g = await fetch_one(conn, "select * from global_daily where day = :d", d=day0)179            assert g["sensors_active"] == 2 and g["companies_active"] == 2 and g["events_by_type"] == {"PRICING": 1}180            assert g["by_country"]["CA"]["meaningful_changes"] == 5 and g["by_industry"]["fintech"]["events"] == 1181            cd = {r["company_id"]: r for r in await fetch_all(conn, "select * from company_daily where day = :d", d=day0)}182            assert cd[a["id"]]["meaningful_changes"] == 5 and cd[b["id"]]["meaningful_changes"] == 0 and cd[b["id"]]["observations"] == 1183    finally:184        await cleanup()185