修复数据采集和清理的bug，新增持仓排名功能，日线数据分页

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

ft-app/app/collector.py

@@ -9,6 +9,7 @@ def fetch_contract_bars(contract_code: str, start_date: str | None = None) -> li
     """Fetch daily OHLCV for a single contract from akshare.
 
     Returns list of {date, open, close, high, low} dicts.
+    Only returns bars on or after start_date when provided.
     """
     import akshare as ak
 
@@ -22,6 +23,8 @@ def fetch_contract_bars(contract_code: str, start_date: str | None = None) -> li
         print(f"[collector] No data returned for {contract_code}")
         return []
 
+    filter_date = date.fromisoformat(start_date) if start_date else None
+
     bars = []
     for _, row in df.iterrows():
         try:
@@ -31,6 +34,8 @@ def fetch_contract_bars(contract_code: str, start_date: str | None = None) -> li
             else:
                 d = date.fromisoformat(str(val)[:10])
 
+            if filter_date and d < filter_date:
+                continue
 
             bars.append({
                 "date": d,
@@ -93,6 +98,7 @@ def _sync_one(db, contract_code: str) -> int:
         return 0
 
     inserted = 0
+    min_date = None
     for bar in bars:
         existing = (
             db.query(DailyBar)
@@ -112,22 +118,130 @@ def _sync_one(db, contract_code: str) -> int:
                 low=bar["low"],
             ))
             inserted += 1
+            if min_date is None or bar["date"] < min_date:
+                min_date = bar["date"]
 
     if inserted > 0:
         db.flush()
-        _recompute_amp(db, contract_code)
+        _recompute_amp(db, contract_code, from_date=min_date)
 
     return inserted
 
 
-def _recompute_amp(db, contract_code: str):
-    """Recompute amp_5d for all bars of a contract."""
+def fetch_position_rankings(contract_code: str, trade_date: str) -> list[dict]:
+    """Fetch top-20 position rankings for a contract on a given date from akshare.
+
+    Calls the API 3 times (volume, long, short) and returns a unified list of dicts:
+    {data_type, rank, institution, value, change}
+    """
+    import akshare as ak
+
+    results = []
+    for sym, dtype in [("成交量", "volume"), ("多单持仓", "long"), ("空单持仓", "short")]:
+        try:
+            df = ak.futures_hold_pos_sina(symbol=sym, contract=contract_code.upper(), date=trade_date)
+        except Exception as e:
+            print(f"[collector] position akshare error for {contract_code} {dtype}: {e}")
+            continue
+
+        if df is None or df.empty:
+            continue
+
+        for _, row in df.iterrows():
+            try:
+                results.append({
+                    "data_type": dtype,
+                    "rank": int(row["名次"]),
+                    "institution": str(row["会员简称"]),
+                    "value": int(row.iloc[2]),
+                    "change": int(row["比上交易增减"]),
+                })
+            except (KeyError, ValueError, TypeError) as e:
+                print(f"[collector] position skip row: {e}")
+                continue
+
+    return results
+
+
+def sync_position_rankings() -> dict:
+    """Sync position rankings for all active contracts. Returns {contract_code: new_rows}."""
+    db = SessionLocal()
+    results = {}
+
+    try:
+        active_contracts = (
+            db.query(Contract).filter(Contract.is_active == True).all()
+        )
+
+        for c in active_contracts:
+            count = _sync_positions_for_contract(db, c.code)
+            results[c.code] = count
+
+        db.commit()
+    finally:
+        db.close()
+
+    return results
+
+
+def _sync_positions_for_contract(db, contract_code: str) -> int:
+    """Sync position rankings for all dates that have bars but no position data."""
+    from app.models import PositionRanking
+
+    code = contract_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()
+    ]
+
+    inserted = 0
+    for d in bar_dates:
+        if d in existing_dates:
+            continue
+        date_str = d.strftime("%Y%m%d")
+        rankings = fetch_position_rankings(code, date_str)
+        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()
+
+    return inserted
+
+
+def _recompute_amp(db, contract_code: str, from_date: date | None = None):
+    """Recompute amp_5d for bars of a contract from from_date onwards."""
     bars = (
         db.query(DailyBar)
         .filter(DailyBar.contract == contract_code)
         .order_by(DailyBar.date)
         .all()
     )
+    start_idx = 0
+    if from_date:
+        for i, bar in enumerate(bars):
+            if bar.date >= from_date:
+                start_idx = i
+                break
     for i, bar in enumerate(bars):
-        if i >= 5:
+        if i >= 5 and i >= start_idx:
             bar.amp_5d = compute_amp_5d([b.diff for b in bars[i - 5 : i]])