Coverage for tests / test_activity_score.py: 100%

1import pytest 

2from datetime import datetime, timedelta 

3import numpy as np 

4from pandas import DataFrame 

5from physiodsp.activity.activity_score import ActivityScore, ActivityScoreSettings 

6 

7 

8@pytest.mark.parametrize( 

9 "baseline_window_days", 

10 [15, 30] 

11) 

12def test_activity_score_excellent(baseline_window_days): 

13 """Test Activity Score for excellent performance (>=85)""" 

14 np.random.seed(42) 

15 settings = ActivityScoreSettings(baseline_window_days=baseline_window_days) 

16 algorithm = ActivityScore(settings=settings) 

17 

18 # Create excellent baseline data with random variation 

19 dates = [datetime.now() - timedelta(days=x) for x in range(baseline_window_days, 0, -1)] 

20 baseline_data = { 

21 'date': dates, 

22 'steps': np.random.randint(12000, 13000, baseline_window_days).tolist(), 

23 'sleep_hours': np.random.uniform(8.0, 8.3, baseline_window_days).tolist(), 

24 'training_minutes': np.random.randint(65, 75, baseline_window_days).tolist(), 

25 'resting_minutes': np.random.randint(540, 570, baseline_window_days).tolist() 

26 } 

27 

28 # Excellent current day (at or above baseline) 

29 current_day = { 

30 'date': datetime.now(), 

31 'steps': np.random.randint(12500, 13500), 

32 'sleep_hours': np.random.uniform(8.1, 8.4), 

33 'training_minutes': np.random.randint(70, 85), 

34 'resting_minutes': np.random.randint(550, 580) 

35 } 

36 

37 all_data = DataFrame({ 

38 **{k: v for k, v in baseline_data.items()}, 

39 'date': baseline_data['date'] + [current_day['date']], 

40 'steps': baseline_data['steps'] + [current_day['steps']], 

41 'sleep_hours': baseline_data['sleep_hours'] + [current_day['sleep_hours']], 

42 'training_minutes': baseline_data['training_minutes'] + [current_day['training_minutes']], 

43 'resting_minutes': baseline_data['resting_minutes'] + [current_day['resting_minutes']] 

44 }) 

45 

46 result = algorithm.run(all_data) 

47 score = result.biomarker_agg.iloc[0]['activity_score'] 

48 

49 assert score >= 85, f"Expected excellent score (>=85), got {score}" 

50 assert algorithm.get_activity_score_interpretation(score) == "Excellent - Outstanding activity and recovery balance" 

51 

52 

53@pytest.mark.parametrize( 

54 "baseline_window_days", 

55 [15, 30] 

56) 

57def test_activity_score_good(baseline_window_days): 

58 """Test Activity Score for good performance (70-84)""" 

59 np.random.seed(43) 

60 settings = ActivityScoreSettings(baseline_window_days=baseline_window_days) 

61 algorithm = ActivityScore(settings=settings) 

62 

63 # Create good baseline data with random variation 

64 dates = [datetime.now() - timedelta(days=x) for x in range(baseline_window_days, 0, -1)] 

65 baseline_data = { 

66 'date': dates, 

67 'steps': np.random.randint(8500, 9500, baseline_window_days).tolist(), 

68 'sleep_hours': np.random.uniform(7.2, 7.8, baseline_window_days).tolist(), 

69 'training_minutes': np.random.randint(40, 50, baseline_window_days).tolist(), 

70 'resting_minutes': np.random.randint(480, 520, baseline_window_days).tolist() 

71 } 

72 

73 # Good current day 

74 current_day = { 

75 'date': datetime.now(), 

76 'steps': np.random.randint(9500, 10500), 

77 'sleep_hours': np.random.uniform(7.8, 8.2), 

78 'training_minutes': np.random.randint(48, 60), 

79 'resting_minutes': np.random.randint(510, 540) 

80 } 

81 

82 all_data = DataFrame({ 

83 **{k: v for k, v in baseline_data.items()}, 

84 'date': baseline_data['date'] + [current_day['date']], 

85 'steps': baseline_data['steps'] + [current_day['steps']], 

86 'sleep_hours': baseline_data['sleep_hours'] + [current_day['sleep_hours']], 

87 'training_minutes': baseline_data['training_minutes'] + [current_day['training_minutes']], 

88 'resting_minutes': baseline_data['resting_minutes'] + [current_day['resting_minutes']] 

89 }) 

90 

91 result = algorithm.run(all_data) 

92 score = result.biomarker_agg.iloc[0]['activity_score'] 

93 

94 assert 70 <= score < 85, f"Expected good score (70-84), got {score}" 

95 assert algorithm.get_activity_score_interpretation(score) == "Good - Healthy activity levels with adequate recovery" 

96 

97 

98@pytest.mark.parametrize( 

99 "baseline_window_days", 

100 [15, 30] 

101) 

102def test_activity_score_fair(baseline_window_days): 

103 """Test Activity Score for fair performance (50-69)""" 

104 np.random.seed(44) 

105 settings = ActivityScoreSettings(baseline_window_days=baseline_window_days) 

106 algorithm = ActivityScore(settings=settings) 

107 

108 # Create fair baseline data with random variation 

109 dates = [datetime.now() - timedelta(days=x) for x in range(baseline_window_days, 0, -1)] 

110 baseline_data = { 

111 'date': dates, 

112 'steps': np.random.randint(7500, 8500, baseline_window_days).tolist(), 

113 'sleep_hours': np.random.uniform(7.5, 8.0, baseline_window_days).tolist(), 

114 'training_minutes': np.random.randint(45, 55, baseline_window_days).tolist(), 

115 'resting_minutes': np.random.randint(500, 540, baseline_window_days).tolist() 

116 } 

117 

118 # Fair current day (slightly below baseline) 

119 current_day = { 

120 'date': datetime.now(), 

121 'steps': np.random.randint(5500, 6500), 

122 'sleep_hours': np.random.uniform(6.8, 7.2), 

123 'training_minutes': np.random.randint(35, 45), 

124 'resting_minutes': np.random.randint(470, 500) 

125 } 

126 

127 all_data = DataFrame({ 

128 **{k: v for k, v in baseline_data.items()}, 

129 'date': baseline_data['date'] + [current_day['date']], 

130 'steps': baseline_data['steps'] + [current_day['steps']], 

131 'sleep_hours': baseline_data['sleep_hours'] + [current_day['sleep_hours']], 

132 'training_minutes': baseline_data['training_minutes'] + [current_day['training_minutes']], 

133 'resting_minutes': baseline_data['resting_minutes'] + [current_day['resting_minutes']] 

134 }) 

135 

136 result = algorithm.run(all_data) 

137 score = result.biomarker_agg.iloc[0]['activity_score'] 

138 

139 assert 50 <= score < 70, f"Expected fair score (50-69), got {score}" 

140 assert algorithm.get_activity_score_interpretation(score) == "Fair - Room for improvement in activity or recovery" 

141 

142 

143@pytest.mark.parametrize( 

144 "baseline_window_days", 

145 [15, 30] 

146) 

147def test_activity_score_poor(baseline_window_days): 

148 """Test Activity Score for poor performance (30-49)""" 

149 np.random.seed(45) 

150 settings = ActivityScoreSettings(baseline_window_days=baseline_window_days) 

151 algorithm = ActivityScore(settings=settings) 

152 

153 # Create poor baseline data with random variation 

154 dates = [datetime.now() - timedelta(days=x) for x in range(baseline_window_days, 0, -1)] 

155 baseline_data = { 

156 'date': dates, 

157 'steps': np.random.randint(5000, 6000, baseline_window_days).tolist(), 

158 'sleep_hours': np.random.uniform(5.5, 6.0, baseline_window_days).tolist(), 

159 'training_minutes': np.random.randint(15, 25, baseline_window_days).tolist(), 

160 'resting_minutes': np.random.randint(420, 460, baseline_window_days).tolist() 

161 } 

162 

163 # Poor current day (significantly below baseline) 

164 current_day = { 

165 'date': datetime.now(), 

166 'steps': np.random.randint(2500, 3500), 

167 'sleep_hours': np.random.uniform(4.0, 4.8), 

168 'training_minutes': np.random.randint(5, 15), 

169 'resting_minutes': np.random.randint(300, 350) 

170 } 

171 

172 all_data = DataFrame({ 

173 **{k: v for k, v in baseline_data.items()}, 

174 'date': baseline_data['date'] + [current_day['date']], 

175 'steps': baseline_data['steps'] + [current_day['steps']], 

176 'sleep_hours': baseline_data['sleep_hours'] + [current_day['sleep_hours']], 

177 'training_minutes': baseline_data['training_minutes'] + [current_day['training_minutes']], 

178 'resting_minutes': baseline_data['resting_minutes'] + [current_day['resting_minutes']] 

179 }) 

180 

181 result = algorithm.run(all_data) 

182 score = result.biomarker_agg.iloc[0]['activity_score'] 

183 

184 assert 30 <= score < 50, f"Expected poor score (30-49), got {score}" 

185 assert algorithm.get_activity_score_interpretation(score) == "Poor - Significant imbalance in activity or recovery" 

186 

187 

188@pytest.mark.parametrize( 

189 "baseline_window_days", 

190 [15, 30] 

191) 

192def test_activity_score_critical(baseline_window_days): 

193 """Test Activity Score for critical performance (<30)""" 

194 np.random.seed(46) 

195 settings = ActivityScoreSettings(baseline_window_days=baseline_window_days) 

196 algorithm = ActivityScore(settings=settings) 

197 

198 # Create critical baseline data with random variation 

199 dates = [datetime.now() - timedelta(days=x) for x in range(baseline_window_days, 0, -1)] 

200 baseline_data = { 

201 'date': dates, 

202 'steps': np.random.randint(2500, 3500, baseline_window_days).tolist(), 

203 'sleep_hours': np.random.uniform(5.0, 5.5, baseline_window_days).tolist(), 

204 'training_minutes': np.random.randint(5, 15, baseline_window_days).tolist(), 

205 'resting_minutes': np.random.randint(380, 420, baseline_window_days).tolist() 

206 } 

207 

208 # Critical current day (significantly below baseline) 

209 current_day = { 

210 'date': datetime.now(), 

211 'steps': np.random.randint(200, 500), 

212 'sleep_hours': np.random.uniform(2.5, 3.5), 

213 'training_minutes': np.random.randint(0, 2), 

214 'resting_minutes': np.random.randint(150, 200) 

215 } 

216 

217 all_data = DataFrame({ 

218 **{k: v for k, v in baseline_data.items()}, 

219 'date': baseline_data['date'] + [current_day['date']], 

220 'steps': baseline_data['steps'] + [current_day['steps']], 

221 'sleep_hours': baseline_data['sleep_hours'] + [current_day['sleep_hours']], 

222 'training_minutes': baseline_data['training_minutes'] + [current_day['training_minutes']], 

223 'resting_minutes': baseline_data['resting_minutes'] + [current_day['resting_minutes']] 

224 }) 

225 

226 result = algorithm.run(all_data) 

227 score = result.biomarker_agg.iloc[0]['activity_score'] 

228 

229 assert score < 30, f"Expected critical score (<30), got {score}" 

230 assert algorithm.get_activity_score_interpretation(score) == "Critical - Urgent attention needed to activity and recovery" 

231 

232 

233def test_activity_score_algorithm_properties(): 

234 """Test basic properties of Activity Score algorithm""" 

235 algorithm = ActivityScore() 

236 

237 assert algorithm.algorithm_name == "ActivityScore" 

238 assert algorithm.version == "v0.1.0" 

239 

240 default_settings = ActivityScoreSettings() 

241 assert default_settings.baseline_window_days == 30 

242 assert default_settings.step_weight == 0.25 

243 assert default_settings.sleep_weight == 0.35 

244 assert default_settings.training_weight == 0.25 

245 assert default_settings.resting_weight == 0.15 

246 

247 

248def test_activity_score_minimum_data(): 

249 """Test that algorithm requires minimum 2 days of data""" 

250 settings = ActivityScoreSettings() 

251 algorithm = ActivityScore(settings=settings) 

252 

253 # Only 1 day of data 

254 single_day = DataFrame({ 

255 'steps': [8000], 

256 'sleep_hours': [8.0], 

257 'training_minutes': [60], 

258 'resting_minutes': [540] 

259 }) 

260 

261 with pytest.raises(ValueError, match="Need at least 2 days of data"): 

262 algorithm.run(single_day) 

263 

264 

265def test_activity_score_personalizes_to_baseline(): 

266 """Test that scores are personalized based on baseline data""" 

267 np.random.seed(47) 

268 settings = ActivityScoreSettings(baseline_window_days=15) 

269 algorithm_low = ActivityScore(settings=settings) 

270 algorithm_high = ActivityScore(settings=settings) 

271 

272 # Low baseline user with random variation 

273 dates = [datetime.now() - timedelta(days=x) for x in range(15, 0, -1)] 

274 low_baseline = DataFrame({ 

275 'date': dates + [datetime.now()], 

276 'steps': np.random.randint(4800, 5200, 15).tolist() + [5500], 

277 'sleep_hours': np.random.uniform(6.3, 6.7, 15).tolist() + [7.0], 

278 'training_minutes': np.random.randint(18, 22, 15).tolist() + [25], 

279 'resting_minutes': np.random.randint(410, 430, 15).tolist() + [450] 

280 }) 

281 

282 # High baseline user with random variation and same current day 

283 high_baseline = DataFrame({ 

284 'date': dates + [datetime.now()], 

285 'steps': np.random.randint(14500, 15500, 15).tolist() + [5500], 

286 'sleep_hours': np.random.uniform(8.3, 8.7, 15).tolist() + [7.0], 

287 'training_minutes': np.random.randint(115, 125, 15).tolist() + [25], 

288 'resting_minutes': np.random.randint(590, 610, 15).tolist() + [450] 

289 }) 

290 

291 result_low = algorithm_low.run(low_baseline) 

292 result_high = algorithm_high.run(high_baseline) 

293 

294 score_low = result_low.biomarker_agg.iloc[0]['activity_score'] 

295 score_high = result_high.biomarker_agg.iloc[0]['activity_score'] 

296 

297 # Same current day should yield different scores based on baseline 

298 assert score_low != score_high 

299 # Low baseline user with same data should score higher (meeting their baseline) 

300 assert score_low > score_high 

301 

302 

303def test_activity_score_output_structure(): 

304 """Test output DataFrame structure""" 

305 settings = ActivityScoreSettings(baseline_window_days=15) 

306 algorithm = ActivityScore(settings=settings) 

307 

308 dates = [datetime.now() - timedelta(days=x) for x in range(15, 0, -1)] 

309 data = DataFrame({ 

310 'date': dates + [datetime.now()], 

311 'steps': [8000] * 16, 

312 'sleep_hours': [8.0] * 16, 

313 'training_minutes': [60] * 16, 

314 'resting_minutes': [540] * 16 

315 }) 

316 

317 result = algorithm.run(data) 

318 

319 # Check biomarker structure 

320 assert hasattr(result, 'biomarker_agg') 

321 assert len(result.biomarker_agg) == 1 

322 

323 # Check required columns 

324 required_cols = ['activity_score', 'step_score', 'sleep_score', 

325 'training_score', 'resting_score', 'baseline_days_used'] 

326 for col in required_cols: 

327 assert col in result.biomarker_agg.columns 

328 

329 # Check score ranges 

330 assert 0 <= result.biomarker_agg.iloc[0]['activity_score'] <= 100 

331 assert 0 <= result.biomarker_agg.iloc[0]['step_score'] <= 100 

332 assert 0 <= result.biomarker_agg.iloc[0]['sleep_score'] <= 100 

333 assert 0 <= result.biomarker_agg.iloc[0]['training_score'] <= 100 

334 assert 0 <= result.biomarker_agg.iloc[0]['resting_score'] <= 100 

335 

336 

337def test_activity_score_weighted_combination(): 

338 """Test that final score is correct weighted combination""" 

339 settings = ActivityScoreSettings( 

340 step_weight=0.25, 

341 sleep_weight=0.35, 

342 training_weight=0.25, 

343 resting_weight=0.15 

344 ) 

345 algorithm = ActivityScore(settings=settings) 

346 

347 dates = [datetime.now() - timedelta(days=x) for x in range(15, 0, -1)] 

348 data = DataFrame({ 

349 'date': dates + [datetime.now()], 

350 'steps': [8000] * 16, 

351 'sleep_hours': [8.0] * 16, 

352 'training_minutes': [60] * 16, 

353 'resting_minutes': [540] * 16 

354 }) 

355 

356 result = algorithm.run(data) 

357 score = result.biomarker_agg.iloc[0] 

358 

359 # Manually calculate expected score 

360 expected_activity_score = ( 

361 score['step_score'] * 0.25 + 

362 score['sleep_score'] * 0.35 + 

363 score['training_score'] * 0.25 + 

364 score['resting_score'] * 0.15 

365 ) 

366 

367 assert np.isclose(score['activity_score'], expected_activity_score, atol=0.5) 

368 

369 

370def test_activity_score_baseline_stats_stored(): 

371 """Test that baseline statistics are computed and stored""" 

372 settings = ActivityScoreSettings(baseline_window_days=15) 

373 algorithm = ActivityScore(settings=settings) 

374 

375 dates = [datetime.now() - timedelta(days=x) for x in range(15, 0, -1)] 

376 data = DataFrame({ 

377 'date': dates + [datetime.now()], 

378 'steps': list(range(5000, 5015)) + [6000], 

379 'sleep_hours': [7.0 + i*0.1 for i in range(15)] + [8.0], 

380 'training_minutes': [30 + i*2 for i in range(15)] + [60], 

381 'resting_minutes': [450 + i*5 for i in range(15)] + [500] 

382 }) 

383 

384 result = algorithm.run(data) 

385 

386 assert result.baseline_stats is not None 

387 assert 'steps_median' in result.baseline_stats 

388 assert 'steps_std' in result.baseline_stats 

389 assert 'sleep_median' in result.baseline_stats 

390 assert 'training_median' in result.baseline_stats 

391 assert 'resting_median' in result.baseline_stats 

392 

393 # Check that baseline stats make sense 

394 assert result.baseline_stats['steps_median'] > 0 

395 assert result.baseline_stats['sleep_median'] > 0