fix(HEL-190): 按真实交易日历补齐最近60日快照，修复断档后只显示当天
保留连续性过滤，新增可审计补档工具与备份步骤；周末/节假日与真缺档分开处理，支持重复执行与部分失败续跑。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
施工员
co-authored by
Cursor
multica-agent
Cursor
multica-agent
parent
8a5e78f022
commit
7ad445bc9f
@@ -3,9 +3,11 @@ from __future__ import annotations
|
||||
import copy
|
||||
import re
|
||||
from datetime import date, datetime, time as dt_time, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from backend.bootstrap.config import (
|
||||
DATA_DIR,
|
||||
normalize_date,
|
||||
tushare_code,
|
||||
validate_stock_code,
|
||||
@@ -13,6 +15,17 @@ from backend.bootstrap.config import (
|
||||
)
|
||||
from backend.data.providers.ifind_client import IfindError
|
||||
from backend.data.providers.tushare_client import TushareClient, TushareError
|
||||
from backend.features.market.backfill_history import (
|
||||
DEFAULT_RECENT_TRADING_DAYS,
|
||||
MAX_RANGE_TRADING_DAYS,
|
||||
build_backfill_audit,
|
||||
calendar_window_start,
|
||||
classify_snapshot_coverage,
|
||||
create_sqlite_backup,
|
||||
display_date,
|
||||
select_open_trade_dates,
|
||||
select_open_trade_dates_in_range,
|
||||
)
|
||||
from backend.features.market.charts import ChartDataError
|
||||
from backend.features.market.insights import MarketInsightsService
|
||||
from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION
|
||||
@@ -890,31 +903,226 @@ class MarketServiceMixin:
|
||||
"intraday": intraday_points,
|
||||
}
|
||||
|
||||
def backfill(self, start_date: str, end_date: str) -> list[dict[str, Any]]:
|
||||
start = datetime.strptime(normalize_date(start_date), "%Y%m%d").date()
|
||||
end = datetime.strptime(normalize_date(end_date), "%Y%m%d").date()
|
||||
if start > end:
|
||||
raise ValueError("开始日期不能晚于结束日期。")
|
||||
weekdays = []
|
||||
current = start
|
||||
while current <= end:
|
||||
if current.weekday() < 5:
|
||||
weekdays.append(current)
|
||||
current += timedelta(days=1)
|
||||
if len(weekdays) > 15:
|
||||
raise ValueError("单次最多回补 15 个工作日。")
|
||||
results = []
|
||||
for day in weekdays:
|
||||
dashboard = self.sync_dashboard(day.strftime("%Y%m%d"))
|
||||
def backfill(
|
||||
self,
|
||||
start_date: str = "",
|
||||
end_date: str = "",
|
||||
*,
|
||||
lookback: int | None = None,
|
||||
dry_run: bool = False,
|
||||
force: bool = False,
|
||||
create_backup: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Backfill dashboard snapshots for real trading days only.
|
||||
|
||||
- Date-range mode keeps the admin UI contract (max 15 open sessions).
|
||||
- Recent mode fills the last N open sessions (default/max 60).
|
||||
Weekends and holidays are reported as skipped non-trading days, not errors.
|
||||
"""
|
||||
if not self.configured:
|
||||
raise ValueError("公共行情尚未配置，无法回补历史快照。")
|
||||
normalized_end = normalize_date(end_date or date.today().isoformat())
|
||||
if lookback is not None or not (start_date and end_date):
|
||||
target_lookback = (
|
||||
DEFAULT_RECENT_TRADING_DAYS if lookback is None else int(lookback)
|
||||
)
|
||||
return self.backfill_recent_trading_days(
|
||||
end_date=normalized_end,
|
||||
lookback=target_lookback,
|
||||
dry_run=dry_run,
|
||||
force=force,
|
||||
create_backup=create_backup,
|
||||
)
|
||||
return self._backfill_date_range(
|
||||
start_date=normalize_date(start_date),
|
||||
end_date=normalized_end,
|
||||
dry_run=dry_run,
|
||||
force=force,
|
||||
create_backup=create_backup,
|
||||
)
|
||||
|
||||
def backfill_recent_trading_days(
|
||||
self,
|
||||
end_date: str = "",
|
||||
lookback: int = DEFAULT_RECENT_TRADING_DAYS,
|
||||
*,
|
||||
dry_run: bool = False,
|
||||
force: bool = False,
|
||||
create_backup: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
normalized_end = normalize_date(end_date or date.today().isoformat())
|
||||
trade_dates = self._load_recent_open_trade_dates(normalized_end, lookback)
|
||||
existing = self.database.list_snapshot_trade_dates(
|
||||
trade_dates[0], trade_dates[-1]
|
||||
)
|
||||
coverage = classify_snapshot_coverage(trade_dates, existing)
|
||||
return self._execute_snapshot_backfill(
|
||||
mode="recent",
|
||||
end_date=normalized_end,
|
||||
lookback=lookback,
|
||||
coverage=coverage,
|
||||
skipped_non_trading_days=[],
|
||||
dry_run=dry_run,
|
||||
force=force,
|
||||
create_backup=create_backup,
|
||||
)
|
||||
|
||||
def _backfill_date_range(
|
||||
self,
|
||||
start_date: str,
|
||||
end_date: str,
|
||||
*,
|
||||
dry_run: bool = False,
|
||||
force: bool = False,
|
||||
create_backup: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
window_start = calendar_window_start(end_date, MAX_RANGE_TRADING_DAYS)
|
||||
calendar_rows = self._tushare_client().query(
|
||||
"trade_cal",
|
||||
{
|
||||
"exchange": "SSE",
|
||||
"start_date": min(window_start, start_date),
|
||||
"end_date": end_date,
|
||||
},
|
||||
"cal_date,is_open,pretrade_date",
|
||||
)
|
||||
trade_dates, skipped = select_open_trade_dates_in_range(
|
||||
calendar_rows,
|
||||
start_date,
|
||||
end_date,
|
||||
maximum=MAX_RANGE_TRADING_DAYS,
|
||||
)
|
||||
if not trade_dates:
|
||||
raise ValueError("选定区间内没有交易日，周末或节假日无需回补。")
|
||||
existing = self.database.list_snapshot_trade_dates(trade_dates[0], trade_dates[-1])
|
||||
coverage = classify_snapshot_coverage(trade_dates, existing)
|
||||
return self._execute_snapshot_backfill(
|
||||
mode="range",
|
||||
end_date=end_date,
|
||||
lookback=None,
|
||||
coverage=coverage,
|
||||
skipped_non_trading_days=skipped,
|
||||
dry_run=dry_run,
|
||||
force=force,
|
||||
create_backup=create_backup,
|
||||
)
|
||||
|
||||
def _load_recent_open_trade_dates(self, end_date: str, lookback: int) -> list[str]:
|
||||
start_date = calendar_window_start(end_date, lookback)
|
||||
calendar_rows = self._tushare_client().query(
|
||||
"trade_cal",
|
||||
{
|
||||
"exchange": "SSE",
|
||||
"start_date": start_date,
|
||||
"end_date": end_date,
|
||||
},
|
||||
"cal_date,is_open,pretrade_date",
|
||||
)
|
||||
return select_open_trade_dates(calendar_rows, end_date, lookback)
|
||||
|
||||
def _execute_snapshot_backfill(
|
||||
self,
|
||||
*,
|
||||
mode: str,
|
||||
end_date: str,
|
||||
lookback: int | None,
|
||||
coverage: dict[str, Any],
|
||||
skipped_non_trading_days: list[str],
|
||||
dry_run: bool,
|
||||
force: bool,
|
||||
create_backup: bool,
|
||||
) -> dict[str, Any]:
|
||||
targets = list(coverage["trade_dates"] if force else coverage["missing"])
|
||||
backup_path: str | None = None
|
||||
if create_backup and not dry_run and targets:
|
||||
backup = create_sqlite_backup(
|
||||
Path(self.database.path),
|
||||
DATA_DIR / "backups",
|
||||
label=f"pre-{mode}-backfill",
|
||||
)
|
||||
backup_path = str(backup)
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
if dry_run:
|
||||
for trade_date in coverage["trade_dates"]:
|
||||
exists = trade_date in coverage["present"]
|
||||
if exists and not force:
|
||||
status = "skipped"
|
||||
action = "exists"
|
||||
else:
|
||||
status = "planned"
|
||||
action = "refresh" if exists else "create"
|
||||
results.append(
|
||||
{
|
||||
"requested_date": display_date(trade_date),
|
||||
"trade_date": display_date(trade_date),
|
||||
"status": status,
|
||||
"action": action,
|
||||
}
|
||||
)
|
||||
return build_backfill_audit(
|
||||
mode=mode,
|
||||
end_date=end_date,
|
||||
lookback=lookback,
|
||||
coverage=coverage,
|
||||
skipped_non_trading_days=skipped_non_trading_days,
|
||||
backup_path=backup_path,
|
||||
dry_run=True,
|
||||
results=results,
|
||||
)
|
||||
|
||||
present_before = set(coverage["present"])
|
||||
for trade_date in targets:
|
||||
existed = trade_date in present_before
|
||||
try:
|
||||
dashboard = self.sync_dashboard(trade_date)
|
||||
actual = normalize_date(
|
||||
str(dashboard.get("meta", {}).get("trade_date") or trade_date)
|
||||
)
|
||||
results.append(
|
||||
{
|
||||
"requested_date": display_date(trade_date),
|
||||
"trade_date": display_date(actual),
|
||||
"status": "success",
|
||||
"action": "refreshed" if existed else "created",
|
||||
"source": dashboard.get("meta", {}).get("source"),
|
||||
"records": self._record_count(dashboard),
|
||||
}
|
||||
)
|
||||
except Exception as exc:
|
||||
results.append(
|
||||
{
|
||||
"requested_date": display_date(trade_date),
|
||||
"trade_date": display_date(trade_date),
|
||||
"status": "failed",
|
||||
"action": "refresh" if existed else "create",
|
||||
"error": str(exc),
|
||||
}
|
||||
)
|
||||
|
||||
for trade_date in coverage["present"]:
|
||||
if force:
|
||||
continue
|
||||
results.append(
|
||||
{
|
||||
"requested_date": day.isoformat(),
|
||||
"trade_date": dashboard["meta"]["trade_date"],
|
||||
"source": dashboard["meta"]["source"],
|
||||
"records": self._record_count(dashboard),
|
||||
"requested_date": display_date(trade_date),
|
||||
"trade_date": display_date(trade_date),
|
||||
"status": "skipped",
|
||||
"action": "exists",
|
||||
}
|
||||
)
|
||||
return results
|
||||
|
||||
results.sort(key=lambda row: str(row.get("requested_date") or ""))
|
||||
return build_backfill_audit(
|
||||
mode=mode,
|
||||
end_date=end_date,
|
||||
lookback=lookback,
|
||||
coverage=coverage,
|
||||
skipped_non_trading_days=skipped_non_trading_days,
|
||||
backup_path=backup_path,
|
||||
dry_run=False,
|
||||
results=results,
|
||||
)
|
||||
|
||||
def _stock_identity(self, code: str, trade_date: str) -> tuple[str, str]:
|
||||
snapshot = self.database.get_snapshot(trade_date) or {}
|
||||
|
||||
Reference in New Issue
Block a user