from __future__ import annotations import json import tempfile import threading import unittest from datetime import datetime from pathlib import Path from typing import Any from unittest.mock import patch from backend.features.market.backfill_history import ( build_backfill_audit, classify_snapshot_coverage, create_sqlite_backup, select_open_trade_dates, select_open_trade_dates_in_range, ) from backend.features.market.service import MarketServiceMixin from backend.features.sentiment.engine import ( build_sentiment_history, latest_contiguous_history, ) from backend.features.sentiment.service import SentimentServiceMixin from database import ReviewDatabase def _snapshot(trade_date: str, previous_trade_date: str) -> dict[str, Any]: display = f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}" previous_display = ( f"{previous_trade_date[:4]}-{previous_trade_date[4:6]}-{previous_trade_date[6:8]}" if previous_trade_date else "" ) return { "meta": { "trade_date": display, "previous_trade_date": previous_display, "source": "tushare", }, "overview": { "up_count": 2500, "down_count": 2000, "flat_count": 100, "amount_billion": 12000, "limit_up_count": 40, "limit_down_count": 5, "broken_count": 10, "seal_rate": 70, "max_height": 3, "second_board_count": 8, "three_plus_count": 4, "previous_limit_count": 35, "previous_positive_rate": 55, "average_previous_change": 1.2, "median_previous_change": 0.8, "advance_rate": 20, "severe_loss_rate": 5, "previous_down_count": 3, "ladder_completeness": 60, "limit_amount_billion": 300, }, "limits": [{"code": "000001"}], "broken": [], "down_limits": [], "yesterday_limits": [], } class _BackfillHarness(MarketServiceMixin, SentimentServiceMixin): def __init__(self, database: ReviewDatabase) -> None: self.database = database self.sync_lock = threading.Lock() self.configured = True self.token = "test-token" self.current_user_id = 1 self._calendar_rows: list[dict[str, Any]] = [] self._fail_dates: set[str] = set() self.sync_calls: list[str] = [] def _tushare_client(self): # type: ignore[override] harness = self class _Client: def query(self, api_name, params, fields=""): assert api_name == "trade_cal" start = str(params["start_date"]) end = str(params["end_date"]) return [ row for row in harness._calendar_rows if start <= str(row["cal_date"]) <= end ] return _Client() def sync_dashboard(self, trade_date: str) -> dict[str, Any]: # type: ignore[override] compact = trade_date.replace("-", "") self.sync_calls.append(compact) if compact in self._fail_dates: raise ValueError(f"simulated failure for {compact}") previous = "" for row in self._calendar_rows: if str(row["cal_date"]) == compact: previous = str(row.get("pretrade_date") or "") break payload = _snapshot(compact, previous) self.database.save_snapshot(compact, "tushare", payload) return payload def _apply_reason_overrides(self, dashboard: dict[str, Any]) -> dict[str, Any]: return dashboard def _with_storage(self, dashboard: dict[str, Any], cached: bool) -> dict[str, Any]: return dashboard class BackfillHistoryHelperTests(unittest.TestCase): def test_select_open_trade_dates_skips_weekends_and_holidays(self) -> None: rows = [ {"cal_date": "20260821", "is_open": 1, "pretrade_date": "20260820"}, {"cal_date": "20260822", "is_open": 0, "pretrade_date": "20260821"}, # Sat {"cal_date": "20260823", "is_open": 0, "pretrade_date": "20260821"}, # Sun {"cal_date": "20260824", "is_open": 1, "pretrade_date": "20260821"}, {"cal_date": "20260825", "is_open": 1, "pretrade_date": "20260824"}, {"cal_date": "20260826", "is_open": 1, "pretrade_date": "20260825"}, {"cal_date": "20260827", "is_open": 1, "pretrade_date": "20260826"}, ] selected = select_open_trade_dates(rows, "20260827", 4) self.assertEqual(selected, ["20260824", "20260825", "20260826", "20260827"]) def test_range_mode_reports_non_trading_days_separately(self) -> None: rows = [ {"cal_date": "20260821", "is_open": 1}, {"cal_date": "20260824", "is_open": 1}, ] open_dates, skipped = select_open_trade_dates_in_range( rows, "20260821", "20260824" ) self.assertEqual(open_dates, ["20260821", "20260824"]) self.assertEqual(skipped, ["20260822", "20260823"]) def test_classify_snapshot_coverage_finds_real_gaps(self) -> None: coverage = classify_snapshot_coverage( ["20260824", "20260825", "20260826", "20260827"], ["20260824", "20260827"], ) self.assertEqual(coverage["missing"], ["20260825", "20260826"]) self.assertEqual(coverage["present"], ["20260824", "20260827"]) class ContiguousHistoryGapTests(unittest.TestCase): def test_missing_previous_trade_day_collapses_to_today(self) -> None: payloads = [ _snapshot("20260824", "20260821"), _snapshot("20260827", "20260826"), # gap: 20260826 missing ] series = latest_contiguous_history(build_sentiment_history(payloads)) self.assertEqual([row["trade_date"] for row in series], ["20260827"]) def test_continuous_history_keeps_full_tail(self) -> None: payloads = [ _snapshot("20260825", "20260824"), _snapshot("20260826", "20260825"), _snapshot("20260827", "20260826"), ] series = latest_contiguous_history(build_sentiment_history(payloads)) self.assertEqual( [row["trade_date"] for row in series], ["20260825", "20260826", "20260827"], ) class SnapshotBackfillServiceTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.db_path = Path(self.temporary.name) / "review.db" self.database = ReviewDatabase(self.db_path) self.service = _BackfillHarness(self.database) self.service._calendar_rows = [ {"cal_date": "20260820", "is_open": 1, "pretrade_date": "20260819"}, {"cal_date": "20260821", "is_open": 1, "pretrade_date": "20260820"}, {"cal_date": "20260822", "is_open": 0, "pretrade_date": "20260821"}, {"cal_date": "20260823", "is_open": 0, "pretrade_date": "20260821"}, {"cal_date": "20260824", "is_open": 1, "pretrade_date": "20260821"}, {"cal_date": "20260825", "is_open": 1, "pretrade_date": "20260824"}, {"cal_date": "20260826", "is_open": 1, "pretrade_date": "20260825"}, {"cal_date": "20260827", "is_open": 1, "pretrade_date": "20260826"}, ] # Sparse history mimicking .11: keep 0824 and today, miss 0825/0826. self.database.save_snapshot("20260824", "tushare", _snapshot("20260824", "20260821")) self.database.save_snapshot("20260827", "tushare", _snapshot("20260827", "20260826")) def tearDown(self) -> None: self.temporary.cleanup() def test_recent_backfill_fills_gap_and_restores_history(self) -> None: before = self.service.sentiment_history("20260827", 20) self.assertEqual(before["available_days"], 1) with patch( "backend.features.market.service.create_sqlite_backup", return_value=Path(self.temporary.name) / "fake-backup.db", ) as backup: audit = self.service.backfill_recent_trading_days( end_date="20260827", lookback=4, dry_run=False, create_backup=True, ) backup.assert_called_once() self.assertEqual(sorted(self.service.sync_calls), ["20260825", "20260826"]) self.assertEqual(audit["missing"], ["2026-08-25", "2026-08-26"]) self.assertEqual(sorted(audit["created_dates"]), ["2026-08-25", "2026-08-26"]) after = self.service.sentiment_history("20260827", 20) self.assertGreaterEqual(after["available_days"], 4) self.assertEqual( [row["trade_date"] for row in after["rows"]], ["20260824", "20260825", "20260826", "20260827"], ) def test_dry_run_does_not_write_snapshots(self) -> None: audit = self.service.backfill_recent_trading_days( end_date="20260827", lookback=4, dry_run=True, create_backup=True, ) self.assertTrue(audit["dry_run"]) self.assertEqual(self.service.sync_calls, []) self.assertIsNone(audit["backup_path"]) self.assertEqual( self.database.list_snapshot_trade_dates("20260824", "20260827"), ["20260824", "20260827"], ) def test_repeat_execution_skips_existing_days(self) -> None: with patch( "backend.features.market.service.create_sqlite_backup", return_value=Path(self.temporary.name) / "fake-backup.db", ): first = self.service.backfill_recent_trading_days( end_date="20260827", lookback=4 ) self.service.sync_calls.clear() second = self.service.backfill_recent_trading_days( end_date="20260827", lookback=4 ) self.assertEqual(first["succeeded_count"], 2) self.assertEqual(self.service.sync_calls, []) self.assertEqual(second["missing_count"], 0) self.assertEqual(second["skipped_count"], 4) self.assertIsNone(second["backup_path"]) def test_partial_failure_continues_remaining_days(self) -> None: self.service._fail_dates.add("20260825") with patch( "backend.features.market.service.create_sqlite_backup", return_value=Path(self.temporary.name) / "fake-backup.db", ): audit = self.service.backfill_recent_trading_days( end_date="20260827", lookback=4 ) self.assertFalse(audit["ok"]) self.assertEqual(audit["failed_count"], 1) self.assertEqual(audit["succeeded_count"], 1) self.assertIn("20260826", self.database.list_snapshot_trade_dates()) self.assertNotIn("20260825", self.database.list_snapshot_trade_dates()) def test_range_backfill_skips_weekend_without_treating_as_error(self) -> None: with patch( "backend.features.market.service.create_sqlite_backup", return_value=Path(self.temporary.name) / "fake-backup.db", ): audit = self.service.backfill( start_date="2026-08-21", end_date="2026-08-24", ) self.assertEqual(audit["mode"], "range") self.assertEqual(audit["skipped_non_trading_days"], ["2026-08-22", "2026-08-23"]) self.assertEqual(sorted(self.service.sync_calls), ["20260821"]) self.assertTrue(audit["ok"]) def test_sqlite_backup_api_creates_restorable_copy(self) -> None: backup_dir = Path(self.temporary.name) / "backups" backup = create_sqlite_backup( self.db_path, backup_dir, label="pre-recent-backfill", stamped_at=datetime(2026, 8, 27, 15, 30, 0), ) self.assertTrue(backup.exists()) self.assertIn("pre-recent-backfill-20260827-153000", backup.name) restored = ReviewDatabase(backup) self.assertEqual( restored.list_snapshot_trade_dates(), ["20260824", "20260827"], ) def test_audit_lists_only_snapshot_related_write_tables(self) -> None: audit = build_backfill_audit( mode="recent", end_date="20260827", lookback=60, coverage={"trade_dates": [], "present": [], "missing": [], "present_count": 0, "missing_count": 0}, ) self.assertEqual( audit["write_tables"], ["dashboard_snapshots", "data_snapshots", "sync_runs"], ) self.assertNotIn("users", audit["write_tables"]) self.assertNotIn("system_settings", audit["write_tables"]) if __name__ == "__main__": unittest.main()