Add launch readiness report

tests/unit/test_api_server.py

@@ -134,6 +134,17 @@ class ApiServerRoutingTests(unittest.TestCase):
         self.assertFalse(payload["ready_for_real_live"])
         self.assertEqual(payload["timeframe"], "5m")
 
+    def test_routes_trading_launch_readiness(self) -> None:
+        status, payload = route_request(
+            "/trading/launch-readiness",
+            {"instrument_id": ["1"], "timeframe": ["5m"]},
+            dependencies={"trading_launch_readiness": lambda **kwargs: {"assisted_trading_ready": False, "timeframe": kwargs["timeframe"]}},
+        )
+
+        self.assertEqual(status, 200)
+        self.assertFalse(payload["assisted_trading_ready"])
+        self.assertEqual(payload["timeframe"], "5m")
+
     def test_routes_trading_broker_state(self) -> None:
         status, payload = route_request(
             "/trading/broker-state",

tests/unit/test_documentation_consistency.py

@@ -17,6 +17,11 @@ TRADING_APP_JS = ROOT / "src" / "web" / "app.js"
 TRAINING_HTML = ROOT / "src" / "web" / "training.html"
 TRAINING_JS = ROOT / "src" / "web" / "training.js"
 OLD_TRAINING_DOC = ROOT / "docs" / "training_case_engine_m1_m6_product_readme.md"
+OPTIONAL_DORMANT_DOCS = {
+    "AI_ICT_盯盘告警开发任务清单_v1.md",
+    "AI_ICT_盯盘告警技术设计稿_v1.md",
+    "AI_ICT_盯盘告警架构图_v1.md",
+}
 
 
 class DocumentationConsistencyTests(unittest.TestCase):
@@ -34,7 +39,7 @@ class DocumentationConsistencyTests(unittest.TestCase):
                 "AI_ICT_实盘辅助决策系统开发方案_v2.md",
                 "ICT交易全书_出版终校版.md",
                 "training_console_public_surface_v1.md",
-            },
+            } | OPTIONAL_DORMANT_DOCS,
         )
         self.assertTrue(TRAINING_PACKAGE_SMOKE_DOC.is_file())
         self.assertFalse(OLD_TRAINING_DOC.exists())

tests/unit/test_launch_readiness_service.py

@@ -0,0 +1,80 @@
+import os
+import unittest
+from unittest.mock import patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
+
+from src.services.readiness import LaunchReadinessService
+
+
+class LaunchReadinessServiceTests(unittest.TestCase):
+    def test_build_summary_reports_assisted_not_ready_without_real_trade_sample(self) -> None:
+        service = LaunchReadinessService(
+            market_state_service=_MarketStateService(),
+            broker_read_model_service=_BrokerReadService(),
+            weekly_review_service=_WeeklyReviewService(),
+            trust_service=_TrustService(),
+            live_readiness_service=_LiveReadinessService(),
+        )
+
+        with patch("src.services.readiness.launch_readiness_service.get_trading_runtime_service", return_value=_RuntimeService()):
+            payload = service.build_summary(instrument_id=1, timeframe="1m")
+
+        self.assertFalse(payload["assisted_trading_ready"])
+        self.assertIn("real_trade_sample_verified", payload["blocking_reasons"])
+        self.assertFalse(payload["real_live_ready"])
+
+
+class _MarketState:
+    freshness = type("Freshness", (), {"code": "market_data_fresh"})()
+    stream_state = type("Stream", (), {"code": "market_stream_connected"})()
+    data_gap_state = type("Gap", (), {"code": "market_data_gap_clear"})()
+    news_window = type("News", (), {"provider_name": "cached", "freshness_code": "news_provider_cached", "error": None})()
+
+    def to_dict(self):
+        return {
+            "freshness": {"code": "market_data_fresh"},
+            "stream_state": {"code": "market_stream_connected"},
+            "data_gap_state": {"code": "market_data_gap_clear"},
+        }
+
+
+class _MarketStateService:
+    def build_market_state(self, **kwargs):
+        return _MarketState()
+
+
+class _BrokerState:
+    status = "broker_state_ok"
+
+    def to_dict(self):
+        return {"status": "broker_state_ok"}
+
+
+class _BrokerReadService:
+    def fetch_snapshot(self, **kwargs):
+        return _BrokerState()
+
+
+class _WeeklyReviewService:
+    def build_summary(self, **kwargs):
+        return {"real_trade_count": 0}
+
+
+class _TrustService:
+    def build_summary(self, **kwargs):
+        return {"whitelist": ["OSOK"], "observe_list": [], "blacklist": []}
+
+
+class _LiveReadinessService:
+    def build_summary(self, **kwargs):
+        return {"ready_for_real_live": False, "blocking_reasons": ["adapter_real_live_not_ready"]}
+
+
+class _RuntimeService:
+    def status(self):
+        return {"scheduler_state": "running", "current_blockers": []}
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_runtime_script_json.py

@@ -23,6 +23,7 @@ from scripts import (
     run_mss,
     run_ranges,
     run_decision_packet,
+    run_launch_readiness,
     run_review_packet,
     run_session_generator,
     run_live_readiness,
@@ -184,6 +185,19 @@ class RuntimeScriptJsonTests(unittest.TestCase):
         self.assertEqual(payload["instrument_id"], 2)
         self.assertEqual(payload["timeframe"], "5m")
 
+    def test_run_launch_readiness_outputs_json(self) -> None:
+        service = type("Service", (), {"build_summary": lambda self, instrument_id, timeframe: {"instrument_id": instrument_id, "timeframe": timeframe, "assisted_trading_ready": False, "real_live_ready": False, "blocking_reasons": ["real_trade_sample_verified"]}})()
+        output = io.StringIO()
+        with patch("scripts.run_launch_readiness.LaunchReadinessService", return_value=service), contextlib.redirect_stdout(output):
+            exit_code = run_launch_readiness.main(
+                env={"LAUNCH_READINESS_OUTPUT_JSON": "1", "LAUNCH_READINESS_INSTRUMENT_ID": "2", "LAUNCH_READINESS_TIMEFRAME": "5m"}
+            )
+
+        payload = json.loads(output.getvalue())
+        self.assertEqual(exit_code, 2)
+        self.assertEqual(payload["instrument_id"], 2)
+        self.assertEqual(payload["timeframe"], "5m")
+
 
 if __name__ == "__main__":
     unittest.main()