"""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, }