From 65d2f54d1bf43e8dcb91e6cf8787bcfadf2ea2f2 Mon Sep 17 00:00:00 2001 From: cirry <812852553@qq.com> Date: Sun, 16 Aug 2026 20:20:59 +0800 Subject: [PATCH] =?UTF-8?q?=E7=9C=8B=E8=82=A1=E5=8A=9F=E8=83=BD=E6=9B=B4?= =?UTF-8?q?=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 36 ++++++--- backend/app/api.py | 118 ++++++++++++++++------------ backend/app/data/repository.py | 3 +- backend/app/indicators.py | 10 +++ backend/app/models.py | 2 +- backend/app/screener/engine.py | 97 +++++++++++++++++------ backend/app/screener/market_sync.py | 115 ++++++++++++++++++++++----- backend/smoke_test.py | 4 +- backend_run.log | 18 +++++ 9 files changed, 296 insertions(+), 107 deletions(-) diff --git a/README.md b/README.md index 057ee8d..3f5dd12 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,25 @@ --- +## 快速启动 + +两个终端分别启动后端与前端： + +```bash +# 终端 1 —— 后端 API（http://localhost:8000） +cd backend +uv run uvicorn app.main:app --reload --port 8000 + +# 终端 2 —— 前端开发服务器（http://localhost:5173） +cd frontend +pnpm dev +``` + +首次运行前先安装依赖：后端 `uv sync`（backend 下）、前端 `pnpm install`（frontend 下）。 +浏览器打开 http://localhost:5173 即可使用。详细说明见下文「快速开始（开发模式）」。 + +--- + ## 功能特性 **已实现（Phase 0 MVP）** @@ -20,7 +39,6 @@ - 回测运行注册表（每次回测落库，可复现/审计的基础） **规划中（见 TECH_STACK.md 路线图）** -- 接入 Tushare/AKShare 真实 A 股数据（替换 DEMO 合成数据） - 防过拟合体检（walk-forward / 样本外 / FDR 多重比较修正） - 基准归因（对比沪深300/中证500：超额、信息比率、beta/alpha） - 复权因子管道、回放式模拟盘 @@ -34,7 +52,7 @@ |---|---| | 后端 | Python 3.12+ · FastAPI · Pydantic v2 · SQLAlchemy 2.0 (async) · uv（包管理） | | 回测引擎 | 自研单一引擎（fast/strict 两档），指标纯 numpy/pandas（生产可换 TA-Lib） | -| 数据库 | MVP 用 SQLite 零配置；生产切 PostgreSQL 16/17 + TimescaleDB | +| 数据库 | PostgreSQL 16/17（后端强制要求）+ Redis 缓存；TimescaleDB 规划中 | | 前端 | Vue 3.5 · Vite · TypeScript · Pinia · PrimeVue 5 · pnpm | | 图表 | lightweight-charts 5（K 线）· ECharts 6（净值） | @@ -60,7 +78,7 @@ stock/ │ ├── commission.py # A 股交易成本（已修正、可配置） │ ├── indicators.py # 指标：MACD/RSI/KDJ/布林/均线（单一事实源） │ ├── api.py # 路由：/health /candles /backtest -│ ├── data/ # DataProvider 适配器 + 合成数据 + 周期聚合 +│ ├── data/ # DataProvider 适配器（Tushare/AKShare）+ 周期聚合 │ └── backtest/ # engine / broker(PaperBroker) / metrics / strategies └── frontend/ # Vue SPA ├── src/ @@ -82,7 +100,7 @@ stock/ | pnpm | ≥ 9（实测 11） | 前端包管理 | | Python | ≥ 3.12（实测 3.14） | 后端 | | uv | 任意（实测 0.12） | 后端包管理，[安装](https://docs.astral.sh/uv/) | -| （可选）PostgreSQL | 16/17 | 生产数据库；MVP 用 SQLite 无需安装 | +| PostgreSQL | 16/17 | 数据库（必装，后端已不再支持 SQLite） | **安装 uv**（若未装）： ```bash @@ -103,7 +121,7 @@ uv sync # 创建 .venv 并安装依赖（首 uv run uvicorn app.main:app --reload --port 8000 ``` -- 首次启动自动建表（SQLite：`backend/stock.db`）并播种约 500 个交易日的合成 K 线（symbol=`DEMO`）。 +- 数据库结构由 Alembic 管理：首次部署/更新代码后先执行 `uv run alembic upgrade head`（见「初始化登录系统」）。 - 交互式 API 文档：http://localhost:8000/docs **自检**（无需起服务器，验证全链路）： @@ -141,7 +159,7 @@ CORS_ORIGINS=http://localhost:5173 # 生产建议关闭接口文档 EXPOSE_API_DOCS=true -# 真实数据源（Tushare Pro；免费版即可。留空则仅 DEMO 合成数据可用） +# 真实数据源（Tushare Pro；免费版即可） TUSHARE_TOKEN=你的token DATA_ADJUST=qfq # 复权：qfq 前复权 / hfq 后复权 / 留空不复权 DATA_DEFAULT_START=20200101 @@ -204,7 +222,7 @@ EXPOSE_API_DOCS=false ```json POST /api/backtest { - "symbol": "DEMO", + "symbol": "000001", "timeframe": "1d", "strategy": "macd_cross", "params": { "fast": 12, "slow": 26, "signal": 9 }, @@ -214,7 +232,7 @@ POST /api/backtest ``` 返回：`candles`（K线）、`indicators`（MACD 三线）、`signals`（买卖点）、`equity`（净值序列）、`metrics`（绩效）、`final_cash`/`final_position`。 -> **关于「标的」**：`DEMO` 为内置合成数据；其余为真实 A 股代码（如 `000001`、`600519`），首次回测时**自动经 Tushare 拉取并本地缓存**（需配置 `TUSHARE_TOKEN` + 联网），二次回测秒出。默认初始资金 100 万，足够交易高价股（如茅台）。 +> **关于「标的」**：标的为真实 A 股代码（如 `000001`、`600519`），首次回测时**自动经 Tushare 拉取并本地缓存**（需配置 `TUSHARE_TOKEN` + 联网），二次回测秒出。默认初始资金 100 万，足够交易高价股（如茅台）。 --- @@ -235,7 +253,7 @@ uv run alembic upgrade head uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 # 或 gunicorn（Linux）：uv run gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker ``` -生产务必设置 `DATABASE_URL` 指向 PostgreSQL，不要用 SQLite。 +后端强制要求 `DATABASE_URL` 使用 PostgreSQL（`postgresql+asyncpg://`），已不再支持 SQLite。 ### 参考：docker-compose（具备 Docker 后使用） ```yaml diff --git a/backend/app/api.py b/backend/app/api.py index fb1d58c..012a1c3 100644 --- a/backend/app/api.py +++ b/backend/app/api.py @@ -17,7 +17,7 @@ from datetime import datetime import pandas as pd from fastapi import APIRouter, Depends, File, HTTPException, UploadFile -from sqlalchemy import delete, select, text +from sqlalchemy import delete, func, select, text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql.elements import TextClause @@ -38,8 +38,8 @@ from .trades import parse_statement from .models import ( AdjFactor, BacktestRun, + Candle, DailySnapshot, - MarketDaily, ScreenerQuery, StockBasic, UserPreference, @@ -780,11 +780,11 @@ async def screener_sync_status(session: AsyncSession = Depends(get_session)) -> @router.get("/screener/preview/{ts_code}", response_model=PreviewResponse) async def screener_preview( ts_code: str, limit: int = 500, adjust: str = "qfq", timeframe: str = "1d", mas: str = "5,10,20,60", - end: str | None = None, + zx: str = "10,20,30,60", end: str | None = None, session: AsyncSession = Depends(get_session), ) -> PreviewResponse: """个股详情预览：日线（candles 不复权底座 + adj_factor 本地换算 bfq/qfq/hfq， - 未缓存自动拉取，失败退 market_daily 近段）+ 全套指标 + 最新截面信息卡。 + 未缓存自动拉取，落后全市场最新交易日则强制刷新）+ 全套指标 + 最新截面信息卡。 timeframe 聚合到周/月/年（先复权再聚合）；mas 指定主图 MA 周期（逗号分隔）。 end=YYYY-MM-DD 时为「向前翻页」：返回该日之前最近 limit 根（含预热计算指标）， has_more 标记窗口前是否还有更早历史，前端据此继续向左滚动加载。""" @@ -798,6 +798,12 @@ async def screener_preview( raise HTTPException(status_code=400, detail="mas 格式应为逗号分隔的数字，如 5,10,20,60") if not ma_periods: ma_periods = [5, 10, 20, 60] + try: + zx_periods = sorted({int(p) for p in zx.split(",") if p.strip().isdigit() and 1 <= int(p) <= 500}) + except ValueError: + raise HTTPException(status_code=400, detail="zx 格式应为逗号分隔的数字，如 10,20,30,60") + if not zx_periods: + zx_periods = [10, 20, 30, 60] limit = max(30, min(limit, 5000)) end_dt: datetime | None = None if end: @@ -807,20 +813,27 @@ async def screener_preview( raise HTTPException(status_code=400, detail="end 格式应为 YYYY-MM-DD") symbol = plain_code(ts_code) - # 先取 market_daily 最新行：既做缓存过期判断，也做信息卡数据源 - md = ( - await session.execute( - select(MarketDaily).where(MarketDaily.ts_code == ts_code).order_by(MarketDaily.trade_date.desc()).limit(1) - ) - ).scalars().first() + # --- Redis 读缓存：历史窗口（end 翻页）只增不改，最新窗口每日由全市场同步推进； + # 键含 ver:candles 版本号（同步完成后自增，旧缓存全部失效），TTL 兜底（cache.py）--- + cache_key = cache.digest( + "preview", ts_code, timeframe, limit, adjust, + end_dt.strftime("%Y-%m-%d") if end_dt else None, ma_periods, zx_periods, + await cache.get_version("candles"), + ) + cached = await cache.cache_get(f"pv:{cache_key}") + if cached is not None: + return PreviewResponse.model_validate(cached) - # --- 日线：candles(不复权底座) 优先；未缓存拉取，缓存落后于全市场最新交易日则强制刷新（每日至多一次） --- - # fetcher 现在只做「不复权」增量 upsert，底座口径恒为 bfq（TDX 全量 + Tushare 增量）， - # 复权（qfq/hfq）读取时按 adj_factor 表本地换算，mode 无需再推断。 - # 每次只取「窗口 + 800 根预热」行（MA250/MACD EMA 在 800 根内充分收敛），不拉全量： + # --- 日线：candles(全量不复权底座)；未缓存拉取，落后于全市场最新交易日则强制刷新 --- + # fetcher 只做「不复权」增量 upsert，底座口径恒为 bfq（TDX 全量 + Tushare 增量）， + # 复权（qfq/hfq）读取时按 adj_factor 表本地换算。 + # 每次只取「窗口 + 400 根预热」行（MA250/MACD EMA 在 400 根内充分收敛），不拉全量： # 首屏 ~500 根秒开，向左滚动时按 end 参数逐页向前翻。 + global_latest = await session.scalar( + select(func.max(Candle.ts)).where(Candle.timeframe == "1d") + ) frame_mult = {"1d": 1, "1w": 6, "1M": 24, "1y": 280}[timeframe] - fetch_n = min(100000, limit * frame_mult + 800) + fetch_n = min(100000, limit * frame_mult + 400) source = "bfq" mode = "bfq" if end_dt is not None: @@ -833,7 +846,7 @@ async def screener_preview( if not rows: await fetcher.sync_symbol(session, symbol, source="auto") rows = await repository.get_recent_candles(session, symbol, "1d", limit=fetch_n) - elif md is not None and rows and rows[-1].ts.date() < md.trade_date.date(): + elif global_latest is not None and rows[-1].ts.date() < global_latest.date(): await fetcher.sync_symbol(session, symbol, source="auto", force=True) rows = await repository.get_recent_candles(session, symbol, "1d", limit=fetch_n) except Exception: # noqa: BLE001 —— tushare/写库失败时回滚会话（否则毒化后兜底查询 500） @@ -842,34 +855,33 @@ async def screener_preview( rows = [] bars = _rows_to_bars(rows) - if not bars and end_dt is None: - source = "market" - res = await session.execute( - select(MarketDaily).where(MarketDaily.ts_code == ts_code).order_by(MarketDaily.trade_date) - ) - bars = [ - Bar( - ts=r.trade_date, 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 else None, # 千元 -> 元 - ) - for r in res.scalars() - ] if not bars and end_dt is None: raise HTTPException(status_code=404, detail=f"无数据: {ts_code}（可先点「同步市场数据」）") + # 信息卡取未聚合的日线最新 bar（聚合后 ts 是周期起点，不适用于「最新交易日」） + last_daily = bars[-1] if bars else None + prev_daily = bars[-2] if len(bars) > 1 else None # 翻页到底（end 之前无数据）：返回空页 + has_more=False，前端停止向前翻页 # --- 复权换算：请求模式与底座模式不同时按 adj_factor 本地换算（无因子则维持原样） --- - if adjust != mode: - factors = ( - await session.execute( - select(AdjFactor).where(AdjFactor.ts_code == ts_code).order_by(AdjFactor.trade_date) - ) - ).scalars().all() + # 只取窗口内因子（qfq 归一还需全局最新因子，追加到最后一行即可，_adjust_bars 取 f_latest=末项） + if adjust != mode and bars: + fq = select(AdjFactor).where(AdjFactor.ts_code == ts_code) + window_end = end_dt if end_dt is not None else bars[-1].ts + if window_end is not None: + fq = fq.where(AdjFactor.trade_date <= window_end) + factors = list((await session.execute(fq.order_by(AdjFactor.trade_date))).scalars().all()) if factors: + latest_f = ( + await session.execute( + select(AdjFactor).where(AdjFactor.ts_code == ts_code) + .order_by(AdjFactor.trade_date.desc()).limit(1) + ) + ).scalars().first() + if latest_f is not None: + factors.append(latest_f) bars = _adjust_bars(bars, factors, mode, adjust) mode = adjust - if source != "market": - source = adjust + source = adjust # --- 周期聚合：复权之后按日历聚合到周/月/年，指标在聚合后的序列上计算 --- bars = resample_bars(bars, timeframe) @@ -897,21 +909,25 @@ async def screener_preview( "rsi24": _series_to_jsonable(ind.rsi(closes, 24)), }, "boll": {k: _series_to_jsonable(boll[k]) for k in ("upper", "mid", "lower")}, + "zx": { + "short": _series_to_jsonable(ind.ema2(closes)), + "duokong": _series_to_jsonable(ind.avg_ma(closes, tuple(zx_periods))), + }, } limit = max(30, min(limit, len(bars))) for group in indicators.values(): for key in group: group[key] = group[key][-limit:] - # --- 信息卡：stock_basic + 最新 market_daily + 与其对齐的快照（避免混用不同交易日） --- + # --- 信息卡：stock_basic + candles 最新日线 bar + 与其对齐的快照（避免混用不同交易日） --- sb = (await session.execute(select(StockBasic).where(StockBasic.ts_code == ts_code))).scalars().first() ds = None - if md is not None: + if last_daily is not None: # 优先取与行情同日的快照；缺当日快照时退最新（字段可能与行情差日期，罕见） ds = ( await session.execute( select(DailySnapshot).where( - DailySnapshot.ts_code == ts_code, DailySnapshot.trade_date == md.trade_date + DailySnapshot.ts_code == ts_code, DailySnapshot.trade_date == last_daily.ts ) ) ).scalars().first() @@ -936,15 +952,16 @@ async def screener_preview( area=sb.area if sb else None, market=sb.market if sb else None, list_date=sb.list_date if sb else None, - trade_date=md.trade_date if md else None, - open=md.open if md else None, - high=md.high if md else None, - low=md.low if md else None, - close=md.close if md else None, - pre_close=md.pre_close if md else None, - pct_chg=md.pct_chg if md else None, - volume_hand=round(md.vol, 0) if md else None, - amount_yi=round(md.amount / 100000, 2) if md else None, # 千元 -> 亿元 + trade_date=last_daily.ts if last_daily else None, + open=last_daily.open if last_daily else None, + high=last_daily.high if last_daily else None, + low=last_daily.low if last_daily else None, + close=last_daily.close if last_daily else None, + pre_close=prev_daily.close if prev_daily else None, + pct_chg=((last_daily.close / prev_daily.close - 1) * 100) + if last_daily and prev_daily and prev_daily.close else None, + volume_hand=round(last_daily.volume / 100, 0) if last_daily else None, # 股 -> 手 + amount_yi=round(last_daily.amount / 1e8, 2) if last_daily and last_daily.amount else None, # 元 -> 亿元 turnover_rate=ds.turnover_rate if ds else None, pe_ttm=ds.pe_ttm if ds else None, pb=ds.pb if ds else None, @@ -957,4 +974,7 @@ async def screener_preview( volume=b.volume, amount=b.amount, turnover=b.turnover) for b in bars[-limit:] ] - return PreviewResponse(ts_code=ts_code, symbol=symbol, source=source, info=info, candles=candles, indicators=indicators, has_more=has_more) + resp = PreviewResponse(ts_code=ts_code, symbol=symbol, source=source, info=info, + candles=candles, indicators=indicators, has_more=has_more) + await cache.cache_set(f"pv:{cache_key}", resp.model_dump(mode="json"), ttl=600) + return resp diff --git a/backend/app/data/repository.py b/backend/app/data/repository.py index 13e70f7..06b2a33 100644 --- a/backend/app/data/repository.py +++ b/backend/app/data/repository.py @@ -1,7 +1,6 @@ """K 线数据访问（从库读）。 -写入由 DataProvider 适配器负责（阶段1 接 Tushare/AKShare）。 -MVP 的数据由 synthetic.seed_if_empty 灌入。 +写入由 DataProvider 适配器负责（Tushare 主 → AKShare 兜底）。 """ from __future__ import annotations diff --git a/backend/app/indicators.py b/backend/app/indicators.py index 248ff1e..1c4cfde 100644 --- a/backend/app/indicators.py +++ b/backend/app/indicators.py @@ -54,3 +54,13 @@ def bollinger(close: pd.Series, period: int = 20, std: float = 2.0) -> pd.DataFr def ma(close: pd.Series, period: int) -> pd.Series: return close.rolling(period, min_periods=1).mean() + + +def ema2(close: pd.Series, span: int = 10) -> pd.Series: + """知行短期趋势线：EMA(EMA(C, span), span)。""" + return ema(ema(close, span), span) + + +def avg_ma(close: pd.Series, periods: tuple[int, ...] = (10, 20, 30, 60)) -> pd.Series: + """知行多空线：(MA(M1)+MA(M2)+MA(M3)+MA(M4))/4。""" + return sum(ma(close, p) for p in periods) / len(periods) diff --git a/backend/app/models.py b/backend/app/models.py index 4bb9179..0ca54c1 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -16,7 +16,7 @@ from .db import Base def _utcnow() -> datetime: - # naive UTC，避免 SQLite 存储时区带来的麻烦 + # naive UTC，统一存储避免时区带来的麻烦 from datetime import timezone return datetime.now(timezone.utc).replace(tzinfo=None) diff --git a/backend/app/screener/engine.py b/backend/app/screener/engine.py index 7668cf5..7931c88 100644 --- a/backend/app/screener/engine.py +++ b/backend/app/screener/engine.py @@ -1,6 +1,6 @@ """选股执行引擎：SQL 快照预筛缩小范围 -> 逐股指标计算过滤。 -性能：预筛在 SQLite 索引上完成（毫秒级）；指标阶段候选集通常数百~数千只 × ~90 根 bar， +性能：预筛在数据库索引上完成（毫秒级）；指标阶段候选集通常数百~数千只 × ~90 根 bar， pandas 逐股计算（复用 app/indicators，指标按 (族, 参数) 去重计算），秒级完成。 """ from __future__ import annotations @@ -14,7 +14,8 @@ from sqlalchemy import and_, func, not_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from .. import indicators as ind -from ..models import DailySnapshot, MarketDaily, StockBasic +from ..data.symbols import plain_code +from ..models import Candle, DailySnapshot, StockBasic from ..schemas import IndicatorCondition, ScreenConditions, SnapshotCondition # 快照字段 -> (DB 列, 中文标签, LLM 值 -> DB 值的换算乘数) @@ -25,7 +26,7 @@ SNAPSHOT_FIELDS: dict[str, tuple[str, str, float]] = { "pe_ttm": ("pe_ttm", "市盈率TTM", 1.0), "pb": ("pb", "市净率", 1.0), "turnover_rate": ("turnover_rate", "换手率%", 1.0), - "close": ("close", "最新价", 1.0), # 实际取 market_daily.close，无需快照表 + "close": ("close", "最新价", 1.0), # 实际取 candles 最新 bar，无需快照表 } @@ -198,7 +199,7 @@ def _snapshot_clause(cond: SnapshotCondition): if cond.field not in SNAPSHOT_FIELDS: raise ValueError(f"未知快照字段: {cond.field}") col_name, _, scale = SNAPSHOT_FIELDS[cond.field] - col = MarketDaily.close if cond.field == "close" else getattr(DailySnapshot, col_name) + col = Candle.close if cond.field == "close" else getattr(DailySnapshot, col_name) lo = cond.value * scale hi = (cond.value2 * scale) if cond.op == "between" and cond.value2 is not None else None if cond.op == "gt": @@ -219,17 +220,22 @@ _CAND_COLS = ["ts_code", "name", "close", "pct_chg", async def _prefilter(session: AsyncSession, conds: ScreenConditions, target_date: datetime) -> pd.DataFrame: - """最新交易日截面预筛：快照条件 + 排除项 + 名称/收盘价。返回候选 DataFrame。""" + """最新交易日截面预筛：candles(全量不复权底座) + stock_basic + daily_snapshot。 + + 只取 target_date 当日有交易的股票（停牌股无当日 bar，自然排除，与原先 market_daily + 的 trade_date == target_date 行为一致）。pct_chg 用前一交易日收盘价计算。 + """ + snap_date = await session.scalar(select(func.max(DailySnapshot.trade_date))) or target_date stmt = ( select( - MarketDaily.ts_code, StockBasic.name, MarketDaily.close, MarketDaily.pct_chg, + Candle.symbol, StockBasic.ts_code, StockBasic.name, Candle.close, DailySnapshot.total_mv, DailySnapshot.circ_mv, DailySnapshot.pe_ttm, DailySnapshot.pb, DailySnapshot.turnover_rate, ) - .join(StockBasic, StockBasic.ts_code == MarketDaily.ts_code) - .outerjoin(DailySnapshot, and_(DailySnapshot.ts_code == MarketDaily.ts_code, - DailySnapshot.trade_date == target_date)) - .where(MarketDaily.trade_date == target_date) + .join(StockBasic, StockBasic.symbol == Candle.symbol) + .outerjoin(DailySnapshot, and_(DailySnapshot.ts_code == StockBasic.ts_code, + DailySnapshot.trade_date == snap_date)) + .where(Candle.timeframe == "1d", Candle.ts == target_date) ) if conds.exclude_delisted: @@ -237,13 +243,39 @@ async def _prefilter(session: AsyncSession, conds: ScreenConditions, target_date if conds.exclude_st: stmt = stmt.where(not_(or_(StockBasic.name.like("%ST%"), StockBasic.name.like("%退%")))) if conds.exclude_bj: - stmt = stmt.where(not_(MarketDaily.ts_code.like("%.BJ"))) + stmt = stmt.where(not_(StockBasic.ts_code.like("%.BJ"))) for c in conds.snapshot: stmt = stmt.where(_snapshot_clause(c)) rows = (await session.execute(stmt)).all() - return pd.DataFrame(rows, columns=_CAND_COLS) + df = pd.DataFrame(rows, columns=["symbol", "ts_code", "name", "close", + "total_mv", "circ_mv", "pe_ttm", "pb", "turnover_rate"]) + if df.empty: + return df[_CAND_COLS] + + # pct_chg：candles 无现成涨跌幅列，用前一交易日的收盘价计算 + prev_dt = await session.scalar( + select(func.max(Candle.ts)).where(Candle.timeframe == "1d", Candle.ts < target_date) + ) + prev_map: dict[str, float] = {} + if prev_dt is not None: + pr = await session.execute( + select(Candle.symbol, Candle.close).where( + Candle.timeframe == "1d", Candle.ts == prev_dt, + Candle.symbol.in_(df["symbol"].tolist()), + ) + ) + prev_map = {r.symbol: r.close for r in pr} + prev = df["symbol"].map(prev_map) + + def _pct(c, p) -> float | None: + if p is None or p != p or float(p) == 0: + return None + return (float(c) / float(p) - 1) * 100 + + df["pct_chg"] = [_pct(c, p) for c, p in zip(df["close"], prev)] + return df[_CAND_COLS] # ---------- 主流程 ---------- @@ -260,18 +292,30 @@ def _max_needed_bars(conds: ScreenConditions) -> int: async def _load_bars(session: AsyncSession, ts_codes: list[str], target_date: datetime, min_date: datetime) -> pd.DataFrame: - """载入候选股的 K 线窗口。候选 <= 2000 用 IN 精确圈定；否则拉全窗口再 pandas 过滤。""" + """载入候选股的 K 线窗口（candles 全量不复权底座）。 + + 候选 <= 2000 用 IN 精确圈定；否则拉全窗口再 pandas 过滤。 + ts_code 按候选集映射回 symbol 查询，pct_chg 用每股收盘价环比计算。 + """ + symbols = [plain_code(t) for t in ts_codes] stmt = select( - MarketDaily.ts_code, MarketDaily.trade_date, MarketDaily.open, MarketDaily.high, - MarketDaily.low, MarketDaily.close, MarketDaily.pct_chg, - ).where(MarketDaily.trade_date >= min_date, MarketDaily.trade_date <= target_date) - if len(ts_codes) <= 2000: - stmt = stmt.where(MarketDaily.ts_code.in_(set(ts_codes))) + Candle.symbol, Candle.ts, Candle.open, Candle.high, Candle.low, Candle.close, + ).where(Candle.timeframe == "1d", Candle.ts >= min_date, Candle.ts <= target_date) + if len(symbols) <= 2000: + stmt = stmt.where(Candle.symbol.in_(set(symbols))) rows = (await session.execute(stmt)).all() - df = pd.DataFrame(rows, columns=["ts_code", "trade_date", "open", "high", "low", "close", "pct_chg"]) - if not df.empty and len(ts_codes) > 2000: - df = df[df["ts_code"].isin(set(ts_codes))] - return df.sort_values(["ts_code", "trade_date"]).reset_index(drop=True) + df = pd.DataFrame(rows, columns=["symbol", "trade_date", "open", "high", "low", "close"]) + if not df.empty and len(symbols) > 2000: + df = df[df["symbol"].isin(set(symbols))] + df = df.sort_values(["symbol", "trade_date"]).reset_index(drop=True) + if df.empty: + df["ts_code"] = pd.Series(dtype=object) + df["pct_chg"] = pd.Series(dtype=object) + else: + df["pct_chg"] = df.groupby("symbol")["close"].pct_change() * 100 + ts_map = {plain_code(t): t for t in ts_codes} + df["ts_code"] = df["symbol"].map(ts_map) + return df[["ts_code", "trade_date", "open", "high", "low", "close", "pct_chg"]] def _f(v) -> float | None: @@ -303,7 +347,9 @@ async def run_screen(session: AsyncSession, conds: ScreenConditions, limit: int) if not conds.indicator and not conds.snapshot: raise ValueError("筛选条件为空") - target_date = await session.scalar(select(func.max(MarketDaily.trade_date))) + target_date = await session.scalar( + select(func.max(Candle.ts)).where(Candle.timeframe == "1d") + ) if target_date is None: raise DataNotReadyError("全市场数据未同步：请先在选股页点击「同步市场数据」") @@ -330,10 +376,11 @@ async def run_screen(session: AsyncSession, conds: ScreenConditions, limit: int) # 纯快照条件：预筛结果即命中 items = [_item_from_row(row) | {"indicators": {}} for _, row in cand.iterrows()] else: - # 圈定 K 线窗口：按已同步交易日序列回溯 needed 根 + # 圈定 K 线窗口：按 candles 全量交易日序列回溯 needed 根（不再受同步窗口限制） need = _max_needed_bars(conds) dates_res = await session.execute( - select(MarketDaily.trade_date).distinct().order_by(MarketDaily.trade_date.desc()).limit(need) + select(Candle.ts).where(Candle.timeframe == "1d").distinct() + .order_by(Candle.ts.desc()).limit(need) ) min_date = min(r[0] for r in dates_res) bars = await _load_bars(session, cand["ts_code"].tolist(), target_date, min_date) diff --git a/backend/app/screener/market_sync.py b/backend/app/screener/market_sync.py index fd562b1..e0ca46c 100644 --- 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 diff --git a/backend/smoke_test.py b/backend/smoke_test.py index 6bb6de9..9ecc127 100644 --- a/backend/smoke_test.py +++ b/backend/smoke_test.py @@ -15,7 +15,7 @@ from sqlalchemy import delete, select from app.auth import hash_password, token_digest from app.db import async_session, engine from app.main import app -from app.models import AuthSession, MarketDaily, User +from app.models import AuthSession, Candle, User async def main() -> None: @@ -100,7 +100,7 @@ async def main() -> None: print("logout and revoke: ok") async with async_session() as db: - has_market_data = bool(await db.scalar(select(MarketDaily.id).limit(1))) + has_market_data = bool(await db.scalar(select(Candle.id).limit(1))) print("market data present:", has_market_data) finally: async with async_session() as db: diff --git a/backend_run.log b/backend_run.log index f5b3360..211325c 100644 --- a/backend_run.log +++ b/backend_run.log @@ -731,3 +731,21 @@ INFO: 127.0.0.1:56621 - "GET /api/auth/me HTTP/1.1" 200 OK INFO: 127.0.0.1:56638 - "GET /api/market/overview HTTP/1.1" 404 Not Found INFO: 127.0.0.1:56637 - "GET /api/preferences HTTP/1.1" 200 OK INFO: 127.0.0.1:56644 - "GET /api/market/overview HTTP/1.1" 404 Not Found +==== restart 2026-08-16 20:03:31 ==== +INFO: Will watch for changes in these directories: ['D:\\Project\\stock\\backend'] +INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) +INFO: Started reloader process [37952] using WatchFiles +INFO: Started server process [45584] +INFO: Waiting for application startup. +INFO: Application startup complete. +INFO: 127.0.0.1:59228 - "GET /api/health HTTP/1.1" 200 OK +INFO: 127.0.0.1:59686 - "GET /api/auth/me HTTP/1.1" 200 OK +INFO: 127.0.0.1:59399 - "GET /api/auth/me HTTP/1.1" 200 OK +INFO: 127.0.0.1:59722 - "GET /api/preferences HTTP/1.1" 200 OK +INFO: 127.0.0.1:59723 - "POST /api/auth/login HTTP/1.1" 200 OK +INFO: 127.0.0.1:59725 - "GET /api/stocks/facets HTTP/1.1" 200 OK +INFO: 127.0.0.1:59724 - "GET /api/stocks?sort=symbol&order=asc&limit=100&offset=0 HTTP/1.1" 200 OK +INFO: 127.0.0.1:59733 - "GET /api/trades?ts_code=000006.SZ HTTP/1.1" 200 OK +INFO: 127.0.0.1:59736 - "GET /api/watchlist HTTP/1.1" 200 OK +WARNING: WatchFiles detected changes in 'debug_gaps.py', 'debug_counts.py', 'debug_future.py', 'debug_preview.py', 'debug_live.py', 'debug_md.py'. Reloading... + \ No newline at end of file