tegwick
dfc1d90c28
src/target_revenue/metrics.py: compute_metrics(manifest, entries, as_of)
is pure/deterministic like fold.py, reusing fold.py/conversion.py
unchanged. Returns facts/calculations/forecasts as three explicitly
separated blocks (TrustServicePRD TS-FR-5), covering the mandatory Q9
set plus the recommended velocity/forecast tier - forecasts are always
null rather than populated once a Phase has converted or velocity is
non-positive, so nothing disguises a projection as a fact.
Adds GET /phases/{id}/metrics (unauthenticated per FR-9/FR-10).
tests/test_metrics.py (6 tests) needs no Docker/Postgres and runs
under plain system Python. One new Docker-gated test in
test_ledger_hosting.py proves the hosted /metrics response exactly
matches compute_metrics() run offline against the same export.
125 lines
4.8 KiB
Python
125 lines
4.8 KiB
Python
"""Public metrics computation (WP-0006-T05).
|
|
|
|
Implements the mandatory public metrics set from
|
|
`specs/OpenQuestions-WorkingDefaults.md` Q9, plus its recommended
|
|
velocity/forecast tier, always labeled fact / calculation / forecast per
|
|
`specs/TargetRevenueLicenseConcept.md` §14.6 and
|
|
`specs/TrustServiceProductRequirementsDocument.md` TS-FR-5:
|
|
|
|
- **fact**: read or summed directly from the Manifest/Ledger, no modeling.
|
|
- **calculation**: a deterministic derivation over facts (e.g. a percentage).
|
|
- **forecast**: a projection that could be wrong; never positioned where a
|
|
fact is expected, and — per TSD §4.1's Metrics component's forbidden
|
|
action — never presented as if it were a ledger fact.
|
|
|
|
Pure and deterministic given `(manifest, entries, as_of)`: the only
|
|
"impure" input is the caller-supplied `as_of` timestamp, kept as an
|
|
explicit parameter (not read from the wall clock internally) so this
|
|
module stays testable and reproducible like `fold.py` (TSD §6.1).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from . import conversion as conversion_module
|
|
from . import fold as fold_module
|
|
|
|
_MATERIAL_PROGRESS_TYPES = frozenset({"development-credit", "remission-credit"})
|
|
|
|
|
|
def _parse(ts: str) -> datetime:
|
|
return datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
|
|
|
|
|
def compute_metrics(
|
|
manifest: dict[str, Any], entries: list[dict[str, Any]], as_of: datetime
|
|
) -> dict[str, Any]:
|
|
"""Compute the full labeled metrics set for one Phase at a point in time."""
|
|
if as_of.tzinfo is None:
|
|
raise ValueError("as_of must be timezone-aware")
|
|
|
|
status = conversion_module.conversion_status(manifest, entries)
|
|
fold_result = fold_module.fold_outstanding_target(
|
|
manifest["phase"]["initial_target"]["amount"], entries
|
|
)
|
|
initial_amount = manifest["phase"]["initial_target"]["amount"]
|
|
last_entry = entries[-1] if entries else None
|
|
|
|
material_entries = [e for e in entries if e["type"] in _MATERIAL_PROGRESS_TYPES]
|
|
first_material = material_entries[0] if material_entries else None
|
|
last_material = material_entries[-1] if material_entries else None
|
|
|
|
facts = {
|
|
"initial_target_amount": initial_amount,
|
|
"initial_target_currency": manifest["phase"]["initial_target"]["currency"],
|
|
"cumulative_development_credit": fold_result.development_credit,
|
|
"cumulative_remission_credit": fold_result.remission_credit,
|
|
"outstanding_target": fold_result.outstanding_target,
|
|
"is_converted": status.is_converted,
|
|
"future_license": status.future_license,
|
|
"last_ledger_entry_id": last_entry["id"] if last_entry else None,
|
|
"longstop_at": manifest["phase"].get("longstop_at"),
|
|
}
|
|
|
|
calculations: dict[str, Any] = {
|
|
"target_satisfaction_percentage": (
|
|
None
|
|
if initial_amount <= 0
|
|
else round(
|
|
100.0
|
|
* min(1.0, (fold_result.development_credit + fold_result.remission_credit) / initial_amount),
|
|
4,
|
|
)
|
|
),
|
|
"development_credit_velocity_per_day": None,
|
|
"remission_credit_velocity_per_day": None,
|
|
"days_since_last_material_progress": None,
|
|
}
|
|
|
|
if first_material and last_material:
|
|
span_days = (_parse(last_material["recognized_at"]) - _parse(first_material["recognized_at"])).total_seconds() / 86400.0
|
|
if span_days > 0:
|
|
dev_total = sum(
|
|
e["amount"] for e in material_entries if e["type"] == "development-credit"
|
|
)
|
|
rem_total = sum(
|
|
e["amount"] for e in material_entries if e["type"] == "remission-credit"
|
|
)
|
|
calculations["development_credit_velocity_per_day"] = round(dev_total / span_days, 6)
|
|
calculations["remission_credit_velocity_per_day"] = round(rem_total / span_days, 6)
|
|
|
|
if last_material:
|
|
calculations["days_since_last_material_progress"] = round(
|
|
(as_of - _parse(last_material["recognized_at"])).total_seconds() / 86400.0, 4
|
|
)
|
|
|
|
forecasts: dict[str, Any] = {"projected_conversion_date": None}
|
|
velocity = (
|
|
(calculations["development_credit_velocity_per_day"] or 0.0)
|
|
+ (calculations["remission_credit_velocity_per_day"] or 0.0)
|
|
)
|
|
if not status.is_converted and velocity > 0:
|
|
days_remaining = fold_result.outstanding_target / velocity
|
|
forecasts["projected_conversion_date"] = (
|
|
as_of.replace(microsecond=0) + _timedelta_days(days_remaining)
|
|
).isoformat()
|
|
|
|
return {
|
|
"phase": manifest["phase"]["id"],
|
|
"as_of": as_of.isoformat(),
|
|
"facts": facts,
|
|
"calculations": calculations,
|
|
"forecasts": forecasts,
|
|
}
|
|
|
|
|
|
def _timedelta_days(days: float):
|
|
from datetime import timedelta
|
|
|
|
return timedelta(days=days)
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|