增加数据同步功能，支持从 akshare 拉取合约日线并自动计算振幅

2026-07-24 21:41:38 +08:00
parent 1ff2aa17e8
commit a53b0fcc4c
3 changed files with 167 additions and 1 deletions
@@ -0,0 +1,135 @@
+"""Data collector — fetch OHLCV from akshare and upsert into daily_bars."""
+from datetime import date
+from app.database import SessionLocal
+from app.models import DailyBar, Contract
+from app.engine.lock_strategy import compute_amp_5d
+
+
+def fetch_contract_bars(contract_code: str, start_date: str | None = None) -> list[dict]:
+    """Fetch daily OHLCV for a single contract from akshare.
+
+    Returns list of {date, open, close, high, low} dicts.
+    """
+    import akshare as ak
+
+    try:
+        df = ak.futures_zh_daily_sina(symbol=contract_code.upper())
+    except Exception as e:
+        print(f"[collector] akshare error for {contract_code}: {e}")
+        return []
+
+    if df is None or df.empty:
+        print(f"[collector] No data returned for {contract_code}")
+        return []
+
+    bars = []
+    for _, row in df.iterrows():
+        try:
+            val = row["date"]
+            if hasattr(val, "strftime"):
+                d = val.date() if hasattr(val, "date") else val
+            else:
+                d = date.fromisoformat(str(val)[:10])
+
+            if start_date and d <= date.fromisoformat(start_date):
+                continue
+
+            bars.append({
+                "date": d,
+                "open": float(row["open"]),
+                "close": float(row["close"]),
+                "high": float(row["high"]),
+                "low": float(row["low"]),
+            })
+        except (KeyError, ValueError, TypeError) as e:
+            print(f"[collector] skip row: {e}")
+            continue
+
+    return bars
+
+
+def sync_active_contracts() -> dict:
+    """Sync all active contracts. Returns {contract_code: new_bars_count}."""
+    db = SessionLocal()
+    results = {}
+
+    try:
+        active_contracts = (
+            db.query(Contract).filter(Contract.is_active == True).all()
+        )
+
+        for c in active_contracts:
+            count = _sync_one(db, c.code)
+            results[c.code] = count
+
+        db.commit()
+    finally:
+        db.close()
+
+    return results
+
+
+def sync_one_contract(contract_code: str) -> int:
+    """Sync a single contract. Returns number of new bars inserted."""
+    db = SessionLocal()
+    try:
+        count = _sync_one(db, contract_code.upper())
+        db.commit()
+        return count
+    finally:
+        db.close()
+
+
+def _sync_one(db, contract_code: str) -> int:
+    """Internal: sync one contract using existing session."""
+    latest = (
+        db.query(DailyBar.date)
+        .filter(DailyBar.contract == contract_code)
+        .order_by(DailyBar.date.desc())
+        .first()
+    )
+    start_date = latest[0].isoformat() if latest else None
+
+    bars = fetch_contract_bars(contract_code, start_date)
+    if not bars:
+        return 0
+
+    inserted = 0
+    for bar in bars:
+        existing = (
+            db.query(DailyBar)
+            .filter(
+                DailyBar.contract == contract_code,
+                DailyBar.date == bar["date"],
+            )
+            .first()
+        )
+        if not existing:
+            db.add(DailyBar(
+                contract=contract_code,
+                date=bar["date"],
+                open=bar["open"],
+                close=bar["close"],
+                high=bar["high"],
+                low=bar["low"],
+            ))
+            inserted += 1
+
+    if inserted > 0:
+        db.flush()
+        _recompute_amp(db, contract_code)
+
+    return inserted
+
+
+def _recompute_amp(db, contract_code: str):
+    """Recompute amp_5d for all bars of a contract."""
+    bars = (
+        db.query(DailyBar)
+        .filter(DailyBar.contract == contract_code)
+        .order_by(DailyBar.date)
+        .all()
+    )
+    for i, bar in enumerate(bars):
+        if i >= 5:
+            bar.amp_5d = compute_amp_5d([b.diff for b in bars[i - 5 : i]])