Buckets:
| from __future__ import annotations | |
| import json | |
| import io | |
| import os | |
| import shutil | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import time | |
| import unittest | |
| from contextlib import redirect_stdout | |
| from unittest import mock | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1] | |
| if str(ROOT) not in sys.path: | |
| sys.path.insert(0, str(ROOT)) | |
| from eval.certification import certify | |
| from eval.best_run_tracker import update_best_run | |
| from eval.evidence_producers import produce_required_eval_evidence | |
| from n21.config import write_json | |
| from eval.frozen_suite import validate_frozen_suite | |
| from eval.hf_frozen_eval import paired_report | |
| from eval.model_quality_gate import evaluate_model_quality_gate | |
| from n21.settings import REPO_ROOT | |
| from n21.settings import IMPLEMENTATION_PRODUCTS_ROOT, SHFT_WORKSPACE_ROOT | |
| from model_policy.selector import select_model | |
| from model_policy.profiles import apply_provider_profile, resolve_model_profile | |
| from model_policy.roadmap import roadmap_report | |
| from approvals.proof_chain import validate_promotion_proof_chain | |
| from rollback.anchor import ensure_last_good_anchor, validate_rollback_anchor | |
| from monitoring.canary import run_canary_monitor | |
| from orchestrator.cycle_controller import run_self_healing_cycles | |
| from orchestrator.continuous_status import write_continuous_status | |
| from orchestrator.human_owner_decision import _send_email_or_outbox, request_human_owner_instruction | |
| from eval.human_spot_check_email import request_human_spot_check_approval | |
| from orchestrator.dataset_provenance import check_resume_provenance | |
| from orchestrator.lifecycle_proof import run_lifecycle_proof | |
| from orchestrator.provider_routing import validate_route | |
| from orchestrator.stall_breakout import run_stall_breakout | |
| from security.secret_scan import scan_text, scan_tree | |
| from training.launch import run_training | |
| from training.iteration_policy import should_stop | |
| from training.start_policy import resolve_training_start | |
| from training.hf_fingpt_train import build_plan, build_trainer_metrics_summary, load_jsonl | |
| from providers import hf_managed | |
| from providers.hf_managed import HFManagedProvider | |
| from providers.hf_staging import stage_hf_dataset | |
| from data_pipeline.learning_pdf_to_jsonl import ( | |
| build_training_jsonl_from_learning, | |
| load_jsonl as load_learning_jsonl, | |
| write_jsonl as write_learning_jsonl, | |
| ) | |
| from data_pipeline.nonrepair_balance import generate_nonrepair_balance_data | |
| from data_pipeline.repair_coverage import evaluate_repair_coverage | |
| from data_pipeline.ingest import ingest_dataset | |
| from data_pipeline.reasoning_data_generation import generate_grounded_reasoning_examples | |
| from data_pipeline.source_intake import intake_public_sources | |
| from data_pipeline.source_quality_certifier import certify_normalized_source_content, certify_source_candidate | |
| from data_pipeline.training_data_validation import validate_training_data | |
| import n21.cli as shft_cli | |
| def _training_certification() -> dict[str, object]: | |
| return { | |
| "schema_version": "source_ai_certification_v1", | |
| "method": "test_certification", | |
| "intended_use": "training", | |
| "training_eligible": True, | |
| "verification_eligible": True, | |
| "score": 6.0, | |
| "rationale": "High-signal worked examples with red-flag, pass/fail, and because-style reasoning.", | |
| } | |
| def _verification_certification() -> dict[str, object]: | |
| return { | |
| "schema_version": "source_ai_certification_v1", | |
| "method": "test_certification", | |
| "intended_use": "verification", | |
| "training_eligible": False, | |
| "verification_eligible": True, | |
| "score": 3.0, | |
| "rationale": "Public source is useful for verification but needs grounded reasoning conversion before training.", | |
| } | |
| class SHFTMVPTests(unittest.TestCase): | |
| def test_fingpt_first_selector(self) -> None: | |
| record = select_model("finance_qa", "dev") | |
| self.assertEqual(record["selected_model"], "linvest21/linvest21_fingpt_v1_000") | |
| self.assertEqual(record["strategy"], "fingpt_first_bootstrap") | |
| self.assertTrue(record["roadmap_evidence"]["known_to_roadmap"]) | |
| self.assertEqual(record["roadmap_evidence"]["weighted_score"], 0.7) | |
| def test_model_roadmap_quantifies_fingpt_and_long_context_path(self) -> None: | |
| report = roadmap_report() | |
| self.assertGreaterEqual(report["candidate_count"], 6) | |
| self.assertEqual(report["current_default_model"], "linvest21/linvest21_fingpt_v1_000") | |
| self.assertEqual(report["current_default_context_tokens"], 8192) | |
| self.assertGreaterEqual(report["current_default_weighted_score"], 0.7) | |
| self.assertEqual(report["highest_context_tokens"], 131072) | |
| self.assertIn("allow_foundation_upgrade_when", report["promotion_rules"]) | |
| evidence = roadmap_report()["ranked_by_score"][0] | |
| self.assertIn("weighted_score", evidence) | |
| def test_training_manifest_records_model_roadmap_evidence(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| result = run_training( | |
| Path(tmp), | |
| run_id="test_roadmap_manifest", | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| train_provider="hf_managed", | |
| infer_provider="hf_managed", | |
| ) | |
| manifest_evidence = result["run_manifest"]["model_roadmap_evidence"] | |
| iteration_evidence = result["iteration_evidence"]["model_roadmap_evidence"] | |
| self.assertTrue(manifest_evidence["known_to_roadmap"]) | |
| self.assertEqual(manifest_evidence["weighted_score"], 0.7) | |
| self.assertEqual(manifest_evidence, iteration_evidence) | |
| def test_resume_provenance_blocks_learning_data_newer_than_training(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| run_path = root / "run_stale" | |
| source = root / "learning" | |
| (run_path / "trainer_state").mkdir(parents=True) | |
| source.mkdir() | |
| handle = run_path / "trainer_state" / "train_handle.json" | |
| learning = source / "synthetic_equity_researcher_critical_reasoning.hf_finetune.jsonl" | |
| handle.write_text("{}", encoding="utf-8") | |
| learning.write_text('{"messages":[]}\n', encoding="utf-8") | |
| now = time.time() | |
| os.utime(handle, (now - 100, now - 100)) | |
| os.utime(learning, (now, now)) | |
| result = check_resume_provenance(run_path, source) | |
| self.assertFalse(result["can_resume"]) | |
| self.assertTrue(result["stale_training_artifacts"]) | |
| self.assertTrue(result["force_new_run_required"]) | |
| self.assertEqual(result["reason"], "learning_corpus_changed_after_training_artifact") | |
| def test_resume_provenance_allows_training_newer_than_learning_data(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| run_path = root / "run_fresh" | |
| source = root / "learning" | |
| (run_path / "trainer_state").mkdir(parents=True) | |
| source.mkdir() | |
| handle = run_path / "trainer_state" / "train_handle.json" | |
| learning = source / "curated.hf_finetune.jsonl" | |
| handle.write_text("{}", encoding="utf-8") | |
| learning.write_text('{"messages":[]}\n', encoding="utf-8") | |
| now = time.time() | |
| os.utime(learning, (now - 100, now - 100)) | |
| os.utime(handle, (now, now)) | |
| result = check_resume_provenance(run_path, source) | |
| self.assertTrue(result["can_resume"]) | |
| self.assertFalse(result["stale_training_artifacts"]) | |
| self.assertEqual(result["reason"], "learning_corpus_not_newer_than_training_artifact") | |
| def test_training_start_policy_defaults_to_bootstrap_adapter(self) -> None: | |
| start = resolve_training_start( | |
| release_id="linvest21_fingpt_equity_researcher_v1_test", | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| start_policy="bootstrap", | |
| ) | |
| self.assertEqual(start["policy"], "bootstrap") | |
| self.assertEqual(start["start_adapter"], "linvest21/linvest21_fingpt_v1_000") | |
| self.assertIsNone(start["continued_from_run_id"]) | |
| def test_qwen3_profile_starts_fresh_lora_without_bootstrap_adapter(self) -> None: | |
| profile = resolve_model_profile("qwen3_32b") | |
| self.assertEqual(profile["model_candidate"], "Qwen/Qwen3-32B") | |
| self.assertEqual(profile["base_model_id"], "Qwen/Qwen3-32B") | |
| self.assertFalse(profile["adapter_bootstrap"]) | |
| start = resolve_training_start( | |
| release_id="linvest21_qwen3_equity_researcher_v1_test", | |
| model_candidate=profile["model_candidate"], | |
| start_policy="bootstrap", | |
| adapter_bootstrap=bool(profile["adapter_bootstrap"]), | |
| ) | |
| self.assertEqual(start["source"], "fresh_base_model_lora") | |
| self.assertIsNone(start["start_adapter"]) | |
| def test_training_start_policy_continue_best_uses_best_checkpoint_only(self) -> None: | |
| release_id = "linvest21_fingpt_equity_researcher_v1_test_start_policy" | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| if best_path.exists(): | |
| best_path.unlink() | |
| try: | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_linvest21_fingpt_equity_researcher_v1_test_best", | |
| "candidate_aggregate": 0.61, | |
| "candidate_critical_pass_rate": 0.72, | |
| "pairwise_win_rate": 0.88, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| start = resolve_training_start( | |
| release_id=release_id, | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| start_policy="continue-best", | |
| ) | |
| self.assertEqual(start["policy"], "continue-best") | |
| self.assertEqual(start["bootstrap_adapter"], "linvest21/linvest21_fingpt_v1_000") | |
| self.assertEqual(start["continued_from_run_id"], "run_linvest21_fingpt_equity_researcher_v1_test_best") | |
| self.assertEqual(start["start_adapter"], "/artifacts/runs/run_linvest21_fingpt_equity_researcher_v1_test_best/adapter") | |
| self.assertEqual(start["best_run_metrics"]["candidate_aggregate"], 0.61) | |
| finally: | |
| if best_path.exists(): | |
| best_path.unlink() | |
| def test_training_start_policy_continue_best_falls_back_when_no_best_exists(self) -> None: | |
| release_id = "linvest21_fingpt_equity_researcher_v1_test_missing_best" | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| if best_path.exists(): | |
| best_path.unlink() | |
| start = resolve_training_start( | |
| release_id=release_id, | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| start_policy="continue-best", | |
| ) | |
| self.assertEqual(start["policy"], "continue-best") | |
| self.assertEqual(start["source"], "no_best_recorded_fallback_bootstrap") | |
| self.assertEqual(start["start_adapter"], "linvest21/linvest21_fingpt_v1_000") | |
| self.assertIsNone(start["continued_from_run_id"]) | |
| def test_training_start_policy_continue_best_all_18_roles_do_not_raise_without_best(self) -> None: | |
| assets = ["equity", "fixed_income", "multi_asset"] | |
| roles = [ | |
| "chief_investment_officer", | |
| "client_portfolio_manager", | |
| "performance_manager", | |
| "portfolio_manager", | |
| "researcher", | |
| "risk_manager", | |
| ] | |
| for asset in assets: | |
| for role in roles: | |
| release_id = f"linvest21_fingpt_{asset}_{role}_v1_test_missing_best" | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| if best_path.exists(): | |
| best_path.unlink() | |
| start = resolve_training_start( | |
| release_id=release_id, | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| start_policy="continue-best", | |
| ) | |
| self.assertEqual(start["source"], "no_best_recorded_fallback_bootstrap") | |
| def test_hf_job_command_records_finetune_start_policy(self) -> None: | |
| command = HFManagedProvider._build_jobs_command( | |
| { | |
| "run_id": "run_test_start_policy", | |
| "model_candidate": "linvest21/linvest21_fingpt_v1_000", | |
| "training_start": { | |
| "policy": "continue-best", | |
| "start_adapter": "/artifacts/runs/run_test_best/adapter", | |
| }, | |
| }, | |
| { | |
| "namespace": "linvest21", | |
| "storage": { | |
| "bucket": "linvest21/shft-artifacts", | |
| "dataset_repo": "linvest21/shft-datasets", | |
| }, | |
| "jobs": { | |
| "flavor": "a100-large", | |
| "timeout": "8h", | |
| "image": "pytorch/pytorch:2.6.0-cuda12.4-cudnn9-devel", | |
| "base_model_id": "meta-llama/Meta-Llama-3-8B", | |
| "training": {"max_steps": 300}, | |
| }, | |
| }, | |
| ) | |
| joined = "\n".join(command) | |
| self.assertIn("SHFT_FINETUNE_START_POLICY=continue-best", joined) | |
| self.assertIn("--start-adapter\n/artifacts/runs/run_test_best/adapter", joined) | |
| self.assertIn("--finetune-start-policy\ncontinue-best", joined) | |
| def test_qwen3_hf_job_command_omits_start_adapter_and_mounts_base(self) -> None: | |
| profile = resolve_model_profile("qwen3_32b") | |
| config = apply_provider_profile( | |
| { | |
| "namespace": "linvest21", | |
| "storage": { | |
| "bucket": "linvest21/shft-artifacts", | |
| "dataset_repo": "linvest21/shft-datasets", | |
| "model_repo": "linvest21/linvest21_fingpt_v1_000", | |
| }, | |
| "jobs": { | |
| "flavor": "a100-large", | |
| "timeout": "8h", | |
| "image": "pytorch/pytorch:2.6.0-cuda12.4-cudnn9-devel", | |
| "base_model_id": "meta-llama/Meta-Llama-3-8B", | |
| "training": {"max_steps": 300}, | |
| }, | |
| }, | |
| profile, | |
| ) | |
| command = HFManagedProvider._build_jobs_command( | |
| { | |
| "run_id": "run_test_qwen3_profile", | |
| "model_candidate": "Qwen/Qwen3-32B", | |
| "training_start": {"policy": "bootstrap", "start_adapter": None}, | |
| }, | |
| config, | |
| ) | |
| joined = "\n".join(command) | |
| self.assertIn("hf://Qwen/Qwen3-32B:/models/base:ro", joined) | |
| self.assertIn("--base-model-id\nQwen/Qwen3-32B", joined) | |
| self.assertNotIn("--start-adapter", command) | |
| def test_provider_route_validation(self) -> None: | |
| self.assertEqual(validate_route("hf_managed", "local_native"), []) | |
| self.assertTrue(validate_route("unknown", "local_native")) | |
| def test_certification_passes_default_fixture(self) -> None: | |
| report = certify("dev", "finance_qa") | |
| self.assertEqual(report["gate_result"], "pass") | |
| self.assertGreater(report["improvement_report"]["improvements"]["aggregate"]["pct"], 0) | |
| def test_certification_can_record_model_roadmap_evidence(self) -> None: | |
| report = certify("dev", "finance_qa", model_candidate="linvest21/linvest21_fingpt_v1_000") | |
| evidence = report["model_roadmap_evidence"] | |
| self.assertTrue(evidence["known_to_roadmap"]) | |
| self.assertEqual(evidence["score_rank"], 1) | |
| self.assertEqual(evidence["context_tokens"], 8192) | |
| def test_self_healing_cycle_records_model_roadmap_evidence(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| run_path = Path(tmp) | |
| summary = run_self_healing_cycles( | |
| run_path, | |
| run_id="test_cycle_roadmap", | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| train_provider="hf_managed", | |
| infer_provider="hf_managed", | |
| max_cycles=1, | |
| ) | |
| evidence = summary["model_roadmap_evidence"] | |
| self.assertTrue(evidence["known_to_roadmap"]) | |
| self.assertEqual(evidence["weighted_score"], 0.7) | |
| self.assertEqual(summary["cycles"][0]["model_roadmap_evidence"], evidence) | |
| def test_promotion_proof_chain_requires_cycle_and_certification(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| run_path = Path(tmp) | |
| run_self_healing_cycles( | |
| run_path, | |
| run_id="test_promotion_proof", | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| train_provider="hf_managed", | |
| infer_provider="hf_managed", | |
| max_cycles=1, | |
| ) | |
| write_json( | |
| run_path / "eval" / "certification_report.json", | |
| certify("dev", "finance_qa", model_candidate="linvest21/linvest21_fingpt_v1_000"), | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "scoring_mode": "deterministic_heuristic_v0", | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.62, "critical_pass_rate": 0.75, "sample_count": 120}, | |
| "candidate": {"aggregate": 0.84, "critical_pass_rate": 0.93, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.22, | |
| "aggregate_pct": 35.4839, | |
| "critical_pass_rate_abs": 0.18, | |
| "pairwise_win_rate": 0.75, | |
| "pairwise_loss_rate": 0.0, | |
| "wins": 90, | |
| "ties": 30, | |
| "losses": 0, | |
| }, | |
| "promotion_gate": {"eligible_for_promotion": True}, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "remote_artifacts" / "training_plan.json", | |
| { | |
| "train_records": 160, | |
| "valid_records": 20, | |
| "hyperparameters": {"max_steps": 300}, | |
| "readiness": {"production_candidate": True, "warnings": []}, | |
| }, | |
| ) | |
| selected_checkpoint = { | |
| "schema_version": "shft_selected_checkpoint_v1", | |
| "selection_metric": "eval_loss", | |
| "selection_metric_value": 1.1, | |
| "selected_checkpoint": "/artifacts/runs/test_promotion_proof/checkpoint-100", | |
| "selected_step": 100, | |
| "candidate_adapter_dir": "/artifacts/runs/test_promotion_proof/adapter", | |
| } | |
| write_json(run_path / "remote_artifacts" / "selected_checkpoint.json", selected_checkpoint) | |
| write_json( | |
| run_path / "remote_artifacts" / "trainer_metrics_summary.json", | |
| { | |
| "schema_version": "shft_trainer_metrics_summary_v1", | |
| "eval_row_count": 3, | |
| "train_log_row_count": 3, | |
| "best_eval_loss": 1.1, | |
| "final_eval_loss": 1.12, | |
| "final_train_loss": 1.0, | |
| "train_eval_loss_gap": 0.12, | |
| "late_eval_loss_regression": 0.02, | |
| "selected_checkpoint": selected_checkpoint, | |
| "overfit_detected": False, | |
| "overfit_flags": [], | |
| }, | |
| ) | |
| write_json( | |
| run_path / "remote_artifacts" / "training_result.json", | |
| { | |
| "status": "completed", | |
| "selected_checkpoint": selected_checkpoint, | |
| "overfit_detected": False, | |
| "overfit_flags": [], | |
| }, | |
| ) | |
| write_json( | |
| run_path / "dataset_snapshot" / "dataset_manifest.json", | |
| { | |
| "quality": {"record_count": 200}, | |
| "split_counts": {"train": 160, "valid": 20, "test": 20}, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "model_judge_report.json", | |
| { | |
| "rubric_version": "model_as_judge_rubric_v1", | |
| "sample_count": 40, | |
| "mean_score": 0.86, | |
| "critical_pass_rate": 0.95, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "human_spot_check_report.json", | |
| {"status": "approved", "sample_count": 12, "critical_failures": 0, "approved": True}, | |
| ) | |
| proof = validate_promotion_proof_chain(run_path) | |
| self.assertTrue(proof["ok"], proof["errors"]) | |
| self.assertTrue(any("fixture/orchestration-only" in item for item in proof["warnings"])) | |
| self.assertEqual(proof["evidence_summary"]["aggregate_improvement_pct"], 2.1493) | |
| self.assertTrue(proof["evidence_summary"]["model_quality_promotion_eligible"]) | |
| def test_model_quality_gate_blocks_missing_judge_and_human_review(self) -> None: | |
| report = evaluate_model_quality_gate( | |
| paired_eval={ | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.6, "critical_pass_rate": 0.8}, | |
| "candidate": {"aggregate": 0.82, "critical_pass_rate": 0.92}, | |
| "improvement": { | |
| "aggregate_abs": 0.22, | |
| "aggregate_pct": 36.6, | |
| "critical_pass_rate_abs": 0.12, | |
| "pairwise_loss_rate": 0.0, | |
| "pairwise_win_rate": 0.8, | |
| }, | |
| }, | |
| training_plan={ | |
| "train_records": 160, | |
| "valid_records": 20, | |
| "hyperparameters": {"max_steps": 300}, | |
| "readiness": {"production_candidate": True, "warnings": []}, | |
| }, | |
| training_result={"selected_checkpoint": {"selection_metric": "eval_loss"}}, | |
| trainer_metrics_summary={ | |
| "eval_row_count": 3, | |
| "train_eval_loss_gap": 0.1, | |
| "late_eval_loss_regression": 0.01, | |
| "overfit_detected": False, | |
| "overfit_flags": [], | |
| }, | |
| selected_checkpoint={ | |
| "selection_metric": "eval_loss", | |
| "selection_metric_value": 1.0, | |
| "selected_checkpoint": "checkpoint-100", | |
| }, | |
| dataset_manifest={"quality": {"record_count": 200}, "split_counts": {"train": 160, "valid": 20, "test": 20}}, | |
| ) | |
| self.assertFalse(report["ok"]) | |
| self.assertTrue(any("model_as_judge_present" in item for item in report["errors"])) | |
| self.assertTrue(any("human_spot_check_present" in item for item in report["errors"])) | |
| def test_trainer_metrics_summary_flags_late_eval_regression(self) -> None: | |
| selected = { | |
| "selection_metric": "eval_loss", | |
| "selection_metric_value": 1.0, | |
| "selected_checkpoint": "checkpoint-50", | |
| "selected_step": 50, | |
| } | |
| summary = build_trainer_metrics_summary( | |
| rows=[ | |
| {"step": 50, "eval_loss": 1.0}, | |
| {"step": 100, "loss": 0.7, "mean_token_accuracy": 0.6}, | |
| {"step": 100, "eval_loss": 1.3}, | |
| ], | |
| selected_checkpoint=selected, | |
| overfit_tolerance=0.10, | |
| ) | |
| self.assertTrue(summary["overfit_detected"]) | |
| self.assertIn("late_eval_loss_regression", summary["overfit_flags"]) | |
| self.assertEqual(summary["best_eval_step"], 50) | |
| def test_model_quality_gate_requires_selected_checkpoint_evidence(self) -> None: | |
| report = evaluate_model_quality_gate( | |
| paired_eval={ | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.7, "critical_pass_rate": 0.8}, | |
| "candidate": {"aggregate": 0.8, "critical_pass_rate": 0.9}, | |
| "improvement": { | |
| "aggregate_abs": 0.1, | |
| "critical_pass_rate_abs": 0.1, | |
| "pairwise_loss_rate": 0.0, | |
| "pairwise_win_rate": 0.8, | |
| }, | |
| }, | |
| training_plan={ | |
| "train_records": 200, | |
| "valid_records": 20, | |
| "hyperparameters": {"max_steps": 300}, | |
| "readiness": {"production_candidate": True}, | |
| }, | |
| dataset_manifest={"quality": {"record_count": 240}, "split_counts": {"train": 200, "valid": 20, "test": 20}}, | |
| model_judge_report={ | |
| "rubric_version": "model_as_judge_rubric_v1", | |
| "sample_count": 40, | |
| "mean_score": 0.9, | |
| "critical_pass_rate": 0.95, | |
| }, | |
| human_review_report={"status": "approved", "sample_count": 12, "critical_failures": 0, "approved": True}, | |
| ) | |
| self.assertFalse(report["ok"]) | |
| self.assertTrue(any("selected_checkpoint_present" in item for item in report["errors"])) | |
| self.assertTrue(any("trainer_metrics_summary_present" in item for item in report["errors"])) | |
| def test_required_eval_evidence_producers_remove_missing_artifact_errors_without_faking_quality(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| run_path = Path(tmp) / "run_test_eval_evidence" | |
| (run_path / "eval").mkdir(parents=True) | |
| paired = { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.34, "critical_pass_rate": 0.30, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.34, | |
| "aggregate_pct": None, | |
| "critical_pass_rate_abs": 0.30, | |
| "pairwise_win_rate": 0.80, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| } | |
| manifest = produce_required_eval_evidence( | |
| run_path, | |
| release_id="linvest21_fingpt_equity_researcher_v1_test", | |
| paired_eval=paired, | |
| ) | |
| self.assertEqual(manifest["baseline_proof"]["proof_mode"], "absolute_only_cold_start") | |
| report = evaluate_model_quality_gate( | |
| paired_eval=paired, | |
| training_plan={ | |
| "train_records": 200, | |
| "valid_records": 20, | |
| "hyperparameters": {"max_steps": 300}, | |
| "readiness": {"production_candidate": True}, | |
| }, | |
| training_result={"selected_checkpoint": {"selection_metric": "eval_loss"}}, | |
| trainer_metrics_summary={ | |
| "eval_row_count": 3, | |
| "train_eval_loss_gap": 0.1, | |
| "late_eval_loss_regression": 0.01, | |
| "overfit_detected": False, | |
| "overfit_flags": [], | |
| }, | |
| selected_checkpoint={ | |
| "selection_metric": "eval_loss", | |
| "selection_metric_value": 1.0, | |
| "selected_checkpoint": "checkpoint-100", | |
| }, | |
| dataset_manifest={"quality": {"record_count": 120}, "split_counts": {"train": 96, "valid": 12, "test": 12}}, | |
| model_judge_report=manifest["model_judge"], | |
| human_review_report=manifest["human_spot_check"], | |
| baseline_proof_report=manifest["baseline_proof"], | |
| ) | |
| self.assertTrue(report["checks"]["nonzero_baseline_for_relative_proof"]["ok"]) | |
| self.assertFalse(report["ok"]) | |
| self.assertFalse(any("model_as_judge_present" in item for item in report["errors"])) | |
| self.assertFalse(any("human_spot_check_present" in item for item in report["errors"])) | |
| self.assertTrue(any("candidate_aggregate_absolute" in item for item in report["errors"])) | |
| def test_required_eval_evidence_can_wait_for_human_email_approval_response_file(self) -> None: | |
| run_id = "run_test_human_review_email_approve" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| review_root = SHFT_WORKSPACE_ROOT / "human_spot_check_reviews" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if review_root.exists(): | |
| shutil.rmtree(review_root) | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| write_json( | |
| run_path / "eval" / "human_spot_check_response.json", | |
| {"decision": "approve", "reviewed_samples": 10, "critical_failures": 0, "reviewer": "test reviewer"}, | |
| ) | |
| paired = { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.91, "critical_pass_rate": 0.95, "sample_count": 120}, | |
| "improvement": {"pairwise_win_rate": 0.88, "pairwise_loss_rate": 0.0, "wins": 105, "ties": 15, "losses": 0}, | |
| } | |
| fake_email = { | |
| "delivered": True, | |
| "delivery_status": "mock_sent", | |
| "to": "david.d.lin@linvest21.com", | |
| "subject": "SHFT human spot-check approval required", | |
| "outbox_path": "mock_outbox.json", | |
| } | |
| with mock.patch("eval.human_spot_check_email._send_email_or_outbox", return_value=fake_email), mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_HUMAN_REVIEW_READ_STDIN": "false", | |
| "SHFT_HUMAN_REVIEW_POLL_SECONDS": "0.01", | |
| }, | |
| clear=False, | |
| ): | |
| manifest = produce_required_eval_evidence( | |
| run_path, | |
| release_id="linvest21_test_release", | |
| paired_eval=paired, | |
| request_human_email=True, | |
| human_email_timeout_seconds=1, | |
| ) | |
| human = manifest["human_spot_check"] | |
| self.assertTrue(human["approved"]) | |
| self.assertEqual(human["critical_failures"], 0) | |
| self.assertEqual(human["status"], "approved") | |
| self.assertEqual(human["review_decision"]["source"], "response_file") | |
| self.assertTrue((run_path / "eval" / "human_spot_check_email_request.json").exists()) | |
| self.assertTrue((run_path / "eval" / "human_spot_check_approval.json").exists()) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if review_root.exists(): | |
| shutil.rmtree(review_root) | |
| def test_human_spot_check_email_timeout_fails_closed(self) -> None: | |
| run_id = "run_test_human_review_email_timeout" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| review_root = SHFT_WORKSPACE_ROOT / "human_spot_check_reviews" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if review_root.exists(): | |
| shutil.rmtree(review_root) | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| paired = { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.91, "critical_pass_rate": 0.95, "sample_count": 120}, | |
| "improvement": {"pairwise_win_rate": 0.88, "pairwise_loss_rate": 0.0}, | |
| } | |
| fake_email = { | |
| "delivered": False, | |
| "delivery_status": "outbox_written_requested", | |
| "to": "david.d.lin@linvest21.com", | |
| "subject": "SHFT human spot-check approval required", | |
| "outbox_path": "mock_outbox.json", | |
| } | |
| with mock.patch("eval.human_spot_check_email._send_email_or_outbox", return_value=fake_email), mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_HUMAN_REVIEW_READ_STDIN": "false", | |
| "SHFT_HUMAN_REVIEW_POLL_SECONDS": "0.01", | |
| }, | |
| clear=False, | |
| ): | |
| approval = request_human_spot_check_approval( | |
| run_dir=run_path, | |
| release_id="linvest21_test_release", | |
| paired_eval=paired, | |
| timeout_seconds=0, | |
| stdout=io.StringIO(), | |
| stderr=io.StringIO(), | |
| ) | |
| self.assertFalse(approval["ok"]) | |
| self.assertFalse(approval["human_spot_check_report"]["approved"]) | |
| self.assertEqual(approval["human_spot_check_report"]["status"], "pending_human_review") | |
| self.assertEqual(approval["decision"]["decision"], "pending") | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if review_root.exists(): | |
| shutil.rmtree(review_root) | |
| def test_paired_report_is_not_enough_for_full_model_quality_promotion(self) -> None: | |
| rows = [] | |
| for idx in range(120): | |
| rows.append( | |
| { | |
| "task": "finance_qa", | |
| "baseline_score": {"score": 0.6, "critical_pass": True}, | |
| "candidate_score": {"score": 0.9, "critical_pass": True}, | |
| } | |
| ) | |
| report = paired_report(rows) | |
| self.assertFalse(report["promotion_gate"]["eligible_for_promotion"]) | |
| self.assertIn("training_plan_present", "\n".join(report["promotion_gate"]["errors"])) | |
| def test_rollback_anchor_is_checksum_valid(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| anchor = ensure_last_good_anchor(Path(tmp), env="stage", model_id="linvest21/linvest21_fingpt_v1_000") | |
| self.assertEqual(validate_rollback_anchor(anchor), []) | |
| self.assertEqual(anchor["type"], "last_good") | |
| self.assertEqual(len(anchor["checksum_sha256"]), 64) | |
| def test_canary_monitor_recommends_rollback_on_threshold_breach(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| run_path = Path(tmp) | |
| ensure_last_good_anchor(SHFT_WORKSPACE_ROOT, env="stage", model_id="linvest21/linvest21_fingpt_v1_000") | |
| write_json( | |
| run_path / "manifests" / "promotion_manifest.json", | |
| {"run_id": "test_canary", "env": "stage", "status": "promotion_planned"}, | |
| ) | |
| report = run_canary_monitor(run_path, run_id="test_canary", env="stage", mode="fail") | |
| self.assertTrue(report["rollback_recommended"]) | |
| self.assertGreaterEqual(len(report["rollback_reasons"]), 1) | |
| self.assertEqual(report["status"], "rollback_recommended") | |
| def test_lifecycle_proof_runs_end_to_end(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| report = run_lifecycle_proof( | |
| Path(tmp), | |
| run_id="test_lifecycle_proof", | |
| env="stage", | |
| max_cycles=1, | |
| canary_mode="pass", | |
| ) | |
| self.assertEqual(report["status"], "rollback_recommended") | |
| self.assertEqual(report["evidence"]["promotion_status"], "promotion_blocked") | |
| self.assertTrue(report["evidence"]["rollback_recommended"]) | |
| self.assertEqual(len(report["lifecycle_map"]), 7) | |
| def test_iteration_stop_policy(self) -> None: | |
| self.assertEqual(should_stop(1, 0, True), (True, "candidate_ready_for_paired_eval")) | |
| self.assertEqual(should_stop(5, 0, False), (True, "max_iterations_reached")) | |
| def test_orchestration_fixture_cycles_do_not_claim_certification(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| with mock.patch("sys.stdout"): | |
| summary = run_self_healing_cycles( | |
| Path(tmp), | |
| run_id="test_fixture_semantics", | |
| model_candidate="linvest21/linvest21_fingpt_v1_000", | |
| train_provider="hf_managed", | |
| infer_provider="hf_managed", | |
| max_cycles=1, | |
| ) | |
| cycle = summary["cycles"][0] | |
| self.assertEqual(cycle["gate_result"], "fixture_pass") | |
| self.assertEqual(cycle["quality_signal"], "orchestration_only") | |
| self.assertEqual(cycle["promotion_recommendation"], "fixture_only_await_paired_eval") | |
| self.assertEqual(cycle["stop_reason"], "candidate_ready_for_paired_eval") | |
| def test_export_release_requires_certified_quality_gate(self) -> None: | |
| args = type( | |
| "Args", | |
| (), | |
| { | |
| "release_id": "test_release_guard", | |
| "source_run_id": "test_uncertified_run", | |
| "export_mode": "adapter_only", | |
| "base_model": "meta-llama/Meta-Llama-3-8B", | |
| "model_id": "test_release_guard", | |
| "asset_class": "equity", | |
| "role": "researcher", | |
| "merged_model_dir": None, | |
| "gguf_model_path": None, | |
| "zip": False, | |
| "allow_uncertified_export": False, | |
| }, | |
| )() | |
| with tempfile.TemporaryDirectory() as tmp: | |
| fake_run = Path(tmp) / "runs" / "test_uncertified_run" | |
| fake_run.mkdir(parents=True) | |
| with mock.patch("n21.cli.run_dir", return_value=fake_run), mock.patch( | |
| "n21.cli.export_release" | |
| ) as export_mock, mock.patch("n21.cli.ensure_workspace"), mock.patch("sys.stdout"): | |
| rc = shft_cli.command_export_release(args) | |
| self.assertEqual(rc, 6) | |
| export_mock.assert_not_called() | |
| def test_secret_scan(self) -> None: | |
| self.assertFalse(scan_text("HF_TOKEN is allowed as a reference")) | |
| self.assertFalse(scan_text("https://example.com/value-at-risk-9780071464956-usa")) | |
| self.assertTrue(scan_text("hf_" + "a" * 30)) | |
| self.assertTrue(scan_text("sk-" + "a" * 30)) | |
| def test_secret_scan_skips_large_generated_workspace_files(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| skipped = root / "impl_codex" / "shft_workspace" / "runs" / "run_x" | |
| skipped.mkdir(parents=True) | |
| (skipped / "log.json").write_text("hf_" + "a" * 30, encoding="utf-8") | |
| included = root / "impl_codex" / "self_healing_finetuning" | |
| included.mkdir(parents=True) | |
| (included / "config.json").write_text("hf_" + "b" * 30, encoding="utf-8") | |
| findings = scan_tree(root) | |
| self.assertEqual(len(findings), 1) | |
| self.assertIn("config.json", next(iter(findings))) | |
| def test_hf_cli_runner_times_out_instead_of_hanging(self) -> None: | |
| with mock.patch("providers.hf_managed.subprocess.run", side_effect=TimeoutError()): | |
| # A non-TimeoutExpired exception should still surface; this guards the mock shape. | |
| with self.assertRaises(TimeoutError): | |
| hf_managed._run_hf(["hf", "jobs", "--help"]) | |
| def test_hf_cli_timeout_expired_returns_124(self) -> None: | |
| expired = subprocess.TimeoutExpired(["hf", "download"], timeout=1, output="out", stderr="err") | |
| with mock.patch("providers.hf_managed.subprocess.run", side_effect=expired): | |
| result = hf_managed._run_hf(["hf", "download"]) | |
| self.assertEqual(result.returncode, 124) | |
| self.assertIn("timed out", result.stderr) | |
| def test_frozen_eval_suite_validates(self) -> None: | |
| report = validate_frozen_suite(REPO_ROOT / "data" / "eval" / "linvest21_frozen_eval_v0_manifest.json") | |
| self.assertTrue(report["ok"], report["errors"]) | |
| self.assertGreaterEqual(report["sample_count"], 100) | |
| def test_hf_trainer_loads_jsonl_and_builds_plan(self) -> None: | |
| path = REPO_ROOT / "data" / "linvest21_train.jsonl" | |
| rows = load_jsonl(path) | |
| self.assertGreaterEqual(len(rows), 100) | |
| args = type( | |
| "Args", | |
| (), | |
| { | |
| "run_id": "test", | |
| "model_candidate": "linvest21/linvest21_fingpt_v1_000", | |
| "base_model_id": "meta-llama/Meta-Llama-3-8B", | |
| "train_jsonl": path, | |
| "valid_jsonl": path, | |
| "output_dir": Path(tempfile.gettempdir()) / "shft_test_plan", | |
| "max_steps": 1, | |
| "per_device_train_batch_size": 1, | |
| "gradient_accumulation_steps": 1, | |
| "learning_rate": 0.00008, | |
| "lora_r": 16, | |
| "lora_alpha": 32, | |
| "lora_dropout": 0.05, | |
| "max_seq_length": 512, | |
| "dry_run": True, | |
| }, | |
| )() | |
| plan = build_plan(args, rows, rows[:2]) | |
| self.assertEqual(plan["train_records"], len(rows)) | |
| self.assertIn("torch", plan["required_packages"]) | |
| self.assertTrue(plan["dataset_provenance"]["ok"]) | |
| def test_hf_dataset_staging_uses_run_scoped_remote_path(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| source = Path(tmp) / "source.jsonl" | |
| source.write_text( | |
| "\n".join( | |
| json.dumps( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"question {idx}"}, | |
| {"role": "assistant", "content": f"answer {idx}"}, | |
| ] | |
| }, | |
| sort_keys=True, | |
| ) | |
| for idx in range(10) | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| dataset_dir = Path(tmp) / "dataset" | |
| ingest_dataset(dataset_dir, dataset_path=source) | |
| plan = stage_hf_dataset("run_unit_dataset_scope", dataset_dir, live=False) | |
| self.assertEqual(plan["path_in_repo"], "runs/run_unit_dataset_scope") | |
| self.assertEqual(plan["job_dataset_dir"], "/data/runs/run_unit_dataset_scope") | |
| upload_command = plan["commands"][1] | |
| self.assertEqual(upload_command[4], "runs/run_unit_dataset_scope") | |
| self.assertIn("train", plan["split_sha256"]) | |
| def test_hf_train_command_uses_run_scoped_dataset_and_expected_hashes(self) -> None: | |
| run_id = "run_unit_hf_command_dataset_scope" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| try: | |
| (run_path / "provider_plans").mkdir(parents=True) | |
| write_json( | |
| run_path / "provider_plans" / "hf_dataset_stage_result.json", | |
| { | |
| "job_dataset_dir": f"/data/runs/{run_id}", | |
| "dataset_manifest_sha256": "m" * 64, | |
| "split_sha256": {"train": "a" * 64, "valid": "b" * 64, "test": "c" * 64}, | |
| }, | |
| ) | |
| manifest = { | |
| "run_id": run_id, | |
| "model_candidate": "linvest21/linvest21_fingpt_v1_000", | |
| "training_start": {"policy": "continue-best", "start_adapter": "linvest21/linvest21_fingpt_v1_000"}, | |
| } | |
| config = { | |
| "namespace": "linvest21", | |
| "storage": {"bucket": "linvest21/shft-artifacts", "dataset_repo": "linvest21/shft-datasets"}, | |
| "jobs": {"training": {}, "base_model_id": "meta-llama/Meta-Llama-3-8B"}, | |
| } | |
| command = HFManagedProvider._build_jobs_command(manifest, config) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| self.assertIn(f"SHFT_EXPECTED_DATASET_DIR=/data/runs/{run_id}", command) | |
| dataset_index = command.index("--dataset-dir") | |
| self.assertEqual(command[dataset_index + 1], f"/data/runs/{run_id}") | |
| self.assertIn("--expected-train-sha256", command) | |
| self.assertEqual(command[command.index("--expected-train-sha256") + 1], "a" * 64) | |
| def test_hf_trainer_blocks_dataset_count_mismatch_before_training(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "source.jsonl" | |
| source.write_text( | |
| "\n".join( | |
| json.dumps( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"question {idx}"}, | |
| {"role": "assistant", "content": f"answer {idx}"}, | |
| ] | |
| }, | |
| sort_keys=True, | |
| ) | |
| for idx in range(10) | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| dataset_dir = root / "dataset" | |
| manifest = ingest_dataset(dataset_dir, dataset_path=source) | |
| manifest["split_counts"]["train"] += 1 | |
| write_json(dataset_dir / "dataset_manifest.json", manifest) | |
| train_rows = load_jsonl(dataset_dir / "train.jsonl") | |
| valid_rows = load_jsonl(dataset_dir / "valid.jsonl") | |
| args = type( | |
| "Args", | |
| (), | |
| { | |
| "run_id": "test", | |
| "model_candidate": "linvest21/linvest21_fingpt_v1_000", | |
| "base_model_id": "meta-llama/Meta-Llama-3-8B", | |
| "train_jsonl": dataset_dir / "train.jsonl", | |
| "valid_jsonl": dataset_dir / "valid.jsonl", | |
| "dataset_dir": str(dataset_dir), | |
| "output_dir": root / "out", | |
| "max_steps": 1, | |
| "per_device_train_batch_size": 1, | |
| "gradient_accumulation_steps": 1, | |
| "learning_rate": 0.00008, | |
| "lora_r": 16, | |
| "lora_alpha": 32, | |
| "lora_dropout": 0.05, | |
| "max_seq_length": 512, | |
| "dry_run": True, | |
| }, | |
| )() | |
| plan = build_plan(args, train_rows, valid_rows) | |
| self.assertFalse(plan["dataset_provenance"]["ok"]) | |
| self.assertIn("manifest_train_records", "\n".join(plan["dataset_provenance"]["errors"])) | |
| def test_learning_jsonl_loader_preserves_unicode_line_separators(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| path = Path(tmp) / "learning.hf_finetune.jsonl" | |
| write_learning_jsonl( | |
| path, | |
| [ | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "line one\u2028line two\u2029line three"}, | |
| {"role": "assistant", "content": "answer"}, | |
| ], | |
| "metadata": {"asset_class": "equity", "role": "risk_manager"}, | |
| } | |
| ], | |
| ) | |
| rows = load_learning_jsonl(path) | |
| self.assertEqual(len(rows), 1) | |
| self.assertIn("line two", rows[0]["messages"][1]["content"]) | |
| def test_build_training_jsonl_caps_repair_rows_when_coverage_can_still_pass(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "role_source.hf_finetune.jsonl" | |
| output = root / "selected_training.jsonl" | |
| rows = [] | |
| for idx in range(800): | |
| rows.append( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"Original source prompt {idx}"}, | |
| {"role": "assistant", "content": f"Original non-repair answer {idx}."}, | |
| ], | |
| "metadata": {"asset_class": "unit_asset", "role": "unit_role", "row_type": "original"}, | |
| } | |
| ) | |
| repair_text = ( | |
| "Reported facts: numeric anchor is 12 percent. Inference: the signal may require " | |
| "investigation rather than certainty because risk and tradeoff evidence can change " | |
| "the critical pass/fail decision." | |
| ) | |
| for idx in range(200): | |
| rows.append( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"Repair source prompt {idx}"}, | |
| {"role": "assistant", "content": repair_text}, | |
| ], | |
| "metadata": { | |
| "asset_class": "unit_asset", | |
| "role": "unit_role", | |
| "synthetic_method": "grounded_template_reasoning_v1", | |
| "task": "grounded_critical_reasoning_sft", | |
| "rubric_target": "deterministic_heuristic_v0_critical_pass", | |
| }, | |
| } | |
| ) | |
| write_learning_jsonl(source, rows) | |
| manifest = build_training_jsonl_from_learning( | |
| source=root, | |
| output_path=output, | |
| asset_class="unit_asset", | |
| role="unit_role", | |
| repair_oversample_factor=2, | |
| max_repair_selected_ratio=0.5, | |
| ) | |
| coverage = evaluate_repair_coverage(selected_training=output) | |
| self.assertTrue(coverage["ok"], coverage) | |
| self.assertTrue(manifest["repair_cap_applied"]) | |
| self.assertEqual(manifest["source_record_count"], 1000) | |
| self.assertEqual(manifest["source_repair_row_count"], 200) | |
| self.assertEqual(manifest["selected_repair_source_rows"], 150) | |
| self.assertEqual(manifest["dropped_repair_source_rows"], 50) | |
| self.assertEqual(manifest["repair_row_count"], 150) | |
| self.assertEqual(manifest["repair_oversampled_record_count"], 150) | |
| self.assertEqual(manifest["record_count"], 1100) | |
| self.assertEqual(coverage["repair_row_count"], 300) | |
| def test_build_training_jsonl_rejects_repair_cap_when_original_rows_are_too_thin(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "thin_role_source.hf_finetune.jsonl" | |
| output = root / "selected_training.jsonl" | |
| repair_text = ( | |
| "Reported facts: numeric anchor is 12 percent. Inference: the signal may require " | |
| "investigation rather than certainty because risk and tradeoff evidence can change " | |
| "the critical pass/fail decision." | |
| ) | |
| rows = [ | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"Original source prompt {idx}"}, | |
| {"role": "assistant", "content": f"Original non-repair answer {idx}."}, | |
| ], | |
| "metadata": {"asset_class": "unit_asset", "role": "thin_unit_role"}, | |
| } | |
| for idx in range(10) | |
| ] | |
| rows.extend( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"Repair source prompt {idx}"}, | |
| {"role": "assistant", "content": repair_text}, | |
| ], | |
| "metadata": { | |
| "asset_class": "unit_asset", | |
| "role": "thin_unit_role", | |
| "synthetic_method": "grounded_template_reasoning_v1", | |
| "task": "grounded_critical_reasoning_sft", | |
| "rubric_target": "deterministic_heuristic_v0_critical_pass", | |
| }, | |
| } | |
| for idx in range(200) | |
| ) | |
| write_learning_jsonl(source, rows) | |
| with self.assertRaisesRegex(ValueError, "Add more original non-repair training rows"): | |
| build_training_jsonl_from_learning( | |
| source=root, | |
| output_path=output, | |
| asset_class="unit_asset", | |
| role="thin_unit_role", | |
| repair_oversample_factor=2, | |
| max_repair_selected_ratio=0.5, | |
| ) | |
| def test_generate_nonrepair_balance_data_closes_repair_cap_deficit(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "thin_role_source.hf_finetune.jsonl" | |
| selected = root / "selected_training.jsonl" | |
| rows = [ | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"Original source prompt {idx}"}, | |
| { | |
| "role": "assistant", | |
| "content": ( | |
| "This curated source note describes process discipline, evidence attribution, " | |
| "and role workflow without creating a repair case." | |
| ), | |
| }, | |
| ], | |
| "metadata": {"asset_class": "unit_asset", "role": "balance_role", "source_title": f"Seed {idx}"}, | |
| } | |
| for idx in range(2) | |
| ] | |
| repair_text = ( | |
| "Reported facts: numeric anchor is 12 percent. Inference: the signal may require " | |
| "investigation rather than certainty because risk and tradeoff evidence can change " | |
| "the critical pass/fail decision." | |
| ) | |
| rows.extend( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"Repair source prompt {idx}"}, | |
| {"role": "assistant", "content": repair_text}, | |
| ], | |
| "metadata": { | |
| "asset_class": "unit_asset", | |
| "role": "balance_role", | |
| "synthetic_method": "grounded_template_reasoning_v1", | |
| "task": "grounded_critical_reasoning_sft", | |
| "rubric_target": "deterministic_heuristic_v0_critical_pass", | |
| }, | |
| } | |
| for idx in range(200) | |
| ) | |
| write_learning_jsonl(source, rows) | |
| report = generate_nonrepair_balance_data( | |
| asset_class="unit_asset", | |
| role="balance_role", | |
| source=root, | |
| output_path=root / "balance.hf_finetune.jsonl", | |
| min_nonrepair_rows=100, | |
| force=True, | |
| ) | |
| manifest = build_training_jsonl_from_learning( | |
| source=root, | |
| output_path=selected, | |
| asset_class="unit_asset", | |
| role="balance_role", | |
| repair_oversample_factor=2, | |
| max_repair_selected_ratio=0.75, | |
| ) | |
| coverage = evaluate_repair_coverage(selected_training=selected) | |
| self.assertTrue(report["ok"], report) | |
| self.assertEqual(report["current_nonrepair_rows_before_generation"], 2) | |
| self.assertEqual(report["generated_count"], 98) | |
| self.assertEqual(report["projected_nonrepair_rows"], 100) | |
| self.assertTrue(coverage["ok"], coverage) | |
| self.assertEqual(manifest["record_count"], 400) | |
| self.assertEqual(coverage["repair_row_count"], 300) | |
| def test_ingest_writes_real_train_valid_test_splits(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| source = Path(tmp) / "source.jsonl" | |
| rows = [ | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": f"question {idx}"}, | |
| {"role": "assistant", "content": f"answer {idx}"}, | |
| ], | |
| "metadata": {"idx": idx}, | |
| } | |
| for idx in range(20) | |
| ] | |
| source.write_text("\n".join(json.dumps(row, sort_keys=True) for row in rows) + "\n", encoding="utf-8") | |
| manifest = ingest_dataset(Path(tmp) / "snapshot", dataset_path=source) | |
| self.assertEqual(manifest["quality"]["record_count"], 20) | |
| self.assertEqual(manifest["split_counts"], {"train": 16, "valid": 2, "test": 2}) | |
| def test_public_source_intake_downloads_but_only_promotes_policy_approved_sources(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| approved = root / "approved.html" | |
| restricted = root / "restricted.html" | |
| approved.write_text( | |
| "<html><body>" | |
| + ( | |
| "equity stock risk manager red flag checklist case study pass fail decision " | |
| "because downside scenario cash flow risk control stress test. " | |
| * 30 | |
| ) | |
| + "</body></html>", | |
| encoding="utf-8", | |
| ) | |
| restricted.write_text("restricted source", encoding="utf-8") | |
| catalog = { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": "equity", | |
| "role": "risk_manager", | |
| "title": "Approved Government Source", | |
| "url": approved.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "US government public information", | |
| "rationale": "approved test source", | |
| "ai_certification": _training_certification(), | |
| }, | |
| { | |
| "asset_class": "equity", | |
| "role": "risk_manager", | |
| "title": "Restricted But Downloadable Source", | |
| "url": restricted.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "copyright notice; no downloading for training", | |
| "rationale": "restricted test source", | |
| "ai_certification": _training_certification(), | |
| }, | |
| ], | |
| } | |
| catalog_path = root / "catalog.json" | |
| catalog_path.write_text(json.dumps(catalog), encoding="utf-8") | |
| manifest = intake_public_sources( | |
| asset_class="equity", | |
| role="risk_manager", | |
| catalog_path=catalog_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| ) | |
| self.assertEqual(manifest["downloaded_count"], 2) | |
| self.assertEqual(manifest["approved_for_training_count"], 1) | |
| self.assertEqual(manifest["review_required_count"], 1) | |
| approved_records = [item for item in manifest["records"] if item.get("train_allowed")] | |
| review_records = [item for item in manifest["records"] if item.get("requires_human_review")] | |
| self.assertTrue(Path(approved_records[0]["training_path"]).exists()) | |
| self.assertTrue(approved_records[0]["training_path"].endswith(".normalized.json")) | |
| self.assertTrue(approved_records[0]["normalized"]["eligibility"]["training"]) | |
| self.assertFalse("training_path" in review_records[0]) | |
| self.assertIn(review_records[0]["decision"], {"downloaded_for_review_not_training", "blocked_ai_certification_not_training_or_verification"}) | |
| def test_public_source_intake_skips_existing_training_urls_and_continues(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| existing = root / "existing.html" | |
| new_source = root / "new.html" | |
| existing.write_text("<html><body>" + ("existing public training source. " * 30) + "</body></html>", encoding="utf-8") | |
| new_source.write_text( | |
| "<html><body>" | |
| + ("new equity researcher case study red flag checklist pass fail decision because valuation quality. " * 30) | |
| + "</body></html>", | |
| encoding="utf-8", | |
| ) | |
| training_dir = root / "learning" / "equity" / "researcher" | |
| training_dir.mkdir(parents=True) | |
| (training_dir / "existing.normalized.json").write_text( | |
| json.dumps( | |
| { | |
| "schema_version": "normalized_source_v1", | |
| "source_url": existing.as_uri(), | |
| "text": "already present", | |
| "eligibility": {"training": True}, | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| catalog_path = root / "catalog.json" | |
| catalog_path.write_text( | |
| json.dumps( | |
| { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": "equity", | |
| "role": "researcher", | |
| "title": "Existing Source", | |
| "url": existing.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "US government public information", | |
| }, | |
| { | |
| "asset_class": "equity", | |
| "role": "researcher", | |
| "title": "New Source", | |
| "url": new_source.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "US government public information", | |
| "ai_certification": _training_certification(), | |
| }, | |
| ], | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| manifest = intake_public_sources( | |
| asset_class="equity", | |
| role="researcher", | |
| catalog_path=catalog_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| max_sources=1, | |
| ) | |
| self.assertEqual(manifest["source_count"], 2) | |
| self.assertEqual(manifest["downloaded_count"], 1) | |
| self.assertEqual(manifest["training_eligible_count"], 1) | |
| self.assertEqual(manifest["records"][0]["decision"], "already_promoted_to_training") | |
| self.assertTrue(manifest["records"][1]["training_path"].endswith(".normalized.json")) | |
| def test_public_source_intake_can_disable_downloads_by_policy(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "approved.html" | |
| source.write_text("public source", encoding="utf-8") | |
| catalog_path = root / "catalog.json" | |
| catalog_path.write_text( | |
| json.dumps( | |
| { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": "equity", | |
| "role": "risk_manager", | |
| "title": "Approved Government Source", | |
| "url": source.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "US government public information", | |
| "ai_certification": _training_certification(), | |
| } | |
| ], | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| policy_path = root / "source_policy.yaml" | |
| policy_path.write_text( | |
| "\n".join( | |
| [ | |
| "auto_download_public_sources: false", | |
| "promote_approved_sources_to_training: true", | |
| "approved_license_statuses: [government_public]", | |
| "review_license_statuses: [unknown]", | |
| "blocked_license_statuses: [blocked_scheme]", | |
| "source_risk:", | |
| " max_download_bytes: 1000000", | |
| " allowed_schemes: [file]", | |
| " disallowed_url_hints: []", | |
| "stall_breakout:", | |
| " max_new_sources_per_round: 10", | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| manifest = intake_public_sources( | |
| asset_class="equity", | |
| role="risk_manager", | |
| catalog_path=catalog_path, | |
| policy_path=policy_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| ) | |
| self.assertEqual(manifest["downloaded_count"], 0) | |
| self.assertEqual(manifest["approved_for_training_count"], 0) | |
| self.assertEqual(manifest["records"][0]["decision"], "download_skipped_by_policy") | |
| def test_source_quality_certifier_rejects_bulk_filings_when_critical_pass_is_blocked(self) -> None: | |
| report = certify_source_candidate( | |
| asset_class="equity", | |
| role="researcher", | |
| title="Form 10-K Annual Report Risk Factors", | |
| url="https://www.sec.gov/example/10-k.htm", | |
| source_type="html", | |
| rationale="raw annual report filing and MD&A text", | |
| quality_errors=["critical_pass_absolute: 0.13 >= 0.70"], | |
| policy={ | |
| "source_quality_certification": { | |
| "min_training_score": 4.0, | |
| "min_verification_score": 2.0, | |
| "require_critical_reasoning_when_gate_fails": True, | |
| } | |
| }, | |
| ) | |
| self.assertEqual(report["intended_use"], "reject") | |
| self.assertFalse(report["training_eligible"]) | |
| self.assertTrue(any("critical pass/fail" in blocker for blocker in report["blockers"])) | |
| def test_source_quality_certifier_accepts_high_signal_reasoned_examples_for_training(self) -> None: | |
| report = certify_source_candidate( | |
| asset_class="equity", | |
| role="researcher", | |
| title="Equity Researcher Red Flag Checklist Case Study", | |
| url="https://www.investor.gov/example/checklist", | |
| source_type="html", | |
| rationale="worked example with pass/fail decision because cash-flow, accounting-quality, and valuation red flags", | |
| quality_errors=["critical_pass_absolute: 0.13 >= 0.70"], | |
| policy={ | |
| "source_quality_certification": { | |
| "min_training_score": 4.0, | |
| "min_verification_score": 2.0, | |
| "require_critical_reasoning_when_gate_fails": True, | |
| } | |
| }, | |
| ) | |
| self.assertEqual(report["intended_use"], "training") | |
| self.assertTrue(report["training_eligible"]) | |
| def test_post_download_content_certifier_blocks_low_signal_extracted_text(self) -> None: | |
| report = certify_normalized_source_content( | |
| asset_class="equity", | |
| role="researcher", | |
| title="Promising Equity Red Flag Case Study", | |
| url="https://www.investor.gov/example", | |
| source_type="html", | |
| text="Form 10-K annual report MD&A risk factors glossary. " * 40, | |
| quality_errors=["critical_pass_absolute: 0.13 >= 0.70"], | |
| policy={ | |
| "source_quality_certification": { | |
| "min_training_score": 4.0, | |
| "min_verification_score": 2.0, | |
| "require_critical_reasoning_when_gate_fails": True, | |
| "min_content_reasoning_density": 0.025, | |
| } | |
| }, | |
| ) | |
| self.assertEqual(report["intended_use"], "reject") | |
| self.assertFalse(report["training_eligible"]) | |
| self.assertTrue(any("content_missing" in blocker for blocker in report["blockers"])) | |
| def test_post_download_content_certifier_accepts_extracted_reasoning_text(self) -> None: | |
| report = certify_normalized_source_content( | |
| asset_class="equity", | |
| role="researcher", | |
| title="Equity Researcher Red Flag Checklist Case Study", | |
| url="https://www.investor.gov/example", | |
| source_type="html", | |
| text=( | |
| "Equity researcher case study: reject the stock because free cash flow is negative, " | |
| "reported earnings quality is weak, and valuation does not compensate for risk. " | |
| "Pass/fail decision: fail because accounting quality and cash flow red flags are present. " | |
| ) | |
| * 25, | |
| quality_errors=["critical_pass_absolute: 0.13 >= 0.70"], | |
| policy={ | |
| "source_quality_certification": { | |
| "min_training_score": 4.0, | |
| "min_verification_score": 2.0, | |
| "require_critical_reasoning_when_gate_fails": True, | |
| "min_content_reasoning_density": 0.025, | |
| } | |
| }, | |
| ) | |
| self.assertEqual(report["intended_use"], "training") | |
| self.assertTrue(report["training_eligible"]) | |
| def test_content_certifier_accepts_all_asset_role_signal_pairs(self) -> None: | |
| asset_terms = { | |
| "equity": "equity stock public company earnings cash flow valuation", | |
| "fixed_income": "fixed income bond yield curve duration credit spread treasury coupon", | |
| "multi_asset": "multi asset asset allocation cross asset diversification correlation rebalancing", | |
| } | |
| role_terms = { | |
| "chief_investment_officer": "investment policy capital market assumptions committee governance policy portfolio strategic allocation", | |
| "client_portfolio_manager": "client explanation suitability risk tolerance client scenario investment objective time horizon", | |
| "performance_manager": "attribution benchmark tracking error information ratio performance attribution selection effect", | |
| "portfolio_manager": "portfolio construction position sizing rebalancing tradeoff risk budget portfolio allocation carry roll down", | |
| "researcher": "valuation moat roic earnings quality base rate financial statement analysis term structure issuer analysis", | |
| "risk_manager": "red flag risk control stress test liquidity risk internal control drawdown market risk credit risk", | |
| } | |
| policy = { | |
| "source_quality_certification": { | |
| "min_training_score": 4.0, | |
| "min_verification_score": 2.0, | |
| "require_critical_reasoning_when_gate_fails": True, | |
| "min_content_reasoning_density": 0.025, | |
| "min_content_words": 120, | |
| } | |
| } | |
| failures: list[str] = [] | |
| for asset_class, asset_text in asset_terms.items(): | |
| for role, role_text in role_terms.items(): | |
| text = ( | |
| f"{asset_text}. {role_text}. " | |
| "Case study checklist: reject or approve with a pass/fail decision because the numeric scenario, " | |
| "risk tradeoff, and fact versus inference separation must be explicit. " | |
| ) * 18 | |
| report = certify_normalized_source_content( | |
| asset_class=asset_class, | |
| role=role, | |
| title=f"{asset_class} {role} red flag decision case study", | |
| url="https://example.gov/case-study", | |
| source_type="html", | |
| text=text, | |
| quality_errors=["critical_pass_absolute: 0.13 >= 0.70"], | |
| policy=policy, | |
| ) | |
| if not report["training_eligible"]: | |
| failures.append(f"{asset_class}/{role}: {report['blockers']}") | |
| self.assertEqual(failures, []) | |
| def test_step0b_converts_training_eligible_normalized_sources_to_jsonl(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| role_dir = root / "data" / "learning" / "equity" / "risk_manager" | |
| role_dir.mkdir(parents=True) | |
| normalized = { | |
| "schema_version": "normalized_source_v1", | |
| "asset_class": "equity", | |
| "role": "risk_manager", | |
| "source_title": "Clean HTML Risk Source", | |
| "source_url": "file:///clean.html", | |
| "format": "html", | |
| "license_status": "government_public", | |
| "text": "Risk should be sized with scenario analysis and downside monitoring. " * 25, | |
| "eligibility": {"training": True, "validation": False, "reasons": ["format_not_validation_enabled:html"]}, | |
| } | |
| (role_dir / "clean_html_risk.normalized.json").write_text(json.dumps(normalized), encoding="utf-8") | |
| import data_pipeline.learning_pdf_to_jsonl as converter | |
| original_repo_root = converter.REPO_ROOT | |
| try: | |
| converter.REPO_ROOT = root | |
| report = converter.convert_learning_role( | |
| asset_class="equity", | |
| role="risk_manager", | |
| chunk_chars=500, | |
| min_text_chars=100, | |
| skip_existing=False, | |
| ) | |
| finally: | |
| converter.REPO_ROOT = original_repo_root | |
| self.assertEqual(report["pdf_count"], 0) | |
| self.assertEqual(report["normalized_source_count"], 1) | |
| self.assertGreater(report["record_count"], 0) | |
| output = role_dir / "clean_html_risk.normalized.hf_finetune.jsonl" | |
| self.assertTrue(output.exists()) | |
| row = json.loads(output.read_text(encoding="utf-8").splitlines()[0]) | |
| self.assertEqual(row["metadata"]["source_format"], "html") | |
| self.assertEqual(row["metadata"]["task"], "normalized_financial_research_sft") | |
| def test_pdf_pointer_noise_filter_keeps_real_pypdf_warnings(self) -> None: | |
| import logging | |
| from data_pipeline.pdf_warning_filter import suppress_known_pypdf_pointer_noise | |
| records: list[str] = [] | |
| class ListHandler(logging.Handler): | |
| def emit(self, record: logging.LogRecord) -> None: | |
| records.append(record.getMessage()) | |
| logger = logging.getLogger("pypdf._reader") | |
| handler = ListHandler() | |
| logger.addHandler(handler) | |
| original_level = logger.level | |
| logger.setLevel(logging.WARNING) | |
| try: | |
| with suppress_known_pypdf_pointer_noise(): | |
| logger.warning("Ignoring wrong pointing object 319 0 (offset 0)") | |
| logger.warning("Unexpected PDF encryption state") | |
| finally: | |
| logger.removeHandler(handler) | |
| logger.setLevel(original_level) | |
| self.assertEqual(records, ["Unexpected PDF encryption state"]) | |
| def test_html_index_downloads_but_does_not_become_training_by_default(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| index = root / "index.html" | |
| index.write_text("<html><body>" + ("search result navigation " * 40) + "</body></html>", encoding="utf-8") | |
| catalog_path = root / "catalog.json" | |
| catalog_path.write_text( | |
| json.dumps( | |
| { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": "equity", | |
| "role": "risk_manager", | |
| "title": "Search Index", | |
| "url": index.as_uri(), | |
| "source_type": "html_index", | |
| "license_hint": "US government public information", | |
| "ai_certification": _training_certification(), | |
| } | |
| ], | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| manifest = intake_public_sources( | |
| asset_class="equity", | |
| role="risk_manager", | |
| catalog_path=catalog_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| ) | |
| self.assertEqual(manifest["downloaded_count"], 1) | |
| self.assertEqual(manifest["approved_for_training_count"], 1) | |
| self.assertEqual(manifest["training_eligible_count"], 0) | |
| record = manifest["records"][0] | |
| self.assertFalse(record["normalized"]["eligibility"]["training"]) | |
| self.assertNotIn("training_path", record) | |
| def test_stall_breakout_writes_transparent_plan_without_certifying_failed_run(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "restricted.html" | |
| source.write_text("restricted source", encoding="utf-8") | |
| catalog_path = root / "catalog.json" | |
| catalog_path.write_text( | |
| json.dumps( | |
| { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": "equity", | |
| "role": "risk_manager", | |
| "title": "Restricted But Downloadable Source", | |
| "url": source.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "copyright notice; no downloading for training", | |
| } | |
| ], | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| policy_path = root / "source_policy.yaml" | |
| policy_path.write_text( | |
| "\n".join( | |
| [ | |
| "auto_download_public_sources: true", | |
| "promote_approved_sources_to_training: true", | |
| "approved_license_statuses: [government_public]", | |
| "review_license_statuses: [copyright_notice, restricted_download, unknown]", | |
| "blocked_license_statuses: [blocked_scheme]", | |
| "source_risk:", | |
| " max_download_bytes: 1000000", | |
| " allowed_schemes: [file]", | |
| " disallowed_url_hints: []", | |
| "source_quality_certification:", | |
| " required_before_download: false", | |
| "live_discovery:", | |
| " enabled: false", | |
| "stall_breakout:", | |
| " max_new_sources_per_round: 10", | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| train_source = root / "training" | |
| train_source.mkdir() | |
| (train_source / "role.hf_finetune.jsonl").write_text( | |
| json.dumps( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "question"}, | |
| {"role": "assistant", "content": "answer"}, | |
| ], | |
| "metadata": {"asset_class": "equity", "role": "risk_manager"}, | |
| } | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| run_id = "run_test_stall_breakout_transparency" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| try: | |
| plan = run_stall_breakout( | |
| run_id=run_id, | |
| asset_class="equity", | |
| role="risk_manager", | |
| train_source=train_source, | |
| quality_gate_exit_code=6, | |
| catalog_path=catalog_path, | |
| policy_path=policy_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| ) | |
| self.assertEqual(plan["status"], "blocked_after_breakout") | |
| self.assertIn("no AI-certified, policy-approved training sources", "\n".join(plan["blockers"])) | |
| self.assertTrue((run_path / "stall_breakout" / "stall_breakout_plan.json").exists()) | |
| self.assertTrue((run_path / "stall_breakout" / "source_intake_manifest.json").exists()) | |
| self.assertTrue( | |
| (run_path / "stall_breakout" / "training_data_validation" / "training_data_validation_report.json").exists() | |
| ) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def test_stall_breakout_uses_live_discovery_after_catalog_exhaustion(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| train_source = root / "training" | |
| train_source.mkdir() | |
| (train_source / "role.hf_finetune.jsonl").write_text( | |
| json.dumps( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "question"}, | |
| {"role": "assistant", "content": "answer"}, | |
| ], | |
| "metadata": {"asset_class": "equity", "role": "researcher"}, | |
| } | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| catalog_path = root / "empty_catalog.json" | |
| catalog_path.write_text(json.dumps({"schema_version": "public_source_catalog_v1", "sources": []}), encoding="utf-8") | |
| policy_path = root / "source_policy.yaml" | |
| policy_path.write_text( | |
| "\n".join( | |
| [ | |
| "auto_download_public_sources: true", | |
| "promote_approved_sources_to_training: true", | |
| "approved_license_statuses: [government_public]", | |
| "review_license_statuses: [unknown]", | |
| "blocked_license_statuses: [blocked_scheme]", | |
| "source_risk:", | |
| " max_download_bytes: 1000000", | |
| " allowed_schemes: [https]", | |
| " disallowed_url_hints: []", | |
| "formats:", | |
| " html:", | |
| " normalize: true", | |
| " train: true", | |
| " validate: false", | |
| " min_train_text_chars: 500", | |
| "source_quality_certification:", | |
| " required_before_download: true", | |
| "live_discovery:", | |
| " enabled: true", | |
| " provider: duckduckgo_html", | |
| " duckduckgo_fallback_enabled: true", | |
| " preferred_domains: [sec.gov]", | |
| "stall_breakout:", | |
| " max_new_sources_per_round: 1", | |
| " max_source_discovery_attempts: 1", | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| run_id = "run_test_stall_breakout_live_discovery" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def fake_download(url: str, output_path: Path, *, max_bytes: int) -> tuple[int, str]: | |
| text = ( | |
| "<html><body>" | |
| + ("equity researcher red flag checklist case study pass fail decision because valuation disclosure risk. " * 40) | |
| + "</body></html>" | |
| ) | |
| output_path.write_text(text, encoding="utf-8") | |
| return output_path.stat().st_size, "text/html" | |
| try: | |
| with mock.patch( | |
| "data_pipeline.live_source_discovery.search_duckduckgo", | |
| return_value=[ | |
| ( | |
| "https://www.sec.gov/investor/example-equity-research.html", | |
| "SEC equity valuation red flag checklist case study decision source", | |
| ) | |
| ], | |
| ), mock.patch("data_pipeline.source_intake._download", side_effect=fake_download), mock.patch( | |
| "orchestrator.stall_breakout.convert_learning_tree", | |
| return_value={"ok": True, "record_count": 2}, | |
| ): | |
| plan = run_stall_breakout( | |
| run_id=run_id, | |
| asset_class="equity", | |
| role="researcher", | |
| train_source=train_source, | |
| quality_gate_exit_code=6, | |
| catalog_path=catalog_path, | |
| policy_path=policy_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| max_sources=1, | |
| ) | |
| self.assertEqual(plan["status"], "ready_for_next_training_run", plan["blockers"]) | |
| self.assertEqual(plan["intake"]["trainable_new_source_count"], 1) | |
| self.assertTrue(plan["live_discovery"]["attempted"]) | |
| self.assertGreaterEqual(plan["live_discovery"]["candidate_count"], 1) | |
| self.assertTrue((run_path / "stall_breakout" / "live_source_intake_manifest.json").exists()) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def test_stall_breakout_retries_live_discovery_until_new_trainable_source(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| train_source = root / "training" | |
| train_source.mkdir() | |
| (train_source / "role.hf_finetune.jsonl").write_text( | |
| json.dumps( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "question"}, | |
| {"role": "assistant", "content": "answer"}, | |
| ], | |
| "metadata": {"asset_class": "equity", "role": "researcher"}, | |
| } | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| existing = root / "existing.html" | |
| new_source = root / "new.html" | |
| existing.write_text("<html><body>" + ("existing equity research source. " * 40) + "</body></html>", encoding="utf-8") | |
| new_source.write_text( | |
| "<html><body>" | |
| + ("new equity researcher red flag checklist case study pass fail decision because valuation disclosure. " * 40) | |
| + "</body></html>", | |
| encoding="utf-8", | |
| ) | |
| training_dir = root / "learning" / "equity" / "researcher" | |
| training_dir.mkdir(parents=True) | |
| (training_dir / "existing.normalized.json").write_text( | |
| json.dumps( | |
| { | |
| "schema_version": "normalized_source_v1", | |
| "source_url": existing.as_uri(), | |
| "text": "already present", | |
| "eligibility": {"training": True, "validation": False}, | |
| } | |
| ), | |
| encoding="utf-8", | |
| ) | |
| catalog_path = root / "empty_catalog.json" | |
| catalog_path.write_text(json.dumps({"schema_version": "public_source_catalog_v1", "sources": []}), encoding="utf-8") | |
| policy_path = root / "source_policy.yaml" | |
| policy_path.write_text( | |
| "\n".join( | |
| [ | |
| "auto_download_public_sources: true", | |
| "promote_approved_sources_to_training: true", | |
| "approved_license_statuses: [government_public]", | |
| "review_license_statuses: [unknown]", | |
| "blocked_license_statuses: [blocked_scheme]", | |
| "source_risk:", | |
| " max_download_bytes: 1000000", | |
| " allowed_schemes: [file]", | |
| " disallowed_url_hints: []", | |
| "formats:", | |
| " html:", | |
| " normalize: true", | |
| " train: true", | |
| " validate: false", | |
| " min_train_text_chars: 500", | |
| "source_quality_certification:", | |
| " required_before_download: true", | |
| "live_discovery:", | |
| " enabled: true", | |
| "stall_breakout:", | |
| " max_new_sources_per_round: 1", | |
| " max_source_discovery_attempts: 2", | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| run_id = "run_test_stall_breakout_retries_live_discovery" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def fake_discover( | |
| *, | |
| asset_class: str, | |
| role: str, | |
| policy: dict[str, object], | |
| output_dir: Path, | |
| exclude_urls: set[str] | None = None, | |
| quality_errors: list[str] | None = None, | |
| discovery_attempt: int = 0, | |
| ) -> dict[str, object]: | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| source = existing if discovery_attempt == 1 else new_source | |
| source_title = "Existing Source" if discovery_attempt == 1 else "New Source" | |
| catalog = { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": asset_class, | |
| "role": role, | |
| "title": source_title, | |
| "url": source.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "US government public information", | |
| "ai_certification": _training_certification(), | |
| } | |
| ], | |
| } | |
| manifest = { | |
| "schema_version": "live_source_discovery_manifest_v1", | |
| "asset_class": asset_class, | |
| "role": role, | |
| "discovery_attempt": discovery_attempt, | |
| "candidate_count": 1, | |
| "catalog_path": str(output_dir / "live_discovered_public_source_catalog.json"), | |
| "errors": [], | |
| } | |
| write_json(output_dir / "live_discovered_public_source_catalog.json", catalog) | |
| write_json(output_dir / "live_source_discovery_manifest.json", manifest) | |
| return {"catalog": catalog, "manifest": manifest} | |
| try: | |
| with mock.patch("orchestrator.stall_breakout.discover_public_sources", side_effect=fake_discover), mock.patch( | |
| "orchestrator.stall_breakout.convert_learning_tree", | |
| return_value={"ok": True, "record_count": 2}, | |
| ): | |
| plan = run_stall_breakout( | |
| run_id=run_id, | |
| asset_class="equity", | |
| role="researcher", | |
| train_source=train_source, | |
| quality_gate_exit_code=6, | |
| catalog_path=catalog_path, | |
| policy_path=policy_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| ) | |
| self.assertEqual(plan["status"], "ready_for_next_training_run", plan["blockers"]) | |
| self.assertEqual(plan["live_discovery"]["attempt_count"], 2) | |
| self.assertEqual(plan["live_discovery"]["attempts"][0]["already_in_training_count"], 1) | |
| self.assertEqual(plan["live_discovery"]["attempts"][1]["intake_training_eligible_count"], 1) | |
| self.assertEqual(plan["intake"]["trainable_new_source_count"], 1) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def test_live_discovery_uses_direct_fallback_when_search_and_seed_pages_empty(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| train_source = root / "training" | |
| train_source.mkdir() | |
| (train_source / "role.hf_finetune.jsonl").write_text( | |
| json.dumps( | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "question"}, | |
| {"role": "assistant", "content": "answer"}, | |
| ], | |
| "metadata": {"asset_class": "equity", "role": "researcher"}, | |
| } | |
| ) | |
| + "\n", | |
| encoding="utf-8", | |
| ) | |
| catalog_path = root / "empty_catalog.json" | |
| catalog_path.write_text(json.dumps({"schema_version": "public_source_catalog_v1", "sources": []}), encoding="utf-8") | |
| run_id = "run_test_stall_breakout_direct_fallback" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def fake_download(url: str, output_path: Path, *, max_bytes: int) -> tuple[int, str]: | |
| text = "<html><body>" + ("form 10-k 10-q financial statement edgar equity research. " * 40) + "</body></html>" | |
| output_path.write_text(text, encoding="utf-8") | |
| return output_path.stat().st_size, "text/html" | |
| try: | |
| with mock.patch("data_pipeline.live_source_discovery.search_duckduckgo", return_value=[]), mock.patch( | |
| "data_pipeline.live_source_discovery._request_text", | |
| side_effect=RuntimeError("seed unavailable"), | |
| ), mock.patch("data_pipeline.source_intake._download", side_effect=fake_download), mock.patch( | |
| "orchestrator.stall_breakout.convert_learning_tree", | |
| return_value={"ok": True, "record_count": 2}, | |
| ): | |
| plan = run_stall_breakout( | |
| run_id=run_id, | |
| asset_class="equity", | |
| role="researcher", | |
| train_source=train_source, | |
| quality_gate_exit_code=6, | |
| catalog_path=catalog_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| max_sources=1, | |
| ) | |
| self.assertEqual(plan["status"], "blocked_after_breakout", plan["blockers"]) | |
| self.assertEqual(plan["intake"]["trainable_new_source_count"], 0) | |
| self.assertEqual(plan["live_discovery"]["candidate_count"], 0) | |
| manifest = json.loads( | |
| (run_path / "stall_breakout" / "live_discovery" / "live_source_discovery_manifest.json").read_text( | |
| encoding="utf-8" | |
| ) | |
| ) | |
| self.assertTrue(any(item.get("provider") == "direct_official_fallback" for item in manifest["fallback_diagnostics"])) | |
| self.assertTrue( | |
| any(item.get("skipped_ai_rejected", 0) >= 1 for item in manifest["fallback_diagnostics"]) | |
| ) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def test_stall_breakout_generates_trainable_reasoning_from_verification_sources(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| train_source = root / "learning" / "equity" / "researcher" | |
| train_source.mkdir(parents=True) | |
| verification_source = root / "verification.html" | |
| verification_source.write_text( | |
| "<html><body>" | |
| + ( | |
| "equity researcher financial statement analysis valuation disclosure cash flow red flag " | |
| "checklist case study pass fail decision because risk and accounting quality. " | |
| * 45 | |
| ) | |
| + "</body></html>", | |
| encoding="utf-8", | |
| ) | |
| catalog_path = root / "empty_catalog.json" | |
| catalog_path.write_text(json.dumps({"schema_version": "public_source_catalog_v1", "sources": []}), encoding="utf-8") | |
| policy_path = root / "source_policy.yaml" | |
| policy_path.write_text( | |
| "\n".join( | |
| [ | |
| "auto_download_public_sources: true", | |
| "promote_approved_sources_to_training: true", | |
| "approved_license_statuses: [government_public]", | |
| "review_license_statuses: [unknown]", | |
| "blocked_license_statuses: [blocked_scheme]", | |
| "source_risk:", | |
| " max_download_bytes: 1000000", | |
| " allowed_schemes: [file]", | |
| " disallowed_url_hints: []", | |
| "formats:", | |
| " html:", | |
| " normalize: true", | |
| " train: true", | |
| " validate: true", | |
| " min_train_text_chars: 500", | |
| " min_validation_text_chars: 500", | |
| "source_quality_certification:", | |
| " required_before_download: true", | |
| " require_critical_reasoning_when_gate_fails: true", | |
| " reject_low_signal_bulk_text_when_critical_pass_fails: true", | |
| "live_discovery:", | |
| " enabled: true", | |
| "stall_breakout:", | |
| " max_new_sources_per_round: 1", | |
| " max_source_discovery_attempts: 1", | |
| " max_breakout_reasoning_records: 6", | |
| ] | |
| ), | |
| encoding="utf-8", | |
| ) | |
| run_id = "run_test_stall_breakout_verification_to_reasoning" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def fake_discover( | |
| *, | |
| asset_class: str, | |
| role: str, | |
| policy: dict[str, object], | |
| output_dir: Path, | |
| exclude_urls: set[str] | None = None, | |
| quality_errors: list[str] | None = None, | |
| discovery_attempt: int = 0, | |
| ) -> dict[str, object]: | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| catalog = { | |
| "schema_version": "public_source_catalog_v1", | |
| "sources": [ | |
| { | |
| "asset_class": asset_class, | |
| "role": role, | |
| "title": "Verification Source", | |
| "url": verification_source.as_uri(), | |
| "source_type": "html", | |
| "license_hint": "US government public information", | |
| "ai_certification": _verification_certification(), | |
| } | |
| ], | |
| } | |
| manifest = { | |
| "schema_version": "live_source_discovery_manifest_v1", | |
| "asset_class": asset_class, | |
| "role": role, | |
| "discovery_attempt": discovery_attempt, | |
| "candidate_count": 1, | |
| "catalog_path": str(output_dir / "live_discovered_public_source_catalog.json"), | |
| "errors": [], | |
| } | |
| write_json(output_dir / "live_discovered_public_source_catalog.json", catalog) | |
| write_json(output_dir / "live_source_discovery_manifest.json", manifest) | |
| return {"catalog": catalog, "manifest": manifest} | |
| try: | |
| with mock.patch("orchestrator.stall_breakout.discover_public_sources", side_effect=fake_discover): | |
| plan = run_stall_breakout( | |
| run_id=run_id, | |
| asset_class="equity", | |
| role="researcher", | |
| train_source=train_source, | |
| quality_gate_exit_code=6, | |
| catalog_path=catalog_path, | |
| policy_path=policy_path, | |
| intake_root=root / "learning_intake", | |
| training_root=root / "learning", | |
| max_sources=1, | |
| ) | |
| self.assertEqual(plan["status"], "ready_for_next_training_run", plan["blockers"]) | |
| self.assertEqual(plan["live_discovery"]["intake_verification_eligible_count"], 1) | |
| self.assertGreaterEqual(plan["reasoning_generation"]["generated_count"], 1) | |
| self.assertEqual(plan["intake"]["trainable_new_source_count"], 1) | |
| generated_path = train_source / "synthetic_equity_researcher_critical_reasoning.hf_finetune.jsonl" | |
| self.assertTrue(generated_path.exists()) | |
| generated_rows = load_learning_jsonl(generated_path) | |
| self.assertGreaterEqual(len(generated_rows), 1) | |
| self.assertEqual(generated_rows[0]["metadata"]["synthetic_method"], "breakout_grounded_template_reasoning_v1") | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def test_preflight_summary_reports_data_thin_and_repair_heavy_roles(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| role = "client_portfolio_manager" | |
| role_dir = root / role | |
| validation_dir = role_dir / "training_data_validation" | |
| validation_dir.mkdir(parents=True) | |
| (role_dir / "selected_training.jsonl").write_text("{}", encoding="utf-8") | |
| write_json( | |
| role_dir / "selected_training.manifest.json", | |
| { | |
| "asset_class": "equity", | |
| "role": role, | |
| "selected_training_sha256": "a" * 64, | |
| "record_count": 416, | |
| "original_record_count": 224, | |
| "required_reasoning_included": True, | |
| }, | |
| ) | |
| write_json( | |
| role_dir / "repair_coverage_gate.json", | |
| { | |
| "ok": True, | |
| "repair_row_count": 384, | |
| "counts": { | |
| "numeric_reasoning": 384, | |
| "fact_inference_separation": 384, | |
| "neutral_language": 384, | |
| "risk_tradeoff": 384, | |
| "critical_reasoning": 384, | |
| }, | |
| "errors": [], | |
| }, | |
| ) | |
| write_json( | |
| validation_dir / "training_data_validation_report.json", | |
| {"ok": True, "record_count": 256, "schema_error_count": 0, "conflict_count": 0}, | |
| ) | |
| write_json(validation_dir / "conflict_report.json", {"conflicts": []}) | |
| write_json(validation_dir / "quarantine_manifest.json", {"records": []}) | |
| output = root / "summary.json" | |
| args = type( | |
| "Args", | |
| (), | |
| { | |
| "asset_class": "equity", | |
| "roles": [role], | |
| "preflight_root": str(root), | |
| "output": str(output), | |
| }, | |
| )() | |
| rc = shft_cli.command_summarize_asset_preflight(args) | |
| self.assertEqual(rc, 0) | |
| summary = json.loads(output.read_text(encoding="utf-8-sig")) | |
| report = summary["role_reports"][0] | |
| self.assertTrue(summary["ok"]) | |
| self.assertEqual(summary["data_thin_roles"], [role]) | |
| self.assertEqual(summary["repair_heavy_roles"], [role]) | |
| self.assertGreaterEqual(summary["corpus_warning_count"], 2) | |
| self.assertEqual(report["repair_to_selected_ratio"], round(384 / 416, 6)) | |
| self.assertEqual(report["repair_to_original_ratio"], round(384 / 224, 6)) | |
| self.assertIn("data_thin_original_records:224<300", report["corpus_warnings"]) | |
| def test_best_run_tracker_keeps_best_measured_checkpoint_after_regression(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| output_path = Path(tmp) / "best.json" | |
| run_good = SHFT_WORKSPACE_ROOT / "runs" / "run_test_best_tracker_good" | |
| run_bad = SHFT_WORKSPACE_ROOT / "runs" / "run_test_best_tracker_bad" | |
| for run_path in [run_good, run_bad]: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| (run_path / "eval").mkdir(parents=True) | |
| try: | |
| write_json( | |
| run_good / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.32, "critical_pass_rate": 0.34, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.32, | |
| "pairwise_win_rate": 0.7, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json( | |
| run_bad / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.22, "critical_pass_rate": 0.20, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.22, | |
| "pairwise_win_rate": 0.6, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| first = update_best_run( | |
| run_id="run_test_best_tracker_good", | |
| release_id="linvest21_fingpt_equity_researcher_v1_test", | |
| output_path=output_path, | |
| ) | |
| second = update_best_run( | |
| run_id="run_test_best_tracker_bad", | |
| release_id="linvest21_fingpt_equity_researcher_v1_test", | |
| output_path=output_path, | |
| ) | |
| self.assertTrue(first["updated"]) | |
| self.assertFalse(second["updated"]) | |
| self.assertEqual(second["best_run"]["run_id"], "run_test_best_tracker_good") | |
| self.assertEqual(second["current_run"]["distance_to_thresholds"]["candidate_aggregate_gap"], 0.38) | |
| self.assertEqual(second["source_batch_acceptance"]["decision"], "rejected_did_not_improve_previous_best") | |
| self.assertFalse(second["source_batch_acceptance"]["accepted_for_future_training"]) | |
| self.assertEqual(second["previous_best_comparison"]["previous_best_run_id"], "run_test_best_tracker_good") | |
| finally: | |
| for run_path in [run_good, run_bad]: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| def test_continuous_status_writes_intelligence_and_next_data_strategy(self) -> None: | |
| run_id = "run_test_continuous_status" | |
| release_id = "linvest21_fingpt_equity_researcher_v1_test" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "remote_artifacts").mkdir(parents=True) | |
| (run_path / "dataset_snapshot").mkdir(parents=True) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.25, "critical_pass_rate": 0.2, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.25, | |
| "pairwise_win_rate": 0.62, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "model_quality_gate.json", | |
| { | |
| "ok": False, | |
| "errors": [ | |
| "candidate_aggregate_absolute: 0.2500 >= 0.6", | |
| "critical_pass_absolute: 0.2000 >= 0.7", | |
| ], | |
| }, | |
| ) | |
| write_json(run_path / "remote_artifacts" / "training_result.json", {"train_loss": 1.01}) | |
| write_json( | |
| run_path / "remote_artifacts" / "training_plan.json", | |
| {"train_records": 180, "valid_records": 20}, | |
| ) | |
| report = write_continuous_status( | |
| run_id=run_id, | |
| release_id=release_id, | |
| asset_class="equity", | |
| role="researcher", | |
| round_index=2, | |
| phase="quality_gate", | |
| ) | |
| self.assertTrue(report["ok"]) | |
| self.assertFalse(report["certified"]) | |
| self.assertEqual(report["current_intelligence"]["candidate_aggregate"], 0.25) | |
| self.assertEqual(report["current_intelligence"]["train_records"], 180) | |
| self.assertIn("source_batch_acceptance", report) | |
| self.assertIn("critical pass/fail", " ".join(report["next_data_strategy"]["actions"])) | |
| self.assertTrue((run_path / "continuous_training_status.json").exists()) | |
| self.assertTrue((run_path / "next_data_strategy.json").exists()) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| def test_continuous_status_halts_when_breakout_repeats_without_trainable_sources(self) -> None: | |
| run_id = "run_test_continuous_status_needs_reasoning_data" | |
| release_id = "linvest21_fingpt_equity_portfolio_manager_v1_test" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_test_previous_best", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.55, | |
| "candidate_critical_pass_rate": 0.65, | |
| "aggregate_abs": 0.55, | |
| "pairwise_win_rate": 0.8, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.30, "critical_pass_rate": 0.25, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.30, | |
| "pairwise_win_rate": 0.7, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "model_quality_gate.json", | |
| { | |
| "ok": False, | |
| "errors": [ | |
| "candidate_aggregate_absolute: 0.3000 >= 0.6", | |
| "critical_pass_absolute: 0.2500 >= 0.7", | |
| ], | |
| }, | |
| ) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": {"attempt_count": 3}, | |
| "validation": {"record_count": 100}, | |
| "blockers": ["no newly AI-certified training sources are trainable by current Step 0b converters"], | |
| }, | |
| ) | |
| report = write_continuous_status( | |
| run_id=run_id, | |
| release_id=release_id, | |
| asset_class="equity", | |
| role="portfolio_manager", | |
| round_index=1, | |
| phase="source_recovery_retry", | |
| ) | |
| control = report["convergence_control"] | |
| self.assertEqual(control["state"], "NEEDS_REASONING_DATA") | |
| self.assertTrue(control["severe_regression"]) | |
| self.assertTrue(control["should_halt_paid_retraining"]) | |
| self.assertTrue(report["next_data_strategy"]["escalation"].endswith("needs high-signal critical-reasoning examples")) | |
| self.assertIn("paired-eval failure repair", " ".join(report["next_data_strategy"]["actions"])) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| def test_continuous_status_keeps_discovery_running_for_minor_regression_before_attempt_floor(self) -> None: | |
| run_id = "run_test_continuous_status_minor_regression" | |
| release_id = "linvest21_fingpt_equity_researcher_v1_minor_regression" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| status_path = SHFT_WORKSPACE_ROOT / "continuous_training" / f"{release_id}_status.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_test_previous_best_minor", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.33, | |
| "candidate_critical_pass_rate": 0.28, | |
| "aggregate_abs": 0.33, | |
| "pairwise_win_rate": 0.7, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.30, "critical_pass_rate": 0.25, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.30, | |
| "pairwise_win_rate": 0.7, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "model_quality_gate.json", | |
| { | |
| "ok": False, | |
| "errors": [ | |
| "candidate_aggregate_absolute: 0.3000 >= 0.6", | |
| "critical_pass_absolute: 0.2500 >= 0.7", | |
| ], | |
| }, | |
| ) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": {"attempt_count": 2}, | |
| "validation": {"record_count": 100}, | |
| "blockers": ["no newly AI-certified training sources are trainable by current Step 0b converters"], | |
| }, | |
| ) | |
| report = write_continuous_status( | |
| run_id=run_id, | |
| release_id=release_id, | |
| asset_class="equity", | |
| role="researcher", | |
| round_index=1, | |
| phase="source_recovery_retry", | |
| ) | |
| control = report["convergence_control"] | |
| self.assertEqual(control["state"], "CONTINUE") | |
| self.assertFalse(control["severe_regression"]) | |
| self.assertFalse(control["should_halt_paid_retraining"]) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| def test_continuous_status_halts_when_retry_returns_zero_candidates(self) -> None: | |
| run_id = "run_test_continuous_status_zero_candidate_retry" | |
| release_id = "linvest21_fingpt_equity_portfolio_manager_v1_zero_candidate_retry" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| status_path = SHFT_WORKSPACE_ROOT / "continuous_training" / f"{release_id}_status.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_test_previous_best_zero_candidate", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.5917, | |
| "candidate_critical_pass_rate": 0.675, | |
| "aggregate_abs": 0.5917, | |
| "pairwise_win_rate": 0.9833, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.5677, "critical_pass_rate": 0.6583, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.5677, | |
| "pairwise_win_rate": 0.8667, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "model_quality_gate.json", | |
| { | |
| "ok": False, | |
| "errors": [ | |
| "candidate_aggregate_absolute: 0.5677 >= 0.6", | |
| "critical_pass_absolute: 0.6583 >= 0.7", | |
| ], | |
| }, | |
| ) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": {"attempt_count": 2, "candidate_count": 0}, | |
| "validation": {"record_count": 43340}, | |
| "blockers": ["no newly AI-certified training sources are trainable by current Step 0b converters"], | |
| }, | |
| ) | |
| report = write_continuous_status( | |
| run_id=run_id, | |
| release_id=release_id, | |
| asset_class="equity", | |
| role="portfolio_manager", | |
| round_index=2, | |
| phase="source_recovery_retry", | |
| ) | |
| control = report["convergence_control"] | |
| self.assertEqual(control["state"], "NEEDS_REASONING_DATA") | |
| self.assertTrue(control["no_candidate_retry_exhausted"]) | |
| self.assertEqual(control["final_live_discovery_candidate_count"], 0) | |
| self.assertTrue(control["should_halt_paid_retraining"]) | |
| self.assertIn("zero candidates", " ".join(control["reasons"])) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| def test_continuous_status_halts_when_candidates_are_not_trainable(self) -> None: | |
| run_id = "run_test_continuous_status_non_trainable_candidates" | |
| release_id = "linvest21_fingpt_equity_portfolio_manager_v1_non_trainable_candidates" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| status_path = SHFT_WORKSPACE_ROOT / "continuous_training" / f"{release_id}_status.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_test_previous_best_non_trainable", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.61, | |
| "candidate_critical_pass_rate": 0.71, | |
| "aggregate_abs": 0.61, | |
| "pairwise_win_rate": 0.9, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "candidate": {"aggregate": 0.59, "critical_pass_rate": 0.69, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.59, | |
| "pairwise_win_rate": 0.8, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json(run_path / "eval" / "model_quality_gate.json", {"ok": False, "errors": []}) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": { | |
| "attempt_count": 2, | |
| "candidate_count": 3, | |
| "intake_training_eligible_count": 0, | |
| "content_ai_rejected_count": 3, | |
| "ai_rejected_count": 0, | |
| }, | |
| "validation": {"record_count": 1000}, | |
| "blockers": ["only verification or rejected material was found"], | |
| }, | |
| ) | |
| report = write_continuous_status( | |
| run_id=run_id, | |
| release_id=release_id, | |
| asset_class="equity", | |
| role="portfolio_manager", | |
| round_index=2, | |
| phase="source_recovery_retry", | |
| ) | |
| control = report["convergence_control"] | |
| self.assertEqual(control["state"], "NEEDS_REASONING_DATA") | |
| self.assertTrue(control["no_trainable_candidate_retry_exhausted"]) | |
| self.assertFalse(control["no_candidate_retry_exhausted"]) | |
| self.assertEqual(control["final_live_discovery_candidate_count"], 3) | |
| self.assertEqual(control["final_live_discovery_training_eligible_count"], 0) | |
| self.assertTrue(control["should_halt_paid_retraining"]) | |
| self.assertIn("none were training-eligible", " ".join(control["reasons"])) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| def test_human_owner_decision_accepts_first_stdin_instruction_and_writes_email_ask(self) -> None: | |
| run_id = "run_test_human_owner_stdin_first" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| decision_root = SHFT_WORKSPACE_ROOT / "human_owner_decisions" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| try: | |
| with mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_RUN_OWNER_EMAIL": "david.d.lin@linvest21.com", | |
| "SHFT_HUMAN_OWNER_ASK_TIMEOUT_SECONDS": "2", | |
| "SHFT_HUMAN_OWNER_ASK_POLL_SECONDS": "0.01", | |
| "SHFT_HUMAN_OWNER_READ_STDIN": "true", | |
| "SHFT_EMAIL_DELIVERY": "outbox", | |
| }, | |
| clear=False, | |
| ): | |
| decision = request_human_owner_instruction( | |
| run_id=run_id, | |
| release_id="linvest21_test_release", | |
| asset_class="equity", | |
| role="portfolio_manager", | |
| reason="test convergence guard", | |
| stdin=io.StringIO("continue\n"), | |
| stdout=io.StringIO(), | |
| stderr=io.StringIO(), | |
| ) | |
| self.assertTrue(decision["ok"]) | |
| self.assertEqual(decision["owner_email"], "david.d.lin@linvest21.com") | |
| self.assertEqual(decision["decision"]["instruction"], "continue") | |
| self.assertEqual(decision["decision"]["source"], "stdin") | |
| self.assertIn("outbox_path", decision["email"]) | |
| self.assertFalse(decision["email"]["delivered"]) | |
| self.assertTrue((run_path / "human_owner_decision.json").exists()) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| def test_human_owner_decision_accepts_email_response_file_before_stdin(self) -> None: | |
| run_id = "run_test_human_owner_email_first" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| decision_root = SHFT_WORKSPACE_ROOT / "human_owner_decisions" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| try: | |
| run_path.mkdir(parents=True) | |
| write_json(run_path / "human_owner_response.json", {"instruction": "exit", "reviewer": "owner"}) | |
| with mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_HUMAN_OWNER_ASK_TIMEOUT_SECONDS": "1", | |
| "SHFT_HUMAN_OWNER_ASK_POLL_SECONDS": "0.01", | |
| "SHFT_HUMAN_OWNER_READ_STDIN": "false", | |
| "SHFT_EMAIL_DELIVERY": "outbox", | |
| }, | |
| clear=False, | |
| ): | |
| decision = request_human_owner_instruction( | |
| run_id=run_id, | |
| release_id="linvest21_test_release", | |
| asset_class="equity", | |
| role="portfolio_manager", | |
| reason="test convergence guard", | |
| stdin=io.StringIO("continue\n"), | |
| stdout=io.StringIO(), | |
| stderr=io.StringIO(), | |
| ) | |
| self.assertTrue(decision["ok"]) | |
| self.assertEqual(decision["decision"]["instruction"], "exit") | |
| self.assertEqual(decision["decision"]["source"], "email_response_file") | |
| self.assertEqual(decision["decision"]["raw_response"]["reviewer"], "owner") | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| def test_human_owner_email_uses_outlook_when_smtp_is_absent(self) -> None: | |
| run_id = "run_test_human_owner_outlook_delivery" | |
| decision_root = SHFT_WORKSPACE_ROOT / "human_owner_decisions" | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| try: | |
| with mock.patch( | |
| "orchestrator.human_owner_decision._send_outlook_com" | |
| ) as send_outlook, mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_EMAIL_DELIVERY": "outlook", | |
| "SHFT_SMTP_HOST": "", | |
| }, | |
| clear=False, | |
| ): | |
| record = _send_email_or_outbox( | |
| to_email="david.d.lin@linvest21.com", | |
| subject="SHFT test owner ask", | |
| body="Reply continue or exit.", | |
| run_id=run_id, | |
| ) | |
| send_outlook.assert_called_once() | |
| self.assertEqual(record["delivery_status"], "sent_outlook_com") | |
| self.assertTrue(record["delivered"]) | |
| self.assertEqual(record["to"], "david.d.lin@linvest21.com") | |
| self.assertTrue(Path(record["outbox_path"]).exists()) | |
| finally: | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| def test_human_owner_email_uses_shcg_smtp_fallbacks(self) -> None: | |
| run_id = "run_test_human_owner_shcg_smtp" | |
| decision_root = SHFT_WORKSPACE_ROOT / "human_owner_decisions" | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| smtp_instance = mock.Mock() | |
| smtp_context = mock.Mock() | |
| smtp_context.__enter__ = mock.Mock(return_value=smtp_instance) | |
| smtp_context.__exit__ = mock.Mock(return_value=False) | |
| try: | |
| with mock.patch("smtplib.SMTP", return_value=smtp_context) as smtp_ctor, mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_EMAIL_DELIVERY": "smtp", | |
| "SHFT_SMTP_HOST": "", | |
| "SHFT_SMTP_FROM": "", | |
| "SHFT_SMTP_USERNAME": "", | |
| "SHFT_SMTP_PASSWORD": "", | |
| "SHCG_ALERT_SMTP_PASSWORD": "test-password", | |
| }, | |
| clear=False, | |
| ): | |
| record = _send_email_or_outbox( | |
| to_email="david.d.lin@linvest21.com", | |
| subject="SHFT test owner ask", | |
| body="Reply continue or exit.", | |
| run_id=run_id, | |
| ) | |
| smtp_ctor.assert_called_once() | |
| self.assertEqual(smtp_ctor.call_args.args[0], "smtp.gmail.com") | |
| self.assertEqual(smtp_ctor.call_args.args[1], 587) | |
| smtp_instance.starttls.assert_called_once() | |
| smtp_instance.login.assert_called_once_with("david.d.lin@linvest21.com", "test-password") | |
| smtp_instance.send_message.assert_called_once() | |
| self.assertEqual(record["delivery_status"], "sent_smtp") | |
| self.assertTrue(record["delivered"]) | |
| self.assertEqual(record["from"], "david.d.lin@linvest21.com") | |
| finally: | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| def test_human_owner_decision_supports_all_asset_role_pairs(self) -> None: | |
| assets = ["equity", "fixed_income", "multi_asset"] | |
| roles = [ | |
| "chief_investment_officer", | |
| "client_portfolio_manager", | |
| "performance_manager", | |
| "portfolio_manager", | |
| "researcher", | |
| "risk_manager", | |
| ] | |
| decision_root = SHFT_WORKSPACE_ROOT / "human_owner_decisions" | |
| touched_run_paths: list[Path] = [] | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| try: | |
| with mock.patch( | |
| "orchestrator.human_owner_decision._send_email_or_outbox", | |
| side_effect=lambda **kwargs: { | |
| "delivery_status": "mock_sent", | |
| "to": kwargs["to_email"], | |
| "subject": kwargs["subject"], | |
| "body": kwargs["body"], | |
| }, | |
| ), mock.patch.dict( | |
| os.environ, | |
| { | |
| "SHFT_RUN_OWNER_EMAIL": "david.d.lin@linvest21.com", | |
| "SHFT_HUMAN_OWNER_ASK_TIMEOUT_SECONDS": "1", | |
| "SHFT_HUMAN_OWNER_ASK_POLL_SECONDS": "0.01", | |
| "SHFT_HUMAN_OWNER_READ_STDIN": "true", | |
| "SHFT_EMAIL_DELIVERY": "outbox", | |
| }, | |
| clear=False, | |
| ): | |
| for asset in assets: | |
| for role in roles: | |
| run_id = f"run_test_owner_decision_{asset}_{role}" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| touched_run_paths.append(run_path) | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| stdout = io.StringIO() | |
| stderr = io.StringIO() | |
| decision = request_human_owner_instruction( | |
| run_id=run_id, | |
| release_id=f"linvest21_fingpt_{asset}_{role}_v1_001", | |
| asset_class=asset, | |
| role=role, | |
| reason="matrix proof", | |
| stdin=io.StringIO("continue\n"), | |
| stdout=stdout, | |
| stderr=stderr, | |
| ) | |
| self.assertTrue(decision["ok"]) | |
| self.assertEqual(decision["decision"]["instruction"], "continue") | |
| self.assertEqual(decision["owner_email"], "david.d.lin@linvest21.com") | |
| self.assertIn(f"Asset/role: {asset}/{role}", decision["question"]) | |
| self.assertIn(f"Asset/role: {asset}/{role}", stdout.getvalue()) | |
| self.assertIn(f"Asset/role: {asset}/{role}", stderr.getvalue()) | |
| self.assertEqual(len(touched_run_paths), 18) | |
| finally: | |
| for run_path in touched_run_paths: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if decision_root.exists(): | |
| shutil.rmtree(decision_root) | |
| def test_continuous_status_enforce_convergence_uses_owner_continue_instruction(self) -> None: | |
| run_id = "run_test_continuous_status_owner_continue" | |
| release_id = "linvest21_fingpt_equity_portfolio_manager_v1_owner_continue" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| status_path = SHFT_WORKSPACE_ROOT / "continuous_training" / f"{release_id}_status.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_previous_best_owner_continue", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.61, | |
| "candidate_critical_pass_rate": 0.71, | |
| "aggregate_abs": 0.61, | |
| "pairwise_win_rate": 0.9, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "candidate": {"aggregate": 0.59, "critical_pass_rate": 0.69, "sample_count": 120}, | |
| "improvement": {"aggregate_abs": 0.59, "pairwise_win_rate": 0.8, "pairwise_loss_rate": 0.0}, | |
| }, | |
| ) | |
| write_json(run_path / "eval" / "model_quality_gate.json", {"ok": False, "errors": []}) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": {"attempt_count": 2, "candidate_count": 0}, | |
| "validation": {"record_count": 1000}, | |
| "blockers": ["no trainable material"], | |
| }, | |
| ) | |
| fake_decision = { | |
| "owner_email": "david.d.lin@linvest21.com", | |
| "decision": {"instruction": "continue", "source": "stdin"}, | |
| "ok": True, | |
| } | |
| args = type( | |
| "Args", | |
| (), | |
| { | |
| "run_id": run_id, | |
| "release_id": release_id, | |
| "asset_class": "equity", | |
| "role": "portfolio_manager", | |
| "round_index": 2, | |
| "phase": "source_recovery_retry", | |
| "enforce_convergence": True, | |
| }, | |
| )() | |
| with mock.patch("n21.cli.request_human_owner_instruction", return_value=fake_decision), redirect_stdout(io.StringIO()): | |
| rc = shft_cli.command_continuous_status(args) | |
| self.assertEqual(rc, 0) | |
| status = json.loads((run_path / "continuous_training_status.json").read_text(encoding="utf-8-sig")) | |
| self.assertEqual(status["human_owner_decision"]["decision"]["instruction"], "continue") | |
| self.assertFalse(status["convergence_control"]["should_halt_paid_retraining"]) | |
| self.assertEqual(status["convergence_control"]["owner_override"], "continue") | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| def test_continuous_status_enforce_convergence_uses_owner_exit_instruction(self) -> None: | |
| run_id = "run_test_continuous_status_owner_exit" | |
| release_id = "linvest21_fingpt_equity_portfolio_manager_v1_owner_exit" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| status_path = SHFT_WORKSPACE_ROOT / "continuous_training" / f"{release_id}_status.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_previous_best_owner_exit", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.61, | |
| "candidate_critical_pass_rate": 0.71, | |
| "aggregate_abs": 0.61, | |
| "pairwise_win_rate": 0.9, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "candidate": {"aggregate": 0.59, "critical_pass_rate": 0.69, "sample_count": 120}, | |
| "improvement": {"aggregate_abs": 0.59, "pairwise_win_rate": 0.8, "pairwise_loss_rate": 0.0}, | |
| }, | |
| ) | |
| write_json(run_path / "eval" / "model_quality_gate.json", {"ok": False, "errors": []}) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": {"attempt_count": 2, "candidate_count": 0}, | |
| "validation": {"record_count": 1000}, | |
| "blockers": ["no trainable material"], | |
| }, | |
| ) | |
| fake_decision = { | |
| "owner_email": "david.d.lin@linvest21.com", | |
| "decision": {"instruction": "exit", "source": "stdin"}, | |
| "ok": True, | |
| } | |
| args = type( | |
| "Args", | |
| (), | |
| { | |
| "run_id": run_id, | |
| "release_id": release_id, | |
| "asset_class": "equity", | |
| "role": "portfolio_manager", | |
| "round_index": 2, | |
| "phase": "source_recovery_retry", | |
| "enforce_convergence": True, | |
| }, | |
| )() | |
| with mock.patch("n21.cli.request_human_owner_instruction", return_value=fake_decision), redirect_stdout(io.StringIO()): | |
| rc = shft_cli.command_continuous_status(args) | |
| self.assertEqual(rc, 10) | |
| status = json.loads((run_path / "continuous_training_status.json").read_text(encoding="utf-8-sig")) | |
| self.assertEqual(status["human_owner_decision"]["decision"]["instruction"], "exit") | |
| self.assertTrue(status["convergence_control"]["should_halt_paid_retraining"]) | |
| self.assertEqual(status["convergence_control"]["owner_override"], "exit") | |
| self.assertEqual(status["convergence_control"]["exit_code"], 10) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| def test_continuous_status_severe_regression_reads_critical_pass_delta_key(self) -> None: | |
| run_id = "run_test_continuous_status_critical_pass_key" | |
| release_id = "linvest21_fingpt_equity_researcher_v1_critical_pass_key" | |
| run_path = SHFT_WORKSPACE_ROOT / "runs" / run_id | |
| best_path = SHFT_WORKSPACE_ROOT / "best_runs" / f"{release_id}.json" | |
| status_path = SHFT_WORKSPACE_ROOT / "continuous_training" / f"{release_id}_status.json" | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| try: | |
| (run_path / "eval").mkdir(parents=True) | |
| (run_path / "stall_breakout").mkdir(parents=True) | |
| write_json( | |
| best_path, | |
| { | |
| "best_run": { | |
| "run_id": "run_test_previous_best_critical", | |
| "paired_eval_present": True, | |
| "model_quality_ok": False, | |
| "candidate_aggregate": 0.31, | |
| "candidate_critical_pass_rate": 0.65, | |
| "aggregate_abs": 0.31, | |
| "pairwise_win_rate": 0.7, | |
| "pairwise_loss_rate": 0.0, | |
| } | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "paired_eval_report.json", | |
| { | |
| "sample_count": 120, | |
| "baseline": {"aggregate": 0.0, "critical_pass_rate": 0.0}, | |
| "candidate": {"aggregate": 0.30, "critical_pass_rate": 0.25, "sample_count": 120}, | |
| "improvement": { | |
| "aggregate_abs": 0.30, | |
| "pairwise_win_rate": 0.7, | |
| "pairwise_loss_rate": 0.0, | |
| }, | |
| }, | |
| ) | |
| write_json( | |
| run_path / "eval" / "model_quality_gate.json", | |
| { | |
| "ok": False, | |
| "errors": [ | |
| "candidate_aggregate_absolute: 0.3000 >= 0.6", | |
| "critical_pass_absolute: 0.2500 >= 0.7", | |
| ], | |
| }, | |
| ) | |
| write_json( | |
| run_path / "stall_breakout" / "stall_breakout_plan.json", | |
| { | |
| "status": "blocked_after_breakout", | |
| "intake": {"trainable_new_source_count": 0}, | |
| "live_discovery": {"attempt_count": 0}, | |
| "validation": {"record_count": 100}, | |
| "blockers": ["no newly AI-certified training sources are trainable by current Step 0b converters"], | |
| }, | |
| ) | |
| report = write_continuous_status( | |
| run_id=run_id, | |
| release_id=release_id, | |
| asset_class="equity", | |
| role="researcher", | |
| round_index=1, | |
| phase="stall_breakout", | |
| ) | |
| control = report["convergence_control"] | |
| self.assertEqual(control["state"], "NEEDS_REASONING_DATA") | |
| self.assertEqual(control["critical_delta_vs_previous_best"], -0.4) | |
| self.assertTrue(control["severe_regression"]) | |
| self.assertTrue(control["should_halt_paid_retraining"]) | |
| finally: | |
| if run_path.exists(): | |
| shutil.rmtree(run_path) | |
| if best_path.exists(): | |
| best_path.unlink() | |
| if status_path.exists(): | |
| status_path.unlink() | |
| def test_grounded_reasoning_generator_writes_certified_examples(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "portfolio_source.hf_finetune.jsonl" | |
| output = root / "synthetic_portfolio_reasoning.hf_finetune.jsonl" | |
| row = { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| { | |
| "role": "user", | |
| "content": "Portfolio construction source excerpt about position sizing and risk budget.", | |
| }, | |
| { | |
| "role": "assistant", | |
| "content": ( | |
| "Equity portfolio construction should connect position sizing to cash flow, valuation, " | |
| "risk budget, rebalancing discipline, and downside tradeoff evidence." | |
| ), | |
| }, | |
| ], | |
| "metadata": {"source_title": "Portfolio Construction Risk Budget Note"}, | |
| } | |
| source.write_text(json.dumps(row) + "\n", encoding="utf-8") | |
| report = generate_grounded_reasoning_examples( | |
| asset_class="equity", | |
| role="portfolio_manager", | |
| source=source, | |
| output_path=output, | |
| max_records=1, | |
| ) | |
| self.assertTrue(report["ok"], report) | |
| self.assertEqual(report["generated_count"], 1) | |
| generated = load_learning_jsonl(output) | |
| self.assertEqual(len(generated), 1) | |
| metadata = generated[0]["metadata"] | |
| self.assertTrue(metadata["synthetic"]) | |
| self.assertTrue(metadata["content_ai_certification"]["training_eligible"]) | |
| self.assertEqual(metadata["rubric_target"], "critical_pass_reasoning") | |
| def test_training_data_validation_reports_concept_conflict_and_quarantine_recommendation(self) -> None: | |
| with tempfile.TemporaryDirectory() as tmp: | |
| root = Path(tmp) | |
| source = root / "role.hf_finetune.jsonl" | |
| rows = [ | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "How should debt be assessed?"}, | |
| {"role": "assistant", "content": "Debt can be optimal when cash flows and maturity are resilient."}, | |
| ], | |
| "metadata": {"source": "majority"}, | |
| }, | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "How should leverage be assessed?"}, | |
| {"role": "assistant", "content": "Leverage can improve returns but must be sized against downside."}, | |
| ], | |
| "metadata": {"source": "majority2"}, | |
| }, | |
| { | |
| "messages": [ | |
| {"role": "system", "content": "system"}, | |
| {"role": "user", "content": "What about debt?"}, | |
| {"role": "assistant", "content": "Debt is always bad and analysts should avoid all debt."}, | |
| ], | |
| "metadata": {"source": "minority"}, | |
| }, | |
| ] | |
| source.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8") | |
| report = validate_training_data( | |
| source=source, | |
| output_dir=root / "validation", | |
| backup_dir=root / "backup", | |
| apply_quarantine=True, | |
| ) | |
| self.assertTrue(report["ok"], report["schema_errors"]) | |
| self.assertEqual(report["conflict_count"], 1) | |
| self.assertEqual(report["quarantine_manifest"]["recommended_quarantine_count"], 1) | |
| self.assertTrue((Path(report["quarantine_manifest"]["backup_dir"]) / "quarantined_records.jsonl").exists()) | |
| def test_portable_release_bundle_exists(self) -> None: | |
| release_id = "linvest21_fingpt_equity_researcher_v1_000" | |
| release = IMPLEMENTATION_PRODUCTS_ROOT / release_id | |
| if not (release / "release_manifest.json").exists(): | |
| self.skipTest(f"portable release fixture is not present: {release}") | |
| manifest = json.loads((release / "release_manifest.json").read_text(encoding="utf-8")) | |
| self.assertEqual(manifest["release_id"], release_id) | |
| self.assertEqual(manifest["model_id"], release_id) | |
| self.assertEqual(manifest["asset_class"], "equity") | |
| self.assertEqual(manifest["role"], "researcher") | |
| self.assertEqual(manifest["model"]["export_mode"], "adapter_only") | |
| self.assertEqual(manifest["runtime"]["serve_api"], "runtime/serve_api.py") | |
| self.assertEqual(manifest["runtime"]["run_api_cpu"], "runtime/run_api_cpu.bat") | |
| self.assertEqual(manifest["runtime"]["api_self_test"], "runtime/self_test_api_contract.py") | |
| self.assertEqual(manifest["distribution_policy"]["tokens_in_bundle"], "forbidden") | |
| self.assertTrue((release / "model" / "adapter" / "adapter_model.safetensors").exists()) | |
| self.assertTrue((release / "runtime" / "chat_console.py").exists()) | |
| self.assertTrue((release / "runtime" / "serve_api.py").exists()) | |
| self.assertTrue((release / "runtime" / "run_api_cpu.bat").exists()) | |
| self.assertTrue((release / "runtime" / "run_api_gpu.bat").exists()) | |
| self.assertTrue((release / "runtime" / "self_test_api_contract.py").exists()) | |
| self.assertTrue((release / "runtime" / "run_api_self_test.bat").exists()) | |
| self.assertTrue((release / "tools" / "merge_hf_model.py").exists()) | |
| self.assertTrue((release / "release_hashes.json").exists()) | |
| chat_config = json.loads((release / "runtime" / "chat_config.json").read_text(encoding="utf-8")) | |
| self.assertEqual(chat_config["model_id"], release_id) | |
| self.assertEqual(chat_config["asset_class"], "equity") | |
| self.assertEqual(chat_config["role"], "researcher") | |
| if __name__ == "__main__": | |
| unittest.main() | |
Xet Storage Details
- Size:
- 159 kB
- Xet hash:
- e538d9d1b1d197c06e4d5618d9646f644a12d8d94381d369e592017bb3b51c4b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.