125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
"""Objective analytics — linear regression for KPI prediction.
|
||
|
||
Given a set of ObjectiveEntry measurements, computes a simple linear
|
||
regression (y = mx + b, where x = days since first entry) and answers:
|
||
|
||
- **predicted_achievement_date**: the date when the trend line crosses
|
||
the KPI target value (or None if the trend is flat or negative).
|
||
- **confidence_score**: a heuristic 0.0–1.0 based on sample count and
|
||
the R² coefficient of determination.
|
||
- **predicted_value_at_date(date)**: the value the trend predicts at a
|
||
given future date.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime
|
||
|
||
|
||
def _to_days(timestamps: list[datetime.datetime]) -> list[float]:
|
||
"""Convert datetimes to float days since the first entry."""
|
||
if not timestamps:
|
||
return []
|
||
t0 = timestamps[0]
|
||
return [(t - t0).total_seconds() / 86400.0 for t in timestamps]
|
||
|
||
|
||
def _r_squared(xs, ys, slope, intercept):
|
||
"""Coefficient of determination (R²)."""
|
||
if len(xs) < 2:
|
||
return 0.0
|
||
mean_y = sum(ys) / len(ys)
|
||
ss_res = sum((y - (slope * x + intercept)) ** 2 for x, y in zip(xs, ys, strict=True))
|
||
ss_tot = sum((y - mean_y) ** 2 for y in ys)
|
||
if ss_tot == 0:
|
||
return 1.0
|
||
return 1.0 - ss_res / ss_tot
|
||
|
||
|
||
def compute_trend(
|
||
timestamps: list[datetime.datetime],
|
||
values: list[float],
|
||
target: float | None,
|
||
) -> dict:
|
||
"""Compute a linear regression trend and prediction.
|
||
|
||
Args:
|
||
timestamps: Entry dates.
|
||
values: Entry values (must be same length as timestamps).
|
||
target: KPI target value, or None for qualitative objectives.
|
||
|
||
Returns:
|
||
dict with keys:
|
||
slope, intercept, r_squared, confidence,
|
||
predicted_achievement_days, predicted_achievement_date,
|
||
predicted_value_at_due, target_value
|
||
"""
|
||
n = len(timestamps)
|
||
if n < 2 or not values:
|
||
return {
|
||
'slope': 0.0,
|
||
'intercept': float(values[0]) if values else 0.0,
|
||
'r_squared': 0.0,
|
||
'confidence': 0.0,
|
||
'predicted_achievement_days': None,
|
||
'predicted_achievement_date': None,
|
||
'predicted_value_at_due': None,
|
||
'target_value': target,
|
||
}
|
||
|
||
xs = _to_days(timestamps)
|
||
ys = [float(v) for v in values]
|
||
|
||
n_f = float(n)
|
||
sum_x = sum(xs)
|
||
sum_y = sum(ys)
|
||
sum_xy = sum(x * y for x, y in zip(xs, ys, strict=True))
|
||
sum_xx = sum(x * x for x in xs)
|
||
|
||
denom = n_f * sum_xx - sum_x * sum_x
|
||
if abs(denom) < 1e-12:
|
||
return {
|
||
'slope': 0.0,
|
||
'intercept': sum_y / n_f,
|
||
'r_squared': 0.0,
|
||
'confidence': 0.0,
|
||
'predicted_achievement_days': None,
|
||
'predicted_achievement_date': None,
|
||
'predicted_value_at_due': None,
|
||
'target_value': target,
|
||
}
|
||
|
||
slope = (n_f * sum_xy - sum_x * sum_y) / denom
|
||
intercept = (sum_y - slope * sum_x) / n_f
|
||
r2 = _r_squared(xs, ys, slope, intercept)
|
||
|
||
# Confidence heuristic: combination of R² and sample count
|
||
sample_confidence = min(1.0, n / 10.0) # 10+ entries = full confidence
|
||
confidence = sample_confidence * r2
|
||
|
||
# Days from first entry to target (if slope > 0 and target > intercept)
|
||
predicted_achievement_days = None
|
||
predicted_achievement_date = None
|
||
if target is not None and slope > 1e-12:
|
||
days_to_target = (target - intercept) / slope
|
||
if days_to_target >= 0:
|
||
predicted_achievement_days = days_to_target
|
||
predicted_achievement_date = timestamps[0] + datetime.timedelta(
|
||
days=days_to_target,
|
||
)
|
||
|
||
return {
|
||
'slope': slope,
|
||
'intercept': intercept,
|
||
'r_squared': round(r2, 4),
|
||
'confidence': round(confidence, 4),
|
||
'predicted_achievement_days': (
|
||
round(predicted_achievement_days, 1) if predicted_achievement_days is not None else None
|
||
),
|
||
'predicted_achievement_date': (
|
||
predicted_achievement_date.isoformat() if predicted_achievement_date else None
|
||
),
|
||
'predicted_value_at_due': None,
|
||
'target_value': target,
|
||
}
|