from __future__ import annotations import copy import json import sys import tempfile import unittest from datetime import date from decimal import Decimal from pathlib import Path ANA_DEV = Path(__file__).resolve().parents[2] if str(ANA_DEV) not in sys.path: sys.path.insert(0, str(ANA_DEV)) from stock_valuation_pipeline.core import ( # noqa: E402 InputError, compute_valuation, load_snapshot, run_pipeline, ) PACKAGE_DIR = ANA_DEV / "stock_valuation_pipeline" FIXTURE = Path(__file__).with_name("fixtures") / "great_wall_military_20260731.json" REGISTRY = PACKAGE_DIR / "source_registry.json" def load_registry() -> dict: with REGISTRY.open("r", encoding="utf-8") as handle: return json.load(handle, parse_float=Decimal, parse_int=Decimal) class PipelineRegressionTests(unittest.TestCase): def setUp(self) -> None: self.snapshot = load_snapshot(FIXTURE) self.registry = load_registry() def test_great_wall_military_regression(self) -> None: result = compute_valuation(self.snapshot, self.registry) metrics = result["metrics"] self.assertEqual(metrics["market_cap"], Decimal("24580311896.00")) self.assertEqual(metrics["ttm_revenue"], Decimal("1573960000")) self.assertEqual(metrics["ttm_attributable_profit"], Decimal("-60250000")) self.assertEqual(metrics["ttm_deduct_profit"], Decimal("-80120000")) self.assertEqual(metrics["annual_fcf"], Decimal("-244770000")) self.assertAlmostEqual(float(metrics["pb"]), 11.26415994, places=7) self.assertAlmostEqual(float(metrics["ps"]), 15.61685932, places=7) self.assertIsNone(metrics["reported_pe"]) self.assertEqual(result["conclusion"]["price_position"], "ABOVE_OPTIMISTIC_RANGE") base = next(row for row in result["scenarios"] if row["role"] == "base") optimistic = next(row for row in result["scenarios"] if row["role"] == "optimistic") self.assertAlmostEqual(float(base["price_low"]), 4.14233963, places=7) self.assertAlmostEqual(float(base["price_high"]), 7.24909435, places=7) self.assertAlmostEqual(float(optimistic["price_high"]), 24.16364782, places=7) reverse_50 = next(row for row in result["reverse_pe"] if row["pe"] == Decimal("50")) self.assertAlmostEqual(float(reverse_50["implied_profit"] / Decimal("100000000")), 4.91606238, places=7) self.assertAlmostEqual(float(result["holding_period"]["required_exit_price"]), 55.91434324, places=7) def test_market_cap_gap_is_blocking_error(self) -> None: snapshot = copy.deepcopy(self.snapshot) snapshot["market"]["platform_market_cap"] = Decimal("20000000000") result = compute_valuation(snapshot, self.registry) self.assertEqual(result["qa"]["status"], "BLOCKED_INPUT_ERRORS") self.assertIn("QA-MARKET-CAP-RECONCILIATION", {row["issue_id"] for row in result["qa"]["issues"]}) def test_stale_institution_forecast_is_excluded(self) -> None: snapshot = copy.deepcopy(self.snapshot) snapshot["institutions"] = { "coverage_status": "available", "forecasts": [ { "institution": "示例证券", "report_date": "2026-07-20", "include": True, "core_assumption": "公告前假设", "source_id": "SRC-INSTITUTION-20260731", "estimates": {"2026": {"profit": 200000000, "eps": 0.276}} } ] } result = compute_valuation(snapshot, self.registry) detail = result["institutions"]["detail"][0] self.assertFalse(detail["included"]) self.assertIn("STALE_BEFORE_LATEST_DISCLOSURE", detail["statuses"]) self.assertEqual(result["institutions"]["summary"], {}) def test_invalid_holding_period_fails_fast(self) -> None: snapshot = copy.deepcopy(self.snapshot) snapshot["valuation"]["holding_period"]["years"] = Decimal("2.5") with self.assertRaises(InputError): compute_valuation(snapshot, self.registry) def test_invalid_source_date_fails_fast(self) -> None: snapshot = copy.deepcopy(self.snapshot) snapshot["sources"][0]["publish_date"] = "2026-13-40" with self.assertRaises(InputError): compute_valuation(snapshot, self.registry) def test_reversed_scenario_range_fails_fast(self) -> None: snapshot = copy.deepcopy(self.snapshot) snapshot["valuation"]["scenarios"][1]["profit_low"] = Decimal("200000000") snapshot["valuation"]["scenarios"][1]["profit_high"] = Decimal("100000000") with self.assertRaises(InputError): compute_valuation(snapshot, self.registry) def test_future_source_is_blocking_error(self) -> None: snapshot = copy.deepcopy(self.snapshot) snapshot["sources"][0]["publish_date"] = date(2026, 8, 1).isoformat() result = compute_valuation(snapshot, self.registry) self.assertEqual(result["qa"]["status"], "BLOCKED_INPUT_ERRORS") self.assertIn("QA-SOURCE-FUTURE-DATE", {row["issue_id"] for row in result["qa"]["issues"]}) def test_identical_run_reuses_verified_cache(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) first = run_pipeline(FIXTURE, output, REGISTRY) second = run_pipeline(FIXTURE, output, REGISTRY) self.assertEqual(first["status"], "GENERATED") self.assertEqual(second["status"], "REUSED") self.assertEqual(first["fingerprint"], second["fingerprint"]) self.assertTrue((output / "valuation_report.md").is_file()) self.assertTrue((output / "valuation_results.json").is_file()) self.assertTrue((output / "run_manifest.json").is_file()) manifest = json.loads((output / "run_manifest.json").read_text(encoding="utf-8")) self.assertRegex(manifest["engine_fingerprint"], r"^[0-9a-f]{64}$") def test_tampered_cache_is_regenerated(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) first = run_pipeline(FIXTURE, output, REGISTRY) report = output / "valuation_report.md" report.write_text(report.read_text(encoding="utf-8") + "被篡改\n", encoding="utf-8") second = run_pipeline(FIXTURE, output, REGISTRY) self.assertEqual(first["status"], "GENERATED") self.assertEqual(second["status"], "GENERATED") self.assertNotIn("被篡改", report.read_text(encoding="utf-8")) def test_generated_markdown_has_closed_fences_and_no_placeholders(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) run_pipeline(FIXTURE, output, REGISTRY) report = (output / "valuation_report.md").read_text(encoding="utf-8") self.assertEqual(report.count("```text") + report.count("```\n"), 4) self.assertNotIn("TODO", report) self.assertNotIn("TBD", report) if __name__ == "__main__": unittest.main()