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()
|
