Spaces:
Runtime error
Runtime error
| 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"] | |
| 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()) | |
| 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 | |
| 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, | |
| } | |
| 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) | |