238 lines
9.3 KiB
Python
238 lines
9.3 KiB
Python
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()
|