from __future__ import annotations import tempfile import unittest from pathlib import Path from types import SimpleNamespace from backend.features.screener.catalog import BUILTIN_STRATEGIES from backend.features.screener.publication import resolve_published_batch from backend.features.screener.service import ScreenerServiceMixin from backend.features.screener.signals import build_candidate_archive, signal_validity from database import ReviewDatabase def _result( trade_date: str, mode: str, strategy_name: str, code: str = "600000", regime: str = "repair", frequency: str = "每日", run_id: int = 1, ) -> dict: candidates = [] if not code else [{ "code": code, "name": "浦发银行", "sector": "银行", "score_display": 72.5, "pct_chg": 1.2, "return_5d": 3.4, }] return { "meta": { "trade_date": trade_date, "mode": mode, "strategy_name": strategy_name, "regime": regime, "run_id": run_id, "health": {"status": "normal" if candidates else "no_signal"}, }, "formula": {"meta": {"frequency": frequency}}, "candidates": candidates, } class ScreenerPublicationTests(unittest.TestCase): def test_incomplete_requested_batch_retains_previous_complete_batch(self): for status in ("running", "failed", "partial"): markers = [ {"trade_date": "20260804", "status": status}, {"trade_date": "20260803", "status": "complete", "finished_at": "done"}, ] published, requested, descriptor = resolve_published_batch( markers, "20260804" ) self.assertEqual(published["trade_date"], "20260803", status) self.assertEqual(requested["status"], status) self.assertTrue(descriptor["is_fallback"]) def test_complete_requested_batch_switches_publication(self): markers = [ {"trade_date": "20260804", "status": "complete"}, {"trade_date": "20260803", "status": "complete"}, ] published, requested, descriptor = resolve_published_batch( markers, "20260804" ) self.assertEqual(published["trade_date"], "20260804") self.assertEqual(requested["status"], "complete") self.assertFalse(descriptor["is_fallback"]) def test_legacy_result_is_inferred_only_when_caller_allows_it(self): published, requested, descriptor = resolve_published_batch( [], "20260804", "20260804" ) self.assertTrue(published["legacy_inferred"]) self.assertEqual(requested["status"], "pending") self.assertFalse(descriptor["is_fallback"]) def test_all_builtin_strategies_declare_signal_validity(self): for strategy in BUILTIN_STRATEGIES: validity = strategy["formula"]["meta"].get("signal_validity") self.assertIsInstance(validity, dict, strategy["name"]) self.assertTrue(validity.get("label"), strategy["name"]) def test_frequency_validity_uses_trading_days(self): dates = [f"202607{day:02d}" for day in range(1, 11)] strategy = { "name": "周度策略", "formula": {"meta": {"library": "curated", "frequency": "每周"}}, } run = _result(dates[0], "curated", "周度策略", frequency="每周") active_day_five, _ = build_candidate_archive( [run], [strategy], dates[:5], dates[4], "repair" ) active_day_six, history = build_candidate_archive( [run], [strategy], dates[:6], dates[5], "repair" ) self.assertEqual(len(active_day_five), 1) self.assertEqual(active_day_six, []) self.assertEqual(history[0]["status"], "已到期") self.assertEqual(signal_validity("curated", strategy["formula"])["days"], 5) def test_stage_signal_expires_when_regime_changes(self): run = _result("20260803", "smart", "修复策略", regime="repair") active, history = build_candidate_archive( [run], [], ["20260803", "20260804"], "20260804", "retreat" ) self.assertEqual(active, []) self.assertEqual(history[0]["status"], "已到期") def test_same_stock_from_multiple_strategies_is_aggregated(self): runs = [ _result("20260803", "curated", "策略甲", run_id=1, frequency="每周"), _result("20260803", "curated", "策略乙", run_id=2, frequency="月度"), ] active, history = build_candidate_archive( runs, [], ["20260803"], "20260803", "repair" ) self.assertEqual(len(active), 1) self.assertEqual(len(history), 1) self.assertEqual(set(active[0]["matched_strategies"]), {"策略甲", "策略乙"}) class ScreenerRepositoryPublicationTests(unittest.TestCase): def setUp(self) -> None: self.temp = tempfile.TemporaryDirectory() self.database = ReviewDatabase(Path(self.temp.name) / "review.db") def tearDown(self) -> None: self.temp.cleanup() def test_marker_history_and_run_dates_are_read_in_publication_order(self): self.database.save_data_snapshot( "screener_auto_v1", "20260803", "system", {"status": "complete"} ) self.database.save_data_snapshot( "screener_auto_v1", "20260804", "system", {"status": "running"} ) self.database.save_screener_run( 0, "20260803", "repair", "阶段策略", {"meta": {}}, _result("20260803", "smart", "阶段策略"), "smart", ) markers = self.database.list_screener_batch_markers("20260804") runs = self.database.screener_runs_for_dates(0, ["20260803"]) self.assertEqual([item["trade_date"] for item in markers], ["20260804", "20260803"]) self.assertEqual(runs[0]["meta"]["trade_date"], "20260803") class ScreenerSetupFallbackTests(unittest.TestCase): def test_setup_uses_previous_successful_automatic_batch(self): smart = { "id": 1, "name": "修复策略", "description": "修复阶段策略", "regimes": ["repair"], "builtin": True, "formula": {"meta": {"library": "smart", "frequency": "每日"}}, } curated = { "id": 2, "name": "周度策略", "description": "周度策略", "regimes": ["repair"], "builtin": True, "formula": {"meta": {"library": "curated", "frequency": "每周"}}, } smart_result = _result("20260803", "smart", "修复策略", run_id=11) curated_result = _result( "20260803", "curated", "周度策略", code="", run_id=12, frequency="每周" ) dates = [f"202607{day:02d}" for day in range(1, 31)] + ["20260803"] markers = [ {"trade_date": "20260804", "status": "running"}, { "trade_date": "20260803", "status": "complete", "regime": {"id": "repair"}, "completed": [ {"mode": "smart", "name": "修复策略"}, {"mode": "curated", "name": "周度策略"}, ], "skipped": [], }, ] database = SimpleNamespace( factor_dates=lambda *_: dates, auction_factor_dates=lambda *_: dates, list_screener_strategies=lambda *_: [smart, curated], list_screener_batch_markers=lambda *_: markers, screener_runs_for_date=lambda _user, value: ( [smart_result, curated_result] if value == "20260803" else [] ), recent_screener_runs=lambda *_: [], screener_runs_for_dates=lambda *_: [smart_result, curated_result], ) health = { "market": True, "auction": True, "benchmark": True, "valuation": True, "fundamental": True, "dividend_history": True, "moneyflow_history": True, "earnings_events": True, "popularity": True, "institutions": True, } screener = SimpleNamespace( detect_regime=lambda *_: { "id": "repair", "label": "修复", "confidence": 80, "reason": "", "evidence": [], "history": [], }, factor_health=lambda *_: health, ) service = ScreenerServiceMixin() service.database = database service.screener = screener service.current_user_id = 7 service._request_context = SimpleNamespace(user_id=7) service.llm_configured = False service.llm_primary_model = "" service.llm_fallback_configured = False service.llm_fallback_model = "" payload = service.screener_setup("20260804") self.assertEqual(payload["published_batch"]["trade_date"], "20260803") self.assertTrue(payload["published_batch"]["is_fallback"]) self.assertEqual(payload["automatic_status"]["status"], "running") self.assertEqual(payload["latest_results"]["smart"]["meta"]["run_id"], 11) self.assertEqual(curated["published_run"]["status"], "no_signal") self.assertEqual(len(payload["active_signals"]), 1) if __name__ == "__main__": unittest.main()