381 lines
12 KiB
Python
381 lines
12 KiB
Python
"""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
|
|
|
|
_progress = {"running": False, "label": "", "done": 0, "total": 0, "finished": False}
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def get_progress() -> dict:
|
|
with _lock:
|
|
return dict(_progress)
|
|
|
|
|
|
def _start_bg(target, label):
|
|
with _lock:
|
|
if _progress["running"]:
|
|
return False
|
|
_progress.update(running=True, finished=False, label=label, done=0, total=0)
|
|
|
|
def _run():
|
|
try:
|
|
target()
|
|
finally:
|
|
with _lock:
|
|
_progress["running"] = False
|
|
_progress["finished"] = True
|
|
_progress["done"] = _progress["total"]
|
|
|
|
threading.Thread(target=_run, daemon=True).start()
|
|
return True
|
|
|
|
|
|
def sync_contract_bars_bg(contract_code: str) -> bool:
|
|
"""Sync bars for one contract in background. Returns True if started."""
|
|
code = contract_code.upper()
|
|
|
|
def _run():
|
|
db = SessionLocal()
|
|
try:
|
|
latest = (
|
|
db.query(DailyBar.date)
|
|
.filter(DailyBar.contract == code)
|
|
.order_by(DailyBar.date.desc())
|
|
.first()
|
|
)
|
|
start_date = latest[0].isoformat() if latest else None
|
|
bars = fetch_contract_bars(code, start_date)
|
|
|
|
with _lock:
|
|
_progress["total"] = len(bars)
|
|
|
|
total_bars = len(bars)
|
|
inserted = 0
|
|
min_date = None
|
|
for i, bar in enumerate(bars):
|
|
existing = (
|
|
db.query(DailyBar)
|
|
.filter(DailyBar.contract == code, DailyBar.date == bar["date"])
|
|
.first()
|
|
)
|
|
if not existing:
|
|
db.add(DailyBar(
|
|
contract=code, date=bar["date"],
|
|
open=bar["open"], close=bar["close"],
|
|
high=bar["high"], low=bar["low"],
|
|
))
|
|
inserted += 1
|
|
if min_date is None or bar["date"] < min_date:
|
|
min_date = bar["date"]
|
|
with _lock:
|
|
_progress["done"] = i + 1
|
|
_progress["label"] = f"同步行情 {code} · {i + 1}/{total_bars} 条"
|
|
|
|
if inserted > 0:
|
|
db.flush()
|
|
_recompute_amp(db, code, from_date=min_date)
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
return _start_bg(_run, f"同步行情 {code}")
|
|
|
|
|
|
def sync_positions_bg(contract_code: str) -> bool:
|
|
"""Sync position rankings for one contract in background. Returns True if started."""
|
|
from app.models import PositionRanking
|
|
code = contract_code.upper()
|
|
|
|
def _run():
|
|
db = SessionLocal()
|
|
try:
|
|
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()
|
|
]
|
|
# Only sync dates within or after existing position date range;
|
|
# akshare may not have data for very old dates
|
|
if existing_dates:
|
|
min_pos = min(existing_dates)
|
|
missing = [d for d in bar_dates if d not in existing_dates and d >= min_pos]
|
|
else:
|
|
missing = [d for d in bar_dates if d not in existing_dates]
|
|
total_missing = len(missing)
|
|
|
|
with _lock:
|
|
_progress["total"] = total_missing
|
|
if total_missing == 0:
|
|
_progress["label"] = f"同步持仓 {code} · 已是最新"
|
|
|
|
inserted = 0
|
|
for i, d in enumerate(missing):
|
|
rankings = fetch_position_rankings(code, d.strftime("%Y%m%d"))
|
|
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:
|
|
_progress["done"] = i + 1
|
|
_progress["label"] = f"同步持仓 {code} · {i + 1}/{total_missing} 天"
|
|
finally:
|
|
db.close()
|
|
|
|
return _start_bg(_run, f"同步持仓 {code}")
|
|
|
|
|
|
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.
|
|
Only returns bars on or after start_date when provided.
|
|
"""
|
|
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 []
|
|
|
|
filter_date = date.fromisoformat(start_date) if start_date else None
|
|
|
|
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 filter_date and d < filter_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
|
|
min_date = None
|
|
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 min_date is None or bar["date"] < min_date:
|
|
min_date = bar["date"]
|
|
|
|
if inserted > 0:
|
|
db.flush()
|
|
_recompute_amp(db, contract_code, from_date=min_date)
|
|
|
|
return inserted
|
|
|
|
|
|
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 and i >= start_idx:
|
|
bar.amp_5d = compute_amp_5d([b.diff for b in bars[i - 5 : i]])
|