import json import math from pathlib import Path from collections import Counter import numpy as np class PhenotypeNextTestRecommender: def __init__(self, classifier, reference_path): self.classifier = classifier self.reference_path = Path(reference_path) if not self.reference_path.exists(): raise FileNotFoundError(f"Reference distribution file not found: {self.reference_path}") with open(self.reference_path, "r", encoding="utf-8") as f: self.ref = json.load(f) self.schema = self.ref["schema"] self.classes = self.ref["classes"] self.default_recommendable_fields = self.ref["default_recommendable_fields"] self.default_exclude_from_next_tests = set(self.ref["default_exclude_from_next_tests"]) self.global_field_value_counts = self.ref["global_field_value_counts"] self.genus_field_value_counts = self.ref["genus_field_value_counts"] self.genus_field_record_counts = self.ref["genus_field_record_counts"] self.field_record_counts = self.ref["field_record_counts"] self.genus_record_counts = self.ref["genus_record_counts"] @staticmethod def entropy(probs): probs = np.asarray(probs, dtype=np.float64) probs = probs[probs > 0] if len(probs) == 0: return 0.0 return float(-(probs * np.log2(probs)).sum()) @classmethod def safe_entropy_normalized(cls, probs): probs = np.asarray(probs, dtype=np.float64) probs = probs[probs > 0] if len(probs) <= 1: return 0.0 h = cls.entropy(probs) max_h = math.log2(len(probs)) if max_h <= 0: return 0.0 return float(h / max_h) def predict_full_distribution(self, features): result = self.classifier.predict(features, top_k=len(self.classifier.classes)) prob_by_genus = { item["genus"]: item["probability"] for item in result["ranked_genera"] } probs = np.array( [prob_by_genus.get(genus, 0.0) for genus in self.classifier.classes], dtype=np.float64, ) total = probs.sum() if total > 0: probs = probs / total return result, probs @staticmethod def top_margin(probs): probs = np.asarray(probs, dtype=np.float64) order = np.argsort(probs)[::-1] if len(order) == 0: return 0.0, 0.0 if len(order) == 1: return float(probs[order[0]]), 0.0 return float(probs[order[0]]), float(probs[order[0]] - probs[order[1]]) def get_counter(self, mapping, *keys): current = mapping for key in keys: if not isinstance(current, dict): return {} current = current.get(key, {}) if isinstance(current, dict): return current return {} def get_global_field_counter(self, field): return self.get_counter(self.global_field_value_counts, field) def get_genus_field_counter(self, genus, field): return self.get_counter(self.genus_field_value_counts, genus, field) def get_genus_field_record_count(self, genus, field): genus_map = self.genus_field_record_counts.get(genus, {}) if not isinstance(genus_map, dict): return 0 return int(genus_map.get(field, 0)) def normalized_top_priors(self, current_probs, top_genera): priors = {} for genus in top_genera: if genus in self.classifier.classes: idx = self.classifier.classes.index(genus) priors[genus] = float(current_probs[idx]) total = sum(priors.values()) if total <= 0: equal = 1.0 / len(top_genera) return {g: equal for g in top_genera} return { genus: prob / total for genus, prob in priors.items() } def get_top_priors_vector(self, current_probs, top_genera): priors = [] for genus in top_genera: if genus in self.classifier.classes: idx = self.classifier.classes.index(genus) priors.append(float(current_probs[idx])) else: priors.append(0.0) priors = np.array(priors, dtype=np.float64) total = priors.sum() if total <= 0: priors = np.ones(len(top_genera), dtype=np.float64) / len(top_genera) else: priors = priors / total return priors def value_probability_given_genus(self, genus, field, value, smoothing=0.25): counter = self.get_genus_field_counter(genus, field) total = sum(counter.values()) possible_count = max(len(self.get_global_field_counter(field)), 1) numerator = counter.get(value, 0) + smoothing denominator = total + smoothing * possible_count return numerator / denominator def global_value_probability(self, field, value, smoothing=0.25): counter = self.get_global_field_counter(field) total = sum(counter.values()) possible_count = max(len(counter), 1) numerator = counter.get(value, 0) + smoothing denominator = total + smoothing * possible_count return numerator / denominator def get_candidate_values_for_field( self, field, top_genera, current_probs, max_values=10, min_global_count=1, ): top_priors = self.normalized_top_priors(current_probs, top_genera) weighted_scores = Counter() for genus, prior in top_priors.items(): counter = self.get_genus_field_counter(genus, field) if not counter: continue total = sum(counter.values()) if total <= 0: continue for value, count in counter.items(): weighted_scores[value] += prior * (count / total) if not weighted_scores: global_counter = self.get_global_field_counter(field) for value, count in Counter(global_counter).most_common(max_values): if count >= min_global_count: weighted_scores[value] = count filtered = [] for value, score in weighted_scores.most_common(max_values * 2): token = f"{field}={value}" if token in self.classifier.vocab or field == "Growth Temperature": filtered.append((value, float(score))) if len(filtered) >= max_values: break if not filtered: filtered = [ (value, float(score)) for value, score in weighted_scores.most_common(max_values) ] return filtered def estimate_candidate_value_weights(self, field, candidate_values, top_genera, current_probs): top_priors = self.normalized_top_priors(current_probs, top_genera) raw_weights = [] for value in candidate_values: weight = 0.0 for genus, prior in top_priors.items(): weight += prior * self.value_probability_given_genus(genus, field, value) weight += 0.05 * self.global_value_probability(field, value) raw_weights.append(weight) raw_weights = np.array(raw_weights, dtype=np.float64) total = raw_weights.sum() if total <= 0: return np.ones(len(candidate_values), dtype=np.float64) / len(candidate_values) return raw_weights / total def smoothed_value_likelihood(self, genus, field, value, candidate_values, smoothing=0.25): counter = self.get_genus_field_counter(genus, field) total = sum(counter.values()) k = max(len(candidate_values), 1) return (counter.get(value, 0) + smoothing) / (total + smoothing * k) def empirical_field_discrimination(self, field, candidate_values, top_genera, current_probs, smoothing=0.25): priors = self.get_top_priors_vector(current_probs, top_genera) prior_entropy = self.entropy(priors) likelihood_matrix = [] for genus in top_genera: row = [] for value in candidate_values: row.append( self.smoothed_value_likelihood( genus=genus, field=field, value=value, candidate_values=candidate_values, smoothing=smoothing, ) ) row = np.array(row, dtype=np.float64) row_total = row.sum() if row_total > 0: row = row / row_total else: row = np.ones(len(candidate_values), dtype=np.float64) / len(candidate_values) likelihood_matrix.append(row) likelihood_matrix = np.vstack(likelihood_matrix) value_probs = priors @ likelihood_matrix value_probs_sum = value_probs.sum() if value_probs_sum > 0: value_probs = value_probs / value_probs_sum else: value_probs = np.ones(len(candidate_values), dtype=np.float64) / len(candidate_values) expected_posterior_entropy = 0.0 posterior_examples = [] for value_idx, value in enumerate(candidate_values): p_value = value_probs[value_idx] if p_value <= 0: continue posterior = priors * likelihood_matrix[:, value_idx] posterior_sum = posterior.sum() if posterior_sum > 0: posterior = posterior / posterior_sum else: posterior = priors.copy() post_entropy = self.entropy(posterior) expected_posterior_entropy += p_value * post_entropy posterior_order = np.argsort(posterior)[::-1] posterior_examples.append({ "value": value, "estimated_value_probability": float(p_value), "posterior_top_genus": top_genera[int(posterior_order[0])], "posterior_top_probability": float(posterior[int(posterior_order[0])]), "posterior_entropy_bits": float(post_entropy), "posterior_distribution": [ { "genus": top_genera[int(i)], "probability": float(posterior[int(i)]), } for i in posterior_order ], }) empirical_ig = prior_entropy - expected_posterior_entropy pairwise_distances = [] for i in range(len(top_genera)): for j in range(i + 1, len(top_genera)): p = likelihood_matrix[i] q = likelihood_matrix[j] tv = 0.5 * np.abs(p - q).sum() weight = priors[i] * priors[j] pairwise_distances.append((tv, weight)) if pairwise_distances: weighted_tv = sum(tv * weight for tv, weight in pairwise_distances) total_weight = sum(weight for _, weight in pairwise_distances) weighted_tv = weighted_tv / total_weight if total_weight > 0 else 0.0 else: weighted_tv = 0.0 value_balance = self.safe_entropy_normalized(value_probs) return { "empirical_prior_entropy_bits": float(prior_entropy), "empirical_expected_posterior_entropy_bits": float(expected_posterior_entropy), "empirical_information_gain_bits": float(empirical_ig), "empirical_pairwise_tv_separation": float(weighted_tv), "empirical_value_balance": float(value_balance), "empirical_value_probabilities": { value: float(prob) for value, prob in zip(candidate_values, value_probs) }, "empirical_posterior_examples": posterior_examples, } @staticmethod def evidence_factor_from_records(evidence_records, soft_cap=300): if evidence_records <= 0: return 0.0 return float(min(1.0, math.log1p(evidence_records) / math.log1p(soft_cap))) def diversity_of_simulated_top_genera(self, simulated_rows): outcome_distribution = Counter() for row in simulated_rows: outcome_distribution[row["top_genus_after"]] += row["estimated_outcome_weight"] if not outcome_distribution: return 0.0, {} probs = np.array(list(outcome_distribution.values()), dtype=np.float64) probs = probs / probs.sum() return self.safe_entropy_normalized(probs), dict(outcome_distribution) def recommend( self, input_features, n_recommendations=5, top_competing_genera=5, max_candidate_values_per_field=8, include_context_fields=False, fields_to_consider=None, ): current_result, current_probs = self.predict_full_distribution(input_features) base_entropy = self.entropy(current_probs) base_top_prob, base_margin = self.top_margin(current_probs) top_genera = [ item["genus"] for item in current_result["ranked_genera"][:top_competing_genera] ] provided_fields = set(current_result["provided_fields"]) missing_fields = [ field for field in self.schema if field not in provided_fields ] if fields_to_consider is not None: candidate_fields = [ field for field in fields_to_consider if field in missing_fields ] else: if include_context_fields: candidate_fields = missing_fields else: candidate_fields = [ field for field in missing_fields if field in self.default_recommendable_fields ] field_results = [] for field in candidate_fields: observed_values = self.get_global_field_counter(field) if not observed_values: continue candidate_value_pairs = self.get_candidate_values_for_field( field=field, top_genera=top_genera, current_probs=current_probs, max_values=max_candidate_values_per_field, ) candidate_values = [value for value, _ in candidate_value_pairs] if not candidate_values: continue candidate_weights = self.estimate_candidate_value_weights( field=field, candidate_values=candidate_values, top_genera=top_genera, current_probs=current_probs, ) simulated_rows = [] expected_entropy = 0.0 expected_top_prob = 0.0 expected_margin = 0.0 top_prediction_stability = 0.0 for value, weight in zip(candidate_values, candidate_weights): simulated_features = dict(input_features) simulated_features[field] = value simulated_result, simulated_probs = self.predict_full_distribution(simulated_features) sim_entropy = self.entropy(simulated_probs) sim_top_prob, sim_margin = self.top_margin(simulated_probs) expected_entropy += float(weight) * sim_entropy expected_top_prob += float(weight) * sim_top_prob expected_margin += float(weight) * sim_margin if simulated_result["top_genus"] == current_result["top_genus"]: top_prediction_stability += float(weight) simulated_rows.append({ "field": field, "simulated_value": value, "estimated_outcome_weight": float(weight), "top_genus_after": simulated_result["top_genus"], "top_probability_after": simulated_result["top_probability"], "margin_after": simulated_result["margin"], "entropy_after": sim_entropy, "top5_after": [ { "genus": item["genus"], "probability": item["probability"], } for item in simulated_result["ranked_genera"][:5] ], }) model_information_gain = base_entropy - expected_entropy top_probability_gain = expected_top_prob - base_top_prob margin_gain = expected_margin - base_margin simulated_outcome_diversity, simulated_outcome_distribution = self.diversity_of_simulated_top_genera( simulated_rows ) challenge_rate = 1.0 - top_prediction_stability evidence_records = sum( self.get_genus_field_record_count(genus, field) for genus in top_genera ) evidence_factor = self.evidence_factor_from_records(evidence_records) empirical_disc = self.empirical_field_discrimination( field=field, candidate_values=candidate_values, top_genera=top_genera, current_probs=current_probs, ) empirical_ig = empirical_disc["empirical_information_gain_bits"] empirical_tv = empirical_disc["empirical_pairwise_tv_separation"] empirical_value_balance = empirical_disc["empirical_value_balance"] confirmation_score = ( model_information_gain + 0.25 * max(0.0, margin_gain) + 0.10 * max(0.0, top_probability_gain) + 0.05 * evidence_factor ) discriminatory_score_raw = ( 0.45 * empirical_ig + 0.35 * empirical_tv + 0.25 * simulated_outcome_diversity + 0.25 * challenge_rate + 0.15 * empirical_value_balance + 0.15 * model_information_gain ) discriminatory_score = discriminatory_score_raw * (0.35 + 0.65 * evidence_factor) if challenge_rate <= 0.001: discriminatory_score *= 0.35 max_outcome_weight = float(np.max(candidate_weights)) if len(candidate_weights) else 1.0 if max_outcome_weight >= 0.98: discriminatory_score *= 0.65 elif max_outcome_weight >= 0.95: discriminatory_score *= 0.80 top_candidate_values = sorted( simulated_rows, key=lambda x: x["estimated_outcome_weight"], reverse=True, ) field_results.append({ "field": field, "confirmation_score": float(confirmation_score), "discriminatory_score": float(discriminatory_score), "model_information_gain_bits": float(model_information_gain), "baseline_entropy_bits": float(base_entropy), "expected_entropy_after_bits": float(expected_entropy), "baseline_top_probability": float(base_top_prob), "expected_top_probability_after": float(expected_top_prob), "top_probability_gain": float(top_probability_gain), "baseline_margin": float(base_margin), "expected_margin_after": float(expected_margin), "margin_gain": float(margin_gain), "top_prediction_stability_probability": float(top_prediction_stability), "challenge_rate": float(challenge_rate), "simulated_outcome_diversity": float(simulated_outcome_diversity), "simulated_outcome_distribution": simulated_outcome_distribution, "empirical_information_gain_bits": float(empirical_ig), "empirical_pairwise_tv_separation": float(empirical_tv), "empirical_value_balance": float(empirical_value_balance), "empirical_prior_entropy_bits": empirical_disc["empirical_prior_entropy_bits"], "empirical_expected_posterior_entropy_bits": empirical_disc["empirical_expected_posterior_entropy_bits"], "empirical_value_probabilities": empirical_disc["empirical_value_probabilities"], "empirical_posterior_examples": empirical_disc["empirical_posterior_examples"], "evidence_records_among_top_genera": int(evidence_records), "evidence_factor": float(evidence_factor), "max_outcome_weight": float(max_outcome_weight), "candidate_values": [ { "value": row["simulated_value"], "estimated_outcome_weight": row["estimated_outcome_weight"], "top_genus_after": row["top_genus_after"], "top_probability_after": row["top_probability_after"], "margin_after": row["margin_after"], "entropy_after": row["entropy_after"], "top5_after": row["top5_after"], } for row in top_candidate_values ], }) confirmation_recommendations = sorted( field_results, key=lambda x: x["confirmation_score"], reverse=True, ) discriminatory_recommendations = sorted( field_results, key=lambda x: x["discriminatory_score"], reverse=True, ) return { "current_prediction": current_result, "baseline_entropy_bits": base_entropy, "top_competing_genera": top_genera, "confirmation_recommendations": confirmation_recommendations[:n_recommendations], "discriminatory_recommendations": discriminatory_recommendations[:n_recommendations], "all_field_scores": field_results, "settings": { "n_recommendations": n_recommendations, "top_competing_genera": top_competing_genera, "max_candidate_values_per_field": max_candidate_values_per_field, "include_context_fields": include_context_fields, }, "note": ( "Confirmation tests strengthen the current top prediction. " "Discriminatory tests separate the current top competing genera." ), } def print_next_test_report(result): current = result["current_prediction"] print() print("=" * 60) print("PHENOTYPECLASSIFIER NEXT-TEST RECOMMENDATIONS") print("=" * 60) print(f"Current top genus: {current['top_genus']}") print(f"Probability: {current['top_probability']:.4f}") print(f"Confidence: {current['confidence']}") print(f"Distinctness: {current['distinctness']}") print(f"Entropy bits: {result['baseline_entropy_bits']:.4f}") print("-" * 60) print("Top competing genera:") for item in current["ranked_genera"][:5]: print(f" - {item['genus']:<25} {item['probability']:.4f}") print() print("Discriminatory next tests:") print("-" * 60) for i, rec in enumerate(result["discriminatory_recommendations"], start=1): print(f"{i}. {rec['field']}") print(f" Discriminatory score: {rec['discriminatory_score']:.4f}") print(f" Model info gain: {rec['model_information_gain_bits']:.4f} bits") print(f" Pairwise separation: {rec['empirical_pairwise_tv_separation']:.4f}") print(f" Challenge rate: {rec['challenge_rate']:.4f}") print(" Likely outcomes:") for val in rec["candidate_values"][:5]: print( f" - {val['value']:<25} " f"weight={val['estimated_outcome_weight']:.3f} | " f"top={val['top_genus_after']} " f"p={val['top_probability_after']:.3f}" ) print() print("Confirmation next tests:") print("-" * 60) for i, rec in enumerate(result["confirmation_recommendations"], start=1): print(f"{i}. {rec['field']}") print(f" Confirmation score: {rec['confirmation_score']:.4f}") print(f" Model info gain: {rec['model_information_gain_bits']:.4f} bits") print(" Likely outcomes:") for val in rec["candidate_values"][:5]: print( f" - {val['value']:<25} " f"weight={val['estimated_outcome_weight']:.3f} | " f"top={val['top_genus_after']} " f"p={val['top_probability_after']:.3f}" ) print() print("-" * 60) print(result["note"]) print("=" * 60)