看股功能更新

2026-08-16 20:20:59 +08:00
parent 63f8ed8b69
commit 65d2f54d1b
9 changed files with 296 additions and 107 deletions
--- a/backend/app/screener/market_sync.py
+++ b/backend/app/screener/market_sync.py
@@ -1,10 +1,11 @@
-"""全市场数据同步（选股专用，未复权；与回测 candles 表隔离）。
+"""全市场数据同步（未复权，写入 candles 全量底座）。

-设计：trade_cal 取近 N 个交易日 -> 逐日 pro.daily(trade_date=...) / pro.daily_basic(trade_date=...)
-一次返回全市场当日数据 -> 按 trade_date 删旧插新批量入库（幂等）。
+设计：trade_cal 取近 N 个交易日 -> 逐日 pro.daily(trade_date=...) 一次返回全市场当日数据
+-> upsert 进 candles（不复权底座，ON CONFLICT 幂等）；daily_basic 仅同步最新交易日到
+DailySnapshot（市值/PE/PB/换手率等截面字段）。
 同步为进程内后台任务（MVP 不引入任务队列），前端轮询 /api/screener/sync/status。

-daily 与 daily_basic 分步独立落库：daily_basic 积分不足时日线仍可用，错误写入状态不中断任务。
+daily 与 daily_basic 分步独立落库：daily_basic 积分不足时快照仍可用，错误写入状态不中断任务。
 """
 from __future__ import annotations

@@ -13,10 +14,13 @@ import time
 from datetime import datetime, timedelta

 from sqlalchemy import delete, func, insert, select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.ext.asyncio import AsyncSession

+from .. import cache
 from ..config import settings
-from ..models import AdjFactor, DailySnapshot, MarketDaily, StockBasic, TradeCalendar
+from ..data.symbols import plain_code
+from ..models import AdjFactor, Candle, DailySnapshot, StockBasic, TradeCalendar
 from .llm import ScreenerError

 # 进程内单例任务状态（uvicorn --reload 单进程场景够用）
@@ -211,6 +215,48 @@ async def _replace_day(session: AsyncSession, model, rows: list[dict], d_str: st
    await session.commit()


+async def _existing_candle_dates(session: AsyncSession) -> set[str]:
+    """candles 表已落库的交易日集合（YYYYMMDD 字符串，便于比对）。"""
+    res = await session.execute(
+        select(func.distinct(func.date(Candle.ts))).where(Candle.timeframe == "1d")
+    )
+    return {r[0].strftime("%Y%m%d") for r in res if r[0] is not None}
+
+
+async def _upsert_candle_day(session: AsyncSession, rows: list[dict], listed: set[str], d_str: str) -> None:
+    """把某交易日全市场日线 upsert 进 candles（不复权底座，幂等）。
+
+    rows 来自 _fetch_daily（ts_code/vol手/amount千元）；只写 stock_basic 在市股票，
+    与 TDX 底座口径一致；amount 已有（TDX 回补）时保留旧值。
+    """
+    batch = [
+        {
+            "symbol": plain_code(r["ts_code"]), "timeframe": "1d",
+            "ts": _parse_d(d_str),
+            "open": r["open"], "high": r["high"], "low": r["low"], "close": r["close"],
+            "volume": r["vol"] * 100.0,          # 手 -> 股
+            "amount": (r["amount"] * 1000.0) if r["amount"] is not None else None,  # 千元 -> 元
+            "turnover": None,                    # 换手率另由 daily_basic 快照维护
+        }
+        for r in rows
+        if plain_code(r["ts_code"]) in listed
+    ]
+    if not batch:
+        return
+    stmt = pg_insert(Candle).values(batch)
+    stmt = stmt.on_conflict_do_update(
+        index_elements=["symbol", "timeframe", "ts"],
+        set_={
+            "open": stmt.excluded.open, "high": stmt.excluded.high,
+            "low": stmt.excluded.low, "close": stmt.excluded.close,
+            "volume": stmt.excluded.volume,
+            "amount": func.coalesce(Candle.amount, stmt.excluded.amount),
+        },
+    )
+    await session.execute(stmt)
+    await session.commit()
+
+
 async def _run_sync(days: int, force: bool) -> None:
    """后台任务主体：stock_basic -> 逐日日线 -> 最新交易日快照。异常写状态。

@@ -239,10 +285,16 @@ async def _run_sync(days: int, force: bool) -> None:
                else:
                    raise

-        # 2) 逐交易日日线（增量；当日未生成则跳过）
+        # 2) 逐交易日全市场日线 -> candles（增量；当日未生成则跳过）
        async with async_session() as session:
            dates = await _recent_trade_dates(session, pro, days)
-            have_daily = set() if force else await _existing_dates(session, MarketDaily)
+            have_daily = set() if force else await _existing_candle_dates(session)
+            # 在市股票集合，限定写入范围（与 TDX 底座口径一致）
+            listed = set(
+                (await session.execute(
+                    select(StockBasic.symbol).where(StockBasic.list_status == "L")
+                )).scalars()
+            )
        todo = [d for d in dates if d not in have_daily]
        _sync_state["total_days"] = len(todo)
        _sync_state["done_days"] = 0
@@ -252,7 +304,7 @@ async def _run_sync(days: int, force: bool) -> None:
            daily_rows = await asyncio.to_thread(_fetch_daily, pro, d)
            if daily_rows:  # 盘前/盘中等未生成数据的日期直接跳过
                async with async_session() as session:
-                    await _replace_day(session, MarketDaily, daily_rows, d)
+                    await _upsert_candle_day(session, daily_rows, listed, d)
            _sync_state["done_days"] += 1

        # 2.5) 复权因子（与日线同窗口增量；历史全量由 scripts/backfill_adj_factor.py 回补）
@@ -266,9 +318,9 @@ async def _run_sync(days: int, force: bool) -> None:
                    await _replace_day(session, AdjFactor, adj_rows, d)

        # 3) 最新「有数据」交易日的快照（daily_basic，仅 1 次调用）
-        #    用 market_daily 实际最大交易日（今天的数据收盘后才生成，日历最新日会拉到空）
+        #    用 candles 实际最大交易日（今天的数据收盘后才生成，日历最新日会拉到空）
        async with async_session() as session:
-            latest_dt = await session.scalar(select(func.max(MarketDaily.trade_date)))
+            latest_dt = await session.scalar(select(func.max(Candle.ts)))
        latest = latest_dt.strftime("%Y%m%d") if latest_dt else None
        if latest:
            async with async_session() as session:
@@ -280,6 +332,8 @@ async def _run_sync(days: int, force: bool) -> None:
                    async with async_session() as session:
                        await _replace_day(session, DailySnapshot, basic_rows, latest)

+        # candles/复权因子已更新：作废旧 K 线预览缓存（键含版本号，自增即全体失效）
+        await cache.bump_version("candles")
        _sync_state["step"] = "同步完成"
    except Exception as e:  # noqa: BLE001
        _sync_state["error"] = f"同步失败：{str(e)[:300]}"
@@ -303,19 +357,42 @@ async def start_sync(session: AsyncSession, days: int, force: bool) -> dict:
    return dict(_sync_state)


+# candles 是千万行表，count 较重；前端每 2s 轮询状态，需 TTL 缓存降载
+_status_stats_cache: dict = {"at": 0.0, "data": None}
+_STATS_TTL = 30.0
+
+
+async def _db_stats(session: AsyncSession) -> dict:
+    """candles/快照/股票列表实况（30s TTL 缓存）。"""
+    now = time.time()
+    if _status_stats_cache["data"] is not None and now - _status_stats_cache["at"] < _STATS_TTL:
+        return _status_stats_cache["data"]
+    stocks = int(await session.scalar(select(func.count()).select_from(StockBasic)) or 0)
+    daily_rows = int(await session.scalar(select(func.count()).select_from(Candle)) or 0)
+    snap_rows = int(await session.scalar(select(func.count()).select_from(DailySnapshot)) or 0)
+    last_daily = await session.scalar(
+        select(func.max(Candle.ts)).where(Candle.timeframe == "1d")
+    )
+    n_dates = int(await session.scalar(
+        select(func.count(func.distinct(func.date(Candle.ts)))).where(Candle.timeframe == "1d")
+    ) or 0)
+    data = {
+        "stocks": stocks, "daily_rows": daily_rows, "snapshot_rows": snap_rows,
+        "last_daily": last_daily, "dates": n_dates,
+    }
+    _status_stats_cache.update(at=now, data=data)
+    return data
+
+
 async def get_sync_status(session: AsyncSession) -> dict:
    """合并任务状态 + DB 实况（最新交易日/行数/ready 标志），与 ScreenerSyncStatus DTO 对齐。"""
-    stocks = int(await session.scalar(select(func.count()).select_from(StockBasic)) or 0)
-    daily_rows = int(await session.scalar(select(func.count()).select_from(MarketDaily)) or 0)
-    snap_rows = int(await session.scalar(select(func.count()).select_from(DailySnapshot)) or 0)
-    last_daily = await session.scalar(select(func.max(MarketDaily.trade_date)))
-    n_dates = int(await session.scalar(select(func.count(func.distinct(MarketDaily.trade_date)))) or 0)
-
+    stats = await _db_stats(session)
    status = dict(_sync_state)
    status.update({
-        "stats": {"stocks": stocks, "daily_rows": daily_rows, "snapshot_rows": snap_rows, "dates": n_dates},
-        "last_trade_date": last_daily,
+        "stats": {"stocks": stats["stocks"], "daily_rows": stats["daily_rows"],
+                  "snapshot_rows": stats["snapshot_rows"], "dates": stats["dates"]},
+        "last_trade_date": stats["last_daily"],
        "last_synced_at": _sync_state.get("finished_at") or _sync_state.get("started_at"),
-        "ready": daily_rows > 0,
+        "ready": stats["daily_rows"] > 0,
    })
    return status