"""Cost rollup↔source drift reconciliation (ai_spending domain). Closes the ROADMAP "Cloud cost/billing accuracy" follow-up: an automatable drift check that cross-references the AI-spend cost tables. The "three cost tables" do NOT form three independent sources to cross- equate, so this is deliberately NOT a "three tables must agree" check: * ``cmdb_cost_facts`` IS the vendor extract for the Azure-Foundry / AWS-Bedrock / GCP-Vertex buckets (the cloud drivers pull straight from AWS CE / Azure CM / GCP BQ into it; ``amount_usd`` is FX-frozen at ingest). There is no second in-platform source for these buckets. * ``bailian_usage_daily`` is the sole source for the ``bailian`` bucket. * seats come from ``SoftwareSeatAllocation`` ⋈ ``SoftwareLicensePlan`` (``category='ai'``), native→USD at the LIVE FX rate. The one reconciliation that IS meaningful and automatable is, for every **closed** month and every bucket: (B) the materialised ``ai_spending_period_rollup`` value ↔ (A) a fresh LIVE recompute from those same sources (``AiSpendingService._compute_period_rows`` — the exact path the dashboard takes for an open month, reused so the口径 can't drift). The rollup is a faithful cache of the live compute, so the two MUST be equal. A supra-threshold gap is a drift signal; the heuristic ``likely_cause`` narrows it: * ``stale_fingerprint`` — the window is materialised, but only under an older fingerprint (whitelist / FX / seat-plan price / semantics version changed and the refresh cron hasn't rebuilt it yet). * ``not_materialized`` — the window (or this specific bucket within it) has no current-fingerprint rollup row at all. * ``late_facts_or_drift`` — both sides present at the current fingerprint but unequal (late-arriving bill facts, or a genuine computation drift). Cloud-provider consoles and the Aliyun Bailian console are NOT reconciled here: they have no API-equivalent口径 and double-counting / pre-paid resource-pack consumption means an honest cross-check is impossible. They are surfaced as a ``data_caveat`` ("treat the console as the source of truth for absolute spend"). This module is **read-only**. It is consumed by: * the ``/admin/reconciliation`` cost lens (``require_admin_read``), and * the daily ``ai_spending.cost_drift_check`` cron, which auto-heals a supra-threshold finding by re-materialising the drifted window (the rollup is a faithful cache — "cache ≠ source" is always fixed by rebuilding the cache) and RAISES only when the re-check still drifts, so the failure_notifier alarms on genuine anomalies only. The healing write lives in the cron handler; this module itself never mutates. Only **closed** months are reconciled. The open month's rollup is by design ≤1 day stale (the cron pre-warms it from yesterday's facts), so comparing it to a fresh live recompute would emit guaranteed false positives. """ from __future__ import annotations from collections import defaultdict from decimal import Decimal from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.domains.ai_spending.models import AiSpendingPeriodRollup from app.domains.ai_spending.service import ( _ROLLUP_EMPTY_MARKER, _SEAT_CURRENCY_USD, AiSpendingService, BucketRow, _current_period, _parse_period, _seat_native_to_usd, _shift_period, ) from app.domains.asset.models import ( AssetSoftwareCatalog, SoftwareLicensePlan, SoftwareSeatAllocation, ) # Drift threshold: a bucket-month is flagged when the absolute gap exceeds # ``max(_DRIFT_ABS_USD, _DRIFT_PCT × live)`` — i.e. tolerate sub-dollar # rounding noise on tiny buckets AND a 0.5% relative wobble on large ones, # flag anything bigger than both. Constants (not magic numbers) so the cron # and the lens share one definition and a future tune is one edit. _DRIFT_ABS_USD = Decimal("1") _DRIFT_PCT = Decimal("0.005") # 0.5% # How many closed months back the lens + cron reconcile by default. Each # month is a full live recompute (a ``cmdb_cost_facts`` scan), so this is the # cost knob — 3 covers "this just-closed month + late-facts tail" without # turning the forensic page into a multi-second wait. _DEFAULT_MONTHS_BACK = 3 # English caveat returned on the wire for non-UI consumers (MCP / export). # The web UI renders its own localized copy via i18n (``reconciliation.cost. # caveat``); this string never reaches a browser directly. _DATA_CAVEAT = ( "Drift is measured as materialised-rollup vs a live recompute from the " "SAME in-platform sources (cmdb_cost_facts / bailian_usage_daily / seat " "allocations). It does NOT reconcile against cloud-provider consoles or " "the Aliyun Bailian console — those have no API-equivalent basis " "(double-counted line items, pre-paid resource-pack consumption not in " "the BSS product bill, etc.). Treat the cloud / Bailian console figures " "as the source of truth for absolute spend; this lens only proves the " "Helix rollup faithfully mirrors the Helix live compute." ) # ── Plan-currency guard (PR-2) ────────────────────────────────────── # # The AI dashboard is USD throughout. A `category='ai'` seat plan whose # ``currency`` is mis-entered as a non-USD code while its ``price`` digits are # actually USD (a real past incident: 4 plans logged CNY, shrinking the seats # bucket ~6.76×) is converted DOWN by FX a second time → understated AI-seat # spend. ``_seat_native_to_usd`` already DROPS unsupported currencies (HKD/EUR/ # …) to 0 with a warning, but CNY is a *supported* currency — a CNY mis-entry # is silently FX-converted, never dropped, so it's the blind spot. This is a # DETECTION (not a hard block — legitimately CNY-priced AI plans exist), surfaced # to admins/finance/auditors so a human can confirm or correct. _PLAN_PER_SEAT_USD_MIN = Decimal("1") _PLAN_PER_SEAT_USD_MAX = Decimal("1000") def _normalized_monthly_native(price: Decimal, billing_cycle: str) -> Decimal | None: """Per-seat MONTHLY native amount for a plan, mirroring ``billing_service._compute_amount``'s cycle semantics (monthly = price, yearly = price/12, one_time = the one-off price). Returns ``None`` for an unknown cycle so the caller skips it rather than reasoning about a bogus per-seat figure.""" cyc = (billing_cycle or "").lower() if cyc == "monthly": return price if cyc == "yearly": return price / Decimal(12) if cyc == "one_time": return price return None async def _ai_plan_active_seat_counts( db: AsyncSession, plan_ids: list, ) -> dict[str, int]: """``{plan_id: active_seat_count}`` for the given AI plans (one grouped query, not N).""" if not plan_ids: return {} rows = (await db.execute( select( SoftwareSeatAllocation.plan_id, func.count().label("n"), ) .where( SoftwareSeatAllocation.plan_id.in_(plan_ids), SoftwareSeatAllocation.status == "active", ) .group_by(SoftwareSeatAllocation.plan_id) )).all() return {str(pid): int(n) for pid, n in rows} async def plan_currency_findings(db: AsyncSession) -> dict[str, Any]: """Detect suspicious ``category='ai'`` seat-plan currencies. Per AI plan, compute the per-seat monthly USD (same cycle + FX path the seats bucket uses) and flag: * ``non_usd_currency_with_active_seats`` — currency != USD AND the plan has active seats (the CNY-misentry blind spot + any non-USD plan whose digits might really be USD), and * ``implausible_per_seat_usd`` — converted per-seat USD outside ``[$1, $1000]`` (a likely mis-price / mis-currency regardless of seats). Detection only — never mutates. A finding carries both the raw plan fields and the converted figure so a human can decide "fix the currency" vs "this is a legitimately CNY-priced plan". """ # Use the None-returning FX source directly (NOT svc._get_cny_rate, which # raises when the rate setting is absent) so THIS function never raises on # a missing rate: it degrades to rate=0 → CNY per-seat converts to 0 (the # non-USD reason is rate-independent and still fires). NB: when invoked via # cost_rollup_drift, the surrounding drift recompute (_rollup_fingerprint / # _bucket_seats) DOES depend on FX and will raise first if the rate is # missing — surfacing as the per-lens error capsule in the reconciliation # overview, or a (legitimate) cron alarm. This standalone robustness only # matters when plan_currency_findings is called on its own. from app.core import fx_rate as _fx_rate rate = await _fx_rate.get_usd_per_cny(db) rate_dec = Decimal(str(rate)) if rate else Decimal("0") ai_software_ids = (await db.execute( select(AssetSoftwareCatalog.id).where( AssetSoftwareCatalog.category == "ai", ) )).scalars().all() if not ai_software_ids: return {"findings": [], "thresholds": _plan_currency_thresholds()} plans = (await db.execute( select(SoftwareLicensePlan) .where(SoftwareLicensePlan.software_id.in_(ai_software_ids)) .order_by(SoftwareLicensePlan.id) )).scalars().all() seat_counts = await _ai_plan_active_seat_counts( db, [p.id for p in plans], ) findings: list[dict[str, Any]] = [] for plan in plans: currency = (plan.currency or _SEAT_CURRENCY_USD).upper() price = Decimal(str(plan.price)) if plan.price is not None else Decimal("0") native = _normalized_monthly_native(price, plan.billing_cycle) if native is None: continue # unknown billing cycle — not a currency question active = seat_counts.get(str(plan.id), 0) per_seat_usd = _seat_native_to_usd( native=native, currency=currency, rate_dec=rate_dec, plan_id=str(plan.id), ) reasons: list[str] = [] if currency != _SEAT_CURRENCY_USD and active > 0: reasons.append("non_usd_currency_with_active_seats") if per_seat_usd > 0 and ( per_seat_usd < _PLAN_PER_SEAT_USD_MIN or per_seat_usd > _PLAN_PER_SEAT_USD_MAX ): reasons.append("implausible_per_seat_usd") if not reasons: continue findings.append({ "plan_id": str(plan.id), "plan_name": plan.name, "software_id": str(plan.software_id), "price": float(round(price, 6)), "currency": currency, "billing_cycle": plan.billing_cycle, "per_seat_usd": float(round(per_seat_usd, 6)), "active_seats": active, "reasons": reasons, }) return {"findings": findings, "thresholds": _plan_currency_thresholds()} def _plan_currency_thresholds() -> dict[str, float]: return { "per_seat_usd_min": float(_PLAN_PER_SEAT_USD_MIN), "per_seat_usd_max": float(_PLAN_PER_SEAT_USD_MAX), } async def has_suspicious_ai_seat_currency(db: AsyncSession) -> bool: """Cheap single-query check: does any ``category='ai'`` plan with ACTIVE seats carry a non-USD currency? Used by the AI-spending dashboard to flag the seats bucket without running the full per-plan ``plan_currency_findings`` on the hot overview path.""" hit = (await db.execute( select(SoftwareLicensePlan.id) .join( AssetSoftwareCatalog, AssetSoftwareCatalog.id == SoftwareLicensePlan.software_id, ) .join( SoftwareSeatAllocation, SoftwareSeatAllocation.plan_id == SoftwareLicensePlan.id, ) .where( AssetSoftwareCatalog.category == "ai", SoftwareSeatAllocation.status == "active", func.upper(func.coalesce(SoftwareLicensePlan.currency, _SEAT_CURRENCY_USD)) != _SEAT_CURRENCY_USD, ) .limit(1) )).first() return hit is not None def _sum_by_bucket(rows: list[BucketRow]) -> dict[str, Decimal]: """Sum ``cost_usd`` per bucket key over a BucketRow list.""" out: dict[str, Decimal] = defaultdict(lambda: Decimal("0")) for r in rows: out[r.bucket] += r.cost_usd return dict(out) async def _window_has_any_rollup( db: AsyncSession, period_start, period_end_excl, ) -> bool: """True if ANY rollup row exists for the window at ANY fingerprint. Distinguishes ``stale_fingerprint`` (materialised, wrong fingerprint) from ``not_materialized`` (never materialised) when the current- fingerprint read misses. """ hit = (await db.execute( select(AiSpendingPeriodRollup.id).where( AiSpendingPeriodRollup.period_start == period_start, AiSpendingPeriodRollup.period_end_excl == period_end_excl, ).limit(1) )).first() return hit is not None def _threshold(live: Decimal) -> Decimal: pct_band = (live if live > 0 else Decimal("0")) * _DRIFT_PCT return _DRIFT_ABS_USD if _DRIFT_ABS_USD > pct_band else pct_band async def cost_rollup_drift( db: AsyncSession, *, months_back: int = _DEFAULT_MONTHS_BACK, ) -> dict[str, Any]: """Reconcile materialised rollup ↔ live recompute for recent closed months. Returns a JSON-serialisable dict:: { "findings": [ {period, bucket, rollup_usd, live_usd, diff_usd, diff_pct, likely_cause}, ... ], "periods_checked": ["2026-05", "2026-04", "2026-03"], "months_back": 3, "thresholds": {"abs_usd": 1.0, "pct": 0.5}, "data_caveat": "...", "checked_from_period": "2026-06", } ``findings`` is empty when every checked bucket-month is within tolerance. Read-only; never commits. """ # Clamp at the chokepoint, not just in each caller: each month is a live # recompute, so an unbounded ``months_back`` (e.g. a mis-set cron config) # would fan out into arbitrarily many heavy scans. 13 matches the rollup- # refresh cron's own lookback — nothing older is materialised to compare. months_back = max(1, min(int(months_back), 13)) svc = AiSpendingService(db) fp = await svc._rollup_fingerprint() current_period = _current_period() findings: list[dict[str, Any]] = [] periods_checked: list[str] = [] # i=0 is the OPEN month — excluded (its rollup is intentionally ≤1 day # stale, comparing it to a live recompute would be a guaranteed false # positive). i=1.. are closed months. for i in range(1, months_back + 1): period = _shift_period(current_period, -i) start, end_excl = _parse_period(period) periods_checked.append(period) live_rows = await svc._compute_period_rows(start, end_excl) live_by_bucket = _sum_by_bucket(live_rows) rollup_rows = await svc._read_rollup(start, end_excl, fp) window_has_current_fp = rollup_rows is not None rollup_by_bucket = ( _sum_by_bucket(rollup_rows) if rollup_rows else {} ) window_has_any = ( window_has_current_fp or await _window_has_any_rollup(db, start, end_excl) ) for bucket in sorted(set(live_by_bucket) | set(rollup_by_bucket)): # The empty-marker bucket is an internal sentinel, never a real # cost bucket — _read_rollup already strips it, but guard anyway. if bucket == _ROLLUP_EMPTY_MARKER: continue live_usd = live_by_bucket.get(bucket, Decimal("0")) rollup_usd = rollup_by_bucket.get(bucket, Decimal("0")) diff = abs(rollup_usd - live_usd) if diff <= _threshold(live_usd): continue if not window_has_current_fp: cause = ( "stale_fingerprint" if window_has_any else "not_materialized" ) elif bucket not in rollup_by_bucket: cause = "not_materialized" else: cause = "late_facts_or_drift" denom = live_usd if live_usd > 0 else Decimal("1") findings.append({ "period": period, "bucket": bucket, "rollup_usd": float(round(rollup_usd, 6)), "live_usd": float(round(live_usd, 6)), "diff_usd": float(round(diff, 6)), "diff_pct": round(float(diff / denom * 100), 4), "likely_cause": cause, }) # Plan-currency sub-block (PR-2): same cost-credibility surface, so it # rides the cost lens rather than a separate page. plan_currency = await plan_currency_findings(db) return { "findings": findings, "periods_checked": periods_checked, "months_back": months_back, "thresholds": { "abs_usd": float(_DRIFT_ABS_USD), "pct": float(_DRIFT_PCT * 100), }, "data_caveat": _DATA_CAVEAT, "checked_from_period": current_period, "plan_currency": plan_currency, } __all__ = [ "cost_rollup_drift", "has_suspicious_ai_seat_currency", "plan_currency_findings", ]