7ad445bc9f
保留连续性过滤，新增可审计补档工具与备份步骤；周末/节假日与真缺档分开处理，支持重复执行与部分失败续跑。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
318 lines
12 KiB
Python
318 lines
12 KiB
Python
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()
|