持仓同步改为后台执行，添加进度条实时反馈

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-07-25 00:10:17 +08:00
parent cd7f6668af
commit b74dccd805
3 changed files with 199 additions and 12 deletions
@@ -1,9 +1,124 @@
 """Data collector — fetch OHLCV from akshare and upsert into daily_bars."""
+import threading
 from datetime import date
 from app.database import SessionLocal
 from app.models import DailyBar, Contract
 from app.engine.lock_strategy import compute_amp_5d

+_sync_progress = {
+    "running": False,
+    "contract": "",
+    "contract_done": 0,
+    "contract_total": 0,
+    "date_done": 0,
+    "date_total": 0,
+    "rows": 0,
+    "finished": False,
+}
+_lock = threading.Lock()
+
+
+def get_sync_progress() -> dict:
+    with _lock:
+        return dict(_sync_progress)
+
+
+def start_sync_positions_background():
+    """Start sync_position_rankings in a background thread. Returns immediately."""
+    with _lock:
+        if _sync_progress["running"]:
+            return False
+        _sync_progress.update({
+            "running": True,
+            "finished": False,
+            "contract": "",
+            "contract_done": 0,
+            "contract_total": 0,
+            "date_done": 0,
+            "date_total": 0,
+            "rows": 0,
+        })
+
+    def _run():
+        try:
+            _do_sync_with_progress()
+        finally:
+            with _lock:
+                _sync_progress["running"] = False
+                _sync_progress["finished"] = True
+
+    threading.Thread(target=_run, daemon=True).start()
+    return True
+
+
+def _do_sync_with_progress():
+    from app.models import PositionRanking
+
+    db = SessionLocal()
+    try:
+        active_contracts = (
+            db.query(Contract).filter(Contract.is_active == True).all()
+        )
+        with _lock:
+            _sync_progress["contract_total"] = len(active_contracts)
+
+        for idx, c in enumerate(active_contracts):
+            with _lock:
+                _sync_progress.update({
+                    "contract": c.code,
+                    "contract_done": idx,
+                    "date_done": 0,
+                    "date_total": 0,
+                })
+
+            code = c.code.upper()
+
+            existing_dates = {
+                r[0] for r in
+                db.query(PositionRanking.date)
+                .filter(PositionRanking.contract_code == code)
+                .distinct()
+                .all()
+            }
+
+            bar_dates = [
+                r[0] for r in
+                db.query(DailyBar.date)
+                .filter(DailyBar.contract == code)
+                .order_by(DailyBar.date)
+                .all()
+            ]
+
+            missing = [d for d in bar_dates if d not in existing_dates]
+            with _lock:
+                _sync_progress["date_total"] = len(missing)
+
+            for date_idx, d in enumerate(missing):
+                date_str = d.strftime("%Y%m%d")
+                rankings = fetch_position_rankings(code, date_str)
+                inserted = 0
+                for r in rankings:
+                    db.add(PositionRanking(
+                        contract_code=code,
+                        institution=r["institution"],
+                        data_type=r["data_type"],
+                        date=d,
+                        rank=r["rank"],
+                        value=r["value"],
+                        change=r["change"],
+                    ))
+                    inserted += 1
+                db.flush()
+                with _lock:
+                    _sync_progress["date_done"] = date_idx + 1
+                    _sync_progress["rows"] += inserted
+
+        db.commit()
+        with _lock:
+            _sync_progress["contract_done"] = len(active_contracts)
+    finally:
+        db.close()
+

 def fetch_contract_bars(contract_code: str, start_date: str | None = None) -> list[dict]:
    """Fetch daily OHLCV for a single contract from akshare.