8e94c7b429
使 python3 tools/backfill_recent_snapshots.py --help 在干净环境下可直接运行，并同步文档运行示例为容器内执行。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Auditable recent trading-day dashboard snapshot backfill.
|
|
|
|
Examples:
|
|
|
|
python tools/backfill_recent_snapshots.py --account admin --dry-run
|
|
python tools/backfill_recent_snapshots.py --account admin --lookback 60
|
|
python tools/backfill_recent_snapshots.py --account admin --end-date 2026-08-27 --force
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
if str(ROOT) not in sys.path:
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from backend.application import SERVICE
|
|
from backend.bootstrap.config import normalize_date
|
|
from backend.features.market.backfill_history import DEFAULT_RECENT_TRADING_DAYS
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Backfill the latest N real trading-day dashboard snapshots"
|
|
)
|
|
parser.add_argument(
|
|
"--account",
|
|
required=True,
|
|
help="Account that can resolve the shared Tushare token",
|
|
)
|
|
parser.add_argument(
|
|
"--end-date",
|
|
default=date.today().isoformat(),
|
|
help="Inclusive end date YYYY-MM-DD (default: today)",
|
|
)
|
|
parser.add_argument(
|
|
"--lookback",
|
|
type=int,
|
|
default=DEFAULT_RECENT_TRADING_DAYS,
|
|
help=f"Number of open trading days to cover (default {DEFAULT_RECENT_TRADING_DAYS}, max 60)",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Plan only: classify missing gaps without writing",
|
|
)
|
|
parser.add_argument(
|
|
"--force",
|
|
action="store_true",
|
|
help="Re-sync days that already have snapshots",
|
|
)
|
|
parser.add_argument(
|
|
"--no-backup",
|
|
action="store_true",
|
|
help="Skip the SQLite backup API step (not recommended)",
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
action="store_true",
|
|
help="Print the full audit payload as JSON",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
user = SERVICE.database.user_by_username(args.account.strip())
|
|
if not user:
|
|
raise SystemExit("account not found")
|
|
SERVICE.bind_user(int(user["id"]))
|
|
|
|
end_date = normalize_date(args.end_date)
|
|
audit = SERVICE.backfill_recent_trading_days(
|
|
end_date=end_date,
|
|
lookback=args.lookback,
|
|
dry_run=args.dry_run,
|
|
force=args.force,
|
|
create_backup=not args.no_backup,
|
|
)
|
|
|
|
if args.json:
|
|
print(json.dumps(audit, ensure_ascii=False, indent=2))
|
|
raise SystemExit(0 if audit.get("ok") else 1)
|
|
|
|
print(
|
|
f"mode={audit['mode']} end={audit['end_date']} lookback={audit['lookback']} "
|
|
f"dry_run={audit['dry_run']}"
|
|
)
|
|
print(
|
|
f"present={audit['present_count']} missing={audit['missing_count']} "
|
|
f"succeeded={audit['succeeded_count']} skipped={audit['skipped_count']} "
|
|
f"failed={audit['failed_count']}"
|
|
)
|
|
if audit.get("backup_path"):
|
|
print(f"backup={audit['backup_path']}")
|
|
if audit.get("missing"):
|
|
print("missing_dates=" + ",".join(audit["missing"]))
|
|
if audit.get("created_dates"):
|
|
print("created_dates=" + ",".join(audit["created_dates"]))
|
|
failed = [row for row in audit.get("results") or [] if row.get("status") == "failed"]
|
|
for row in failed:
|
|
print(f"failed {row.get('requested_date')}: {row.get('error')}")
|
|
if not audit.get("ok"):
|
|
raise SystemExit(1)
|
|
print("backfill complete")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|