Coverage for dataexcept/datascience_exceptions/training.py: 90%

1"""Model training, evaluation, and inference errors."""

3from __future__ import annotations

5from typing import Any, Optional

7from .._validation import is_number

8from .base import DataScienceError

11class ModelTrainingError(DataScienceError):

12 """

13 Raised when model training fails.

15 Attributes:

16 model_type: model class or name.

17 epoch: optional epoch index.

18 """

20 def __init__(

21 self,

22 model_type: str,

23 epoch: Optional[int] = None,

24 message: Optional[str] = None,

25 ) -> None:

26 if not isinstance(model_type, str): 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

27 raise TypeError(f"model_type must be str, got {type(model_type).__name__}")

28 if epoch is not None and not isinstance(epoch, int):

29 raise TypeError(f"epoch must be int or None, got {type(epoch).__name__}")

30 if message is not None and not isinstance(message, str): 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 raise TypeError(

32 f"message must be str or None, got {type(message).__name__}"

33 )

35 if message is None:

36 msg = f"Training failed for model '{model_type}'"

37 if epoch is not None:

38 msg += f" at epoch {epoch}" # include epoch

39 else:

40 msg = message

42 self.model_type = model_type

43 self.epoch = epoch

44 super().__init__(msg)

46 def __str__(self) -> str:

47 base = f"{self.model_type}"

48 if self.epoch is not None:

49 base += f"@{self.epoch}"

50 return f"[ModelTrainingError:{base}] {self.message}"

53class ConvergenceError(ModelTrainingError):

54 """

55 Raised when optimization fails to converge.

57 Attributes:

58 iterations: number of iterations run.

59 """

61 def __init__(

62 self, model_type: str, iterations: int, message: Optional[str] = None

63 ) -> None:

64 if not isinstance(iterations, int): 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true

65 raise TypeError(f"iterations must be int, got {type(iterations).__name__}")

67 if message is None: 67 ↛ 74line 67 didn't jump to line 74 because the condition on line 67 was always true

68 message = (

69 f"Model '{model_type}' failed to converge after "

70 f"{iterations} iterations"

71 )

72 # Assigned before super(): DataExceptError.__init__ sweeps the stored

73 # strings for URLs, and anything set afterwards escapes that.

74 self.iterations = iterations

75 super().__init__(model_type=model_type, epoch=None, message=message)

77 def __str__(self) -> str:

78 return f"[ConvergenceError] {self.message}"

81class TrainingTimeoutError(ModelTrainingError):

82 """Raised when model training exceeds a time limit."""

84 def __init__(self, model_type: str, timeout: float) -> None:

85 if not is_number(timeout):

86 raise TypeError(f"timeout must be a number, got {type(timeout).__name__}")

87 message = f"Training '{model_type}' exceeded timeout of {timeout} seconds"

88 self.timeout = float(timeout)

89 super().__init__(model_type=model_type, epoch=None, message=message)

91 def __str__(self) -> str:

92 return f"[TrainingTimeoutError] {self.message}"

95class HyperparameterError(DataScienceError):

96 """

97 Raised for invalid hyperparameter settings.

99 Attributes:

100 param: name of hyperparameter.

101 value: invalid value.

102 """

103

104 def __init__(self, param: str, value: Any, message: Optional[str] = None) -> None:

105 if not isinstance(param, str):

106 raise TypeError(f"param must be str, got {type(param).__name__}")

107

108 if message is None: 108 ↛ 111line 108 didn't jump to line 111 because the condition on line 108 was always true

109 message = f"Invalid hyperparameter '{param}': {value!r}"

110

111 self.param = param

112 self.value = value

113 super().__init__(message)

114

115 def __str__(self) -> str:

116 return f"[HyperparameterError:{self.param}] {self.message}"

117

118

119class ModelEvaluationError(DataScienceError):

120 """

121 Raised during evaluation metrics computation.

122

123 Attributes:

124 metric: name of the metric.

125 value: computed value.

126 """

127

128 def __init__(

129 self, metric: str, value: float, message: Optional[str] = None

130 ) -> None:

131 if not isinstance(metric, str): 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 raise TypeError(f"metric must be str, got {type(metric).__name__}")

133 if not is_number(value):

134 raise TypeError(f"value must be number, got {type(value).__name__}")

135

136 if message is None: 136 ↛ 139line 136 didn't jump to line 139 because the condition on line 136 was always true

137 message = f"Failed to compute metric '{metric}', got {value}"

138

139 self.metric = metric

140 self.value = float(value)

141 super().__init__(message)

142

143 def __str__(self) -> str:

144 return f"[ModelEvaluationError:{self.metric}] {self.message}"

145

146

147class PredictionError(DataScienceError):

148 """

149 Raised when making predictions fails.

150

151 Attributes:

152 model_type: model used.

153 inputs: input data snapshot.

154 """

155

156 def __init__(

157 self, model_type: str, inputs: Any, message: Optional[str] = None

158 ) -> None:

159 if not isinstance(model_type, str): 159 ↛ 160line 159 didn't jump to line 160 because the condition on line 159 was never true

160 raise TypeError(f"model_type must be str, got {type(model_type).__name__}")

161

162 if message is None: 162 ↛ 167line 162 didn't jump to line 167 because the condition on line 162 was always true

163 message = (

164 f"Prediction failed for model '{model_type}' " f"with inputs {inputs!r}"

165 )

166

167 self.model_type = model_type

168 self.inputs = inputs

169 super().__init__(message)

170

171 def __str__(self) -> str:

172 return f"[PredictionError:{self.model_type}] {self.message}"

173

174

175class FeatureSelectionError(DataScienceError):

176 """Failure in feature selection procedure."""

177

178 def __init__(self, technique: str, details: Optional[str] = None) -> None:

179 if not isinstance(technique, str): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true

180 raise TypeError(f"technique must be str, got {type(technique).__name__}")

181 msg = f"Feature selection failed using {technique}" + (

182 f": {details}" if details else ""

183 )

184 self.technique = technique

185 super().__init__(msg)

186

187

188class DimensionalityReductionError(DataScienceError):

189 """Error applying dimensionality reduction method."""

190

191 def __init__(self, method: str, components: Optional[int] = None) -> None:

192 if not isinstance(method, str): 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true

193 raise TypeError(f"method must be str, got {type(method).__name__}")

194 if components is not None and not isinstance(components, int): 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true

195 raise TypeError(

196 ("components must be int or None, " f"got {type(components).__name__}")

197 )

198 msg = f"Dimensionality reduction '{method}' failed" + (

199 f" for {components} components" if components else ""

200 )

201 self.method = method

202 self.components = components

203 super().__init__(msg)

204

205

206class CrossValidationError(DataScienceError):

207 """Failure during cross-validation procedure."""

208

209 def __init__(self, folds: int, cause: Optional[str] = None) -> None:

210 if not isinstance(folds, int): 210 ↛ 211line 210 didn't jump to line 211 because the condition on line 210 was never true

211 raise TypeError(f"folds must be int, got {type(folds).__name__}")

212 msg = f"Cross-validation failed on {folds} folds" + (

213 f": {cause}" if cause else ""

214 )

215 self.folds = folds

216 super().__init__(msg)

217

218

219class HyperparameterTuningError(DataScienceError):

220 """Error during hyperparameter search or tuning."""

221

222 def __init__(self, method: str, details: Optional[str] = None) -> None:

223 if not isinstance(method, str): 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true

224 raise TypeError(f"method must be str, got {type(method).__name__}")

225 msg = f"Hyperparameter tuning ({method}) failed" + (

226 f": {details}" if details else ""

227 )

228 self.method = method

229 super().__init__(msg)

230

231

232class ExperimentTrackingError(DataScienceError):

233 """Issues logging or retrieving experiment metadata."""

234

235 def __init__(self, run_id: str, cause: Optional[str] = None) -> None:

236 if not isinstance(run_id, str): 236 ↛ 237line 236 didn't jump to line 237 because the condition on line 236 was never true

237 raise TypeError(f"run_id must be str, got {type(run_id).__name__}")

238 msg = f"Experiment tracking failed for run '{run_id}'" + (

239 f": {cause}" if cause else ""

240 )

241 self.run_id = run_id

242 super().__init__(msg)

243

244

245class GPUOutOfMemoryError(DataScienceError):

246 """Model or tensor exceeds GPU memory capacity."""

247

248 def __init__(self, device: str, required: str, available: str) -> None:

249 if not all(isinstance(v, str) for v in (device, required, available)): 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 raise TypeError("device, required, available must be str")

251 msg = f"GPU OOM on {device}: required={required}, available={available}"

252 self.device = device

253 self.required = required

254 self.available = available

255 super().__init__(msg)

256

257

258class ModelInferenceError(DataScienceError):

259 """Raised when model inference fails.

260

261 Args:

262 model_type: Identifier of the model used for inference.

263 original: Underlying exception raised by the model.

264 """

265

266 def __init__(self, model_type: str, original: Exception) -> None:

267 if not isinstance(model_type, str):

268 raise TypeError(f"model_type must be str, got {type(model_type).__name__}")

269 if not isinstance(original, Exception):

270 raise TypeError(

271 f"original must be Exception, got {type(original).__name__}"

272 )

273 msg = f"Inference failed for model '{model_type}': {original}"

274 self.model_type = model_type

275 self.original = original

276 super().__init__(msg)

277

278 def __str__(self) -> str:

279 return f"[ModelInferenceError:{self.model_type}] {self.message}"

280

281

282class ModelCompatibilityError(DataScienceError):

283 """Raised when a model is incompatible with the runtime environment.

284

285 Args:

286 expected_version: Required model version.

287 found_version: Detected model version.

288 message: Optional custom message.

289 """

290

291 def __init__(

292 self,

293 expected_version: str,

294 found_version: str,

295 message: Optional[str] = None,

296 ) -> None:

297 if not isinstance(expected_version, str):

298 raise TypeError(

299 "expected_version must be str, got "

300 f"{type(expected_version).__name__}"

301 )

302 if not isinstance(found_version, str):

303 raise TypeError(

304 "found_version must be str, got " f"{type(found_version).__name__}"

305 )

306 if message is not None and not isinstance(message, str):

307 raise TypeError(

308 f"message must be str or None, got {type(message).__name__}"

309 )

310

311 if message is None: 311 ↛ 317line 311 didn't jump to line 317 because the condition on line 311 was always true

312 msg = (

313 f"Model requires version {expected_version}, "

314 f"but found {found_version}"

315 )

316 else:

317 msg = message

318

319 self.expected_version = expected_version

320 self.found_version = found_version

321 super().__init__(msg)

322

323 def __str__(self) -> str:

324 return f"[ModelCompatibilityError] {self.message}"

325

326

327class OverfittingError(DataScienceError):

328 """Raised when a model is overfitting the training data.

329

330 Args:

331 train_metric: Metric value on the training set.

332 val_metric: Metric value on the validation set.

333 """

334

335 def __init__(self, train_metric: float, val_metric: float) -> None:

336 if not is_number(train_metric):

337 raise TypeError(

338 ("train_metric must be numeric, got " f"{type(train_metric).__name__}")

339 )

340 if not is_number(val_metric):

341 raise TypeError(

342 f"val_metric must be numeric, got {type(val_metric).__name__}"

343 )

344

345 self.train_metric = float(train_metric)

346 self.val_metric = float(val_metric)

347 msg = (

348 f"Overfitting detected: train={self.train_metric}, "

349 f"val={self.val_metric}"

350 )

351 super().__init__(msg)

352

353 def __str__(self) -> str:

354 return f"[OverfittingError] {self.message}"

355

356

357class UnderfittingError(DataScienceError):

358 """Raised when a model fails to capture patterns in the data.

359

360 Args:

361 train_metric: Metric value on the training set.

362 threshold: Minimum acceptable metric value.

363 """

364

365 def __init__(self, train_metric: float, threshold: float) -> None:

366 if not is_number(train_metric):

367 raise TypeError(

368 ("train_metric must be numeric, got " f"{type(train_metric).__name__}")

369 )

370 if not is_number(threshold):

371 raise TypeError(

372 f"threshold must be numeric, got {type(threshold).__name__}"

373 )

374

375 self.train_metric = float(train_metric)

376 self.threshold = float(threshold)

377 msg = (

378 f"Underfitting detected: training metric {self.train_metric} "

379 f"< threshold {self.threshold}"

380 )

381 super().__init__(msg)

382

383 def __str__(self) -> str:

384 return f"[UnderfittingError] {self.message}"

385

386

387class EarlyStoppingError(DataScienceError):

388 """Raised when training stops early based on a stopping criterion.

389

390 Args:

391 epoch: Epoch index where training stopped.

392 reason: Optional reason for stopping.

393 """

394

395 def __init__(self, epoch: int, reason: Optional[str] = None) -> None:

396 if not isinstance(epoch, int):

397 raise TypeError(f"epoch must be int, got {type(epoch).__name__}")

398 if reason is not None and not isinstance(reason, str):

399 raise TypeError(f"reason must be str or None, got {type(reason).__name__}")

400

401 msg = f"Training stopped early at epoch {epoch}"

402 if reason:

403 msg += f": {reason}"

404

405 self.epoch = epoch

406 self.reason = reason

407 super().__init__(msg)

408

409 def __str__(self) -> str:

410 return f"[EarlyStoppingError:{self.epoch}] {self.message}"

411

412

413class BiasDetectionError(DataScienceError):

414 """Raised when algorithmic bias exceeds an acceptable threshold.

415

416 Args:

417 feature: Feature or group where bias was detected.

418 bias_score: Calculated bias metric.

419 threshold: Maximum acceptable bias metric.

420 message: Optional custom message.

421 """

422

423 def __init__(

424 self,

425 feature: str,

426 bias_score: float,

427 threshold: float,

428 message: Optional[str] = None,

429 ) -> None:

430 if not isinstance(feature, str):

431 raise TypeError(f"feature must be str, got {type(feature).__name__}")

432 if not is_number(bias_score):

433 raise TypeError(

434 f"bias_score must be numeric, got {type(bias_score).__name__}"

435 )

436 if not is_number(threshold):

437 raise TypeError(

438 f"threshold must be numeric, got {type(threshold).__name__}"

439 )

440 if message is not None and not isinstance(message, str):

441 raise TypeError(

442 f"message must be str or None, got {type(message).__name__}"

443 )

444

445 if message is None: 445 ↛ 451line 445 didn't jump to line 451 because the condition on line 445 was always true

446 msg = (

447 f"Bias detected in '{feature}': score={bias_score:.3f} > "

448 f"threshold={threshold:.3f}"

449 )

450 else:

451 msg = message

452

453 self.feature = feature

454 self.bias_score = float(bias_score)

455 self.threshold = float(threshold)

456 super().__init__(msg)

457

458 def __str__(self) -> str:

459 return f"[BiasDetectionError:{self.feature}] {self.message}"

460

461

462class ExplainabilityError(DataScienceError):

463 """Raised when generating model explanations fails.

464

465 Args:

466 method: Explanation technique identifier.

467 details: Optional description of the failure.

468 """

469

470 def __init__(self, method: str, details: Optional[str] = None) -> None:

471 if not isinstance(method, str):

472 raise TypeError(f"method must be str, got {type(method).__name__}")

473 if details is not None and not isinstance(details, str):

474 raise TypeError(

475 f"details must be str or None, got {type(details).__name__}"

476 )

477 msg = f"Explainability using '{method}' failed"

478 if details:

479 msg += f": {details}"

480 self.method = method

481 self.details = details

482 super().__init__(msg)

483

484 def __str__(self) -> str:

485 return f"[ExplainabilityError:{self.method}] {self.message}"

486

487

488class FeatureScalingError(DataScienceError):

489 """Raised when scaling or standardization of features fails.

490

491 Args:

492 scaler: Name of the scaler or transformation used.

493 details: Optional explanation of the failure.

494 """

495

496 def __init__(self, scaler: str, details: Optional[str] = None) -> None:

497 if not isinstance(scaler, str):

498 raise TypeError(f"scaler must be str, got {type(scaler).__name__}")

499 if details is not None and not isinstance(details, str):

500 raise TypeError(

501 f"details must be str or None, got {type(details).__name__}"

502 )

503 msg = f"Feature scaling with '{scaler}' failed"

504 if details:

505 msg += f": {details}"

506 self.scaler = scaler

507 self.details = details

508 super().__init__(msg)

509

510 def __str__(self) -> str:

511 return f"[FeatureScalingError:{self.scaler}] {self.message}"