diff options
Diffstat (limited to 'experiments/analyze_contrastive_bias_b1.py')
| -rw-r--r-- | experiments/analyze_contrastive_bias_b1.py | 35 |
1 files changed, 27 insertions, 8 deletions
diff --git a/experiments/analyze_contrastive_bias_b1.py b/experiments/analyze_contrastive_bias_b1.py index 527b2ba..d750abf 100644 --- a/experiments/analyze_contrastive_bias_b1.py +++ b/experiments/analyze_contrastive_bias_b1.py @@ -19,16 +19,30 @@ def cell_path(cell_id): def valid_record(record, cell): history = record.get("history") or {} - return ( + base_valid = ( record.get("status") == "completed" and record.get("cell_id") == cell["cell_id"] and record.get("kind") == cell["kind"] and record.get("rule") == cell["rule"] and record.get("ratio") == cell["ratio"] - and history.get("finite") is True - and history.get("epochs_completed") == 20 and math.isnan(float(history.get("test_accuracy", float("nan")))) ) + if not base_valid: + return False + if history.get("finite") is True: + return history.get("epochs_completed") == 20 + # The protocol explicitly retains a differential raw condition that + # becomes nonfinite. No other condition may use this exception. + if not (cell["rule"] == "raw" and cell["kind"] in ("fixed", "activity")): + return False + completed = history.get("epochs_completed") + curves = history.get("curves") or {} + losses = list(curves.get("train_loss", [])) + list(curves.get("val_loss", [])) + return ( + isinstance(completed, int) and 1 <= completed <= 20 + and len(losses) >= 2 * completed + and any(not math.isfinite(float(value)) for value in losses) + ) def metric_max(record, key): @@ -69,11 +83,12 @@ def main(): candidates = [] table = [] for ratio in (0.25, 1.0, 4.0): - tag = rate_tag = f"{ratio:g}".replace(".", "p") + tag = f"{ratio:g}".replace(".", "p") row = {"ratio": ratio} for rule in ("raw", "innovation", "oracle"): record = records[f"activity-r{tag}-{rule}"] row[rule] = float(record["history"]["final_validation_accuracy"]) + row[rule + "_finite"] = bool(record["history"]["finite"]) row["raw_degradation"] = clean_acc - row["raw"] row["innovation_clean_gap"] = abs(row["innovation"] - clean_acc) row["innovation_oracle_gap"] = abs(row["innovation"] - row["oracle"]) @@ -81,7 +96,8 @@ def main(): records[f"activity-r{tag}-innovation"], "post_bias_raw_bias_rms_ratio") row["passes"] = ( - row["raw_degradation"] >= 5.0 + (not row["raw_finite"] or row["raw_degradation"] >= 5.0) + and row["innovation_finite"] and row["oracle_finite"] and row["innovation_clean_gap"] <= 2.0 and row["innovation_oracle_gap"] <= 1.0 and row["innovation_post_bias_ratio_max"] <= 1e-3 @@ -101,16 +117,19 @@ def main(): registry_values = {row["registry_sha256"] for row in records.values()} checks["single_source_lock"] = len(source_values) == 1 checks["single_registry_lock"] = len(registry_values) == 1 + gate = "pass" if all(checks.values()) else "fail" report = { - "stage": "contrastive_bias_b1", "gate": ( - "pass" if all(checks.values()) else "fail"), + "stage": "contrastive_bias_b1", "gate": gate, "checks": checks, "clean_final_validation_accuracy": clean_acc, "common_final_validation_accuracy": common_acc, "common_maximum_difference_relative_error": common_error, "innovation_maximum_post_bias_ratio": innovation_post_max, "predictor_maximum_instruction_observations": predictor_instruction_max, "activity_table": table, - "selected_confirmation_ratio": max(candidates) if candidates else None, + "largest_core_passing_activity_ratio": ( + max(candidates) if candidates else None), + "selected_confirmation_ratio": ( + max(candidates) if candidates and gate == "pass" else None), "num_expected_records": 17, "num_audited_records": len(records), "source": clean["source"], "registry_sha256": clean["registry_sha256"], } |