diff --git a/ft-app/app/collector.py b/ft-app/app/collector.py index 24dd932..235676c 100644 --- a/ft-app/app/collector.py +++ b/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]]) diff --git a/ft-app/app/models.py b/ft-app/app/models.py index cf292a8..bc0802d 100644 --- a/ft-app/app/models.py +++ b/ft-app/app/models.py @@ -51,18 +51,18 @@ class DailyBar(Base): return self.high - self.low -class PositionSnapshot(Base): - __tablename__ = "position_snapshots" - __table_args__ = (UniqueConstraint("contract_code", "institution", "direction", "date"),) +class PositionRanking(Base): + __tablename__ = "position_rankings" + __table_args__ = (UniqueConstraint("contract_code", "institution", "data_type", "date"),) id: Mapped[int] = mapped_column(primary_key=True) - contract_code: Mapped[str] = mapped_column(String(10), index=True, default="FG") - institution: Mapped[str] = mapped_column(String(20), index=True) - direction: Mapped[str] = mapped_column(String(10)) + contract_code: Mapped[str] = mapped_column(String(10), index=True) + institution: Mapped[str] = mapped_column(String(30), index=True) + data_type: Mapped[str] = mapped_column(String(10), index=True) date: Mapped[date] = mapped_column(Date, index=True) - position: Mapped[int] = mapped_column(Integer) - delta: Mapped[int] = mapped_column(Integer, default=0) - avg_cost: Mapped[float] = mapped_column(Float) + rank: Mapped[int] = mapped_column(Integer) + value: Mapped[int] = mapped_column(Integer) + change: Mapped[int] = mapped_column(Integer) class User(Base): diff --git a/ft-app/app/routers/admin.py b/ft-app/app/routers/admin.py index 2f61c3f..d705e74 100644 --- a/ft-app/app/routers/admin.py +++ b/ft-app/app/routers/admin.py @@ -2,8 +2,8 @@ from fastapi import APIRouter, Depends, Form, Request from fastapi.responses import HTMLResponse, RedirectResponse from sqlalchemy.orm import Session from app.database import get_db -from app.models import Product, Contract, DailyBar, PositionSnapshot -from app.collector import sync_active_contracts, sync_one_contract +from app.models import Product, Contract, DailyBar, PositionRanking +from app.collector import sync_active_contracts, sync_one_contract, sync_position_rankings router = APIRouter(prefix="/admin", tags=["admin"]) @@ -113,7 +113,7 @@ def delete_contract(contract_id: int, db: Session = Depends(get_db)): c = db.query(Contract).filter(Contract.id == contract_id).first() if c: db.query(DailyBar).filter(DailyBar.contract == c.code).delete() - db.query(PositionSnapshot).filter(PositionSnapshot.contract_code == c.code).delete() + db.query(PositionRanking).filter(PositionRanking.contract_code == c.code).delete() db.delete(c) db.commit() return RedirectResponse("/admin/?tab=contract", status_code=303) @@ -125,6 +125,7 @@ def delete_product(product_id: int, db: Session = Depends(get_db)): if p: for c in p.contracts: db.query(DailyBar).filter(DailyBar.contract == c.code).delete() + db.query(PositionRanking).filter(PositionRanking.contract_code == c.code).delete() db.delete(p) db.commit() return RedirectResponse("/admin/?tab=product", status_code=303) @@ -144,6 +145,14 @@ def sync_single(contract_code: str): return RedirectResponse(f"/admin/?tab=sync&synced={count}", status_code=303) +@router.post("/sync-positions") +def sync_positions(request: Request): + results = sync_position_rankings() + total = sum(results.values()) + print(f"[sync] Position rankings: {total} rows across {len(results)} contracts") + return RedirectResponse(f"/admin/?tab=sync&pos_synced={total}", status_code=303) + + @router.post("/sync/product/{product_id}") def sync_product(product_id: int, request: Request, db: Session = Depends(get_db)): contracts = db.query(Contract).filter( diff --git a/ft-app/app/routers/auth.py b/ft-app/app/routers/auth.py index bc52f36..65fe984 100644 --- a/ft-app/app/routers/auth.py +++ b/ft-app/app/routers/auth.py @@ -9,16 +9,6 @@ router = APIRouter(prefix="/auth", tags=["auth"]) SESSION_COOKIE = "ft_session" -def get_current_user(request: Request, db: Session = Depends(get_db)) -> User | None: - user_id = request.cookies.get(SESSION_COOKIE) - if not user_id: - return None - try: - return db.query(User).filter(User.id == int(user_id)).first() - except (ValueError, TypeError): - return None - - @router.get("/login", response_class=HTMLResponse) def login_page(request: Request): template = request.app.state.templates.get_template("login.html") diff --git a/ft-app/app/routers/contracts.py b/ft-app/app/routers/contracts.py index d000023..9d5fc51 100644 --- a/ft-app/app/routers/contracts.py +++ b/ft-app/app/routers/contracts.py @@ -1,13 +1,15 @@ +import math from datetime import date, timedelta -from fastapi import APIRouter, Depends, Request +from fastapi import APIRouter, Depends, Request, Query from fastapi.responses import HTMLResponse from sqlalchemy.orm import Session from app.database import get_db -from app.models import DailyBar, Contract +from app.models import DailyBar, Contract, PositionRanking router = APIRouter(prefix="/contracts", tags=["contracts"]) WEEKDAY_ZH = {0: "周一", 1: "周二", 2: "周三", 3: "周四", 4: "周五", 5: "周六", 6: "周日"} +PAGE_SIZE = 7 def get_active_contracts(db: Session) -> list[str]: @@ -47,7 +49,13 @@ def contract_index(request: Request, db: Session = Depends(get_db)): @router.get("/{contract}", response_class=HTMLResponse) -def contract_detail(request: Request, contract: str, db: Session = Depends(get_db)): +def contract_detail( + request: Request, + contract: str, + pos_date: str | None = None, + page: int = Query(1, ge=1), + db: Session = Depends(get_db), +): active_contracts = get_active_contracts(db) bars = ( db.query(DailyBar) @@ -56,9 +64,17 @@ def contract_detail(request: Request, contract: str, db: Session = Depends(get_d .all() ) + total = len(bars) + total_pages = max(1, math.ceil(total / PAGE_SIZE)) + page = min(page, total_pages) + start = (page - 1) * PAGE_SIZE + page_bars = bars[start:start + PAGE_SIZE] + rows = [] - for bar in bars: + for i, bar in enumerate(page_bars): + global_idx = start + i rows.append({ + "global_idx": global_idx, "date": bar.date.strftime("%Y/%-m/%-d"), "weekday": WEEKDAY_ZH.get(bar.date.weekday(), ""), "open": int(bar.open) if bar.open else "-", @@ -72,6 +88,45 @@ def contract_detail(request: Request, contract: str, db: Session = Depends(get_d latest = bars[0] if bars else None + # Dates that have position data (for date picker) + pos_dates = [ + r[0] for r in + db.query(PositionRanking.date) + .filter(PositionRanking.contract_code == contract.upper()) + .distinct() + .order_by(PositionRanking.date.desc()) + .all() + ] + + # Determine which date to show position rankings for + if pos_date: + try: + selected_date = date.fromisoformat(pos_date) + except ValueError: + selected_date = pos_dates[0] if pos_dates else None + else: + selected_date = pos_dates[0] if pos_dates else None + + # Position rankings for the selected date + pos_data = {"volume": [], "long": [], "short": []} + if selected_date: + rankings = ( + db.query(PositionRanking) + .filter( + PositionRanking.contract_code == contract.upper(), + PositionRanking.date == selected_date, + ) + .order_by(PositionRanking.data_type, PositionRanking.rank) + .all() + ) + for r in rankings: + pos_data[r.data_type].append({ + "rank": r.rank, + "institution": r.institution, + "value": r.value, + "change": r.change, + }) + # Predict next trading day amplitude: mean of latest 5 diffs # Compute next trading date next_date = latest.date + timedelta(days=1) if latest else None @@ -91,9 +146,17 @@ def contract_detail(request: Request, contract: str, db: Session = Depends(get_d contracts=active_contracts, rows=rows, latest=latest, + pos_dates=pos_dates, + selected_pos_date=selected_date, + pos_data=pos_data, next_amp=next_amp, next_date=next_date.strftime("%Y/%-m/%-d") if next_date else None, next_weekday=WEEKDAY_ZH.get(next_date.weekday(), "") if next_date else "", row_count=len(rows), + total_rows=total, + page=page, + total_pages=total_pages, + pos_date=pos_date or "", + WEEKDAY_ZH=WEEKDAY_ZH, ) ) diff --git a/ft-app/app/templates/admin.html b/ft-app/app/templates/admin.html index 8068cac..7baee2c 100644 --- a/ft-app/app/templates/admin.html +++ b/ft-app/app/templates/admin.html @@ -32,10 +32,16 @@
振幅 = 近 5 日 (最高−最低) 均值取整 · 点击振幅值查看计算过程
- {% if rows|length > 7 %} - - {% endif %} + 共 {{ total_rows }} 条| 日期 | 星期 | 开盘 | 收盘 | 最高 | 最低 | 波幅 | 5日均振幅 |
|---|---|---|---|---|---|---|---|
| {{ next_date }} | {{ next_weekday }} | @@ -31,7 +29,7 @@||||||
| {{ row.date }} | {{ row.weekday }} | {{ row.open }} | @@ -51,48 +49,101 @@