Coverage for physiodsp / activity / activity_score.py: 89%
146 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-02-26 21:14 +0000
1from typing import Literal
3import numpy as np
4from pandas import DataFrame
5from pydantic import BaseModel, Field, PositiveInt, PositiveFloat
7from physiodsp.base import BaseAlgorithm
10class ActivityScoreSettings(BaseModel):
11 """Configuration settings for Activity Score algorithm"""
13 # Baseline window (number of days to use for personalization)
14 baseline_window_days: PositiveInt = Field(default=30, description="Number of days to use for computing baselines")
16 # Factor weights (must sum to 1.0)
17 step_weight: PositiveFloat = Field(default=0.25, description="Weight for step count factor")
18 sleep_weight: PositiveFloat = Field(default=0.35, description="Weight for sleep factor")
19 training_weight: PositiveFloat = Field(default=0.25, description="Weight for training time factor")
20 resting_weight: PositiveFloat = Field(default=0.15, description="Weight for resting time factor")
22 # Step targets (personalized)
23 baseline_daily_steps: PositiveInt = Field(default=8000, description="User's baseline daily steps")
24 step_ceiling: PositiveInt = Field(default=15000, description="Maximum steps for optimal score")
26 # Sleep targets (hours)
27 min_sleep_hours: PositiveFloat = Field(default=6.0, description="Minimum healthy sleep duration")
28 optimal_sleep_hours: PositiveFloat = Field(default=8.0, description="Optimal sleep duration")
29 max_sleep_hours: PositiveFloat = Field(default=10.0, description="Maximum sleep before penalty")
31 # Training targets (minutes)
32 min_training_minutes: PositiveInt = Field(default=0, description="Minimum training minutes per day")
33 optimal_training_minutes: PositiveInt = Field(default=60, description="Optimal training minutes per day")
34 max_training_minutes: PositiveInt = Field(default=120, description="Maximum training before overtraining penalty")
36 # Resting targets (minutes)
37 min_resting_minutes: PositiveInt = Field(default=480, description="Minimum resting minutes (8 hours)")
38 optimal_resting_minutes: PositiveInt = Field(default=540, description="Optimal resting minutes (9 hours)")
39 max_resting_minutes: PositiveInt = Field(default=720, description="Maximum resting minutes (12 hours)")
41 # Scoring method
42 scoring_method: Literal["gaussian", "sigmoid", "linear"] = Field(
43 default="gaussian",
44 description="Method to map normalized values to scores"
45 )
48class ActivityScore(BaseAlgorithm):
49 """
50 Activity Score Algorithm - Personalized Daily Activity Assessment
52 Combines multiple health metrics (steps, sleep, training, resting) into a single
53 0-100 score tailored to individual user baselines and targets.
55 Based on principles from Oura Ring and WHOOP band scoring algorithms.
56 """
58 _algorithm_name = "ActivityScore"
59 _version = "v0.1.0"
61 def __init__(self, settings: ActivityScoreSettings = ActivityScoreSettings()) -> None:
62 self.settings = settings
63 self._validate_weights()
64 self.user_stats = None
65 self.daily_scores = None
66 self.baseline_stats = None
67 return None
69 def _validate_weights(self) -> None:
70 """Validate that weights sum to 1.0"""
71 total_weight = (
72 self.settings.step_weight +
73 self.settings.sleep_weight +
74 self.settings.training_weight +
75 self.settings.resting_weight
76 )
77 if not np.isclose(total_weight, 1.0, atol=0.01):
78 raise ValueError(f"Weights must sum to 1.0, got {total_weight}")
80 def _compute_baseline_stats(self, baseline_data: DataFrame) -> dict:
81 """
82 Compute personalized baseline statistics from historical data.
84 Args:
85 baseline_data: DataFrame with previous N-1 days
87 Returns:
88 Dictionary with computed statistics
89 """
90 return {
91 'steps_median': baseline_data['steps'].median(),
92 'steps_std': baseline_data['steps'].std(),
93 'steps_p75': baseline_data['steps'].quantile(0.75),
94 'sleep_median': baseline_data['sleep_hours'].median(),
95 'sleep_std': baseline_data['sleep_hours'].std(),
96 'training_median': baseline_data['training_minutes'].median(),
97 'training_std': baseline_data['training_minutes'].std(),
98 'resting_median': baseline_data['resting_minutes'].median(),
99 'resting_std': baseline_data['resting_minutes'].std()
100 }
102 def _score_steps(self, daily_steps: int, baseline_stats: dict = None) -> float:
103 """
104 Score daily step count (0-100).
106 Gaussian distribution centered at user's median baseline.
107 """
108 if baseline_stats is None:
109 baseline = self.settings.baseline_daily_steps
110 ceiling = self.settings.step_ceiling
111 else:
112 # Use personalized baseline from historical data
113 baseline = baseline_stats['steps_median']
114 # Ceiling is 75th percentile or 1.5x median, whichever is higher
115 ceiling = max(baseline * 1.5, baseline_stats['steps_p75'])
117 if self.settings.scoring_method == "gaussian":
118 if baseline == 0:
119 return 0.0
120 normalized = daily_steps / baseline
121 peak_normalized = ceiling / baseline
122 std_dev = (peak_normalized - 1.0) / 2.0
123 score = 100 * np.exp(-((normalized - 1.0) ** 2) / (2 * std_dev ** 2))
124 score = np.clip(score, 0, 100)
125 else:
126 if ceiling == 0:
127 return 0.0
128 score = min(100, (daily_steps / ceiling) * 100)
130 return float(score)
132 def _score_sleep(self, sleep_hours: float, baseline_stats: dict = None) -> float:
133 """
134 Score sleep duration (0-100).
136 Centered at user's median sleep, with personalized thresholds.
137 """
138 if baseline_stats is None:
139 optimal = self.settings.optimal_sleep_hours
140 min_sleep = self.settings.min_sleep_hours
141 max_sleep = self.settings.max_sleep_hours
142 else:
143 median_sleep = baseline_stats['sleep_median']
144 std_sleep = baseline_stats['sleep_std'] or 0.5
145 optimal = median_sleep
146 min_sleep = max(4.0, median_sleep - std_sleep * 1.5)
147 max_sleep = min(11.0, median_sleep + std_sleep * 1.5)
149 if sleep_hours < min_sleep:
150 deficit = (min_sleep - sleep_hours) / min_sleep
151 score = max(0, 50 - deficit * 50)
152 elif sleep_hours <= optimal:
153 score = ((sleep_hours - min_sleep) / (optimal - min_sleep)) * 100
154 elif sleep_hours <= max_sleep:
155 score = 100 - ((sleep_hours - optimal) / (max_sleep - optimal)) * 30
156 else:
157 excess = sleep_hours - max_sleep
158 score = max(0, 70 - excess * 10)
160 return float(np.clip(score, 0, 100))
162 def _score_training(self, training_minutes: int, baseline_stats: dict = None) -> float:
163 """
164 Score training time (0-100).
166 Rewards consistent moderate training relative to user's baseline.
167 """
168 if baseline_stats is None:
169 optimal = self.settings.optimal_training_minutes
170 max_train = self.settings.max_training_minutes
171 else:
172 median_training = baseline_stats['training_median']
173 std_training = baseline_stats['training_std'] or 15
174 optimal = max(30, median_training)
175 max_train = optimal + std_training * 2
177 if training_minutes < 5:
178 score = 50
179 elif training_minutes <= optimal:
180 score = 50 + (training_minutes / optimal) * 50
181 elif training_minutes <= max_train:
182 excess_ratio = (training_minutes - optimal) / (max_train - optimal)
183 score = 100 + (excess_ratio * 10) - 10
184 else:
185 excess = training_minutes - max_train
186 score = max(0, 100 - excess * 0.5)
188 return float(np.clip(score, 0, 100))
190 def _score_resting(self, resting_minutes: int, baseline_stats: dict = None) -> float:
191 """
192 Score resting/recovery time (0-100).
194 Rewards recovery relative to user's baseline.
195 """
196 resting_hours = resting_minutes / 60
198 if baseline_stats is None:
199 optimal_hours = self.settings.optimal_resting_minutes / 60
200 min_hours = self.settings.min_resting_minutes / 60
201 max_hours = self.settings.max_resting_minutes / 60
202 else:
203 median_resting = baseline_stats['resting_median'] / 60
204 std_resting = baseline_stats['resting_std'] / 60 or 0.5
205 optimal_hours = median_resting
206 min_hours = max(6.0, median_resting - std_resting * 1.5)
207 max_hours = min(12.0, median_resting + std_resting * 1.5)
209 if resting_hours < min_hours:
210 deficit = (min_hours - resting_hours)
211 score = max(0, 50 - deficit * 15)
212 elif resting_hours <= optimal_hours:
213 score = ((resting_hours - min_hours) / (optimal_hours - min_hours)) * 100
214 elif resting_hours <= max_hours:
215 excess_hours = resting_hours - optimal_hours
216 max_excess = max_hours - optimal_hours
217 score = 100 - (excess_hours / max_excess) * 25
218 else:
219 excess_hours = resting_hours - max_hours
220 score = max(0, 75 - excess_hours * 10)
222 return float(np.clip(score, 0, 100))
224 def run(self, daily_activity_data: DataFrame):
225 """
226 Calculate Activity Score for the most recent day using baseline statistics.
228 Args:
229 daily_activity_data: DataFrame with all available data, sorted by date (oldest to newest).
230 Must contain columns:
231 - 'date' or index as date
232 - 'steps': Daily step count
233 - 'sleep_hours': Total sleep duration in hours
234 - 'training_minutes': Total training time in minutes
235 - 'resting_minutes': Total resting/recovery time in minutes
237 Should have at least baseline_window_days of data.
239 Returns:
240 self with biomarker containing score for the most recent day
241 """
242 if not all(col in daily_activity_data.columns for col in ['steps', 'sleep_hours', 'training_minutes', 'resting_minutes']):
243 raise ValueError("DataFrame must contain: steps, sleep_hours, training_minutes, resting_minutes")
245 if len(daily_activity_data) < 2:
246 raise ValueError(f"Need at least 2 days of data, got {len(daily_activity_data)}")
248 # Ensure we have the right number of baseline days
249 window_size = min(self.settings.baseline_window_days, len(daily_activity_data) - 1)
251 # Split into baseline (all but last) and current day (last)
252 baseline_data = daily_activity_data.iloc[:-1].tail(window_size)
253 current_day_data = daily_activity_data.iloc[-1]
255 # Compute personalized baseline statistics
256 self.baseline_stats = self._compute_baseline_stats(baseline_data)
258 # Calculate individual scores using personalized baselines
259 step_score = self._score_steps(current_day_data['steps'], self.baseline_stats)
260 sleep_score = self._score_sleep(current_day_data['sleep_hours'], self.baseline_stats)
261 training_score = self._score_training(current_day_data['training_minutes'], self.baseline_stats)
262 resting_score = self._score_resting(current_day_data['resting_minutes'], self.baseline_stats)
264 # Weighted combination
265 activity_score = (
266 (step_score * self.settings.step_weight) +
267 (sleep_score * self.settings.sleep_weight) +
268 (training_score * self.settings.training_weight) +
269 (resting_score * self.settings.resting_weight)
270 )
272 # Create output DataFrame
273 score_record = {
274 'date': current_day_data.get('date', daily_activity_data.index[-1]) if 'date' in daily_activity_data.columns else daily_activity_data.index[-1],
275 'activity_score': round(activity_score, 1),
276 'step_score': round(step_score, 1),
277 'sleep_score': round(sleep_score, 1),
278 'training_score': round(training_score, 1),
279 'resting_score': round(resting_score, 1),
280 'steps': current_day_data['steps'],
281 'sleep_hours': current_day_data['sleep_hours'],
282 'training_minutes': current_day_data['training_minutes'],
283 'resting_minutes': current_day_data['resting_minutes'],
284 'baseline_days_used': window_size
285 }
287 self.biomarker_agg = DataFrame([score_record])
288 self.daily_scores = self.biomarker_agg.copy()
290 return self
292 def get_activity_score_interpretation(self, activity_score: float) -> str:
293 """
294 Return interpretation of Activity Score.
296 Args:
297 activity_score: Activity score (0-100)
299 Returns:
300 Interpretation string
301 """
302 if activity_score >= 85:
303 return "Excellent - Outstanding activity and recovery balance"
304 elif activity_score >= 70:
305 return "Good - Healthy activity levels with adequate recovery"
306 elif activity_score >= 50:
307 return "Fair - Room for improvement in activity or recovery"
308 elif activity_score >= 30:
309 return "Poor - Significant imbalance in activity or recovery"
310 else:
311 return "Critical - Urgent attention needed to activity and recovery"