"""Platform aggregates: `/stats`, `/stats/history`, `/system`, `/pulse`, `/index`, `/methodology`.""" from __future__ import annotations from typing import Any from fastapi import APIRouter, Query, Request from companyatlas.api import aggregates as agg from companyatlas.api import queries as q from companyatlas.api.common import cached, cached_response from companyatlas.config import settings from companyatlas.db import connection from companyatlas.taxonomy import CCI_FORMULA_VERSION, CCI_WEIGHTS, EVENT_SUBTYPES, METRICS_FORMULA_VERSION, EventType, Metric ORDER = 10 router = APIRouter(prefix="/api/v1", tags=["platform"]) LAUNCH_SUBTYPES = ["PRODUCT_LAUNCH", "NEW_PRODUCT", "FEATURE_LAUNCH", "AI_LAUNCH", "API_LAUNCH", "SDK_RELEASE"] @router.get("/stats", summary="Platform counters (cached 60 s)") async def stats(request: Request) -> Any: return cached_response(request, await agg.cached_global_stats(), 60) @router.get("/stats/history", summary="Global daily history") async def stats_history(request: Request, days: int = Query(90, ge=1, le=730)) -> Any: async def produce() -> dict[str, Any]: async with connection() as conn: return {"items": await agg.global_daily_rows(conn, days)} return cached_response(request, await cached(f"stats:history:{days}", 300, produce), 300) @router.get("/system", summary="Public aggregate health") async def system(request: Request) -> Any: async def produce() -> dict[str, Any]: async with connection() as conn: return await agg.system_health(conn) return cached_response(request, await cached("system", 30, produce), 30) @router.get("/index", summary="Global Corporate Activity Index") async def index(request: Request, days: int = Query(365, ge=30, le=1095)) -> Any: async def produce() -> dict[str, Any]: async with connection() as conn: return await agg.activity_index(conn, days) return cached_response(request, await cached(f"index:{days}", 300, produce), 300) @router.get("/pulse", summary="Homepage aggregate (cached 60 s)") async def pulse(request: Request) -> Any: async def produce() -> dict[str, Any]: stats_payload = await agg.cached_global_stats() async with connection() as conn: live = await agg.live_events(conn, 12) movers = await agg.ranking_cards(conn, "most_active", "7d", limit=10, sparkline=True) hiring = await agg.ranking_cards(conn, "hiring_growth", "30d", limit=8) launches = await agg.live_events(conn, 8, event_subtypes=LAUNCH_SUBTYPES) pricing = await agg.live_events(conn, 8, event_type="PRICING") ai = await agg.ranking_cards(conn, "ai_active", "30d", limit=8) trending = await agg.trend_rows(conn, 7, 10) idx = await agg.activity_index(conn, 30) buckets = await agg.map_buckets(conn, "events_30d") industries = (await agg.cached_industry_rows())[:12] countries = (await agg.cached_country_rows())[:12] return {"stats": stats_payload, "live": live, "movers": movers, "hiring": hiring, "launches": launches, "pricing": pricing, "ai": ai, "industries": industries, "countries": countries, "trending": trending, "activity_index": {"value": idx["value"], "delta_7d": idx["delta_7d"], "series": idx["series"][-30:]}, "map": buckets[:300]} return cached_response(request, await cached("pulse", 60, produce), 60) METRIC_DOCS: dict[str, tuple[str, list[str]]] = { Metric.ACTIVITY_SCORE: (("Company Activity Score 0–100: weighted volume and significance of detected changes across monitored surfaces (website, products, " "news, jobs, leadership, documentation, pricing) over rolling windows, normalised by sensor coverage."), ["changes", "meaningful_changes", "events", "surfaces", "sensor_coverage"]), Metric.HIRING_MOMENTUM_7D: ("Hiring momentum over 7 days: net change of publicly listed open positions (%); negative when listings disappear.", ["jobs_open", "jobs_new", "jobs_removed"]), Metric.HIRING_MOMENTUM_30D: ("Hiring momentum over 30 days (%).", ["jobs_open", "jobs_new", "jobs_removed"]), Metric.HIRING_MOMENTUM_90D: ("Hiring momentum over 90 days (%).", ["jobs_open", "jobs_new", "jobs_removed"]), Metric.OPEN_JOBS: ("Number of publicly listed open positions currently observed on monitored career surfaces.", ["jobs_open"]), Metric.AI_ADOPTION: (("AI Adoption Score 0–100 from observable public signals only: AI products, AI job postings, documentation, marketing, " "partnerships, research and leadership roles. Never claims internal use without evidence."), ["ai_jobs", "ai_products", "ai_docs", "ai_news", "ai_leadership"]), Metric.PRODUCT_VELOCITY: ("Product Velocity 0–100: cadence of product launches, updates, renames and removals detected on product surfaces.", ["product_events", "changelog_entries", "feature_launches"]), Metric.GEO_EXPANSION: ("Geographic Expansion 0–100: new locations, new countries and geographic hiring spread.", ["new_locations", "new_countries", "job_countries"]), Metric.DEVELOPER_MOMENTUM: ("Developer Momentum 0–100: API/SDK launches, documentation and changelog activity.", ["developer_events", "doc_changes", "api_changes"]), Metric.COMMUNICATION_ACTIVITY: ("Communication Activity 0–100: newsroom, blog and feed publication cadence.", ["news_items", "blog_posts"]), Metric.PRICING_ACTIVITY: ("Pricing Activity 0–100: detected plan and price changes.", ["pricing_events"]), Metric.LEADERSHIP_ACTIVITY: ("Leadership Activity 0–100: additions, removals and title changes on monitored leadership pages.", ["leadership_events"]), Metric.CORPORATE_CHANGE_INDEX: (f"Corporate Change Index: weighted composite — {', '.join(f'{k} {v:.2f}' for k, v in CCI_WEIGHTS.items())}.", list(CCI_WEIGHTS)), Metric.ANOMALY_SCORE: ("Unusual activity 0–100: z-score of current activity against the company's own baseline (56-day window).", ["baseline_mean", "baseline_stddev", "current"]), Metric.HISTORICAL_COVERAGE: ("Historical Completeness 0–100: sensor uptime, continuity and surface coverage of the record.", ["sensor_uptime", "continuity", "surface_coverage", "failed_periods"]), } @router.get("/methodology", summary="How metrics, significance and confidence are computed") async def methodology(request: Request) -> Any: payload = { "metrics": [{"metric": m.value, "formula_version": CCI_FORMULA_VERSION if m is Metric.CORPORATE_CHANGE_INDEX else METRICS_FORMULA_VERSION, "description": METRIC_DOCS[m][0], "inputs": METRIC_DOCS[m][1]} for m in Metric], "significance_bands": {"noise": [0.0, settings.noise_threshold], "minor": [settings.noise_threshold, settings.meaningful_threshold], "meaningful": [settings.meaningful_threshold, settings.major_threshold], "major": [settings.major_threshold, settings.critical_threshold], "critical": [settings.critical_threshold, 1.0]}, "event_types": [t.value for t in EventType], "event_subtypes": [{"event_subtype": k, "event_type": v[0].value, "default_importance": v[1]} for k, v in EVENT_SUBTYPES.items()], "confidence_labels": [{"label": "VERIFIED", "min_confidence": 0.95}, {"label": "HIGH_CONFIDENCE", "min_confidence": 0.85}, {"label": "LIKELY", "min_confidence": 0.7}, {"label": "INFERRED", "min_confidence": 0.5}, {"label": "LOW_CONFIDENCE", "min_confidence": 0.0}], "windows": list(q.WINDOWS), "cci_weights": dict(CCI_WEIGHTS), "baseline_window_days": settings.baseline_window_days, "anomaly_z": settings.anomaly_z, "language": "Interpretive language is careful by design: events say 'no longer listed', never 'fired'; inferred facts carry a confidence label.", } return cached_response(request, payload, 3600)